问答交流

上手开发一个可视线性策略的简单模块

由bqf5z4z7创建,最终由bqf5z4z7 被浏览 29 用户

1、模板创建

首先通过命令行创建一个模块模板,其中 get_data_with_handled 是自定义的模块名,可以更替为其他的内容。

bq module init get_data_with_handled


然后在命令行中手动输入本模块的描述信息:

this is a module which can get data retained n decimal places.



至此,模块模板即可创建成功。

\

2、模块功能实现

打开位于 ~/work/get_data_with_handled/src/get_data_with_handled 中的 init.py 文件,即可看的模块开发的模板内容。

在 metadata 中,可以编辑本模块的一些基本信息,方便后期使用和维护。

重点关注 run() 函数,这里实现了模块的功能,其基本格式如下(函数输入处的方括号中的内容为可选,而输出处的方括号代表输出的结果被封装为列表):

def run(
  传入参数标识符1: 要求的传入参数类型(对传入参数的描述[, 一些具体的设置]) [= 默认值] [,
  传入参数标识符1: 要求的传入参数类型(对传入参数的描述[, 一些具体的设置]) [= 默认值], 
  ...]
) -> [
  输出参数的类型(对输出结果的描述, 其他设置)
]:
  执行操作

为达成快速熟悉模块开发流程的目的,本模块设置的功能也相对简单,以求能够快速上手,帮助入门。

本模块 run() 函数的基本定义为:

  1. 接受参数
    • param1: 字符串,接收一个数据表名(必选)
    • param2: 整数,接收一个n,以指示精确到小数点后 n 位 (0 <= n <= 6) (可选,n 默认为2)
    • param3: 字符串,接收一个字符串,字符串包含需要进行处理的字段名,字段名之间以逗号隔开 (可选,默认为所有的浮点数字段)
    • start_date: 字符串,接收一个字符串,指示数据选取的开始日期 (可选,格式要求为 yyyy-mm-dd,默认为今天)
    • end_date: 字符串,接收一个字符串,指示数据选取的开始日期 (可选,格式要求为 yyyy-mm-dd,默认为今天)
  2. 输出结果
    • 返回保留了小数点 n 位的数据 (封装成DataSource)

既然已经搞清楚了定义,那就首先让我们先实现 run() 函数的基本框架:

def run(
    param1: I.str("输入表名"),
    param2: I.int("输入精确位数: 精确到小数点后 n 位 (0 <= n <= 6, n 默认为2)", min = 0, max = 6) = 2,
    param3: I.str("输入进行处理的字段: 字段间以逗号隔开 (默认为所有的浮点数字段)") = "",
    start_date: I.str("输入开始日期: yyyy-mm-dd (默认为今天)") = _today,
    end_date: I.str("输入结束日期: yyyy-mm-dd (默认为今天)") = _today
) -> [I.port("输出DataSource", "data")]:
  # TODO

通过 I.str, I.int, I.port等来指定数据类型。注意到,对于传入参数为 I.int 类型的,我们可以通过 min 和 max 来指定基本范围,有助于防止使用者输入一个不合理的值。

注意有无默认值在定义时的区别:

# 无默认值
param1: I.str("输入表名"),
# 有默认值
param2: I.int("输入精确位数: 精确到小数点后 n 位 (0 <= n <= 6, n 默认为2)", min = 0, max = 6) = 2,

在后面使用模块时,有默认值的参数,使用者可以传入一个空值,而没有默认值的参数,使用者一定要进行输入,否则模块将无法使用。

还注意到,在设置 start_date, end_date 时,使用了 _today, 这是因为我们无法直接以硬编码的方式写入“今天”的日期,所以我们需要定义一个变量,以在 run() 函数执行前,先获取今天的日期。

from datetime import datetime as dt 
_today = dt.today().strftime('%Y-%m-%d')

