qmt-tcp-bridge

QMT 量化交易桥接库 —— 通过 TCP+NDJSON 协议连接 QMT 交易终端。

支持股票/期货行情查询、股票/期货交易（包括开平）、多账户类型。

你的 Python 脚本/策略
        ↕ TCP:18180 (股票) / 18880 (期货)
┌─────────────────────────┐
│  QMT 编辑器              │
│  (qmt_bridge/server/    │  ← pip install 后部署
│   loader.py 策略)        │
│        ↕ xtquant        │
│  QMT 柜台               │
└─────────────────────────┘

安装

pip install qmt-tcp-bridge

或从源码安装：

git clone git@gitee.com:m2land/qmt-tcp-bridge.git
cd qmt-tcp-bridge
pip install -e .

零外部依赖，仅需 Python >= 3.10（纯行情通道额外需要 redis 库：pip install qmt-tcp-bridge[redis]）。

使用步骤

第 1 步：部署服务端到 QMT

在运行 QMT 终端的机器上操作（如无网络，可先在有网的机器上导出）：

# 在有网的机器上安装并导出服务端文件
pip install qmt-tcp-bridge
python -m qmt_bridge deploy D:\qmt\server

然后把 D:\qmt\server 整个目录拷到 QMT 电脑上。服务端目录自包含（不依赖客户端库，QMT 内嵌 Python 3.6 可直接运行），qmt_bridge 库只需装在客户端一侧。

接着在 QMT 中配置策略：

1. 进入 QMT 的 模型研究 页面，在“我的策略”中点击 新建策略，分别创建股票和期货 Python 策略，例如“股票桥接”和“期货桥接”。

   

2. 点击策略的 编辑，将 qmt_bridge/server/loader.py 的全部代码粘贴到策略编辑器中，然后保存策略。

   

3. 找到 QMT 安装目录下的 python\formulaLayout\ 目录。将 qmt_bridge/server/loader.xml 复制到此目录，并把 XML 文件名改成对应的策略名称，例如 股票桥接.xml 和 期货桥接.xml。策略名称与 XML 文件名必须一致。

   

4. 进入 QMT 的 模型交易 页面，点击 新建策略交易。

   

5. 在”编辑策略交易”窗口中选择策略、账号类型和资金账号，并配置以下参数：

   • 服务端口：股票推荐使用 18180，期货推荐使用 18880
   • 服务端目录：填写 server.py 所在目录，例如 D:\qmt\server

   

   如果窗口中没有出现“服务端口”和“服务端目录”，先点击“加载参数”，如果仍未出现再检查第 3 步中的 XML 目录及文件名是否正确。

   

6. 保存配置并启动策略交易。股票和期货需要分别创建实例，并使用不同端口。启动后确认“策略状态”为 运行中，同时在 策略日志 中看到 server loaded、QMT Server 启动 和 QMT Server 就绪 等信息。

   

loader 支持热重载：修改 qmt_bridge/server/ 中的 Python 文件后，只需在 QMT 中停止并重新启动对应的策略交易实例即可生效。

第 2 步：编写客户端连接 QMT

股票交易

from qmt_bridge import QmtGateway, OrderAction, Trade, Order

gw = QmtGateway(host="192.168.1.xxx", port=18180, account="39987072")
gw.connect()

# 下单
coid = gw.send_order(OrderAction.BUY, "000001.SZ", 10.0, 100)
print(f"委托编号: {coid}")

# 查询
positions = gw.positions()
account = gw.account()
orders = gw.orders()
trades = gw.trades()

# 行情快照
snap = gw.snapshot(["000001.SZ", "600519.SH"])
for code, tick in snap.items():
    print(f"{code}: 最新价={tick.last_price} 时间={tick.timetag}")

# 实时回调（由服务端推送）
def on_trade(trade: Trade):
    print(f"成交: {trade.code} {trade.volume}@{trade.price}")

def on_order(order: Order):
    print(f"委托: {order.code} 状态={order.status}")

gw.on_trade(on_trade)
gw.on_order(on_order)

期货交易

期货需要连接到期货账户实例（默认 18880），account_type="FUTURE"：

from qmt_bridge import QmtGateway, OrderAction

gw = QmtGateway(host="192.168.1.xxx", port=18880, account="", account_type="FUTURE")
gw.connect()

