# zmq_live

**Repository Path**: mkee/zmq_live

## Basic Information

- **Project Name**: zmq_live
- **Description**: No description available
- **Primary Language**: Python
- **License**: MulanPSL-2.0
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-10-29
- **Last Updated**: 2026-10-29

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# zmq_live —— 基于 ZMQ 的 A股 QMT 实盘交易框架

用 **ZeroMQ** 做进程解耦，面向 **单机多进程 · 多策略 · 多账户** 的生产化交易框架。
行情/交易走大 QMT RPC 桥（`bigqmt`），策略以**目标权重信号文件**接入，
与产出信号的项目（如 `E:\quant\etf_rotation`）**无代码依赖**。

> 当前形态：`mode=live`，supervisor 常驻 6 个组件 + QMT 看门狗，
> 部署仅占 **3 个 ZMQ 端口**（5555 / 5551 / 5557），`doctor --connect` 全绿。

> 文档导航：本文件（使用向：配置/运行/上线） ·
> [`docs/DEVELOPMENT.md`](docs/DEVELOPMENT.md)（开发向：架构/机制/扩展/调试） ·
> [`docs/LIVE_CHECKLIST.md`](docs/LIVE_CHECKLIST.md)（实盘上线清单） ·

## 设计原则

1. **策略与执行解耦**：策略侧只产出「目标权重 JSON 文件」，本框架负责读文件、
   算股数、先卖后买、下单与对账。策略项目可随时替换，不需要改本仓库代码。
2. **账户真相唯一**：持仓/资金/挂单由 `trade_gateway` 持有；策略只给目标，不碰账户。
3. **单 QMT 会话一个网关**：同一 `userdata_mini`/账户由单进程持有，避免会话冲突与重复下单。
4. **崩溃可恢复**：订单 WAL + 启动对账 + 撤残留挂单 + 单实例锁 + 信号幂等库。
5. **A股规则集中**：交易日历、T+1、整手、涨跌停、集合竞价、交易成本。
6. **本仓库不依赖 hou_stock**：`QmtBroker` 直接用 `xtquant`，风控为内置实现。

## 进程拓扑

```
                     supervisor（守护/重启/熔断/预检）
                              │
  外部策略（任意项目）── 写 JSON ──> data/signals/inbox
                              │                    │
                              │        signal_executor × N（每 [[signals]] 一个）
                              │                    │ RPC place_order
  md_gateway ──PUB──> [msg_broker] ──SUB──> risk_service / monitor
  (唯一 xtdata)        XSUB 5555 / XPUB 5551        │
                              │   行情 md.* 与事件 evt.*/sys.* 共用这一对端口，按前缀过滤
  trade_gateway ──RPC──> RPC 代理 5557 ──> md | trade:<账户> | risk | admin
  (每账户一个)
```

**部署只占 3 个端口**：`5555`(XSUB) + `5551`(XPUB) + `5557`(RPC 代理)。
`use_broker=false` 时是另一种形态（各通道直连 bind），仅供单进程直连调试使用。

| 组件 | 进程 | ZMQ 端口(默认) | 职责 |
|---|---|---|---|
| `msg_broker` | 1 | **bind** 5555 / 5551 / 5557 | pub·sub 转发 + RPC 代理（supervisor 强制开启） |
| `md_gateway` | 1/QMT实例 | connect 5555（发）+ 5557（RPC `md`） | 唯一 xtdata 会话，行情快照/查价 |
| `trade_gateway` | 1/账户 | connect 5555（发）+ 5557（RPC `trade:<账户>`） | 下单/撤单/对账/WAL |
| `risk_service` | 1 | connect 5555（发）/ 5551（收）+ 5557（RPC `risk`） | 组合风控、熔断广播 |
| `signal_executor` | 1/`[[signals]]` | 5557（RPC `trade`/`md`） | 读信号 → 算计划 → 09:25 先卖后买 → 写报告 |
| `strategy_host` | N（可选） | connect 5551/5555 + 5557 | 原生策略宿主（`engine="native"`，当前未启用） |
| `recorder` | 1（可选） | connect 5551 | 行情落 Parquet |
| `monitor` | 1 | connect 5557，Web 8080 | 实时看板 + 控制 |

## 目录结构