接下来,一步步实现模块的功能

  1. 获取数据

    import dai
    # 编写sql
    sql = f"SELECT * FROM {param1} WHERE date BETWEEN '{start_date}' AND '{end_date}'"
    # 通过dai获取数据
    original_data = dai.query(sql).df()
    
  2. 筛选出浮点数类型的字段

    由于对非浮点类型的数据进行保留操作没有意义,因此我们要筛选出表中的浮点数类型字段。

    # 如果用户没有指定字段,则筛选出所有的浮点数类型的字段;
    # 若指定了,则从指定的字段中进行筛选
     if param3 == "":
       df = original_data
     else:
      df = original_data[param3.split(",")]
    float_columns = df.select_dtypes(include=['float']).columns.tolist()
    
  3. 对目标列进行小数点保留操作

    由于可能要处理多列,所以我们可以先定义一个辅助函数,让该函数实现对一列数据的小数点保留操作。

    # 在 run() 外定义一个辅助函数
    import math
    import numpy as np
    import pandas as pd
    
    def _keep_float_n(s: pd.Series, n: int) -> pd.Series: 
      """传入一列数据, 保留小数点后 n 位, 然后输出该列"""
      factor = math.pow(10, n)
      return np.floor(s * factor + 0.5) / factor
    

    然后在 run() 中使用该函数。

    for col in float_columns:
      original_data[col] = _keep_float_n(original_data[col], param2)
    
  4. 返回结果

    # 封装成 DataSource 类型, 以便其他模块使用
    return I.Outputs(data = dai.DataSource.write_bdb(original_data))
    

至此,本模块的功能就实现了,完整的代码如下:

"""get_data_with_handled package.

this is a module which can get data retained n decimal places.
"""
import math
from datetime import datetime as dt

import dai
import numpy as np
import pandas as pd
from bigmodule import I

# metadata
# 模块作者
author = "BigQuant"
# 模块分类
category = "示例模块"
# 模块显示名
friendly_name = "get_data_with_handled"
# 文档地址, optional
doc_url = "https://bigquant.com/wiki/"
# 是否自动缓存结果
cacheable = True

_today = dt.today().strftime('%Y-%m-%d')

def _keep_float_n(s: pd.Series, n: int) -> pd.Series: 
    """传入一列数据, 保留小数点后 n 位, 然后输出该列"""
    factor = math.pow(10, n)
    return np.floor(s * factor + 0.5) / factor

def run(
    param1: I.str("输入表名"),
    param2: I.int("输入精确位数: 精确到小数点后 n 位 (0 <= n <= 6, n 默认为2)", min = 0, max = 6) = 2,
    param3: I.str("输入进行处理的字段: 字段间以逗号隔开 (默认为所有的浮点数字段)") = "",
    start_date: I.str("输入开始日期: yyyy-mm-dd (默认为今天)") = _today,
    end_date: I.str("输入结束日期: yyyy-mm-dd (默认为今天)") = _today
) -> [I.port("输出DataSource", "data")]:
    # 获取数据
    sql = f"SELECT * FROM {param1} WHERE date BETWEEN '{start_date}' AND '{end_date}'"
    original_data = dai.query(sql).df()
    # 筛选出浮点数类型的字段
    if param3 == "":
        df = original_data
    else:
        df = original_data[param3.split(",")]
    float_columns = df.select_dtypes(include=['float']).columns.tolist()
    # 进行小数点位数的保留操作
    for col in float_columns:
        original_data[col] = _keep_float_n(original_data[col], param2)
    print(original_data.head(5)) # 该print以供测试使用, 可以删去
    return I.Outputs(data = dai.DataSource.write_bdb(original_data))


def post_run(outputs):
    """后置运行函数"""
    return outputs

3、模块使用

在完成模块功能的编写后,我们需要进行实际使用,首先要进行一些准备工作,在命令行终端中执行一下命令:

cd get_data_with_handled # 进入模块创建文件夹
bq module install --dev # 在本地环境安装模块

新建或打开一个可视化线性策略文件,在左侧导航栏中的“开发”→“示例模块”中查看我们所开发的模块,双击以在可视化策略使用:

在右侧,输入参数即可:

接下来简单的测试一下,表名定为 cn_stock_bar1d (该表的浮点数字段有 adjust_factor, pre_close, open, close, high, low, change_ratio, turn, upper_limit, lower_limit)

  • 点击可视化策略左上方的运行按钮,直接执行,结果如下所示:

    \

  • 设置精确位数为4,然后执行,结果如下所示:

    \

  • 设置精确位数为4,列为 open,close,high,low,结果如下所示:

  • 设置精确位数为2,列为 date, 结果如下所示(由于 date 不为浮点数类型,所以不会执行任何精度保留操作,返回的数据就是读取的原始数据):


可以发现,结果正如我们所预期的那样,那么我们暂时可以确定该模块是可以正常使用的。

但本模块在考虑上仍有些不足,例如当使用者输入了一张不存在的表或一些不存在的字段时,该模块会直接报错,没有进行相应的处理或防护。

如果希望更多人使用,那么执行以下命令:

#  测试完成后卸载开发环境模块
bq module uninstall --dev

#  发布模块到模块库,以用于正式使用
bq module publish

\