3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在 QMT 极速策略交易系统中,ContextInfo.get_full_tick() 是获取指定合约最新分笔(Tick)行情的核心接口。然而,在实盘或高频模拟交易中,如果频繁、无节制地调用该接口,或者一次性传入过多的股票列表,极易导致策略运行卡顿、主线程阻塞,甚至触发柜台的接口限频(Rate Limit)保护。
以下是针对 get_full_tick 的多维度性能优化与防限频方案:
问题根源:当 get_full_tick(stock_code=[]) 传入空列表时,默认会尝试获取当前主图或全市场的最新快照,数据量极大,极易造成 I/O 阻塞。
优化方案:始终明确指定需要获取行情的股票池列表,缩小单次请求的数据范围。
# 错误示范:获取全市场或不确定范围
# result = ContextInfo.get_full_tick([])
# 正确示范:只获取核心股票池的行情
my_universe = ['600000.SH', '000001.SZ', '000002.SZ']
result = ContextInfo.get_full_tick(my_universe)
subscribe_quote 采用事件驱动,替代主动轮询问题根源:在 handlebar 中每年来一个 tick 就调用一次 get_full_tick,或者在定时器中高频轮询,会产生大量无效的重复请求。
优化方案:使用 ContextInfo.subscribe_quote 订阅行情,通过回调函数(Callback)被动接收数据推送。只有当行情真正发生变化时,才触发逻辑处理,从而彻底避免主动轮询带来的性能开销和限频风险。
def on_quote(datas):
# datas 结构为 {code: dataframe}
for code, df in datas.items():
last_price = df['close'].iloc[-1] # 获取最新价
# 执行交易逻辑...
def init(ContextInfo):
# 订阅沪深300指数的日线或tick行情,通过 on_quote 回调驱动
ContextInfo.subscribe_quote('000300.SH', period='tick', callback=on_quote)
run_time 的执行间隔问题根源:如果必须使用轮询,定时器设置的毫秒级别太高(如 100nMilliSecond),会导致 CPU 占用率飙升并迅速触发限频。
优化方案:根据策略的实际需求,将定时器间隔放宽。对于大多数日内波段或套利策略,1nSecond(1秒)或 5nSecond(5秒)的频次已足够,且能大幅降低接口压力。
def init(ContextInfo):
# 限制为每 2 秒执行一次行情检查,降低请求频次
ContextInfo.run_time("check_market_data", "2nSecond", "2023-01-01 09:30:00", "SH")
def check_market_data(ContextInfo):
my_universe = ContextInfo.get_universe()
if my_universe:
ticks = ContextInfo.get_full_tick(my_universe)
# 处理行情...
问题根源:QMT 策略在启动时会从第 0 根 K 线开始回放运行历史 Bar,如果在历史 Bar 运行期间高频调用 get_full_tick,不仅毫无意义(历史 Bar 无法获取实时 Tick),还会严重拖慢初始化速度。
优化方案:在 handlebar 中增加 is_last_bar() 判断,只在最后一根 K 线(即实时盘中阶段)才执行 get_full_tick。
def handlebar(ContextInfo):
# 过滤历史 K 线回放阶段,仅在最新实时 Bar 执行
if not ContextInfo.is_last_bar():
return
# 实时盘中行情获取
ticks = ContextInfo.get_full_tick(['600000.SH'])
优化方案:如果策略需要频繁读取多只股票的快照,可以在 Python 中建立一个全局字典作为缓存。每次调用 get_full_tick 后,对比时间戳或成交量,只有当数据有更新时才进行后续的复杂计算,减少重复计算带来的卡顿。