```
zmq_live/
├── configs/
│   └── config.toml            # 主配置（端点/账户/信号/风控），也是生成新配置的模板
├── zlive/
│   ├── common/                # config/log/schema/bus/codes/timeutil/wal/metrics/signal_schema
│   ├── compat/                # bigqmt（大QMT RPC 垫片）/ backend（后端分发）
│   ├── gateways/              # md_gateway / trade_gateway / brokers(QmtBroker,SimBroker) / risk_guard / recorder
│   ├── services/              # signal_executor / risk_service / strategy_host / monitor / supervisor / doctor / msg_broker / watchdog
│   └── strategy/              # base / rotation（native 示例，当前未启用）
├── data/
│   ├── signals/inbox|done|reports   # 信号投递、归档、执行报告
│   └── signal_state.duckdb          # 信号幂等库（signal_id 唯一）
├── docs/LIVE_CHECKLIST.md     # QMT 实盘联调清单
├── deploy/run_supervisor.py   # 计划任务入口（单实例守卫 → 拉起 supervisor）
└── .venv/                     # Python 3.11（与 xtquant 兼容）
```

## 快速开始

```powershell
# 1) 环境
.venv\Scripts\python.exe -m pip install -r requirements.txt

# 2) 查看解析后的配置（打印最终生效的 JSON，含所有默认值）
.venv\Scripts\python.exe -m zlive.common.config

# 3) 预检（--connect 会真实连接柜台，查询账户/持仓/行情）
.venv\Scripts\python.exe -m zlive.services.doctor --connect

# 4) 启动（live；启动前强制 doctor 预检，FAIL 则拒绝启动）
.venv\Scripts\python.exe -m zlive.services.supervisor --source xtdata --broker qmt --preflight-connect --with-monitor
```

配置查找顺序（`zlive/common/config.py::load_config`）：显式 `path` 参数 →
环境变量 `ZLIVE_CONFIG` → 默认 `configs/config.toml`。

## 信号接入（`zmq_live.signal.v1`）

外部策略只需按契约写一个 JSON 文件到 `[signal_ingest] dir`（默认
`data/signals/inbox`），其余全部由 `signal_executor` 完成。

```json
{
  "schema": "zmq_live.signal.v1",
  "signal_id": "etf_rotation-20261028-a1b2c3",
  "producer": "etf_rotation",
  "strategy": "etf_rotation",
  "account": "010400010251",
  "signal_date": "20261028",
  "execute_date": "20261029",
  "mode": "target_weight",
  "cash_buffer": 0.02,
  "liquidate_unlisted": true,
  "weights": {"511010.SH": 0.0, "159941.SZ": 0.4},
  "meta": {"note": "可选"},
  "created_at": "2026-10-28T15:06:12+08:00"
}
```

契约要点（实现唯一基准：`zlive/common/signal_schema.py`）：

- **语义是目标权重**：`weights[code]` = 目标市值 / 总资产，不是买卖指令。
  `sum(weights) <= 1.0`，单票 `0..1`；`0` 表示清仓。未列出的持仓在
  `liquidate_unlisted=true` 时清仓。
- **代码格式**：`^\d{6}\.(SH|SZ|BJ)$`（小写 `sh/sz`、裸码会被规范化；
  两个键规范化后相同时权重相加，不静默丢弃）。
- **日期**：`signal_date`（决策日）/`execute_date`（执行日）均为 `YYYYMMDD`；
  早于今天或错过的窗口会被拒绝并落报告。
- **原子投递**：先写 `.tmp` 再 `os.replace` 到 inbox；执行器只读非 `.tmp` 文件。
- **幂等**：同一 `signal_id` 只执行一次（`data/signal_state.duckdb`）。
  `signal_id` 建议 `策略-信号日-内容指纹前6位`。
- **失败整单拒绝**：任何字段不合法 → 不执行、写报告（`status=failed`）、移入 done。

执行节奏（`zlive/services/signal_executor.py`）：

| 时间 | 行为 |
|---|---|
| 09:20 | 读账户/持仓/参考价，生成计划（卖/买股数、限价） |
| 09:25 - 09:45 | 先卖：发卖单 → 等终态（`sell_wait_sec`，超时撤单，仍不终态则放弃买入） |
| 卖单终态后 | 用**卖后实际现金 × (1 - cash_buffer)** 重新封顶买入额度 → 发买单 |
| 完成 | 写 `data/signals/reports/<signal_id>.json`，信号移入 `done` |

