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

零外部依赖，仅需 Python >= 3.10。

使用步骤

第 1 步：部署服务端到 QMT

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

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

然后把 D:\qmt\server 整个目录拷到 QMT 电脑上。在 QMT 电脑上也需要安装 qmt-bridge（如无网络，直接把 qmt_bridge 包目录拷到 Python 路径下即可）。

接着在 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)

单独使用 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 共用以下接口：

委托方向（OrderAction）

数据模型

QmtGateway 构造参数

QmtGateway(
    host="127.0.0.1",       # QMT 服务端 IP
    port=18180,             # 端口（股票 18180，期货 18880）
    account="",             # 账号（为空则服务端自动发现）
    account_type="STOCK",   # 账户类型：STOCK / FUTURE
)

开发

# 运行测试（不含连接真实 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）

项目结构

qmt_bridge/
├── pyproject.toml
├── LICENSE
├── README.md
├── doc/                       ← 文档截图
│   ├── 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 实盘网关）
│       ├── 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 运行时封装
│           └── loader.xml     → QMT 策略参数模板
└── tests/                     ← 测试