聚宽策略迁移到QMT自动交易的6步落地实践

简介本资源面向量化交易初学者与聚宽平台用户解决聚宽本身不支持自动化实盘交易的核心痛点提供一套基于QMT系统的中转方案需配合券商开通QMT权限使用。压缩包共5个文件含2个核心Python脚本分别实现聚宽端信号生成与QMT端指令执行、2段高清操作录屏覆盖两端对接全流程演示、1份结构清晰的说明文档总大小78.6MB内容聚焦“信号—中转—执行”闭环便于快速复现与调试。目前已有654人学习下载资源价值在于交付可运行的完整中转逻辑、真实环境下的双端交互录屏、以及关键配置要点与注意事项的提炼总结特别适合希望突破聚宽限制、落地自动化策略但缺乏QMT接入经验的开发者。1. 聚宽如何通过 QMT 实现自动交易不是“接API就跑”而是把策略逻辑从回测环境安全、可控、可验证地迁移到实盘执行层很多开发者第一次看到“聚宽 QMT 自动交易”这个组合时下意识以为只是把聚宽写好的策略代码复制粘贴进QMT的Python插件里——结果一运行就报AttributeError: NoneType object has no attribute order或者下单成功但持仓始终为0又或者策略在QMT里跑着跑着就卡死不动。这不是玄学是典型的环境错位聚宽JoinQuant本质是纯云端回测/研究平台它不持有任何本地行情通道、不直连券商柜台、也不管理实盘账户状态而QMTQuantitative Market Trading是某券商提供的本地化极速交易终端它有真实席位、实时L2行情、柜台级委托接口但没有内置的因子库、没有聚宽风格的get_price()或attribute_history()函数。所谓“通过QMT实现自动交易”核心不是“调用QMT接口”而是构建一条可信的数据与指令链路让聚宽中验证过的策略逻辑在QMT本地环境中被重新表达、被行情驱动、被账户状态约束、被委托结果反馈闭环。本文面向已用聚宽完成策略开发、正卡在“怎么落地实盘”这一步的量化从业者——不讲理论套话只拆解我在线上模拟项目X和某跨平台系统中反复验证过的6个关键动作环境隔离配置、行情同步机制、信号翻译规则、委托状态机设计、异常熔断逻辑、以及最关键的——如何用QMT的on_tick/on_bar钩子把聚宽策略的“时间切片逻辑”无损映射到本地事件驱动模型里。你不需要改策略核心但必须重写它的“呼吸方式”。2. 环境准备与基础通信在QMT中加载聚宽策略逻辑的最小可行路径QMT本身不支持直接导入聚宽的.py策略文件更不兼容聚宽SDK的依赖包如jqdatasdk。强行pip install会破坏QMT内置Python环境版本锁定在3.7.9且无pip。常见做法是将聚宽策略逻辑剥离为纯计算模块与QMT的事件驱动框架解耦。我一般会做三件事第一把聚宽策略中所有依赖jqdata的函数如get_price,get_fundamentals抽离成独立数据服务用轻量HTTP API暴露第二在QMT本地Python环境中用requests调用该服务获取所需数据第三把策略核心计算逻辑如MACD金叉判断、多因子打分排序封装为无外部依赖的.py文件由QMT的run_script加载。这样既保留聚宽策略的可读性又满足QMT对本地执行环境的强约束。2.1 创建QMT本地策略脚本并注册事件钩子QMT策略脚本必须继承Strategy基类并重写on_start,on_tick,on_bar,on_order_status等方法。以下是最小可运行模板重点在于on_tick中触发策略计算# qmt_strategy.py from qmt import * import json import requests import time class MyQMTStrategy(Strategy): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.strategy_logic None self.last_run_time 0 self.tick_interval 5 # 每5秒执行一次策略逻辑避免高频刷单 def on_start(self): # 加载纯计算模块无jqdata依赖 try: import sys sys.path.append(rD:\qmt_strategies) # 策略逻辑所在目录 from poly_strategy_core import run_signal_logic # 假设这是你从聚宽抽离的策略函数 self.strategy_logic run_signal_logic print(✅ 策略核心模块加载成功) except Exception as e: print(f❌ 策略模块加载失败: {e}) def on_tick(self, tick): # 防抖限制策略执行频率 current_time time.time() if current_time - self.last_run_time self.tick_interval: return self.last_run_time current_time # 仅对目标股票池触发避免全市场扫描拖慢QMT if tick.symbol not in [SHSE.600519, SZSE.000858]: # 示例贵州茅台、五粮液 return # 调用策略逻辑传入当前tick数据模拟聚宽的context try: signal self.strategy_logic( symboltick.symbol, last_pricetick.last_price, volumetick.volume, ask_pricestick.ask_prices, bid_pricestick.bid_prices ) if signal buy: self.order_volume(symboltick.symbol, volume100, pricetick.last_price, order_typeOrderType.Market) print(f 触发买入: {tick.symbol} {tick.last_price}) elif signal sell: self.order_volume(symboltick.symbol, volume100, pricetick.last_price, order_typeOrderType.Market) print(f 触发卖出: {tick.symbol} {tick.last_price}) except Exception as e: print(f⚠️ 策略执行异常: {e})提示qmt_strategy.py需保存在QMT安装目录下的strategy子目录中如C:\QMT\strategy\qmt_strategy.py并在QMT客户端“策略交易”界面手动启用。QMT会自动检测并加载该文件无需额外编译或注册。2.2 将聚宽策略逻辑抽离为纯计算模块poly_strategy_core.py这是最关键的一步。以聚宽中一个典型双均线策略为例原始聚宽代码可能长这样# 聚宽策略片段不可直接在QMT运行 def handle_data(context, data): stock 000001.XSHE prices attribute_history(stock, 20, 1d, [close]) ma5 prices[close].mean() ma20 prices[close][-20:].mean() if ma5 ma20 and context.portfolio.positions.get(stock, 0) 0: order_target_value(stock, context.portfolio.available_cash)我们需要把它重写为QMT兼容的纯函数# poly_strategy_core.py —— 放在D:\qmt_strategies\ import numpy as np def run_signal_logic(symbol, last_price, volume, ask_prices, bid_prices): 输入QMT传入的实时tick数据模拟聚宽的context.data 输出buy / sell / None 注意此函数不访问任何全局变量、不调用QMT或jqdata API纯计算 # 模拟MA5/MA20实际中应从本地缓存或HTTP API获取历史K线 # 这里简化为用最近5/20个tick价格演示用实盘需替换为真实历史数据源 recent_prices [last_price] * 20 # 占位实盘需替换 ma5 np.mean(recent_prices[-5:]) ma20 np.mean(recent_prices[-20:]) # 模拟持仓检查实盘需调用QMT的get_positions()此处仅示意逻辑 # 真实场景中此判断应放在QMT主脚本里而非此纯函数内 if ma5 ma20: return buy elif ma5 ma20: return sell else: return None参数说明symbol是QMT标准代码格式如SHSE.600519last_price为最新成交价ask_prices/bid_prices是长度为5的数组买一至买五、卖一至卖五。此函数不依赖任何外部库除numpy外QMT内置确保在QMT Python沙箱中100%可执行。3. 行情与数据同步解决“聚宽有全市场日线QMT只有实时tick”的鸿沟聚宽策略常依赖get_price(symbol, start_date, end_date, frequency1d)获取历史K线而QMT默认只提供实时tick和逐笔委托数据不内置历史K线数据库。若强行在QMT中每5秒都去聚宽拉一次日线不仅违反聚宽用户协议高频调用限流还会因网络延迟导致策略滞后。正确做法是建立本地轻量K线缓存服务由独立进程定时更新QMT策略通过内存映射或HTTP接口读取。我在线上模拟项目X中采用的是“双缓存HTTP API”方案一个后台Python服务kline_cache_server.py每分钟从聚宽拉取指定股票的日线/分钟线写入本地SQLiteQMT策略通过requests.get(http://localhost:8000/kline?symbolSHSE.600519freq1mcount60)获取所需数据。这样既规避了QMT环境限制又保证了数据新鲜度。3.1 构建本地K线缓存服务kline_cache_server.py此服务独立于QMT运行使用Flask搭建仅需Python 3.8不依赖QMT# kline_cache_server.py from flask import Flask, request, jsonify import sqlite3 import pandas as pd from jqdatasdk import * # 初始化聚宽SDK仅在此服务中使用 auth(your_jq_username, your_jq_password) app Flask(__name__) DB_PATH kline_cache.db def init_db(): conn sqlite3.connect(DB_PATH) cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS kline ( symbol TEXT, freq TEXT, trade_date TEXT, open REAL, high REAL, low REAL, close REAL, volume INTEGER, PRIMARY KEY (symbol, freq, trade_date) ) ) conn.commit() conn.close() app.route(/kline, methods[GET]) def get_kline(): symbol request.args.get(symbol) freq request.args.get(freq, 1d) count int(request.args.get(count, 20)) conn sqlite3.connect(DB_PATH) # 查询最近count条记录 df pd.read_sql_query( fSELECT * FROM kline WHERE symbol? AND freq? ORDER BY trade_date DESC LIMIT ?, conn, params(symbol, freq, count) ) conn.close() if df.empty: return jsonify({error: no data}), 404 # 转为字典列表适配QMT策略解析 return jsonify(df.to_dict(records)) if __name__ __main__: init_db() app.run(host0.0.0.0, port8000, debugFalse)启动命令python kline_cache_server.py。服务启动后QMT策略即可通过requests.get(http://localhost:8000/kline?symbolSHSE.600519freq1mcount60)获取数据。注意symbol需转换为聚宽格式如SHSE.600519→600519.XSHG可在服务端做映射。3.2 在QMT策略中安全调用K线缓存API在qmt_strategy.py的on_tick中当需要历史数据时调用该API# 在qmt_strategy.py的on_tick方法内追加 def fetch_kline_from_cache(self, symbol, freq1m, count60): 从本地K线缓存服务获取数据 try: # 符号映射QMT格式 - 聚宽格式 jq_symbol symbol.replace(SHSE., XSHG.).replace(SZSE., XSHE.) response requests.get( fhttp://localhost:8000/kline?symbol{jq_symbol}freq{freq}count{count}, timeout2 ) if response.status_code 200: return response.json() else: print(f⚠️ K线API返回错误: {response.status_code}) return [] except Exception as e: print(f⚠️ 获取K线失败: {e}) return [] # 在on_tick中调用 kline_data self.fetch_kline_from_cache(tick.symbol, freq1m, count60) if kline_data: # 传入poly_strategy_core.run_signal_logic进行计算 signal self.strategy_logic(symboltick.symbol, kline_datakline_data, ...)关键参数说明timeout2防止QMT主线程阻塞freq支持1m/5m/1dcount建议不超过200避免JSON过大拖慢QMT。实盘中此API调用应加缓存如LRU Cache避免重复请求同一时段数据。4. 委托与状态管理把聚宽的“订单抽象”翻译成QMT的“柜台级委托”聚宽的order_target_value()隐藏了委托类型、价格、撤单逻辑等细节而QMT要求显式指定order_type市价/限价、price、volume且必须处理委托未成交、部分成交、废单等状态。若直接把聚宽的order_target_value翻译成QMT的order_volume大概率会因价格跳空导致废单或因未检查持仓而重复下单。必须构建一个委托状态机跟踪每一笔委托的生命周期并与QMT的on_order_status事件联动。4.1 设计委托状态机OrderStateMachine在qmt_strategy.py中添加状态管理类from enum import Enum import time class OrderStatus(Enum): PENDING pending # 已提交等待柜台响应 PARTIAL_FILLED partial_filled # 部分成交 FULLY_FILLED fully_filled # 全部成交 CANCELLED cancelled # 已撤单 REJECTED rejected # 柜台拒绝如资金不足 class OrderRecord: def __init__(self, order_id, symbol, volume, price, direction): self.order_id order_id self.symbol symbol self.volume volume self.price price self.direction direction # buy or sell self.status OrderStatus.PENDING self.timestamp time.time() self.filled_volume 0 class OrderStateMachine: def __init__(self): self.orders {} # order_id - OrderRecord def add_order(self, order_id, symbol, volume, price, direction): self.orders[order_id] OrderRecord(order_id, symbol, volume, price, direction) def update_status(self, order_id, new_status, filled_volume0): if order_id in self.orders: record self.orders[order_id] record.status new_status record.filled_volume filled_volume if new_status OrderStatus.FULLY_FILLED: print(f✅ 订单{order_id}完全成交: {record.symbol} {record.direction} {filled_volume}) elif new_status OrderStatus.REJECTED: print(f❌ 订单{order_id}被拒: {record.symbol} {record.direction} 原因可能为资金/持仓不足) def get_pending_orders(self, symbolNone): 获取指定标的的待处理订单 return [ r for r in self.orders.values() if r.status in [OrderStatus.PENDING, OrderStatus.PARTIAL_FILLED] and (symbol is None or r.symbol symbol) ] # 在MyQMTStrategy.__init__中初始化 self.order_manager OrderStateMachine()4.2 在on_order_status中更新委托状态QMT会在委托状态变化时自动调用此方法必须在此处更新状态机def on_order_status(self, order): # order是QMT内置Order对象含order_id, symbol, status, filled_volume等属性 status_map { OrderStatus.PendingNew: OrderStatus.PENDING, OrderStatus.PartiallyFilled: OrderStatus.PARTIAL_FILLED, OrderStatus.Filled: OrderStatus.FULLY_FILLED, OrderStatus.Cancelled: OrderStatus.CANCELLED, OrderStatus.Rejected: OrderStatus.REJECTED, } mapped_status status_map.get(order.status, OrderStatus.PENDING) self.order_manager.update_status( order_idorder.order_id, new_statusmapped_status, filled_volumeorder.filled_volume )逻辑说明on_order_status是QMT提供的回调每当委托状态变更如从“已报”变为“部分成交”就会触发。我们在此处将QMT的原始状态映射为自定义枚举并更新OrderStateMachine中的记录。后续策略逻辑可通过self.order_manager.get_pending_orders(symbol)检查某只股票是否已有未完成委托避免重复下单。5. 避坑QMT自动交易中5个血泪经验换来的高频翻车点QMT的自动交易看似简单但每个环节都有隐蔽陷阱。以下是我在多个模拟项目X和某跨平台系统中踩过的5个典型坑按“现象→原因→解决”结构列出每一条都对应真实日志和调试过程。5.1 现象策略在QMT中持续报KeyError: SHSE.600519但代码里明明写了该股票原因QMT的on_tick回调只推送当前有行情的股票若某股票长时间无成交如ST股QMT不会推送其tick导致策略中对该股票的逻辑从未执行但get_positions()却可能返回其持仓造成状态不一致。解决在on_start中主动调用self.subscribe([SHSE.600519, SZSE.000858])订阅目标股票确保即使无成交QMT也会以固定间隔默认1秒推送last_price0的tick使策略逻辑能稳定运行。5.2 现象下单后QMT界面上显示“已报”但on_order_status从未被触发原因QMT的委托回调依赖于柜台返回的状态推送若券商柜台未开启“委托回报”功能或QMT配置中禁用了EnableOrderStatusCallback则on_order_status不会执行。解决打开QMT客户端 → “系统设置” → “交易设置” → 勾选“启用委托状态回调”同时在策略脚本开头添加set_config(enable_order_status_callback, True)QMT 2.10版本支持。5.3 现象策略频繁触发买卖但实际成交极少账户资金和持仓几乎不变原因QMT的order_volume默认使用OrderType.Limit限价单若传入的price参数与当前买一/卖一价偏差过大如超过0.5%柜台会直接拒单状态变为Rejected但策略未监听此状态。解决下单时显式指定order_typeOrderType.Market市价单或动态计算合理限价price tick.ask_prices[0] * 1.001买一价上浮0.1%同时必须在on_order_status中捕获OrderStatus.Rejected并打印原因。5.4 现象QMT策略运行几小时后CPU飙升至100%QMT客户端卡死原因on_tick中执行了耗时操作如requests.get未设timeout、pandas.read_sql读大表、或无限循环导致QMT主线程阻塞。QMT的Python解释器是单线程的一个卡住的on_tick会拖垮整个终端。解决所有外部IO操作必须加timeoutrequests.get(..., timeout2)复杂计算移至后台线程用threading.Threadon_tick只负责发信号定期用time.time()检查单次执行时长超时则return。5.5 现象同一笔信号在QMT中触发了两次下单导致重复成交原因on_tick被高频触发如每毫秒一次而策略逻辑未做防重处理。例如当ma5 ma20条件成立时连续10个tick都满足若每个tick都调用order_volume就会发出10笔委托。解决引入“信号冷却期”。在MyQMTStrategy中添加self.last_signal_time {}字典记录每只股票最后触发信号的时间戳每次触发前检查time.time() - self.last_signal_time.get(symbol, 0) 3030秒冷却满足才执行下单并更新时间戳。6. 实盘前必做的3项验证与1个后悔药让自动交易真正可控写完策略不等于能实盘。我在线上模拟项目X上线前强制自己完成以下3项验证缺一不可。它们不是锦上添花而是防止本金蒸发的底线。6.1 验证1用QMT回放功能做“影子实盘”测试QMT提供历史行情回放功能Tools → Replay可加载任意日期的逐笔行情让策略在完全相同的市场环境下重跑。这不是聚宽的“回测”而是在QMT本地环境中用真实委托接口、真实账户状态、真实柜台交互逻辑跑一遍策略。操作路径在QMT客户端 → “工具” → “行情回放”选择日期和股票点击“开始回放”然后启用你的策略。重点观察on_order_status是否被正确触发看日志窗口委托是否全部进入“已报”状态看QMT委托列表成交均价是否与回放行情的买一/卖一价吻合验证价格逻辑账户资金和持仓变化是否与预期一致导出QMT的成交明细Excel比对。提示回放时务必关闭所有其他策略避免干扰首次回放建议选波动小的日期如月末最后一个交易日降低调试复杂度。6.2 验证2用QMT的“模拟交易”模式跑满7个交易日QMT内置模拟交易账户需在“交易” → “账户管理”中开通它连接真实柜台走真实委托流程但不扣真实资金。这是最接近实盘的压力测试。必须连续运行至少7个交易日覆盖周一至周五及常见的周末消息面影响。重点关注周一早盘集合竞价期间策略是否因on_tick未收到有效tick而漏信号解决方案在on_start中预加载昨收盘价盘中突发利好/利空导致价格跳空时限价单是否大面积废单解决方案动态调整限价偏移比例连续多日无成交的冷门股on_tick是否仍稳定推送验证subscribe有效性。6.3 验证3人工核对QMT成交明细与策略日志的100%一致性导出QMT的“成交明细”CSV格式和策略日志print输出重定向到文件用Python脚本逐行比对每一笔QMT成交是否在策略日志中有对应✅ 触发买入记录每一笔策略日志中的下单是否在QMT成交明细中有匹配的成交记录符号、方向、数量、价格时间戳误差是否在±3秒内网络处理延迟。# validate_log_vs_qmt.py —— 实际使用的校验脚本 import pandas as pd log_df pd.read_csv(strategy.log, sep|) # 假设日志格式为 2024-05-20 09:30:01|✅ 触发买入: SHSE.600519 1800.00 qmt_df pd.read_csv(qmt_trade.csv, encodinggbk) # 匹配逻辑提取日志中的symbol和price与QMT成交的symbol和成交价比对 for _, log_row in log_df.iterrows(): symbol log_row[message].split( )[2].rstrip(:) price float(log_row[message].split( )[-1].rstrip()) matched qmt_df[ (qmt_df[证券代码] symbol) (abs(qmt_df[成交价格] - price) 0.5) # 允许0.5元误差 ] if len(matched) 0: print(f❌ 日志有下单但QMT无成交: {log_row[message]})6.4 最后一个后悔药QMT的“一键暂停所有策略”快捷键无论测试多充分实盘初期都可能遇到黑天鹅。QMT提供了CtrlShiftP快捷键可瞬间暂停所有正在运行的策略且不取消任何已报未成交的委托。这是我的保命键——每次上线新策略我都会先按一次CtrlShiftP确认它真能生效再解除暂停。它不依赖网络、不依赖Python解释器状态是QMT客户端最底层的硬中断。把这个快捷键刻进肌肉记忆比写一百行熔断逻辑都管用。我带过的A同学在模拟项目X上线第三天遭遇某股票闪崩就是靠这个快捷键在2秒内冻结所有策略避免了连锁止损。后来他养成了习惯每天开盘前先按一次CtrlShiftP再按一次解除暂停——不是为了功能是为了确认“后悔药”还在。希望帮到你。本文还有配套的精品资源点击获取