3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在 PTrade 交易模块中,对于需要秒级监控(如监控盘口快照、动态止盈止损、捕捉高频套利机会)的策略,常规的 handle_data(分钟/日线级别)无法满足频率要求。此时可以使用 PTrade 提供的 run_interval 接口来实现指定时间间隔(秒级)的周期性任务。
run_interval 接口说明initialize 函数中调用。seconds),最小时间间隔为 3 秒。若设置小于 3 秒,系统默认调整为 3 秒。以下是一个完整的 PTrade 策略源码示例,展示如何使用 run_interval 每 3 秒获取一次股票最新行情快照并触发交易逻辑:
def initialize(context):
# 1. 设置标的股票
g.security = '600570.SS'
set_universe(g.security)
# 2. 绑定秒级定时任务,每3秒执行一次 custom_interval_handle 函数
run_interval(context, custom_interval_handle, seconds=3)
log.info("已成功注册3秒定时轮询任务")
def custom_interval_handle(context):
"""
自定义定时轮询函数
"""
security = g.security
# 获取标的当前行情快照
snapshot = get_snapshot(security)
if not snapshot or security not in snapshot:
log.warning("获取行情快照失败")
return
last_px = snapshot[security].get('last_px', 0)
bid_1 = snapshot[security].get('bid_grp', {}).get(1, [0])[0]
ask_1 = snapshot[security].get('offer_grp', {}).get(1, [0])[0]
log.info(f"[3秒轮询] {security} 最新价: {last_px}, 买一价: {bid_1}, 卖一价: {ask_1}")
# 示例策略逻辑:当最新价突破阈值且无持仓时买入
# current_pos = get_position(security).amount
# if last_px > 50.0 and current_pos == 0:
# order(security, 100)
def handle_data(context, data):
# 必选主函数,若无分钟/日线逻辑可保持 pass
pass
多线程并发安全:
run_interval 是通过独立的线程运行的。如果同时设置了多个 run_interval 任务或与 tick_data、handle_data 配合使用,需特别注意全局变量 g 的读写冲突与订单重复下达问题。
接口并发与流控限制:
在秒级轮询函数中,应尽量避免高频调用带有 HTTP 限制或柜台限制的在线查询接口(如 get_fundamentals 财务接口、频繁的订单全量查询等),优先使用本地缓存或快照接口 get_snapshot。
服务器重启保护:
实盘环境中若服务器发生重启,策略重新加载时会再次触发 initialize。如果策略包含状态变量,建议结合 set_parameters(not_restart_trade="1") 进行防重复挂单保护,或通过 pickle 进行持久化存储。