问题描述
比如run_daily(buy_sig, time='10:30')
我是强行设置了10点30分,其实我想从9点30,9点45,10点等等,每隔15分钟,去跑一下回测,看看哪个时间点进入会有相对较好的收益。
如果我把g.t=15作为一个全局变量,上面这句不知道怎么表达
run_daily(buy_sig, time='10:30')该如何改??请教!!
解决方案
在 JoinQuant (聚宽) 平台中,run_daily 的 time 参数接收的是一个字符串(如 '10:30')。如果你想通过修改全局变量(如 g.t)来动态改变回测的时间点,或者想在同一个策略中每隔 15 分钟执行一次,可以通过Python字符串格式化或循环注册来实现。
以下为你提供三种常见的解决方案,满足你不同的优化需求:
方案一:每次回测测试一个特定的时间点(参数化单次执行)
如果你希望每次回测只在一个特定的时间点执行(比如这次回测测 9:45,下次测 10:00),你可以将时间直接定义为字符串变量,或者通过计算生成字符串:
def initialize(context):
# 假设你想测试 10:15 这个时间点
g.hour = 10
g.minute = 15
# 使用 Python 的 f-string 格式化时间字符串,保证两位数对齐(如 09:05)
run_time = f"{g.hour:02d}:{g.minute:02d}"
# 动态传入时间参数
run_daily(buy_sig, time=run_time)
def buy_sig(context):
log.info(f"当前执行时间: {context.current_dt}")
# 你的买入逻辑
优化建议:在研究环境中,你可以使用 create_backtest API 批量创建回测,每次传入不同的 extras 参数来自动遍历所有时间点。
方案二:在同一次回测中,每隔 15 分钟执行一次(循环注册)
如果你想在同一次回测中,让策略在 9:30、9:45、10:00 等每隔 15 分钟都跑一次 buy_sig,你可以利用 Python 的 for 循环在 initialize 中批量注册 run_daily:
def initialize(context):
g.t = 15 # 设置间隔为15分钟
# 遍历交易时间的每一个小时和分钟
for hour in (9, 10, 11, 13, 14):
for minute in range(0, 60, g.t):
# 过滤掉非交易时间段
if hour == 9 and minute < 30:
continue
if hour == 11 and minute > 30:
continue
# 格式化时间字符串
time_str = f"{hour:02d}:{minute:02d}"
# 批量注册定时任务
run_daily(buy_sig, time=time_str)
def buy_sig(context):
log.info(f"触发信号,当前时间: {context.current_dt}")
# 你的买入逻辑
注意:run_daily 允许在同一个策略中被多次调用,每次调用都会生成一个新的定时任务。
方案三:使用分钟级回测 + 逻辑判断(最灵活)
如果你希望策略以分钟级(every_bar)运行,并在 handle_data 或每分钟执行的函数中通过 g.t 来控制触发频率,可以这样做:
def initialize(context):
g.t = 15 # 每 15 分钟触发一次
# 每分钟运行一次 check_signal
run_daily(check_signal, time='every_bar')
def check_signal(context):
# 获取当前的分钟数
current_minute = context.current_dt.minute
# 如果当前分钟数能被 g.t 整除(例如 0, 15, 30, 45),则执行买入逻辑
if current_minute % g.t == 0:
buy_sig(context)
def buy_sig(context):
log.info(f"满足 {g.t} 分钟间隔,执行买入逻辑: {context.current_dt}")
# 你的买入逻辑
总结
- 如果你是为了寻找最佳的单一入场时间,推荐使用方案一,结合批量回测工具进行参数寻优。
- 如果你是为了在盘中高频多次捕捉机会,推荐使用方案二或方案三。