热门搜索
首页 » 行业资讯 » 突破券商封锁!手把手教你用“里应外合”架构实现QMT全自动交易

行业资讯

加入亿拓客·流量大师 撬动财富之门！！！

突破券商封锁!手把手教你用“里应外合”架构实现QMT全自动交易

wang 2026-03-22 行业资讯
突破券商封锁!手把手教你用“里应外合”架构实现QMT全自动交易
很多量化交易者可能都经历过这样的无奈：自己精心编写的Python监控系统明明探测到了绝佳的套利机会，却因为券商不开放MiniQMT等外部API权限，只能眼睁睁看着机会溜走，然后手忙脚乱地切换到交易软件手动下单。等你完成这一系列操作，市场早就变了天。
今天，我将分享一个在量化圈内非常经典且成熟的解决方案——“里应外合”Socket架构。这个方法不需要向券商申请任何特殊的高级权限，只要你的电脑能登录普通的QMT客户端，就能让你外部的独立Python交易系统，无缝对接并实时操控你的实盘账户。

缘起：一道无形的墙

事情的起因很简单。我搭建了一套外部的LOF基金套利监控系统，能够实时计算溢价并发出信号。但当信号出现时，我却卡在了最后一步——交易。我所在的银河证券，其标准版QMT（俗称“大QMT”）为了合规，禁止了外部Python程序的直接调用，所有策略代码必须在其内置的Python编辑器中编写和运行。
这意味着，我那套运行在VSCode或PyCharm里的、功能强大的监控系统，成了一位“场外观众”，看得见机会，却下不了单。虽然存在一个叫MiniQMT的极简版本支持外部调用，但并非所有券商或所有客户都会开放此权限。

核心思路：安插一个“内应”

既然券商不允许我们从“外部”攻进去，那我们就从“内部”想办法。QMT客户端内部自带一个Python策略运行环境，这是我们的突破口。
“里应外合”架构的核心思想可以概括为三步：
安插内应：在QMT内部编写并运行一个特殊的策略。这个策略本身不进行任何行情分析和交易决策，它唯一的任务就是在你电脑本地的某个端口（例如8888）上启动一个Socket服务器，像一个忠实的通信兵，静静地等待外部指令。
外部发令：我们原有的、运行在QMT外部的监控系统（可以是任何Python程序、网站后端甚至Excel）保持不变。一旦它计算出交易信号，就通过Socket网络通信，向本机的127.0.0.1:8888发送一条简单的文本指令，例如：“BUY,512000.SH,100,2.550”（含义：买入、代码、数量、价格）。
内部执行：QMT内部的“内应”策略收到这条文本指令后，立即在其内部调用QMT原生的passorder()下单函数。由于这个调用发生在QMT进程内部，完全合规，因此订单会被瞬间、无阻碍地送达交易所。
就这样，我们通过一个本地Socket隧道，巧妙地绕过了券商对外部调用的限制，实现了外部系统对实盘的间接控制。

关键踩坑与实战代码

思路很清晰，但实现过程中有几个“坑”必须避开，否则系统会极不稳定。
第一个坑：子线程操作GUI/C++核心崩溃。
如果在Socket监听线程中直接调用passorder()或查询持仓的函数，有很大概率会导致QMT底层C++引擎崩溃闪退。这是因为QMT的很多核心函数必须在主线程中调用。
解决方案：我们建立一个全局的任务队列order_queue。Socket线程只负责接收指令并将其放入队列。然后，我们利用QMT策略的ContextInfo.run_time()定时器函数，绑定一个在主线程中运行的函数（如check_orders），由这个函数从队列中取出指令并执行下单操作。
第二个坑：端口占用与策略重启问题。
当你停止QMT策略然后重新运行时，之前的Socket线程可能没有完全释放，导致端口8888被占用，新策略无法启动（报错10048）。
解决方案：实现一个“优雅自杀释放机制”。在每次策略初始化(init函数)时，先尝试连接本地的8888端口。如果连接成功，就向那个“幽灵”线程发送一条“SHUTDOWN”指令，让它安全关闭并释放端口，然后再启动新的监听线程。
下面是经过实战检验的核心代码，分为QMT内部策略和外部调用两部分。

📍 Part 1: QMT内部“内应”策略代码

将此代码复制粘贴到QMT策略编辑器中运行。

encoding: gbk

import socket
import threading
import time

全局订单队列和查询结果变量

order_queue = []
query_result = None
defsocket_server_thread():
"""Socket服务器线程，负责接收外部指令"""
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

设置端口复用

try:
server.bind(('127.0.0.1', 8888))
server.listen(5)
except Exception:
return

绑定失败则退出线程

whileTrue:
try:
conn, addr = server.accept()
data = conn.recv(1024).decode('utf-8')
if data:
if data == "SHUTDOWN":

收到自杀指令，优雅关闭服务器

