自定义模块教程 (副本)

由qxiao创建，最终由qxiao更新于2024-05-15 02:10 被浏览 9 用户

导语

本文将详细的展示如何开发自定义模块。我们希望开发一个模块，用来修改数据的列名：

输入：有一个输入数据源，为输入数据
参数：列名映射，格式为：原列名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}

添加自定义模块

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

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

一边开发一边测试

为了便于开发，我们新建一个代码单元，先在代码单元里，一边开发，一边测试。

从自定模块主函数里拷贝代码到代码单元： {w:100}{w:100}{w:100}

定义好接口

代码如下：

input_ds：输入数据源
columns：列名映射
Outputs.data：输出数据源

# 修改数据列名
def bigquant_run(input_ds, columns):
    # 示例代码如下。在这里编写您的代码
    # df =  pd.DataFrame({'data': [1, 2, 3]})
    # data_1 = DataSource.write_df(df)
    # data_2 = DataSource.write_pickle(df)
    return Outputs(data=None)

# 在m18上测试
bigquant_run(m18.data, 'avg_amount_0:new_avg_amount_0|rank_return_0/rank_return_5:return_change_5')

代码实现

# 修改数据列名
def bigquant_run(input_ds, columns):
    # 解析列映射为dict, TODO: 验证输入是否有效
    columns = dict(c.split(':') for c in columns.split('|'))
    print('列名映射：', columns)
    # 输出数据源
    dataset_ds = DataSource()
    output_store = dataset_ds.open_df_store()

    for key, df in input_ds.iter_df():
        df.columns = [columns.get(c, c) for c in df.columns]
        df.to_hdf(output_store, key)
        row_count = len(df)
        print('%s: %s' % (key, len(df)))

    dataset_ds.close_df_store()
    return Outputs(data=dataset_ds)

# 6. 在m4上测试
mx = bigquant_run(m4.data, 'avg_amount_0:new_avg_amount_0|rank_return_0/rank_return_5:return_change_5')