- 权重 → 股数：`总资产(现金+市值) × 权重 ÷ 参考价`，按整手（ETF/股票 100，科创板 200）取整；
- 价格：限价 `参考价 × (1 ± slip_bps/1e4)`，并夹在涨跌停价内；
- 保护：参考价缺失、涨跌停/停牌（幅度 ≥ 98% 限幅）的标的跳过并记 `skipped`。

### 生产端示例：etf_rotation

`E:\quant\etf_rotation\signal_export\export_signal.py` 是零侵入导出插件
（读 `data\回测结果\<策略卡>\最新选股结果.csv`，写契约 JSON）：

```powershell
cd E:\quant\etf_rotation
python signal_export\export_signal.py --strategy-card "全球资产ETF三池扰动集成轮动" --dry-run
python signal_export\export_signal.py --strategy-card "全球资产ETF三池扰动集成轮动"
```

**时序**：**16:00** 的计划任务先把当日 K 线同步入库，`最新选股结果.csv` 随后才更新到
当日，因此导出必须**顺延到同步完成之后**（建议 16:10 起）：

| 时间 | 动作 |
|---|---|
| 16:00 | 计划任务同步当日 K 线（完成后选股结果才落到当日） |
| 16:10 起 | 本插件导出信号 → `data/signals/inbox/`，执行器 5s 内接管 |
| 次日 09:25 | 执行（时序见上一节） |

若同步未完成就导出，`execute_date` 会停在旧交易日：次日执行器判「信号已过期」直接拒绝
（不会误下单），但当日等于空转——**宁可晚跑，不可早跑**。契约细节见该目录下 `README.md`。

### 运维命令

```powershell
# 单信号演练（不下单、不写状态库、不移动文件）
.venv\Scripts\python.exe -m zlive.services.signal_executor --signal-id etf_rotation --once --signal data\signals\inbox\x.json --dry-run --date 20261029

# 查看执行报告 / 幂等状态
Get-Content data\signals\reports\<signal_id>.json
```

## 交易网关（`trade_gateway`）

```powershell
# 实盘柜台（原生 xtquant；需 QMT 客户端在线）
.venv\Scripts\python.exe -m zlive.gateways.trade_gateway --broker qmt

# 内存模拟柜台（联调用，不下真实单）
.venv\Scripts\python.exe -m zlive.gateways.trade_gateway --broker sim --initial-cash 1000000
```

**RPC**（经 RPC 代理 `tcp://127.0.0.1:5557`，service = `trade:<账户>`）：
`place_order` / `cancel_order` / `cancel_all` / `query_account` / `query_positions` /
`query_orders` / `snapshot` / `kill_switch` / `set_risk_block` / `health`。
下单事件发布到 `evt.<account>.order`、`evt.<account>.fill`。

预交易硬风控（本地拒绝，fail-closed）：整手、价格、涨跌停、资金、T+1 可卖；
`CapitalAllocator` 按策略预算防超买；订单 WAL（独立 DuckDB）+
启动撤残留挂单 + 账户级单实例锁。

## 风控（内置）

`risk_service` 自主轮询账户（`--poll-sec`，0 为被动更新），规则见 `[risk]`：

| 键 | 含义 |
|---|---|
| `max_drawdown` | 峰值回撤熔断（暂停新开仓） |
| `max_daily_loss` | 单日亏损比例熔断 |
| `max_consecutive_losses` / `max_consecutive_loss_streak_pct` | 连亏次数/累计幅度 |
| `max_position_concentration` / `max_sector_concentration` | 单票/单行业集中度（当前 `1.0` = 不限制，目标权重由策略卡决定） |
| `max_drawdown_reattempt_days` | 回撤熔断后的重试间隔 |

运行态持久化到 `<data_dir>/risk_state_<app.name>.json`，重启不清零。

## 行情链 / 录制 / 回放

```powershell
.venv\Scripts\python.exe -m zlive.gateways.md_gateway --source xtdata --codes 600519.SH --interval 0.5
.venv\Scripts\python.exe -m zlive.gateways.recorder --flush-interval 5
.venv\Scripts\python.exe -m zlive.services.replay --speed 50
```

`--source mock` 无需 QMT。若 `[bus] use_broker = true`，另起 `python -m zlive.services.msg_broker`。

## 监控与守护

```powershell
# Web 监控台 http://127.0.0.1:8080
.venv\Scripts\python.exe -m zlive.services.monitor

# 一键守护（自动起 broker + 全组件 + 重启；live 强制预检）
.venv\Scripts\python.exe -m zlive.services.supervisor --source xtdata --broker qmt --preflight-connect --with-monitor
```

