这类工具最值得先看的不是功能列表而是能不能在普通环境里稳定跑起来。大QMT的信号桥接方案核心解决的是量化交易场景下如何让外部Python脚本与QMT交易终端进行稳定、可靠的双向数据交换。很多人一上来就想搞复杂的策略交互但往往卡在第一步——通信链路不通。基于文件通信的方案听起来简单但真正要落地到生产环境需要处理的细节远比想象的多比如文件锁、读写时序、异常处理和性能瓶颈。我建议先从最小样例开始确认基础通信能跑通再考虑策略逻辑。下面按实际落地顺序拆一遍。1. 先搞清楚“信号桥接”到底要解决什么问题很多人看到“信号桥接”、“双向交互”会觉得很高深其实拆开看就两个核心需求信号怎么从Python发到QMT以及QMT的状态和结果怎么回传给Python。1.1 为什么不用直接的APIQMT本身提供了Python API但直接调用有几个现实问题。第一如果你的策略逻辑复杂或者需要引入一些QMT Python环境不支持的第三方库比如某些特定的机器学习库直接写进QMT的策略编辑器里会很麻烦。第二策略的研发、调试、版本管理如果完全在QMT内部进行流程上不够灵活。第三有些场景下你需要一个外部的“大脑”来协调多个账户、多个策略或者对接外部数据源这时就需要一个独立的Python进程。所以信号桥接的本质是解耦。把策略计算放在一个独立、可控的外部环境只把最终的交易指令信号通过一个可靠的通道传递给QMT执行。1.2 文件通信方案的优缺点基于文件通信是众多桥接方案里最“土”但往往最稳定的一种。它的优点很明显环境依赖极简几乎不需要额外的网络配置或中间件只要双方能读写同一个目录就行。跨平台兼容性好Windows、Linux服务器上都能用。规避网络风险没有端口、防火墙、连接超时等问题。调试直观信号就写在文件里出问题了直接打开看一目了然。但缺点同样突出这也是实操中必须处理好的地方性能有上限高频交易比如毫秒级不适合文件IO是瓶颈。但对于秒级、分钟级的策略完全够用。需要处理并发读写两个进程同时读写一个文件会出问题需要引入文件锁或采用“写完即移走”的模式。需要设计通信协议文件里写什么格式JSON、CSV还是自定义二进制如何区分不同类型的信号下单、撤单、查询需要处理异常和清理进程崩溃了残留的临时文件怎么办如何防止旧信号被重复执行理解了这些你就知道接下来的重点不是写Python代码而是设计一套健壮的文件交互规则。2. 环境准备与目录结构设计在写第一行代码之前先把地盘规划好。混乱的目录是后期排查问题的噩梦。2.1 基础环境确认QMT终端确保已安装并可以正常运行。知道其安装目录特别是用户数据目录通常包含bin,data,log等子目录。外部Python环境建议使用Anaconda或Miniconda创建一个独立的虚拟环境。这能避免与系统Python或其他项目的库冲突。conda create -n qmt_bridge python3.8 conda activate qmt_bridge共享目录这是通信的核心。必须确保QMT进程和外部Python进程都有读写权限。通常有两种选择使用QMT用户数据目录下的一个子目录如%QMT_PATH%/userdata_mini/bridge/。好处是路径固定QMT脚本内好引用。使用一个双方都能访问的独立目录如D:/qmt_bridge/或/home/user/qmt_bridge/。好处是干净不影响QMT自身文件。我个人的习惯是选择第二种创建一个独立的bridge目录结构如下qmt_bridge/ ├── config/ # 存放配置文件如标的列表、参数等 ├── signals/ # 外部Python - QMT 的信号存放目录 │ ├── pending/ # 待处理信号外部Python写入 │ ├── processing/ # 处理中信号QMT读取后移入 │ └── archive/ # 已处理信号处理完成后移入按日期归档 ├── feedback/ # QMT - 外部Python 的反馈存放目录 │ ├── pending/ # 待处理反馈QMT写入 │ ├── processing/ # 处理中反馈外部Python读取后移入 │ └── archive/ # 已处理反馈 ├── logs/ # 双方进程的日志文件 └── temp/ # 临时文件用于锁文件或缓存这个结构通过目录状态来管理信号生命周期避免了直接覆盖或删除文件带来的风险。2.2 关键权限与依赖权限在Windows上如果你以管理员身份运行过QMT而Python脚本是普通用户运行的可能会遇到权限问题。最好统一两者运行的用户身份。Python库外部Python环境只需要标准库json,time,os,shutil等最多加一个watchdog库用于监控目录变化非必须。不需要安装QMT的Python API包。pip install watchdog3. 通信协议与文件格式设计这是整个方案的核心设计得好后续扩展和维护会非常轻松。3.1 信号文件内容设计一个信号文件例如signal_20240520_102301_abc123.json应该包含足够的信息。我常用JSON格式因为它易读、易解析。{ signal_id: order_20240520_102301_abc123, timestamp: 1716178981.123456, type: place_order, account: 模拟账户, symbol: 000001.SZ, exchange: SZ, order_type: limit, side: buy, price: 14.25, volume: 100, strategy_name: ma_crossover_v1, extra: { remark: 金叉信号 } }关键字段说明signal_id: 全局唯一标识用于跟踪和去重。建议包含时间戳和随机字符串。timestamp: 信号生成时间Unix时间戳带毫秒。type: 信号类型。除了place_order下单还可能有cancel_order撤单、query_position查询持仓、heartbeat心跳等。account: QMT中的账户名确保信号发到正确的账户。symbol,exchange,order_type,side,price,volume: 标准的下单要素。strategy_name: 策略名称用于日志记录和绩效分析。extra: 扩展字段存放任何自定义信息。3.2 反馈文件内容设计QMT处理完信号后需要将结果写回。反馈文件也采用JSON格式。{ feedback_id: fb_20240520_102305_def456, signal_id: order_20240520_102301_abc123, timestamp: 1716178985.654321, type: order_response, status: success, data: { order_id: 1234567890, entrust_no: 987654321 }, error_msg: }或出错时{ feedback_id: fb_..., signal_id: order_..., timestamp: ..., type: order_response, status: failed, data: {}, error_msg: 账户资金不足 }3.3 文件命名与流转规则清晰的命名和流转规则是避免混乱的关键。信号生成外部Python:在signals/pending/目录下创建文件命名如{signal_type}_{timestamp}_{uuid}.json使用json.dump确保写入完整。写入完成后立即关闭文件句柄。这是很多新手忽略的点不关闭文件QMT端可能无法立即读取到完整内容。信号读取与处理QMT端Python脚本:定时扫描signals/pending/目录例如每1秒或3秒。使用os.listdir获取文件列表按文件名中的时间戳排序处理最早的文件。读取文件内容后立即将文件移动到signals/processing/目录。这个“移动”操作是原子性的在同一个文件系统内相当于加锁防止被重复处理。解析JSON调用QMT的place_order,cancel_order等API执行操作。根据API返回结果生成反馈文件写入feedback/pending/目录。最后将处理完的信号文件从processing/移动到archive/目录并按日期建立子目录归档。反馈读取外部Python:类似地外部Python进程定时扫描feedback/pending/目录。读取反馈更新策略状态例如知道订单是否成功。将处理完的反馈文件移入feedback/archive/。这套“pending - processing - archive”的流水线是保证至少一次at-least-once处理语义的关键能有效应对进程意外重启。4. 核心代码实现拆解理论讲完我们来看具体代码怎么写。我会分成外部Python端和QMT端两部分。4.1 外部Python端信号生成器这个脚本运行在独立的Python环境中负责策略计算和生成信号文件。import json import os import time import uuid from datetime import datetime from pathlib import Path class SignalSender: def __init__(self, bridge_root_dir): self.bridge_dir Path(bridge_root_dir) self.signal_pending_dir self.bridge_dir / signals / pending self.signal_pending_dir.mkdir(parentsTrue, exist_okTrue) def generate_signal_id(self): 生成唯一的信号ID return f{int(time.time()*1000)}_{uuid.uuid4().hex[:8]} def send_order_signal(self, account, symbol, exchange, order_type, side, price, volume, strategy_name, **extra): 发送下单信号 signal { signal_id: self.generate_signal_id(), timestamp: time.time(), type: place_order, account: account, symbol: symbol, exchange: exchange, order_type: order_type, side: side, price: price, volume: volume, strategy_name: strategy_name, extra: extra } filename forder_{signal[signal_id]}.json filepath self.signal_pending_dir / filename # 原子化写入先写入临时文件再重命名 temp_filepath filepath.with_suffix(.tmp) try: with open(temp_filepath, w, encodingutf-8) as f: json.dump(signal, f, ensure_asciiFalse, indent2) os.replace(temp_filepath, filepath) # 原子操作替换 print(f[Sender] 信号已发送: {filename}) return signal[signal_id] except Exception as e: print(f[Sender] 写入信号文件失败: {e}) if temp_filepath.exists(): temp_filepath.unlink() # 清理临时文件 return None def monitor_feedback(self, callback): 监控反馈目录回调处理反馈 feedback_pending_dir self.bridge_dir / feedback / pending feedback_pending_dir.mkdir(parentsTrue, exist_okTrue) processing_dir self.bridge_dir / feedback / processing processing_dir.mkdir(parentsTrue, exist_okTrue) while True: try: files sorted(feedback_pending_dir.glob(*.json)) for filepath in files: # 移动到处理中目录防止并发读取 processing_path processing_dir / filepath.name try: filepath.rename(processing_path) except FileExistsError: continue # 已被其他进程处理跳过 # 读取并处理反馈 with open(processing_path, r, encodingutf-8) as f: feedback json.load(f) callback(feedback) # 用户自定义的回调函数 # 归档 archive_dir self.bridge_dir / feedback / archive / datetime.now().strftime(%Y%m%d) archive_dir.mkdir(parentsTrue, exist_okTrue) processing_path.rename(archive_dir / processing_path.name) except Exception as e: print(f[Sender] 监控反馈时出错: {e}) time.sleep(1) # 扫描间隔 # 使用示例 if __name__ __main__: sender SignalSender(D:/qmt_bridge) # 模拟策略逻辑生成一个买入信号 signal_id sender.send_order_signal( account模拟账户, symbol000001.SZ, exchangeSZ, order_typelimit, sidebuy, price14.25, volume100, strategy_nametest_strategy, remark测试订单 ) # 定义处理反馈的回调函数 def handle_feedback(fb): print(f[Callback] 收到反馈: 信号ID{fb[signal_id]}, 状态{fb[status]}) if fb[status] success: print(f 订单号: {fb[data].get(order_id)}) else: print(f 错误信息: {fb[error_msg]}) # 启动反馈监控在实际应用中这个监控循环应该在独立线程中运行 print(开始监控反馈...) sender.monitor_feedback(handle_feedback)关键点generate_signal_id使用了时间戳和UUID确保唯一性。send_order_signal中采用了“先写临时文件(.tmp)再原子替换”的方式避免写入过程中文件被读取到不完整内容。monitor_feedback函数在一个循环中扫描反馈目录。处理文件时先将其移动到processing/目录这相当于一个简单的锁机制。回调函数callback让使用者可以自定义如何处理反馈比如更新数据库、触发风控等。4.2 QMT端信号处理器这个脚本需要放在QMT的bin/xxx/userdata_mini/下的某个目录并通过QMT的策略编辑器或定时任务来触发运行。# 在QMT的Python环境中运行 import json import os import shutil import time from datetime import datetime from pathlib import Path from typing import Dict, Any import warnings warnings.filterwarnings(ignore) # 忽略QMT环境的一些警告 # 导入QMT交易API try: from xtquant import xtdata from xtquant.xttype import StockAccount from xtquant import xttrader except ImportError as e: print(f无法导入QMT API请检查环境: {e}) exit(1) class SignalProcessor: def __init__(self, bridge_root_dir, account_id): self.bridge_dir Path(bridge_root_dir) self.signal_pending_dir self.bridge_dir / signals / pending self.signal_processing_dir self.bridge_dir / signals / processing self.signal_archive_dir self.bridge_dir / signals / archive self.feedback_pending_dir self.bridge_dir / feedback / pending # 初始化QMT交易接口 self.acc StockAccount(account_id) self.session_id -1 self.trader None self.connect_to_qmt() # 创建必要目录 for d in [self.signal_processing_dir, self.signal_archive_dir, self.feedback_pending_dir]: d.mkdir(parentsTrue, exist_okTrue) def connect_to_qmt(self): 连接QMT交易服务 try: self.trader xttrader.XtQuantTrader(./, self.session_id) self.trader.start() # 订阅账户这里需要根据实际情况调整 # self.trader.subscribe(self.acc) print(f[Processor] 已连接QMT交易服务账户: {self.acc.account_id}) except Exception as e: print(f[Processor] 连接QMT交易服务失败: {e}) self.trader None def send_feedback(self, signal_data: Dict[str, Any], status: str, result_data: Dict None, error_msg: str ): 发送处理反馈 if result_data is None: result_data {} feedback { feedback_id: ffb_{int(time.time()*1000)}, signal_id: signal_data.get(signal_id, unknown), timestamp: time.time(), type: f{signal_data.get(type)}_response, status: status, data: result_data, error_msg: error_msg } filename ffeedback_{feedback[feedback_id]}.json filepath self.feedback_pending_dir / filename temp_filepath filepath.with_suffix(.tmp) try: with open(temp_filepath, w, encodingutf-8) as f: json.dump(feedback, f, ensure_asciiFalse, indent2) os.replace(temp_filepath, filepath) print(f[Processor] 反馈已发送: {filename}) except Exception as e: print(f[Processor] 写入反馈文件失败: {e}) if temp_filepath.exists(): temp_filepath.unlink() def process_order_signal(self, signal_data: Dict[str, Any]): 处理下单信号 print(f[Processor] 处理下单信号: {signal_data[signal_id]}) if not self.trader: self.send_feedback(signal_data, failed, {}, QMT交易服务未连接) return try: # 提取订单参数 order_type signal_data[order_type] # limit 或 market side 23 if signal_data[side] buy else 24 # QMT中23为买入24为卖出 price signal_data[price] volume signal_data[volume] stock_code f{signal_data[symbol]}.{signal_data[exchange]} # 调用QMT下单API (此处为示例实际API调用可能不同请参考最新QMT文档) # 注意以下代码为示意需要根据实际QMT API调整 order_id self.trader.order_stock(self.acc, stock_code, side, volume, order_type, price) # 假设order_id为返回的订单编号 result_data {order_id: order_id} self.send_feedback(signal_data, success, result_data) print(f[Processor] 订单已提交订单号: {order_id}) except Exception as e: error_msg str(e) print(f[Processor] 下单失败: {error_msg}) self.send_feedback(signal_data, failed, {}, error_msg) def process_signal_file(self, filepath: Path): 处理单个信号文件 # 1. 移动到处理中目录 processing_path self.signal_processing_dir / filepath.name try: filepath.rename(processing_path) except FileExistsError: print(f[Processor] 文件已被其他进程处理: {filepath.name}) return except Exception as e: print(f[Processor] 移动文件失败 {filepath.name}: {e}) return # 2. 读取并解析信号 signal_data None try: with open(processing_path, r, encodingutf-8) as f: signal_data json.load(f) print(f[Processor] 读取信号: {signal_data.get(signal_id)}) except json.JSONDecodeError as e: print(f[Processor] 解析JSON失败 {processing_path.name}: {e}) self.send_feedback({signal_id: unknown}, failed, {}, f无效的JSON格式: {e}) # 将损坏文件移到归档目录避免阻塞 self._archive_file(processing_path) return except Exception as e: print(f[Processor] 读取文件失败 {processing_path.name}: {e}) return # 3. 根据信号类型分发处理 try: signal_type signal_data.get(type) if signal_type place_order: self.process_order_signal(signal_data) elif signal_type cancel_order: # 实现撤单逻辑 pass elif signal_type heartbeat: # 心跳信号简单回复即可 self.send_feedback(signal_data, success, {msg: alive}) else: self.send_feedback(signal_data, failed, {}, f未知的信号类型: {signal_type}) except Exception as e: print(f[Processor] 处理信号过程中出错: {e}) self.send_feedback(signal_data, failed, {}, f处理异常: {e}) # 4. 归档已处理文件 self._archive_file(processing_path) def _archive_file(self, filepath: Path): 归档文件 today_str datetime.now().strftime(%Y%m%d) archive_dir self.signal_archive_dir / today_str archive_dir.mkdir(parentsTrue, exist_okTrue) try: filepath.rename(archive_dir / filepath.name) except Exception as e: print(f[Processor] 归档文件失败 {filepath.name}: {e}) def run(self, interval_seconds3): 主循环定时扫描信号目录 print(f[Processor] 信号处理器启动扫描间隔: {interval_seconds}秒) while True: try: # 获取待处理信号文件按修改时间排序 pending_files list(self.signal_pending_dir.glob(*.json)) pending_files.sort(keyos.path.getmtime) for filepath in pending_files: self.process_signal_file(filepath) time.sleep(interval_seconds) except KeyboardInterrupt: print([Processor] 用户中断退出。) break except Exception as e: print(f[Processor] 主循环发生未知错误: {e}) time.sleep(interval_seconds) # 避免错误导致疯狂循环 # 在QMT中运行的主函数 if __name__ __main__: # 配置参数 BRIDGE_ROOT D:/qmt_bridge # 必须与发送端配置一致 ACCOUNT_ID 你的模拟或实盘账户ID # 替换为你的账户 processor SignalProcessor(BRIDGE_ROOT, ACCOUNT_ID) processor.run(interval_seconds2) # 每2秒扫描一次关键点SignalProcessor类封装了连接QMT、处理信号、发送反馈的完整逻辑。process_signal_file是核心方法遵循“移动-读取-处理-归档”的流程确保每个信号只被处理一次。send_feedback方法将处理结果写回反馈目录格式与发送端约定一致。run方法是一个无限循环定时扫描pending目录。在实际部署中你可能需要将这个脚本设置为QMT的定时任务比如每分钟运行一次或者作为一个常驻脚本在QMT中启动。重要提示示例中的下单API调用self.trader.order_stock是示意代码务必根据你使用的QMT版本和官方文档调整正确的API调用方式。API的导入、参数顺序、返回值可能不同。5. 部署、运行与监控代码写好了怎么让它真正跑起来5.1 部署步骤创建共享目录在磁盘上创建好D:/qmt_bridge及其所有子目录。部署外部Python脚本将SignalSender类集成到你的策略代码中。策略计算部分独立运行在需要交易时调用send_order_signal方法。部署QMT处理脚本将SignalProcessor类的代码保存为一个.py文件例如qmt_bridge_processor.py放入QMT的脚本目录例如userdata_mini/script/。配置QMT定时任务在QMT客户端找到“量化交易”或“策略交易”相关界面。创建一个新的定时任务设定运行周期例如每分钟运行一次。在任务中调用execfile(‘你的脚本路径/qmt_bridge_processor.py’)或者将主逻辑封装成函数后调用。更优的做法将处理器脚本改写为一个循环次数有限的脚本例如处理完当前所有信号后退出然后让QMT的定时任务高频调用如每10秒一次这比在QMT内运行一个无限循环的脚本更稳定因为QMT可能会清理长时间运行的脚本。5.2 运行验证流程不要一上来就跑实盘。按这个顺序验证第一步测试文件读写手动在signals/pending/目录下放一个写好的JSON信号文件。手动运行QMT的处理脚本观察文件是否被正确移动到processing/然后到archive/同时feedback/pending/里是否生成了反馈文件。手动检查反馈文件内容是否正确。第二步测试外部Python发送运行你的外部策略脚本先注释掉真正的策略逻辑只调用一次send_order_signal发送一个测试订单。观察信号文件是否生成QMT端是否处理并返回反馈。外部Python端是否能监听到反馈。第三步模拟下单使用模拟账户在QMT处理脚本中将账户改为模拟账户。发送一个模拟订单价格可以设得偏离市价很多避免成交。在QMT的委托记录中查看是否出现了这笔订单。查看反馈文件中的order_id是否与QMT中一致。第四步加入简单策略逻辑在外部Python端实现一个简单的策略比如每分钟检查一次价格满足条件发信号。观察整个链路是否能持续、稳定地运行一段时间比如半小时。5.3 监控与日志生产环境必须要有监控。日志代码中所有的print语句最好替换为标准的logging模块输出到logs/目录下的文件按日期滚动。要记录信号ID、处理时间、成功/失败状态。心跳机制外部Python可以定期比如每30秒发送一个type为heartbeat的信号。QMT处理器收到后回复一个成功反馈。如果外部Python长时间收不到心跳回复可以报警。目录健康检查可以写一个简单的监控脚本定期检查pending/目录下的文件数量是否积压或者文件修改时间是否过于陈旧这可能是处理器挂掉的迹象。反馈超时外部Python发送信号后可以启动一个计时器如果在预期时间内比如30秒没有收到对应signal_id的反馈则记录警告并可以考虑重新发送需要设计幂等性。6. 性能调优、边界条件与常见问题6.1 性能考量扫描间隔interval_seconds不宜设置过小如0.1秒会给磁盘带来不必要的压力。对于分钟级策略2-5秒的间隔足够。秒级策略可以考虑1秒。文件数量如果信号量非常大os.listdir或Path.glob可能会成为瓶颈。可以考虑使用watchdog库进行事件驱动监听而不是轮询。归档清理archive/目录下的文件会越来越多需要定期清理例如保留最近7天的数据。可以写一个简单的清理脚本加入系统定时任务。6.2 边界条件与容错信号重复发送网络抖动或外部Python超时未收到反馈可能导致同一信号发送多次。QMT处理器端需要根据signal_id做去重检查可以在内存中维护一个已处理ID的集合并定期持久化。处理器崩溃重启如果QMT处理器在移动文件到processing/后崩溃这个文件会一直留在processing/目录。解决方案处理器启动时先扫描processing/目录将这些“僵尸”文件重新处理或移到一个error/目录人工检查。磁盘空间满这是致命错误。代码中应有try-except捕获OSError并记录严重日志甚至停止发送新信号。QMT账户未登录或断开在process_order_signal中下单前应检查self.trader的连接状态。断开时应尝试重连并发送“失败”反馈而不是静默丢弃信号。6.3 常见问题排查清单当通信失败时按这个顺序查路径权限双方进程是否有权读写共享目录的所有子目录在Windows上特别注意用户账户控制UAC和杀毒软件可能拦截。文件锁是否在写入文件后没有立即关闭句柄用with open() as f可以确保。JSON格式错误手动打开一个pending/下的文件用在线JSON校验工具检查格式是否正确。QMT脚本未运行检查QMT的定时任务是否启用日志是否有报错。可以在脚本开头加一句print(‘QMT处理器启动’)并查看QMT的输出窗口。账户错误信号中的account字段是否与QMT中登录的账户名完全一致大小写、空格都要注意。API调用错误这是最复杂的一环。确保你使用的xttrader等API对象和方法与你的QMT版本匹配。仔细阅读官方文档或示例代码。我个人更建议先把单任务跑稳再考虑批量和接口。这个基于文件的桥接方案真正落地时最该盯住的不是功能列表而是输入格式、目录流转状态和异常处理。它可能不是性能最高的但对于需要将复杂策略逻辑与交易执行解耦的场景提供了一个极其简单、稳定、可视化的解决方案。一旦这套机制跑通你可以在此基础上扩展出更复杂的路由、风控和监控模块。
