3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
非常理解您在实盘中遇到这种数据延迟导致错误止损的心情。在量化交易中,数据源的准确性和及时性是策略的生命线。针对您提到的 get_full_tick 在特定时刻返回了 10 分钟前的过期数据的问题,这在基于本地客户端架构的量化软件中确实偶有发生。以下是深度原因分析以及相应的解决方案。
get_full_tick 会返回过期数据?在 QMT 的底层架构中,get_full_tick 函数的作用是直接从本地客户端的内存中读取该股票最新的分笔快照。出现数据不准确(延迟)通常由以下几个原因导致:
get_full_tick 是一种“主动拉取(Polling)”的方法。如果在 handlebar 中调用,它只反映调用那一瞬间本地缓存的状态,缺乏对数据新鲜度的校验。为了避免再次发生“卖飞”或错误止损,建议对策略的数据获取模块进行以下升级:
subscribe_quote 进行事件驱动(强烈推荐)不要主动去拉取数据,而是让系统在收到最新 tick 时主动通知策略。使用 ContextInfo.subscribe_quote 订阅目标股票的 tick 数据,并在回调函数中处理止损逻辑。这样可以确保你处理的永远是刚刚到达的最新数据。
def init(ContextInfo):
# 订阅目标股票的 tick 行情,设置回调函数
ContextInfo.subscribe_quote('002984.SZ', period='tick', dividend_type='none', callback=on_quote)
def on_quote(datas):
# 当有最新 tick 到达时,此函数会被触发
for stock_code, data in datas.items():
if not data.empty:
# 获取最新的一条 tick 数据
latest_tick = data.iloc[-1]
latest_price = latest_tick['close'] # tick周期下的close即为最新价
# 在这里执行你的止损判断逻辑
# print(f"{stock_code} 最新推送价格: {latest_price}")
def handlebar(ContextInfo):
# 使用了 subscribe_quote 后,常规的监控逻辑可以转移到回调函数中
pass
get_market_data_ex 替代并开启订阅如果您仍然希望在 handlebar 中按周期(如 1 分钟线)进行判断,建议废弃 get_full_tick,改用更健壮的 get_market_data_ex,并务必将 subscribe 参数设置为 True。这会在后台强制建立行情订阅链路,保证本地数据的实时性。
def handlebar(ContextInfo):
# 获取最新 tick 数据,强制开启订阅 (subscribe=True)
tick_data = ContextInfo.get_market_data_ex(
fields=['close'],
stock_code=['002984.SZ'],
period='tick',
count=1,
subscribe=True
)
if '002984.SZ' in tick_data and not tick_data['002984.SZ'].empty:
# 获取最新价
latest_price = tick_data['002984.SZ']['close'].iloc[-1]
# 获取该数据的时间戳索引,用于校验数据新鲜度
data_time = tick_data['002984.SZ'].index[-1]
# 增加时间戳校验防线:如果数据时间与当前系统时间相差过大(例如超过30秒),则拒绝执行交易
# (此处需引入 datetime 模块进行时间差计算)
# 执行止损逻辑...
在实盘中,特别是涉及止损、清仓等不可逆的敏感操作时,永远不要相信单一的数据源。如果 get_full_tick 触发了止损条件,可以在代码中增加一步交叉验证:
def handlebar(ContextInfo):
target_stock = '002984.SZ'
tick_dict = ContextInfo.get_full_tick([target_stock])
if target_stock in tick_dict:
price_from_tick = tick_dict[target_stock]['lastPrice']
# 假设 15.15 是止损线
if price_from_tick < 15.15:
# 【防线】不要立刻下单,用 get_market_data_ex 再查一次 1分钟线的收盘价或最新 tick
verify_data = ContextInfo.get_market_data_ex(['close'], [target_stock], period='1m', count=1)
if not verify_data[target_stock].empty:
price_from_kline = verify_data[target_stock]['close'].iloc[-1]
# 如果两者偏差过大(例如超过 1%),说明数据异常,放弃本次操作
if abs(price_from_tick - price_from_kline) / price_from_kline > 0.01:
print(f"警告:数据异常!Tick价格 {price_from_tick} 与 K线价格 {price_from_kline} 偏差过大,拒绝止损。")
return
# 验证通过,执行下单
# passorder(...)
get_full_tick 获取非主图品种的最新价。subscribe_quote 事件驱动模式。