你肯定遇到过这种情况在量化交易里想用 Python 写一个复杂的策略逻辑但发现 QMT 内置的策略编辑器写起来不够顺手调试也麻烦。或者你有一个用 Python 开发好的成熟策略想直接接入 QMT 进行实盘交易却发现两者之间像隔着一堵墙数据进不去信号出不来。这时候一个常见的想法是能不能让外部的 Python 程序独立运行然后和 QMT 进行“对话”让 Python 负责复杂的计算和决策QMT 负责稳定的行情接收和订单执行。这个想法很自然但实现起来你会发现官方并没有提供一个现成的、高并发的 API 让你随意调用。于是很多人卡在了“如何通信”这一步。今天要聊的“基于文件通信的信号桥接方案”就是解决这个问题的。它听起来可能有点“土”——用文件来传递信息似乎不够“高科技”。但恰恰是这种简单直接的方式在很多场景下成为了最可靠、最可控的桥梁。它不依赖于复杂的网络协议或可能变化的接口核心就是让 Python 把信号“写”到一个文件里QMT 定时去“读”这个文件并执行反过来QMT 把执行结果或行情快照“写”到另一个文件Python 再去读取完成闭环。这个方案真正解决的不是“能不能通信”的问题而是“如何以一种低依赖、高稳定、易调试的方式实现外部复杂逻辑与内部交易执行引擎的可靠对接”。下面我们就从为什么需要它开始一步步拆解这个方案的实现逻辑、实操细节以及那些决定成败的工程化考量。1. 为什么是文件通信重新理解“可靠”与“解耦”在追求高性能、低延迟的量化领域提到“文件通信”很多人的第一反应可能是“太慢了”。确实相比内存共享、Socket 甚至 RPC文件的 I/O 操作有物理磁盘的延迟。但我们需要先厘清一个关键点策略信号生成的频率和交易系统对延迟的敏感度往往是两个不同量级的事情。对于大部分非高频策略例如分钟级、小时级甚至日级别的信号信号生成本身可能几秒甚至几分钟才发生一次。这时文件读写那几十毫秒的额外开销几乎可以忽略不计。真正的瓶颈和风险往往出现在系统的复杂性、稳定性和可维护性上。1.1 对比其他通信方式的隐形成本我们快速对比几种常见的交互方式内存共享 / 进程间通信 (IPC)效率最高但实现复杂严重依赖平台且 QMT 作为闭源商业软件通常不提供此类底层接口。强行注入风险极高。网络 Socket / HTTP API需要 QMT 暴露网络服务端口这涉及到安全策略、防火墙配置并且增加了网络抖动、连接中断等不确定性。对于部署在券商服务器或严格内网环境的 QMT开通端口可能很困难。数据库看似规范但引入了 MySQL、Redis 等中间件部署和维护成本陡增。对于简单的信号传递属于“杀鸡用牛刀”。消息队列 (如 RabbitMQ, Kafka)功能强大适用于大型分布式系统但架构过重学习和运维成本高不适合个人或小团队快速验证策略。文件通信方案的优势在这个时候就凸显出来了零外部依赖只需要操作系统和文件系统这是任何环境都具备的。跨平台兼容无论是在 Windows、Linux 还是 macOS 上Python 和 QMT如果支持都能读写文件。极强的可调试性信号内容以明文如 JSON、CSV形式保存在文件中任何时候都可以直接打开查看、修改或回放这对策略开发和问题排查是巨大的便利。自然的解耦Python 端和 QMT 端完全独立。Python 程序可以随时重启、升级甚至用其他语言重写只要遵守文件格式约定QMT 端无需任何改动。反之亦然。规避进程崩溃连锁反应一方崩溃不会直接导致另一方崩溃最多是读取不到新信号或旧信号被重复处理这可以通过设计来规避。1.2 文件通信的核心挑战与设计目标当然选择文件方案并不意味着把问题简单化。它把网络通信的复杂性转移到了文件读写的一致性、时序性和错误处理上。我们的设计必须围绕以下几点原子性如何确保 QMT 读取时Python 已经完整地写完了信号文件避免读到“半成品”。信号去重与防遗漏如何防止同一个信号被重复执行又如何确保没有信号被遗漏状态反馈QMT 执行订单后如何将状态成交、拒绝可靠地通知给 Python 端容错与监控当一方出现延迟或故障时另一方如何感知并采取安全策略如暂停交易理解了这些我们才能开始构建一个健壮的桥接方案而不是一个充满漏洞的临时脚本。2. 桥接方案核心架构双通道与状态机一个完整的双向交互至少需要两条独立的“通道”指令通道 (Python - QMT)用于传递交易信号。例如{“action”: “buy”, “symbol”: “000001.SZ”, “price”: 15.20, “volume”: 100}。反馈通道 (QMT - Python)用于传递执行状态、账户快照或行情切片。例如{“order_id”: “123456”, “status”: “filled”, “filled_price”: 15.19, “filled_volume”: 100}。最简单的实现是使用两个文件signal_to_qmt.json和feedback_from_qmt.json。但这远远不够我们需要引入状态机来管理文件的生命周期。2.1 基于文件状态的三段式协议一个经过实践检验的可靠模式是“三段式”文件协议准备文件 (.ready或.tmp)写入方先将要传递的数据写入一个临时文件如signal.tmp。这个文件对读取方不可见或通过命名规则忽略。就绪文件 (.json或.dat)当数据完整写入临时文件后将其重命名Rename为目标文件如signal.json。在大多数操作系统中重命名是原子操作。完成文件 (.done或.ack)读取方成功读取并处理完目标文件后将其重命名为完成文件如signal.json.done或直接删除。这标志着该信号已被消费。这个流程确保了读取方永远不会读到不完整的文件。同时通过检查是否存在“滞留”过久的.json文件即未被消费可以监控通信是否中断。2.2 具体文件结构与命名约定我们可以为每个信号或每个时间批次创建唯一的文件避免覆盖。例如指令通道 /signal_bridge/outgoing/20240520_103005_abc123.tmp - 写入中 /signal_bridge/outgoing/20240520_103005_abc123.json - 就绪等待QMT读取 /signal_bridge/outgoing/20240520_103005_abc123.json.done - QMT已处理 反馈通道 /signal_bridge/incoming/20240520_103010_def456.tmp - QMT写入中 /signal_bridge/incoming/20240520_103010_def456.json - 就绪等待Python读取 /signal_bridge/incoming/20240520_103010_def456.json.done - Python已处理文件名中的时间戳和唯一ID如UUID用于排序和去重。3. Python 端实现信号生成与文件写入Python 端是策略大脑它的职责是生成信号并按照协议可靠地写入指令通道。3.1 基础写入模块import json import os import time import uuid from pathlib import Path from typing import Dict, Any class SignalBridgeSender: def __init__(self, bridge_dir: str): self.outgoing_dir Path(bridge_dir) / outgoing self.incoming_dir Path(bridge_dir) / incoming # 确保目录存在 self.outgoing_dir.mkdir(parentsTrue, exist_okTrue) self.incoming_dir.mkdir(parentsTrue, exist_okTrue) def send_signal(self, signal_data: Dict[str, Any]) - str: 发送一个交易信号到QMT。 返回生成的文件名不含路径用于后续跟踪。 # 1. 生成唯一文件名 timestamp time.strftime(%Y%m%d_%H%M%S) unique_id str(uuid.uuid4())[:8] base_name f{timestamp}_{unique_id}.json temp_file self.outgoing_dir / f{base_name}.tmp ready_file self.outgoing_dir / base_name # 2. 写入临时文件 try: with open(temp_file, w, encodingutf-8) as f: json.dump(signal_data, f, ensure_asciiFalse, indent2) except IOError as e: raise Exception(fFailed to write signal temp file: {e}) # 3. 原子性重命名为就绪文件 try: os.rename(temp_file, ready_file) except OSError as e: # 如果重命名失败尝试清理临时文件 if temp_file.exists(): temp_file.unlink() raise Exception(fFailed to rename signal file: {e}) print(f[Sender] Signal written to: {ready_file}) return base_name def check_feedback(self) - list: 检查反馈通道读取所有未被处理的反馈文件。 处理完成后将其标记为.done。 feedbacks [] for fb_file in self.incoming_dir.glob(*.json): if fb_file.suffix .json and not fb_file.with_suffix(.json.done).exists(): try: with open(fb_file, r, encodingutf-8) as f: data json.load(f) feedbacks.append(data) # 标记为已处理 done_file fb_file.with_suffix(.json.done) os.rename(fb_file, done_file) except (IOError, json.JSONDecodeError) as e: print(f[Sender] Error reading feedback {fb_file}: {e}) # 可以选择跳过或报警 return feedbacks # 示例用法 if __name__ __main__: bridge SignalBridgeSender(./signal_bridge) # 模拟生成一个买入信号 buy_signal { strategy_id: ma_crossover_001, timestamp: time.time(), action: buy, symbol: 000001.SZ, price: 15.25, # 建议价格QMT可按此或市价下单 volume: 100, remark: 金叉信号触发 } try: file_sent bridge.send_signal(buy_signal) print(fSignal sent: {file_sent}) except Exception as e: print(fFailed to send signal: {e}) # 检查是否有来自QMT的反馈 fb_list bridge.check_feedback() for fb in fb_list: print(fReceived feedback: {fb})3.2 策略集成与信号管理在实际策略中你需要在主循环或事件驱动框架中集成这个发送器。信号生成在你的策略逻辑判断需要交易时调用send_signal。反馈处理定期例如每秒调用check_feedback根据反馈更新策略状态。例如收到“成交”反馈后可以更新本地持仓状态避免重复发单。信号去重可以在信号数据中加入一个signal_id并在 Python 端维护一个已发送未成交的列表结合反馈来清理。4. QMT 端实现信号读取与订单执行QMT 端使用 VBA或支持的其他脚本语言来监控指令通道读取信号并执行交易。这里是方案中最需要谨慎处理的部分因为涉及到真金白银的交易。4.1 QMT VBA 脚本核心逻辑假设我们在 QMT 中创建一个定时任务例如每秒执行一次来扫描信号目录。‘ 假设此VBA脚本在QMT中定时运行 Sub ReadSignalAndTrade() On Error GoTo ErrorHandler Dim bridgePath As String bridgePath C:\qmt_bridge\outgoing\ ‘ 修改为你的实际路径 Dim fso As Object, folder As Object, file As Object Dim signalFile As String, content As String, json As Object Dim action As String, symbol As String, price As Double, volume As Long Dim orderId As String Set fso CreateObject(Scripting.FileSystemObject) If Not fso.FolderExists(bridgePath) Then Exit Sub Set folder fso.GetFolder(bridgePath) ‘ 遍历.outgoing目录下所有.json文件排除.done文件 For Each file In folder.Files If LCase(fso.GetExtensionName(file.Name)) json Then signalFile file.Path ‘ 检查是否已被处理过存在同名的.done文件 If Not fso.FileExists(signalFile .done) Then ‘ 读取信号文件 Set ts fso.OpenTextFile(signalFile, 1) ‘ 1ForReading content ts.ReadAll ts.Close ‘ 解析JSON - 这里需要借助VBA的JSON解析库如ScriptControl或引用外部库 ‘ 以下为示例性伪代码实际解析需根据所用库调整 Set json ParseJson(content) ‘ 假设ParseJson是一个自定义的解析函数 action json(action) symbol json(symbol) price CDbl(json(price)) volume CLng(json(volume)) ‘ --- 核心交易逻辑 --- If action buy Then ‘ 调用QMT下单接口这里用伪代码表示 orderId PlaceOrder(symbol, “Buy”, price, volume) ElseIf action sell Then orderId PlaceOrder(symbol, “Sell”, price, volume) ElseIf action cancel Then ‘ 撤单逻辑 CancelOrder(json(“order_id”)) End If ‘ -------------------- ‘ 记录日志重要 LogMessage “SignalProcessed”, “File: ” file.Name “, Action: ” action “, Symbol: ” symbol ‘ 发送反馈到Python端写入incoming目录 SendFeedback bridgePath, “..\incoming\”, file.Name, orderId, “submitted” ‘ 标记信号文件为已处理 fso.MoveFile signalFile, signalFile “.done” End If End If Next file Exit Sub ErrorHandler: LogMessage “Error”, “ReadSignalAndTrade failed: ” Err.Description ‘ 错误时不要标记文件为.done以便下次重试 End Sub ‘ 发送反馈的辅助函数 Sub SendFeedback(bridgePath As String, incomingRelPath As String, originalFileName As String, orderId As String, status As String) Dim fso As Object, ts As Object Dim fbData As String, fbFilePath As String Dim timeStr As String Set fso CreateObject(“Scripting.FileSystemObject”) timeStr Format(Now, “yyyymmdd_hhnnss”) ‘ 构建反馈数据 fbData “{“ Chr(34) “order_id” Chr(34) “:” Chr(34) orderId Chr(34) _ “,” Chr(34) “status” Chr(34) “:” Chr(34) status Chr(34) _ “,” Chr(34) “original_signal” Chr(34) “:” Chr(34) originalFileName Chr(34) “}” fbFilePath fso.BuildPath(bridgePath, incomingRelPath) timeStr “_fb.json” ‘ 先写.tmp再重命名 Set ts fso.CreateTextFile(fbFilePath “.tmp”, True) ts.Write fbData ts.Close fso.MoveFile fbFilePath “.tmp”, fbFilePath End Sub重要提示上述 VBA 代码是高度简化的概念展示。实际应用中你需要实现或引入一个可靠的 JSON 解析函数如使用ScriptControl解析Eval或引用Microsoft Scripting Runtime等。替换PlaceOrder,CancelOrder为 QMT 实际提供的下单、撤单 API 函数。实现健壮的LogMessage函数将运行日志写入文件或 QMT 日志窗口。考虑 QMT 的定时任务机制确保脚本按预期频率稳定运行。4.2 QMT 脚本的健壮性设计错误处理必须用On Error GoTo包裹核心逻辑任何文件操作、JSON解析、API调用失败都不能导致脚本崩溃。错误应记录日志并尽可能保持文件状态不变以支持重试。信号幂等性确保即使脚本意外重启同一个.json信号文件也不会被重复执行。.done文件标记是关键。订单状态管理QMT 脚本在发单后应持续监控订单状态成交、部分成交、拒单等并通过反馈通道将最终状态发送给 Python 端。这可能需要另一个定时任务来查询订单列表。资源清理定期如每天开盘前清理过旧的.done文件避免磁盘空间被占满。5. 从“跑通”到“可靠”工程化与监控让信号在本地跑通一次演示和让它在生产环境稳定运行一周完全是两回事。以下是必须考虑的工程化问题。5.1 部署与目录结构规范建立一个清晰、独立的桥接目录与策略代码和 QMT 安装目录分离。/qmt_signal_bridge/ ├── config.ini # 配置文件包含路径、日志级别等 ├── sender.py # Python发送端主模块 ├── strategy_logic.py # 你的策略逻辑 ├── logs/ │ ├── python_sender.log # Python端日志 │ └── qmt_script.log # QMT端日志由VBA写入 ├── data/ │ ├── outgoing/ # Python - QMT │ │ ├── 20240520_103005_abc123.json │ │ └── 20240520_103005_abc123.json.done │ └── incoming/ # QMT - Python │ ├── 20240520_103010_def456.json │ └── 20240520_103010_def456.json.done └── archive/ # 按日期归档已处理的信号和反馈 └── 2024-05-20/5.2 监控与报警机制心跳检测Python 端可以定期如每10秒向outgoing目录写入一个特殊的心跳信号文件如heartbeat.json内容包含时间戳和进程ID。QMT 端脚本读取后在incoming目录回复一个心跳确认。双方都可以通过检查心跳是否超时来判断对方是否存活。文件积压报警监控outgoing目录下.json文件的数量和“年龄”。如果超过一定数量如10个或最老的文件存在时间过长如超过30秒说明 QMT 端可能已停止处理需要触发报警邮件、短信、桌面通知。日志分级记录 INFO、WARN、ERROR 级别的日志。INFO 记录正常信号发送/接收WARN 记录重试、短暂超时ERROR 记录解析失败、API调用失败、心跳丢失等。5.3 容错与恢复策略信号重发Python 端在发送信号后如果在超时时间内如60秒未收到对应的成交反馈可以根据策略决定是否重发。重发必须非常谨慎需要结合signal_id和订单状态判断避免重复下单。QMT 重启恢复QMT 重启后脚本应能自动处理之前未处理的.json文件即没有.done标记的文件。但需要判断信号的“新鲜度”对于过于陈旧的信号如前一天的可能应该直接标记为过期并丢弃而不是执行。文件锁替代方案在 Linux 等系统上可以考虑使用fcntl或portalocker实现真正的文件锁。但在 Windows 和跨平台场景下原子重命名方案通常更简单可靠。6. 适用边界与进阶思考这个方案不是银弹它有非常明确的适用场景。6.1 最适合的场景中低频策略信号生成间隔在秒级及以上。策略原型快速验证在投入大量精力封装原生 API 之前先用文件桥接验证策略逻辑的盈利能力。复杂策略逻辑外置策略涉及机器学习、复杂统计等重型计算需要在强大的 Python 生态中完成仅将交易指令交给 QMT。多平台信号聚合你的信号可能来自多个来源其他行情软件、网站、API用一个 Python 程序统一处理后通过一个通道发给 QMT。6.2 不适用或需要谨慎的场景高频交易 (HFT)文件 I/O 延迟无法满足微秒或毫秒级要求。对延迟极度敏感的套利策略同样的原因。需要极高频状态同步的策略例如需要实时知道每一笔挂单的排队情况。文件通信的轮询方式会有延迟。无人值守且对可靠性要求极高的生产环境虽然本方案可以做到很可靠但毕竟依赖文件系统和定时任务其稳定性低于专业的消息中间件。如果这是核心交易链路需要有更强大的基础设施和更严格的监控。6.3 可能的进阶方向如果这个基本方案运行良好但遇到了瓶颈可以考虑以下演进方向改用内存映射文件对于同一台机器上的进程使用mmap可以实现更快的共享内存式通信但仍需处理同步问题。引入轻量级消息队列例如 Redis 的 Pub/Sub 或者 ZeroMQ。这增加了外部依赖但获得了更好的实时性和并发能力。Python 和 QMT如果支持对应客户端可以分别作为发布者和订阅者。封装为标准服务将 Python 端封装成一个 REST API 服务QMT 通过 HTTP 客户端调用。这样通信更标准但引入了网络栈的复杂性。信号压缩与批量处理如果信号非常多可以对一个批次内的信号进行压缩后再写入单个文件减少 I/O 次数。文件桥接方案的价值在于它用最小的复杂度和依赖打通了一条从外部智能到内部执行的可靠路径。它让你能够将最前沿的 Python 数据分析、AI 模型与经过市场检验的交易执行环境QMT结合。在追求技术炫酷之前先让策略稳定地跑起来这个朴素的方案往往是最踏实的起点。当你通过它验证了策略的有效性并且对延迟和吞吐量有了更精确的要求后再去迭代通信层的基础设施方向会更加清晰。