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

由bqf5z4z7创建，最终由bqf5z4z7更新于2024-05-13 02:45 被浏览 50 用户

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() 函数的基本定义为：

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

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

获取数据

import dai
# 编写sql
sql = f"SELECT * FROM {param1} WHERE date BETWEEN '{start_date}' AND '{end_date}'"
# 通过dai获取数据
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()

对目标列进行小数点保留操作

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

# 在 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)

返回结果

# 封装成 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