# 开多
coid = gw.send_order(OrderAction.OPEN_LONG, "IF2609.IF", 4545.0, 1)
print(f"开多委托: {coid}")

# 平今多
gw.send_order(OrderAction.CLOSE_TODAY_LONG, "IF2609.IF", 4546.0, 1)

# 平昨多（今仓失败时的 fallback）
gw.send_order(OrderAction.CLOSE_HISTORY_LONG, "IF2609.IF", 4546.0, 1)

# 开空 / 平空
gw.send_order(OrderAction.OPEN_SHORT, "IF2609.IF", 4546.0, 1)
gw.send_order(OrderAction.CLOSE_TODAY_SHORT, "IF2609.IF", 4545.0, 1)

第 3 步（可选）：实时行情分发 —— 行情并入交易实例 + Relay + Redis

默认用法中 snapshot() 走 TCP RPC，3 秒一帧，适合轮询；需要**实时 tick 行情 （多客户端、低延迟）**时，按本步部署"交易实例 + Relay + Redis"三级通道 （详见 doc/market-data-fanout-design.md）。交易通道协议不变， Redis/Relay 故障只停行情、不影响交易。

为什么合并进交易实例？ QMT 所有策略实例共享同一个线程（设计文档 §4.1）， 拆实例不提供执行隔离，只带来独立启停与逻辑隔离。因此行情直接挂在交易实例上， 不需要单独的行情实例：

交易品种	实例数	布局
只做股票	1	股票交易 + 行情（18180），Relay 连接 18180
股票 + 期货	2	股票交易 + 行情（18180）；期货交易（18880）

行情取数用 get_full_tick，全市场通用、不绑定资金账号——期货代码的实时 行情也从股票实例推送，客户端订阅任意市场代码即可，期货实例不参与行情。 （仍想单独开行情实例的，可把 纯行情模式 设为 1，协议不变。）

QMT 终端（所有策略共享一个线程）
├─ 交易实例 :18180     ← 交易 + 行情合一（每 200ms get_full_tick → diff → 非阻塞推送）
└─ 交易实例 :18880     ← 期货交易（无订阅者时行情为空操作）
        │ TCP NDJSON（Relay 是行情消费者）
        ▼
Relay 进程（独立进程，系统 Python 运行）
        │ HSET + PUBLISH + SETEX 心跳
        ▼
Redis（本机 loopback，关持久化）
        │ SUBSCRIBE / HMGET
        ▼
你的 Python 脚本 × N

1. 启动 Redis（必须放本机、必须关持久化）

docker run -d --name qmt-redis -p 6379:6379 --restart unless-stopped `
  m.daocloud.io/docker.io/library/redis:7-alpine `
  redis-server --save "" --appendonly no

--save "" --appendonly no 是刻意的：Redis 只作行情缓存，关闭持久化可 消除 RDB fork 造成的尾延迟。必须与 QMT 同机（loopback）部署： 少一跳、无网络抖动、运维简单。

2. 在股票交易实例上开启行情（合并模式）

按第 1 步创建的"股票桥接"实例，在"编辑策略交易"中点击 加载参数 并配置：

• 服务端口：18180（不变）
• 纯行情模式：保持 0（交易服务，同时推送行情）
• 行情推送周期(ms)：200
• Redis 地址 / Redis 端口：本机 Redis，默认 127.0.0.1:6379

不需要新建实例。重启策略后，日志会看到 md timer注册成功 period=200ms 和 行情推送: 合并模式。

3. 启动 Relay 进程（QMT 机器上，独立于 QMT 终端）

python relay.py --port 18180

用系统 Python 运行（QMT 自带解释器无控制台，日志不可见）。relay.py 在部署目录 server/ 内，先 cd 到该目录再运行。

Relay 是行情实例的唯一消费者：把增量 tick 帧写入 Redis（Hash + Pub/Sub + 心跳），并按 md:subs:* 订阅声明把并集同步回实例。断线自动重连， 重连后实例强制全量重发。建议用 NSSM 或 Windows 计划任务守护。

已有独立行情实例（纯行情模式=1）的部署无需改动即可继续使用。迁移到 合并模式：把 Relay 改连交易端口（--port 18180）→ 在交易实例上加载参数并 重启 → 停掉行情实例即可。

4. 客户端用法（三种，§5.8.2）

纯行情（不创建 QmtGateway，不碰交易）：

from qmt_bridge import RedisMarketData