- supervisor 的 admin RPC：`health` / `kill_switch` / `restart` / `stop`
  （service `admin`，端口 5557；`stop` 为优雅停止，会依次终止子进程）。
- **QMT 看门狗**（`[watchdog]`）双判据（进程 `XtItClient.exe` + RPC ping），
  连续 `fail_threshold` 次失败 → 全局 kill_switch；恢复后不自动解除，需人工确认。
- 心跳统一走事件总线 `sys.heartbeat`，supervisor 聚合到 `health.heartbeats` / `health.stale`。

### 开机自启（计划任务）

计划任务 `zmq_live_supervisor` 由 `deploy/run_supervisor.py` 拉起（**单实例守卫生效**：
先用 admin RPC 探测，已运行则直接退出 0，不会重复起一套进程）：

- 触发器：登录后 2 分钟（`AtLogOn`，当前用户、交互式会话） + 每日 `08:45`（错过则 `StartWhenAvailable` 补跑）；
- 动作：`.venv\Scripts\pythonw.exe deploy\run_supervisor.py`（无窗口；supervisor 输出落入
  `logs/supervisor.task.log`，任务状态 `Running` ⇔ supervisor 存活）；
- 设置：`MultipleInstances=IgnoreNew`、无运行时长上限、失败重试 3 次×1 分钟。

```powershell
# 查看 / 启停 / 手动触发
Get-ScheduledTaskInfo -TaskName zmq_live_supervisor     # State + LastTaskResult
Start-ScheduledTask -TaskName zmq_live_supervisor
# 删除
Unregister-ScheduledTask -TaskName zmq_live_supervisor -Confirm:$false
```

> 任务只负责拉起 supervisor，**不会自动登录 QMT 客户端**：live 预检（`--preflight-connect`）
> 失败会以 rc=1 退出并触发 3 次重试；若 QMT 未登录，请登录后手动
> `Start-ScheduledTask -TaskName zmq_live_supervisor`。

## 后端：bigqmt（唯一实盘通道）

`[app] backend = "bigqmt"`。券商侧 **miniQMT 已于 2026-10-23 被封禁**，本仓库不再
提供 miniqmt 直连路径，也不再作为可选项维护。

`xtquant-big-convert` 在 QMT 内常驻 RPC 服务，方法名与 xtquant 兼容；
`zlive.compat.bigqmt` 把 `bigqmt_signal_trader.xtquant_compat` 伪装成 `xtquant`，
`QmtBroker` 与 `XtDataSource` 无需感知差异。

```powershell
pip install xtquant-big-convert
.venv\Scripts\python.exe -m zlive.services.doctor --connect
```

详见 [`docs/LIVE_CHECKLIST.md`](docs/LIVE_CHECKLIST.md) 附录 A。

> **2026-10-23 QMT 沙箱收紧：传输必须是 `pipe`**
> QMT 自 10-23 起在内置 Python 策略中禁用本地文件读写与外部网络连接（含
> TCP/IP 端口与相关三方库），miniQMT 直连同时被封，bigqmt 桥成为唯一实盘
> 通道，其传输由 `zmq` 改为 `pipe`（Windows 命名管道，纯 ctypes/kernel32，
> 无 socket、无文件 IO）。三处必须一致：
> 1. `configs/config.toml` → `[bigqmt] transport = "pipe"`
> 2. `bigqmt_signal_trader_client_config.py` → `"transport": 'pipe'`（且 `host`
>    置空，否则客户端会按"显式 redis 配置"去拨 6379）
> 3. 服务端单文件 `D:\...\python\BIGQMT_DRYRUN_NO_REDIS_FLAT_ALL_IN_ONE.py`
>    顶部配置块 → `"transport": 'pipe'`（改后需在 QMT 策略编辑器重新加载）
>
> 附带效果：pipe 无推送通道，`md_gateway` 订阅 tick 时自动降级为
> `get_full_tick` 轮询出价（已测）；成交回报由客户端按 1s 轮询合成。⚠️
> `bigqmt-init` 不能生成 pipe 单文件（单文件部署强制 zmq），
> **重跑生成器会把服务端配置块改回 zmq**，生成后须手工改回 pipe 再重新加载。

### 生成 `BIGQMT_DRYRUN_NO_REDIS_FLAT_ALL_IN_ONE.py`

