3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在 ptrade 量化交易平台中,若需要实现最小频率达到 3 秒一次的 Tick 级高频交易,通常有两种核心实现机制:tick_data 事件驱动函数 与 run_interval 定时轮询函数。
| 特性 | tick_data 事件函数 |
run_interval 定时函数 |
|---|---|---|
| 触发机制 | 盘中 Tick 行情推送触发(约 3 秒一次) | 固定时间间隔线程轮询(最小设为 3 秒) |
| 适用场景 | 精确依赖 Tick 行情、L2 逐笔数据的盘口策略 | 基于固定时间间隔查询快照、高频轮询下单等策略 |
| 下单接口 | 必须使用 order_tick |
支持 order、order_market 等通用下单接口 |
| 数据包含 | order(逐笔委托)、tick、transcation(逐笔成交) |
需要通过 get_snapshot() 主动获取盘口快照 |
| 定义入口 | 顶层回调函数 def tick_data(context, data) |
在 initialize 中通过 run_interval 注册 |
tick_data 实现 3 秒 Tick 策略实操(附源码)tick_data 函数在交易时间(09:30 - 14:59)内由系统行情推送自动触发,频率约为 3 秒一次。在 tick_data 中,下单操作需使用专门的 order_tick 接口。
def initialize(context):
# 设置目标标的
g.security = '600570.SS'
set_universe(g.security)
def tick_data(context, data):
security = g.security
# 安全性检查:确保行情数据存在
if security not in data or 'tick' not in data[security]:
return
# 获取当前 Tick 的盘口买一价
bid_grp = data[security]['tick'].get('bid_grp', {})
if not bid_grp or 1 not in bid_grp:
return
bid_1_price = bid_grp[1][0] # 买一档价格
log.info(f"当前 {security} 买一价: {bid_1_price}")
# 策略逻辑示例:买一价满足条件时,按买一档下单买入 100 股
if bid_1_price > 0:
# priceGear: '1' 表示买一档,'-1' 表示卖一档
order_tick(security, 100, priceGear="1")
def handle_data(context, data):
pass
run_interval 实现 3 秒轮询策略实操(附源码)如果策略需要定期(如每 3 秒)检查盘口快照或资金状态,可以在 initialize 中使用 run_interval 挂载后台定时任务。
def initialize(context):
g.security = '600570.SS'
set_universe(g.security)
# 注册周期处理函数,设置间隔为 3 秒
run_interval(context, interval_handle, seconds=3)
def interval_handle(context):
security = g.security
# 获取实时行情快照
snapshot = get_snapshot(security)
if not snapshot or security not in snapshot:
return
last_price = snapshot[security].get('last_px', 0)
log.info(f"高频轮询 - 当前最新价: {last_price}")
# 触发交易逻辑
cash = context.portfolio.cash
if cash > last_price * 100:
order(security, 100, limit_price=last_price)
def handle_data(context, data):
pass
tick_data 与 run_interval 均仅在交易环境生效,日线/分钟线回测环境不支持此级别运行。tick_data 中下单必须使用 order_tick;在 run_interval 中可使用常规 order 系列接口。tick_data 传入的数据中若需获取逐笔委托(order)与逐笔成交(transcation),需开通 Level2 行情权限。