配置简介
这里列出常见配置项与说明,如果看不懂不知道该怎么用,可以直接去常用模版里复制完整模版。在模版的基础上修改比较方便。
字段
通用字段配置
| 参数名 | 说明 | 默认值 | 参数要求 |
|---|---|---|---|
| strategy_id | 策略标识 | 自动生成 | 唯一值,软件版本判断策略的唯一性 |
| strategy_type | 策略类型 | StrategyType.SELECT | STRATEGY:策略模式 SELECT:选择模式 |
| select_period | 选择器数据周期 | Period.D1 | 周期字典 选择器周期必须大于D1 |
| next_period | next数据跳动周期 | Period.D1 | 周期字典 数据跳动的周期 |
| channel_period | 渠道周期数据 | 默认和next_period一致 | 实时数据获取的周期 |
| select_extend_len | 选择器数据延长天数 | 30 | on_select中所需的数据向前延长天数 |
| next_extend_len | next数据延长天数 | 0 | 如果select和next的period一致会取最大值 |
| max_workers | 程序数 | 1 | 大于1会启用多进程 |
回测字段配置
| 参数名 | 说明 | 默认值 | 参数要求 |
|---|---|---|---|
| backtest_date_range | 回测时间 | 无 | 单日期如'2024-03-05' 范围日期如: ['2024-03-05','2024-06-05'] |
| backtest_money | 回测初始资金 | 100000 | 回测初始资金 |
| backtest_signal_analysis | 回测信号分析 | 无 | 如果设置[1,5,20] 则生成发生信号以后的1,5,20日收益率 |
| backtest_db_is_clear_history | 清理历史数据 | True | 回测时是否清除上次执行的历史数据,默认清理 |
函数
on_symbol(标的设置函数)
设置比如需要回测的标的【"标的"是股票,期货等代码的概念】
python
def on_symbol(so: SymbolOp):
so.set_symbol(['sz.000001', 'sz.000002'])on_select(选择器函数)
- 可以通过【select_period】属性来设置so的周期数据
- 该函数比较特殊,为了避免未来数据,这个函数只能拿到"昨天"之前的数据
- so.last() 是从“昨天”开始获取数据。【on_next】函数的so.last()是从“今天”开始的
- 例如当回测到2025-08-06。 选票的概念是你基于昨天的数据去选票也就是2025-08-05。
- 所以last只能拿到昨天以前的数据。
- 每次进来都是一个标的,通过so.allow()代表选中当前进来的标的
python
def on_select(so: SelectOp):
# 这里填写选票判断条件
so.allow() # 同意该标的on_filter_select(选择结果过滤器函数)
- 该函数是基于对【on_select】的集中过滤,一般用于排名取出前N个票。
- 一般用fso.filter_df或者fso.filter_ls 获取"so.allow({xx:xx})"里传入的数据。然后进行排名
python
def on_filter_select(fso: FilterSelectOp):
# 按照上涨天数排序
df_sorted = fso.filter_df.sort_values(by=['上涨天数', "涨停天数"], ascending=False)
# 取前20个
top_20 = df_sorted.head(20)
# 存储筛选结果
fso.set_symbol(top_20.symbol.to_list())on_next(数据跳动函数)
- 每个标的的每个数据跳动都会进到这个方法,跳动周期基于【next_period】设置的值,核心策略编写函数。
python
# 当【next_period】周期的数据每次跳动都会执行,
def on_next(no: NextOp):
# 这里编写买入卖出的代码
pass
# 另类写法,代表当时间大于13:00的时候会执行一次以后就不会执行了
# 1300可以是任何时间
def on_next_1300(no: NextOp):
# 这里编写买入卖出的代码
passon_filter_next_xxxx(数据跳动信号过滤器函数)
- 一般使用场景,on_next函数中有10个标的满足我下单的条件。但是我余额可能不够。我希望按某种条件进行分配下单。
- 在【on_next】里调用no.allow({xx:xx}) 传入数据后
- 集中排名数据,然后统一在该函数进行集中下单
python
# 后4位数是时间会在该时间后触发函数
def on_filter_next_1450(nfo: FilterNextOp):
passon_bar_xx(bar合成器函数)
- 该函数的功能基本和on_next一致,当数据合成bar完成后调用。
- PackOp是对应周期的bar操作类。和NextOp一样可以使用po.last()函数等数据获取。
python
# 其中m5是周期的缩写,代表合成5分钟的bar。
def on_bar_m5(no: NextOp, po: PackOp):
pass