单文件服务端（QMT 策略编辑器加载的那份）由 `xtquant-big-convert` 自带的
`bigqmt-init` 生成（入口 `bigqmt_signal_trader.init_config:main`）。
0.3.x 的 wheel 已内置构建器（`bigqmt_signal_trader/_singlefile/*.txt`），
**不需要源码 checkout**；`--repo-root` 仅在使用源码树时才需要。

```powershell
cd E:\quant\zmq_live
.\.venv\Scripts\bigqmt-init.exe          # 加 --force 则直接覆盖、不再逐文件确认
```

交互答案（本机生产值）：

| # | 提示 | 输入 | 说明 |
|---|---|---|---|
| 1 | 资金账号 | `010400010251` | 必填 |
| 2 | 账号类型 | `1`（STOCK） | 回车 = 默认 |
| 3 | 传输方式 | `2`（zmq） | 下一步会被强制改 zmq；选 1 还会多问 Redis 地址/db/密码 |
| 4 | QMT 终端地址 | `127.0.0.1` | 端口自动推导 **15611**；跨机填 QMT 主机 IP |
| 5 | 允许远程下单/撤单 | `y` | 不开则 RPC 只能查询，信号执行器无法下单 |
| 6 | 部署方式 | `3`（单文件·明文代码，强制 zmq） | 见下方「必须改回 pipe」 |
| 7 | QMT 的 python 目录 | `D:\AppData\中航证券QMT实盘-交易端\python` | 服务端单文件落点 |
| 8 | 客户端配置目录 | `E:\quant\zmq_live` | 生成 `bigqmt_signal_trader_client_config.py` |

产出三个文件：

1. `<QMT python 目录>\BIGQMT_DRYRUN_NO_REDIS_FLAT_ALL_IN_ONE.py` —— **GBK** 单文件服务端
   （配置块内嵌，运行时不再读任何本地配置文件）；
2. `bigqmt_signal_trader_client_config.py` —— 客户端配置（含资金账号）；
3. `<QMT python 目录>\bigqmt_signal_trader_local_config.py` —— 标准包部署用，
   单文件模式下**不会被读取**，可忽略。

生成后必做（生成器把 `transport` 固定写成 `zmq`，而沙箱已禁 TCP）：

```python
# 服务端单文件顶部 BIGQMT_REDIS_CONFIG（GBK 文件，勿存成 UTF-8）
"transport": "pipe",
"host": "",          # 非 redis 传输下 host 非空会被判为“显式 redis 配置”，事件线程去拨 6379
# 客户端 bigqmt_signal_trader_client_config.py 同样改成 "pipe" + host 空
```

然后在 QMT 策略编辑器**重新加载**该策略，再 `doctor --connect` 验证：管道名
`\\.\pipe\bigqmt_rpc_010400010251`、`FAIL=0`。

> 只换账户/端口时不必重跑生成器：直接改单文件顶部 `BIGQMT_ACCOUNT_ID` 与
> `BIGQMT_REDIS_CONFIG` 更快，且不会把 transport 打回 zmq。生成物含资金账号，
> 勿提交版本库（当前生产件 1,446,727 字节 / 2026-10-26 13:54）。

## 消息协议

多帧：`[topic][header][payload]`，`msgspec` MessagePack。

- 主题：`md.tick.<code>`、`md.bar.<period>.<code>`、`evt.<account>.order|fill|position|account`、
  `evt.risk`、`sys.heartbeat`
- `OrderIntent.client_order_id` 为幂等键（信号执行器为 `sig:<signal_id>:<code>:<side>:<seq>`）；
  重启时从订单 WAL 重建幂等缓存（仅近 3 天已受理单），跨重启不重复下单。
- 订阅按 `header.seq` 检缺口 → RPC 拉快照 resync。

## 架构变更（2026-10-28）

- **策略接入改为文件契约**：`[signal_ingest]` + `[[signals]]` + `signal_executor`，
  生产者与消费者仅共享 `zmq_live.signal.v1` JSON schema。
- 上游策略项目（etf_rotation 等）以 `signal_export` 插件产出信号，改动不影响本仓库。
- 风控集中度阈值放开为 `1.0`（目标权重由策略决定），组合级风险由单日亏损/回撤/熔断兜底。

## 免责声明

仅供学习与研究，实盘风险自负。任何券商、交易所无关。