conn.close()
server.close()
return
elif data == "QUERY_POS":

查询持仓请求

global query_result
query_result = None
order_queue.append(data)

将查询请求也放入主线程队列

等待主线程处理并填充query_result

for _ inrange(40):

等待最多4秒

if query_result isnotNone:
break
time.sleep(0.1)

将结果发送回外部调用者

conn.sendall((query_result or"TIMEOUT").encode('utf-8'))
else:

普通交易指令

order_queue.append(data)
conn.sendall("OK".encode('utf-8'))
conn.close()
except:
time.sleep(1)
definit(ContextInfo):
"""策略初始化函数"""

设置你的资金账号

ContextInfo.set_account('你的资金账号')

【关键】优雅自杀：启动前先尝试关闭可能存在的旧线程

try:
killer = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
killer.settimeout(0.5)
if killer.connect_ex(('127.0.0.1', 8888)) == 0:
killer.sendall("SHUTDOWN".encode('utf-8'))
killer.close()
time.sleep(1)

等待端口释放

except:
pass

无旧线程，正常继续

启动Socket监听线程

threading.Thread(target=socket_server_thread, daemon=True).start()

注册主线程定时任务，防止C++崩溃

ContextInfo.run_time("check_orders", "1nSecond", "2020-01-01 09:30:00")
defcheck_orders(ContextInfo):
"""主线程定时任务：处理订单队列"""
global order_queue, query_result
whilelen(order_queue) > 0:
cmd = order_queue.pop(0)
parts = cmd.split(',')
if parts[0] == 'QUERY_POS':

查询持仓

holdings = get_trade_detail_data('你的资金账号', 'STOCK', 'position')
if holdings:
query_result = " | ".join([f"[{p.m_strInstrumentID}] 可卖: {p.m_nCanUseVolume}"for p in holdings])
else:
query_result = "空仓"
elif parts[0] in ['BUY', 'SELL'] andlen(parts) >= 4:

处理买卖指令

opType = 23if parts[0] == 'BUY'else24

23买，24卖

调用QMT原生下单函数

passorder(opType, 1101, '你的资金账号', parts[1], 11,
float(parts[3]), int(parts[2]), 'SocketTrade', 1, "API", ContextInfo)

📍 Part 2: 外部Python系统的调用代码

你原有的监控系统在需要交易或查询时，使用如下函数与QMT通信。
import socket
defsend_to_qmt(cmd_str):
"""发送指令到QMT内部策略"""
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.settimeout(4.0)

设置超时

try:
client.connect(('127.0.0.1', 8888))
client.sendall(cmd_str.encode('utf-8'))
response = client.recv(4096).decode('utf-8')
print(f"收到QMT回执: {response}")
return response
except Exception as e:
print(f"指令发送失败: {e}")
returnNone
finally:
client.close()

使用示例

if __name__ == "__main__":

1. 查询实时持仓

pos_info = send_to_qmt("QUERY_POS")
if pos_info:
print(f"当前持仓: {pos_info}")

2. 发出买入指令 (格式: 操作,代码,数量,价格)

例如：买入512000，100股，价格2.55元

send_to_qmt("BUY,512000.SH,100,2.550")

3. 发出卖出指令

send_to_qmt("SELL,512000.SH,50,2.560")

效果与总结

运行上述代码后，你的外部系统和QMT就建立了一条高速、稳定的“地下通道”。
当你的监控系统发现机会时，会立即打印类似指令：
收到QMT回执: [512000.SH] 可卖: 1000
收到QMT回执: OK
同时，在QMT客户端的日志窗口中，你会看到实实在在的下单记录：
Order Sent: BUY 512000.SH 100 @ 2.550
通过这套“里应外合”的架构，我们成功实现了以下目标：
零权限自动化：无需向券商申请特殊的MiniQMT或外部API权限。
系统完全独立：你的核心监控、算法系统可以继续用你最熟悉的框架（Django、Flask、Pandas等）开发，与券商客户端彻底解耦。
稳定可靠：解决了多线程和端口占用的核心难题，适合长期运行。
灵活扩展：Socket指令格式可以自定义，轻松扩展出条件单、批量单、查询资金等各种功能。
这个方法本质上是利用合规工具（QMT内部Python环境）搭建了一个桥梁，实现了自动化交易的目的。它思路巧妙，代码简洁，堪称量化交易者在有限条件下的“神兵利器”。如果你也受困于券商API的限制，不妨试试这个方案，用最极客的方式，扣动量化交易的扳机。
以上活动仅是一个起点。Web3领域充满机会，但需深入学习和资源支持。如果你想系统了解区块链、空投策略和更多实战技巧，欢迎加入我们的社区：添加闲云微信：请扫描下方二维码（备注“进群”获取专属链接），解锁更多Web3干货。

猜你喜欢

发表评论

发表评论: