这两年做A股量化最常被问的一句话是你的策略代码写在聚宽上怎么让它在QMT里真实下单聚宽研究环境在云端算完信号只能看不能碰券商通道QMT装在本地能下单但写策略又没那么顺手。于是很多人卡在最后这一公里信号怎么从策略平台安全、及时地送到交易终端。我用的方案很简单也很稳Redis做桥聚宽负责出信号QMT本地用xtquant接口消费信号并执行交易。中间的关键环节就是信号的数据结构设计、消息队列的选择、以及xtquant连接QMT时那些说不清道不明的坑。这篇文章把我的完整做法、踩坑记录、代码模板都摊开讲适合正在搭“聚宽研究QMT实盘”这套组合的个人开发者参考。1. 方案拆解聚宽出信号QMT下单Redis当中间人1.1 为什么需要中间层而不是直接调接口聚宽的策略代码跑在它的云端研究环境里QMT终端在你自己的电脑上两者不在同一个网络。QMT虽然也支持Python写策略但很多人的回测和研究习惯都在聚宽上不想全部重写而聚宽的模拟交易和实盘对接能力又不如QMT直接尤其面对券商极速柜台时最终交易通道还是要走QMT。直接让聚宽调用QMT的接口不现实隔着公网券商交易接口也不是用来给云端程序随意连的。所以需要一个中间层把“信号”这份数据从云端搬到本地。这个中间层要满足几个条件写入简单、延迟低、能排队、下游断开时不丢消息。对比了几种常见的方案最终我全在用Redis理由很实在。1.2 信号传递方案的取舍数据库轮询、消息队列、还是Redis早期有人用MySQL或PostgreSQL来做信号传递聚宽把信号insert一张表QMT侧每秒轮询一次。这种做法能跑通但存在明显问题轮询太频繁压力大轮询间隔太长信号延迟高还要自己维护状态字段去重、过期、断线补偿都得手写。消息中间件如RabbitMQ、Kafka又偏重部署和维护成本对个人量化来说不划算。Redis在这场景里几乎是量身定做。它足够轻单个实例就能跑它支持List、Pub/Sub、Stream等多种消息模型延迟是毫秒级它自带过期时间可以给每个信号设置有效期它还有现成的持久化机制开启AOF后即使重启也不会丢数据。更关键的是聚宽环境可以安装redis-py客户端QMT侧同样用redis-py读取两边语言一致代码写起来非常顺。1.3 整体数据流和部署拓扑我的实际部署是这样的聚宽研究环境运行策略代码通过!pip install redis安装 redis-py。云服务器上部署一个Redis实例开启密码和AOF持久化安全组只放行聚宽出口IP和本地QMT的IP。QMT本机跑一个常驻Python脚本先连接QMT交易通道再连接同一个Redis用brpop阻塞消费信号队列。信号以JSON字符串形式入队包含消息ID、股票代码、方向、数量、价格类型、创建时间、过期时间等字段。QMT侧先做去重和过期检查再调用xtquant的order_stock下单并把委托回报写进日志。这套链路下来聚宽只管“想买什么、想卖什么”QMT只管“按信号执行”职责清晰。延迟通常在几十毫秒级别足够覆盖日级甚至分钟级的交易频率。2. Redis信号管道消息结构、队列选型和序列化避坑2.1 信号消息的字段设计宁可多写不要漏写我发现很多人第一次搭这个桥的时候只往Redis里塞一句话“buy 600519 100股”。结果QMT侧一跑就发现缺这缺那没有消息ID没法去重没有时间戳没法判断信号是否过期没有价格类型不知道该市价还是限价下单。信号这份数据是两套系统的唯一沟通语言字段必须设计完整。我常用的信号结构大致长这样{ message_id: a1b2c3d4..., # 唯一ID聚宽侧生成用于QMT去重 strategy_id: test_strategy, # 策略标识方便日志区分 symbol: 600519.XSHG, # 聚宽代码格式 side: buy, # buy / sell volume: 100, # 股数必须100的整数倍 price_type: market, # market / limit price: 0.0, # 限价单的价格市价单填0 created_at: 2025-01-01T09:30:00, # 聚宽时间ISO8601格式 expire_after: 60 # 信号有效期单位秒 }message_id是最不能省的一个字段。聚宽侧用股票代码、方向、数量、毫秒级时间戳拼接后做MD5基本能保证唯一。QMT侧拿到消息先查这个ID有没有处理过处理过就跳过这是防止重复下单的第一道防线。expire_after也很重要比如一个信号本来打算在9点30分买入结果网络抖动QMT在9点40分才收到这时候市场价格可能已经变了信号就不该继续执行。2.2 List队列、Pub/Sub、Stream到底该用哪个Redis的几种消息模型我都试过最直接的建议是别用Pub/Sub做交易信号。Pub/Sub是“发了就完”如果QMT侧脚本当时没在线消息直接丢失没有任何补偿机制。交易信号不像聊天消息丢了就丢了丢了就意味着该买的没买、该卖的没卖代价很大。相比之下List队列天然支持lpush写入、brpop阻塞消费QTM断开期间消息会积压在队列里重新连上后还能继续消费。对于信号量不大的个人量化场景List已经完全够用。如果你对可靠性要求更高或者准备同时跑多套策略、多个账户可以考虑Redis Stream。Stream支持消费组、ACK确认、消息持久化相当于把Kafka的能力塞进了Redis。QMT消费一条信号后要主动XACK没ACK的消息会在重启后重新投递这比“手工维护已处理ID集合”更严谨。缺点是实现复杂度高一些个人刚开始搭建建议先上List跑顺了再迁移到Stream。2.3 Redis安装和安全加固别让端口裸奔在公网既然聚宽在云端Redis就必须能被公网访问这就意味着你的Redis一旦配置不当等于给全网的人送了一个内存数据库。我见过有些教程让用户把bind 0.0.0.0一写、密码一设就完事这远远不够。以Linux云服务器为例我的redis.conf里至少会有这些配置bind 0.0.0.0 protected-mode yes port 6379 requirepass YourStrongPasswordHere appendonly yes appendfsync everysec maxmemory 512mb maxmemory-policy noeviction rename-command FLUSHALL rename-command CONFIG rename-command的作用是把高危命令改名或禁用防止攻击者连进来后一键清空数据。maxmemory-policy noeviction表示内存满了宁可写入报错也不自动淘汰任何key对交易信号来说这点非常重要——信号数据不能因为内存淘汰策略凭空消失。安全组和防火墙也要同步设置只放行聚宽出口IP和你本地QMT所在的公网IP其他来源一律拒绝。Redis的密码建议用随机生成的长串不要用生日、手机号、123456这类容易被爆破的。Windows用户如果想把Redis装在本地可以直接用tporadowski的redis-windows发行版或者用WSL、Docker跑一个Linux容器。日常调试信号的时候我习惯开一个Another Redis Desktop Manager来盯着队列变化能看到quant:signal:queue的积压数量、quant:signal:done的去重集合比敲redis-cli直观得多。3. 聚宽侧实战策略产生信号推送到Redis3.1 聚宽环境里安装和连接Redis聚宽研究环境本质是一个云端Notebook可以通过!pip install redis安装第三方库。装好之后连接方式和你本地Python完全一致。要注意聚宽环境默认是Python3redis-py新老版本的参数有差异我建议统一用比较新的版本连接时加上decode_responsesTrue这样读出来的字符串就是str而不是bytes后面处理JSON会省很多事。import redis r redis.Redis( hostyour.redis.server, port6379, passwordYourStrongPasswordHere, db0, decode_responsesTrue ) # 测试连通 print(r.ping())如果ping()返回True说明聚宽环境能正常访问你的Redis服务。这时候不要把连接代码写在策略函数外面直接执行最好封装成一个send_signal函数每次触发交易逻辑时调用。3.2 信号生成与写入的完整代码示例下面是一个可以直接抄走的示例。它定义了一个send_signal函数聚宽策略里只要调用它就能把信号推进Redis队列。import redis import json import hashlib from datetime import datetime REDIS_HOST your.redis.server REDIS_PORT 6379 REDIS_PASSWORD YourStrongPasswordHere REDIS_QUEUE_KEY quant:signal:queue r redis.Redis( hostREDIS_HOST, portREDIS_PORT, passwordREDIS_PASSWORD, db0, decode_responsesTrue ) def build_message_id(symbol, side, volume): raw f{symbol}-{side}-{volume}-{datetime.now().strftime(%Y%m%d%H%M%S%f)} return hashlib.md5(raw.encode()).hexdigest() def send_signal(symbol, side, volume, price_typemarket, price0.0, expire_after60): msg { message_id: build_message_id(symbol, side, volume), strategy_id: my_strategy_v1, symbol: symbol, side: side, volume: volume, price_type: price_type, price: price, created_at: datetime.now().isoformat(timespecseconds), expire_after: expire_after, } r.lpush(REDIS_QUEUE_KEY, json.dumps(msg, ensure_asciiFalse)) print(fsignal pushed: {msg[message_id]} {symbol} {side} {volume})调用的时候长这样# 买入100股贵州茅台市价成交 send_signal(600519.XSHG, buy, 100, price_typemarket) # 卖出100股平安银行限价12.5元 send_signal(000001.XSHE, sell, 100, price_typelimit, price12.5)这里有个细节聚宽的股票代码格式是600519.XSHG、000001.XSHEQMT侧需要的是600519.SH、000001.SZ两边代码后缀不一样必须在QMT侧做一次转换。我见过有人直接在聚宽侧就把代码改写成QMT格式结果回测和研究时就乱了不建议这么干保持聚宽原生格式到QMT侧再转换是最不容易出错的。3.3 聚宽侧的联动思路和建议如果你不想每次手动触发send_signal想让策略每天自动跑可以研究聚宽的定时运行功能。聚宽的研究环境一般通过run_daily之类的方式挂任务在设定的时间点执行你的策略代码把当天的买卖信号推送到Redis。实际过程中我发现一个不算问题但很影响体验的点聚宽环境访问公网Redis有时候不稳定可能是网络抖动也可能是云服务商对出站连接有限制。所以我在聚宽侧加了异常捕获推送失败时把信号写入本地csv之后可以人工核查有没有漏推。别小看这一步信号漏推在实盘里往往比推错更麻烦。4. QMT侧实战xtquant连接与真实下单4.1 连接xtquant之前先搞定client is nullxtquant是QMT自带的一套Python接口库主要模块是xttrader和xtdata前者负责交易后者负责行情。QMT终端安装目录下的bin/xtquant路径里就有现成的库Python脚本里把这个路径加入sys.path就能导入。连接过程中遇到最多的报错就是热词搜出来的那个client is null。我一开始也被这个坑折腾了两天后来把原因总结成几类QMT终端没有保持登录状态。xtquant连接的是本机的QMT交易进程如果QMT主程序没打开、没登录券商账户进程不存在或接口没就绪客户端连接就拿不到有效引用。路径不对。连接时指定的path参数应该是QMT安装目录下的userdata_mini文件夹不同券商的安装目录可能不同不能想当然直接照抄别人的路径。权限没开通。极速交易权限、QMT接口权限这类功能需要向券商申请很多券商默认不给你开。xtquant版本和QMT终端版本不匹配。有些券商定制版QMT的xtquant接口比公共版本旧混用就会出现诡异问题。更稳妥的做法是优先使用miniQMT模式也就是极简客户端。miniQMT不需要打开完整的图形界面登录后可以在后台持续运行xtquant连接它更稳定重启策略也不影响终端状态。路径上找userdata_mini而不是userdata这点很多新人不注意。4.2 消费Redis信号并调用order_stock下单下面的代码是QMT侧的核心消费逻辑流程是连接QMT - 订阅账户 - 连接Redis - 循环brpop - 去重 - 检查过期 - 转换代码 - 下单 - 记录处理结果。import sys sys.path.append(rD:\QMT\bin\xtquant) from xtquant.xttrader import XtQuantTrader from xtquant.xttype import StockAccount from xtquant import xtconstant import redis import json import time from datetime import datetime, timedelta # ---------- 连接QMT ---------- path rD:\QMT\userdata_mini session_id 20250101 xt_trader XtQuantTrader(path, session_id) xt_trader.start() connect_result xt_trader.connect() if connect_result ! 0: print(QMT connect failed, code:, connect_result) exit(1) account StockAccount(你的资金账号) xt_trader.subscribe(account) print(QMT connected, account subscribed) # ---------- 连接Redis ---------- r redis.Redis( hostyour.redis.server, port6379, passwordYourStrongPasswordHere, db0, decode_responsesTrue ) REDIS_QUEUE quant:signal:queue REDIS_DONE quant:signal:done def to_qmt_symbol(joinquant_symbol): code, exchange joinquant_symbol.split(.) if exchange XSHG: return f{code}.SH elif exchange XSHE: return f{code}.SZ return joinquant_symbol def is_processed(message_id): return r.sismember(REDIS_DONE, message_id) def mark_processed(message_id): r.sadd(REDIS_DONE, message_id) r.expire(REDIS_DONE, 86400) def place_order(signal): qmt_symbol to_qmt_symbol(signal[symbol]) action xtconstant.STOCK_BUY if signal[side] buy else xtconstant.STOCK_SELL price_type xtconstant.FIX_PRICE if signal[price_type] limit else xtconstant.LATEST_PRICE price float(signal.get(price, 0)) volume int(signal[volume]) # 简单风控单笔限价买入金额上限防止参数配错导致大额下单 if action xtconstant.STOCK_BUY and price_type xtconstant.FIX_PRICE and price * volume 50000: print(f风控拒绝单笔金额超上限 {signal[message_id]}) return order_id xt_trader.order_stock( account, qmt_symbol, action, volume, price_type, price, strategy, signal[message_id] ) print(f委托提交{signal[message_id]} {qmt_symbol} {signal[side]} {volume}股 order_id{order_id}) print(开始消费信号队列按 CtrlC 退出) while True: item r.brpop(REDIS_QUEUE, timeout5) if not item: continue signal json.loads(item[1]) if is_processed(signal[message_id]): print(f重复信号跳过{signal[message_id]}) continue created datetime.fromisoformat(signal[created_at]) if datetime.now() - created timedelta(secondssignal.get(expire_after, 60)): print(f信号已过期放弃{signal[message_id]}) continue place_order(signal) mark_processed(signal[message_id])下单成功后order_stock会返回一个order_id。需要注意的是order_id返回不代表成交只是委托被券商接受了后续成交回报要通过回调获知。所以我还建议注册一个回调类实时打印委托状态和成交信息这样在实盘调试时能第一时间发现问题。from xtquant.xttrader import XtQuantTraderCallback class MyCallback(XtQuantTraderCallback): def on_stock_order(self, order): print(委托回调, order.order_id, order.order_status) def on_stock_trade(self, trade): print(成交回调, trade.order_id, trade.traded_volume, trade.traded_price) xt_trader.register_callback(MyCallback())4.3 实盘安全不只看Redis下单之前还要做几道检查很多人把信号推到Redis就以为万事大吉实际上QMT侧拿到信号后在下单前还要做几道本地检查否则很容易因为一些小问题产生意外交易。第一股票代码格式必须正确。QMT的股票代码是600519.SH、000001.SZ这种格式指数还要额外处理我上面的to_qmt_symbol函数只覆盖了A股股票的情况。如果你的策略涉及ETF、可转债后缀可能变成SH、SZ还是不变的但代码位数和逻辑要再核对。第二委托数量必须是100股的整数倍。聚宽侧如果出现手误把volume写成50QMT侧下单会直接失败所以最好在QMT侧也做一次取整或拒绝处理。第三涨跌停和停牌检查。市价单遇到涨跌停、停牌股票可能出现无法成交或者价格滑点巨大的情况。简化方案是下单前用xtdata拉一下最新行情判断当前价格是否在涨跌停范围内不在就不下单。第四资金和持仓检查。QMT接口本身有查询函数买入前查一下可用资金卖出前查一下持仓数量不满足直接跳过信号并记录原因。这层检查在策略逻辑不完善的时候特别有用能避免资金不足反复触发废单。5. 常见问题与排雷实录5.1 QMT侧高频报错速查表现象可能原因处理方式client is nullQMT未登录、未使用miniQMT模式、userdata_mini路径错误、xtquant版本不匹配、极速交易权限未开通手动打开miniQMT并登录一次确认路径正确向券商确认接口权限connect_result非0端口被占用、QMT进程卡死、session_id与已有进程冲突换一个session_id重启QMT终端检查任务管理器委托返回负值账户未订阅、资金不足、股票代码错误、价格类型不合法打印返回码对照xtquant文档逐项排查委托提交成功但没有成交回调账户资金冻结失败、股票停牌、涨跌停封板查询委托状态和成交记录确认实际持仓情况QMT跑一段时间后策略脚本退出内存泄漏、网络断线未重连、Redis连接超时给脚本加异常捕获和自动重启机制用supervisor或Windows计划任务守护5.2 Redis侧的典型问题聚宽推信号QMT收不到是最常被问的场景。先从最简单的地方排查用Redis Desktop Manager连上去看quant:signal:queue有没有数据增长。如果队列长度一直在涨但QMT没反应说明QMT侧消费脚本没正常工作如果队列长度是0且QMT侧也没日志可能是聚宽根本没推成功检查聚宽Notebook里的r.ping()是否返回True。公网Redis在高并发下也可能成为瓶颈但对个人量化来说每天几百条信号根本谈不上压力。真正要注意的是AOF持久化是否开启。我遇到过云服务器重启后队列消息全部消失的情况原因就是当时只开了默认的RDB快照快照周期没到内存数据全丢了。开启appendonly yes并设置appendfsync everysec之后再没出现过这类丢数据问题。信号里的时间字段也容易出幺蛾子。聚宽服务器时间默认是北京时间但QMT所在电脑可能时区设置不对如果两边时区不一致datetime.now()减去created_at得到的过期判断就会出错。更稳妥的做法是让聚宽侧生成时间时带时区信息或者统一用带时区的ISO8601字符串QMT侧比较时先转换到同一时区。5.3 几个值得长期坚持的实战习惯这个方案跑通之后我给自己定了几条纪律每一条都是在真金白银的教训里换来的。第一所有信号和委托必须留痕。聚宽推送时打印消息IDQMT收到时打印消息内容下单后打印order_id成交后打印成交记录。看似啰嗦但排查问题的时候这些日志就是唯一可靠的证据链。第二实盘前先用模拟盘或小资金跑通全流程。我一开始直接用100股测试虽然金额不大但至少验证了代码转换、下单路径、账单反馈都是通的。等链路完全稳定后再逐步放大仓位能省掉很多不必要的学费。第三QMT侧脚本要能自愈。我用的是常驻Python脚本配合Windows任务计划做了一个简单的看门狗检测到进程退出就自动拉起。Redis断线也在消费循环里加了重连逻辑不至于网络抖一下就整个停摆。第四尽量不要在信号里夹带策略逻辑。Redis队列里只放“买什么、卖什么、多少量、什么价格”所有需要判断的规则都放在聚宽侧研究好QMT侧只负责执行和基础风控。这样做的好处是两边职责单一任何一边出问题都好排查。最后分享一个我踩过的坑有段时间我发现QMT侧偶尔会漏单查了半天发现是brpop的超时时间设得太短Redis连接在空闲一段时间后被云端防火墙断开脚本接着消费时连接已经失效消息就一直积压在队列里。后来我在消费循环里加了连接检查和自动重连这个问题才算彻底解决。所以如果你也用这套架构记得把Redis断线重连当成一个正经功能来写而不是指望网络永远稳定。这套“聚宽信号 Redis桥 QMT执行”的架构我用了大半年整体稳定足够支撑日频甚至分钟频的策略。往后的扩展方向也很明确把List队列升级成Redis Stream做更严谨的ACKQMT侧接入更多账户做分仓策略端增加信号撤销和合并的逻辑。先把今天这套跑稳后面每一步都好说。