md = RedisMarketData("redis://127.0.0.1:6379/0")
md.on_tick(lambda t: print(t.code, t.last_price))          # 实时回调
md.on_status(lambda alive: print("行情", "正常" if alive else "陈旧"))
snap = md.subscribe(["000001.SZ", "600519.SH"])            # 立即返回快照
# md.unsubscribe([...]) / md.is_stale() / md.close()

双通道（交易走 TCP，行情走 Redis）：

from qmt_bridge import QmtGateway, RedisMarketData, OrderAction

md = RedisMarketData("redis://127.0.0.1:6379/0")
gw = QmtGateway(host="192.168.1.xxx", port=18180, account="39987072", md=md)
gw.connect()
snap = gw.subscribe(["000001.SZ"], lambda t: print(t.last_price))  # 返回快照
gw.send_order(OrderAction.BUY, "000001.SZ", 10.0, 100)             # 交易走 TCP

纯交易（不配 md）：行为与旧版完全一致，snapshot() 回退到 TCP RPC， subscribe()/unsubscribe() 抛 NotImplementedError 给出明确提示。 FakeGateway 提供对等实现（subscribe/unsubscribe/is_stale， set_tick() 触发订阅回调），可用于回测。

RedisMarketData 惰性导入：未安装 redis 也能 import qmt_bridge （交易功能不受影响），实例化时才会提示安装。

多客户端共享账号：用 remark 打标签，本地过滤

同一端口上的所有客户端共享一个资金账号，服务端广播全部委托/成交回报 （§5.9）。给 send_order(..., remark=...) 打前缀标签，在回调里按 client_order_id 前缀过滤即可：

PREFIX = "myclient-"

def on_order(order):
    if order.client_order_id.startswith(PREFIX):
        ...

单独使用 FakeGateway（测试/回测）

FakeGateway 与 QmtGateway 实现完全相同的接口，无需部署服务端。支持两种模式：

自动撮合模式（默认）： 下单立即全部成交。

from qmt_bridge import FakeGateway, OrderAction, OrderStatus

gw = FakeGateway()
gw.connect()
gw.send_order(OrderAction.BUY, "000001.SZ", 10.0, 100)
assert gw.orders()[0].status == OrderStatus.ALL_TRADED

异步撮合模式： 调用 set_tick() 设置盘口后，match_pending() 触发成交，按对手价撮合。

gw = FakeGateway(init_cash=1_000_000, t_plus_1=False)
gw.set_tick("000001.SZ", Tick(code="000001.SZ", ask_price=[10.0]*5, bid_price=[9.90]*5))
gw.send_order(OrderAction.BUY, "000001.SZ", 10.0, 100)
gw.match_pending(latency=0.0)

assert gw.orders()[0].status == OrderStatus.ALL_TRADED
assert gw.positions()[0].volume == 100

接口一览

QmtGateway 和 FakeGateway 共用以下接口：

协议	方法	说明
Broker	send_order()	下单（参数非法抛 ValueError，未连接抛 ConnectionError，服务端拒绝走 on_order 回报 REJECTED）
	cancel_order()	撤单
	positions()	查询持仓
	orders()	查询当日委托
	trades()	查询当日成交
	account()	查询资金账户
MarketData	snapshot()	行情快照（含 timetag 时间戳）
	subscribe()	订阅实时行情（QmtGateway 需配 md，见第 3 步；FakeGateway 直接可用）
	unsubscribe()	退订行情
	is_stale()	行情是否陈旧（有 md 时查 Redis 心跳）
Clock	now()	当前时间戳

委托方向（OrderAction）

枚举	适用	说明
BUY / SELL	股票	买入 / 卖出
OPEN_LONG / CLOSE_TODAY_LONG / CLOSE_HISTORY_LONG	期货	开多 / 平今多 / 平昨多
OPEN_SHORT / CLOSE_TODAY_SHORT / CLOSE_HISTORY_SHORT	期货	开空 / 平今空 / 平昨空

数据模型

模型	关键字段
Tick	code, last_price, bid_price[], ask_price[], volume, amount, last_close, settle_price, timetag
Order	code, action, price, volume, volume_traded, status, order_id, client_order_id, created_at
Trade	code, action, price, volume, amount, trade_time
Position	code, volume, can_use_volume, avg_price, float_profit
Account	balance, available, frozen_cash, market_value

QmtGateway 构造参数

