【历史文档】算子样例-自定义Python模块

主函数：编写主处理函数。
后处理函数：编写后处理代码，即使勾选了启用缓存加速功能，后处理的函数也会被执行。
模块输入端：封装成模块后暴露出来的输入端口，需要跟主函数的参数名称保持一致。多个参数用逗号分隔。
模块参数：封装成模块后暴露出来的参数，需要跟主函数的参数名称保持一致，字典格式，实例：
```
{
    'columns': 'deal_number:new_deal_number|close:new_close',
    'keep_old_columns': True
}
```
模块输出端：封装成模块后暴露出来的输出端口，需要跟主函数的return时返回的参数名称保持一致。

输出端

输出1：第一个输出端口，支持输出任意格式的数据，一般为DataSource类型。
输出2：第二个输出端口，支持输出任意格式的数据，一般为DataSource类型。
输出3：第三个输出端口，支持输出任意格式的数据，一般为DataSource类型。

运行结果

支持模块id.封装的端口名来查看模块的输出。

使用样例

本样例详细展示如何开发自定义模块，用来修改数据的列名，此模块的输入、参数、输出如下：

输入：有一个输入数据源，为输入数据
参数：列名映射，格式为：原列名1:新列名1|原列名2:新列名2|原列名3:新列名3|..
输出：有一个输出数据源，为结果数据

准备输入数据

为了有测试数据方便开发调试，我们直接新建 “可视化策略-空白”，拖入数据源模块，并配置如下，用作测试数据：

数据源ID：bar1d_CN_STOCK_A
开始日期：2018-01-01
结束日期：2018-02-01

{w:100}{w:100}{w:100}{w:100}{w:100}{w:100}{w:100}

其数据如下：

{w:100}{w:100}{w:100}{w:100}{w:100}{w:100}{w:100}

添加自定义模块

从左侧自定义模块拖拽自定义Python模块到画布中，并如下图连接

{w:90}{w:100}

定义好接口

输入端：input_ds：输入数据源，columns：列名映射，keep_old_columns：是否保留老的列名
输出端：Outputs.data：输出修改列名后的数据

代码实现

可以新建一个代码单元，先在代码单元里进行功能编写和测试。

# 修改数据列名
def bigquant_run(input_ds, columns, keep_old_columns):
    # 解析列映射为dict, TODO: 验证输入是否有效
    columns = dict(c.split(':') for c in columns.split('|'))
    print('列名映射：', columns)
    # 输出数据源
    df = input_ds.read()
    old_column_set = set(df.columns)
    for old_col, new_col in columns.items():
        if old_col not in old_column_set:
            print('警告：列 %s 不存在' % old_col)
    if keep_old_columns:
        for old_col, new_col in columns.items():
            if old_col in old_column_set:
                df[new_col] = df[old_col]
    else:
        df.columns = [columns.get(c, c) for c in df.columns]

    return Outputs(data=DataSource.write_df(df))
# 6. 在m4上测试
out_data = bigquant_run(m4.data, 'deal_number:new_deal_number|close:new_close',keep_old_columns=False)

查看输入结果，确保符合预期。

{w:100}{w:100}{w:100}