QmtGateway(
    host="127.0.0.1",       # QMT 服务端 IP
    port=18180,             # 端口（股票 18180，期货 18880）
    account="",             # 账号（为空则服务端自动发现）
    account_type="STOCK",   # 账户类型：STOCK / FUTURE
    trade_addr=None,        # 或直接传 (host, port) 元组，等价于上面两项
    md=None,                # RedisMarketData 实例（双通道行情，见第 3 步）
)

开发

# 运行测试（不含连接真实 QMT 的集成测试）
pytest tests/ -v --ignore=tests/contract/test_qmt_gateway.py

# 运行全部测试（需要 QMT 服务端正在运行）
pytest tests/ -v

# 实盘环境变量：
#   QMT_TEST=1             运行需要真 QMT Server 的只读测试
#   QMT_PAPER=1            运行下单冒烟和开平测试（会发真实委托）
#   QMT_FUTURES_PORT=18880 期货端口（默认 18880）

测试层次

tests/
├── contract/base.py           ← 协议契约基类（ReadOnlyContractTests）
├── contract/test_fake_gateway.py   ← FakeGateway 跑全部契约 + 撮合验证
├── contract/test_qmt_gateway.py    ← QmtGateway 跑全部只读契约 + 行情深度校验 + 下单冒烟
├── stock/test_market.py        ← FakeGateway 股票行情单元测试
├── stock/test_trade.py         ← FakeGateway 股票交易单元测试
├── futures/test_market.py      ← FakeGateway 期货行情单元测试
├── futures/test_trade.py       ← FakeGateway 期货交易单元测试
├── test_models.py, test_events.py, test_actions.py
├── test_qmt_gateway_send.py    ← QmtGateway send_order 语义（Mock TCP）
├── test_md_publisher.py        ← 行情发布器（真 Redis，Docker）
├── test_md_server.py           ← MD_ONLY 行情实例（subscribe/unsubscribe/md_tick）
├── test_relay.py               ← Relay 进程（假 MD 实例 + 真 Redis）
└── test_redis_md.py            ← RedisMarketData 客户端 + QmtGateway 双通道

项目结构

qmt_bridge/
├── pyproject.toml
├── LICENSE
├── README.md
├── doc/                       ← 文档
│   ├── market-data-fanout-design.md ← 行情分发架构设计（含实测数据）
│   ├── readme01.png           ← 在模型研究中新建策略
│   ├── readme02.png           ← 在策略编辑器中粘贴 loader 代码
│   ├── readme03.png           ← 在模型交易中新建策略交易
│   ├── readme04.png           ← 配置策略、账号和服务端参数
│   ├── readme05.png           ← 将参数 XML 放入 formulaLayout
│   ├── readme06.png           ← 确认服务端口和服务端目录
│   └── readme07.png           ← 从策略日志确认服务端就绪
├── src/
│   └── qmt_bridge/            ← PyPI 安装的库
│       ├── __init__.py        ← 统一导出
│       ├── __main__.py        ← CLI（python -m qmt_bridge deploy）
│       ├── actions.py         ← OrderAction/OrderStatus 枚举
│       ├── models.py          ← Tick/Order/Position/Trade/Account
│       ├── interfaces.py      ← Broker/MarketData/Clock 协议
│       ├── events.py          ← EventBus
│       ├── fake.py            ← FakeGateway（模拟网关）
│       ├── gateway.py         ← QmtGateway（TCP 实盘网关）
│       ├── redis_md.py        ← RedisMarketData（纯行情客户端，[redis] extra）
│       ├── tcp_client.py      ← TCP+NDJSON 客户端
│       └── server/            ← 部署到 QMT（python -m qmt_bridge deploy 导出，自包含）
│           ├── loader.py      → 粘贴到 QMT 策略编辑器
│           ├── server.py      → TCP 服务端
│           ├── handlers.py    → 协议处理
│           ├── transport.py   → TCP+NDJSON 传输
│           ├── runtime.py     → xtquant 运行时封装
│           ├── actions.py     → 自包含副本（QMT 3.6 无法 import 客户端库）
│           ├── md_publisher.py ← 行情发布器（纯逻辑，Relay 进程内运行）
│           ├── md_server.py   ← QMT 行情胶水（MdState/md_tick）
│           ├── relay.py       ← Relay 进程（QMT python.exe 运行）
│           └── loader.xml     → QMT 策略参数模板
└── tests/                     ← 测试