From 992f53d4db99e8ce2180c31a8b2319ee96ef91d6 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Wed, 15 Jul 2026 18:57:20 +0800 Subject: [PATCH] =?UTF-8?q?feat(trader):=20Phase2=E2=80=94vnpy=5Fqmt=20?= =?UTF-8?q?=E8=BF=9B=E7=A8=8B=E5=86=85=E6=89=A7=E8=A1=8C=E5=AE=A2=E6=88=B7?= =?UTF-8?q?=E7=AB=AF,=E6=9B=BF=20HTTP=20bridge?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - qmt_gateway_client.py: QmtGatewayClient(同 BridgeClient 接口 place_order/ get_account/get_positions/get_orders/cancel_order),底层进程内 QmtGateway 直连 miniQMT(单例 _QmtExec: EventEngine+QmtGateway+账本缓存,懒连接复用)。 实证:连真实 miniQMT 读账户(9997081)+持仓(600000/000001),place_order 落单 QMT.xxxxx#1。 - live_orchestrator.py: _make_exec_client 工厂,SANGUO_USE_QMT_GATEWAY=1 → QmtGatewayClient (bridge 已退),否则 HTTP bridge 兼容。shadow + reconcile 两处替换。 同机部署后 HTTP 跳无必要,brain→vnpy_qmt→xtquant→miniQMT 零跳直链。 bridge+sanguo-caddy schtasks 可 disable(vnpy_qmt 路径已实证)。 --- sanguo_trader/live_orchestrator.py | 50 ++++---- sanguo_trader/qmt_gateway_client.py | 173 ++++++++++++++++++++++++++++ 2 files changed, 201 insertions(+), 22 deletions(-) create mode 100644 sanguo_trader/qmt_gateway_client.py diff --git a/sanguo_trader/live_orchestrator.py b/sanguo_trader/live_orchestrator.py index a720f32..241de33 100644 --- a/sanguo_trader/live_orchestrator.py +++ b/sanguo_trader/live_orchestrator.py @@ -32,6 +32,26 @@ from .strategy_runner import StrategyRunner logger = logging.getLogger(__name__) +def _make_exec_client(cfg): + """构造执行客户端（Phase 2）。 + + SANGUO_USE_QMT_GATEWAY=1 → 进程内 vnpy_qmt 直连 miniQMT（bridge 已退）； + 否则 → HTTP bridge（需 bridge_url + token，缺则 warning 返回 None）。 + 返回 None 表示跳过本次影子/reconcile（不阻断 live_step）。 + """ + if os.environ.get("SANGUO_USE_QMT_GATEWAY", "").strip().lower() in ("1", "true", "yes", "on"): + from .qmt_gateway_client import QmtGatewayClient + return QmtGatewayClient() + live_cfg = getattr(cfg, "live", None) or {} + token = live_cfg.get("bridge_token") or os.environ.get("BRIDGE_TOKEN") + url = live_cfg.get("bridge_url") + if not token or not url: + logger.warning("影子下单：bridge 模式但 bridge_url/token 未配（或设 SANGUO_USE_QMT_GATEWAY=1 走 vnpy_qmt），跳过") + return None + from .bridge_client import BridgeClient + return BridgeClient(url, token) + + def _get_account(db_path: str, account_id: int) -> dict: with sqlite3.connect(db_path) as conn: conn.row_factory = sqlite3.Row @@ -187,18 +207,11 @@ def _shadow_trades_to_bridge(db_path: str, account_id: int, today: str, cfg) -> live_cfg = getattr(cfg, "live", None) or {} if not live_cfg.get("enabled"): return - token = live_cfg.get("bridge_token") or os.environ.get("BRIDGE_TOKEN") - if not token: - logger.warning("live_step %s: 影子下单启用但 bridge_token（config 或 BRIDGE_TOKEN env）未设，跳过", account_id) - return - url = live_cfg.get("bridge_url") - if not url: - logger.warning("live_step %s: 影子下单启用但 bridge_url 未配，跳过", account_id) - return + from .bridge_client import to_bridge_code - from .bridge_client import BridgeClient, to_bridge_code - - client = BridgeClient(url, token) + client = _make_exec_client(cfg) + if client is None: + return with sqlite3.connect(db_path) as conn: conn.row_factory = sqlite3.Row cur = conn.execute( @@ -248,18 +261,11 @@ def reconcile_from_bridge(db_path: str, account_id: int, today: str, return if not live_cfg.get("mode_b"): return - token = live_cfg.get("bridge_token") or os.environ.get("BRIDGE_TOKEN") - if not token: - logger.warning("reconcile %s: mode_b 启用但 bridge_token（config 或 BRIDGE_TOKEN env）未设，跳过", account_id) - return - url = live_cfg.get("bridge_url") - if not url: - logger.warning("reconcile %s: mode_b 启用但 bridge_url 未配，跳过", account_id) - return + from .bridge_client import from_bridge_code - from .bridge_client import BridgeClient, from_bridge_code - - client = BridgeClient(url, token) + client = _make_exec_client(cfg) + if client is None: + return # 1. 校正资金（bridge /account 为准） acc_resp = client.get_account() diff --git a/sanguo_trader/qmt_gateway_client.py b/sanguo_trader/qmt_gateway_client.py new file mode 100644 index 0000000..761ec37 --- /dev/null +++ b/sanguo_trader/qmt_gateway_client.py @@ -0,0 +1,173 @@ +"""vnpy_qmt 进程内执行客户端（Phase 2：替掉 HTTP bridge）。 + +与 ``bridge_client.BridgeClient`` 同接口（place_order/get_account/get_positions/ +get_orders/cancel_order），但底层是**进程内 QmtGateway 直连 miniQMT**（同机同会话， +xtquant 进程内调用），不经 HTTP bridge。 + +设计： +- ``_QmtExec`` 单例持有长生命周期的 EventEngine + QmtGateway，懒连接（首次用即连）， + 跨 live_step 调用复用。gateway 的定时器周期性 query account/position/order 并经事件回调 + 推入缓存，故 get_account/get_positions 读缓存即同步返回。 +- ``QmtGatewayClient`` 包装单例 + 缓存，方法签名/返回与 BridgeClient 对齐，live_orchestrator + 可零改调用方代码替换。 + +配置（env）：SANGUO_QMT_ACCOUNT（交易账号，默认 66639661）、SANGUO_QMT_PATH（userdata_mini +路径，不设则扫 C:\\ 自动发现，避免中文路径字面量）。 + +依赖：vnpy 4.4.0 源码（sys.path）+ vnpy_qmt 0.3.3（pip）+ miniQMT 同机运行。 +""" +from __future__ import annotations + +import logging +import os +import threading +from typing import Any + +logger = logging.getLogger(__name__) + +_singleton: _QmtExec | None = None +_lock = threading.Lock() + + +def _parse_code(code: str) -> tuple[str, Any]: + """bridge code（sh600000/sz000001）→ (symbol, Exchange)。兼容纯数字码。""" + from vnpy.trader.constant import Exchange + c = code.lower() + if c.startswith("sh"): + return code[2:], Exchange.SSE + if c.startswith("sz"): + return code[2:], Exchange.SZSE + # 纯数字码兜底（与 bridge_client.to_bridge_code 一致） + if code.startswith(("60", "68", "51", "56", "58")): + return code, Exchange.SSE + return code, Exchange.SZSE + + +def _to_bridge_code(symbol: str, exchange: Any) -> str: + """(symbol, Exchange) → bridge code（sh600000），get_positions 输出对齐 bridge。""" + from vnpy.trader.constant import Exchange + return f"sh{symbol}" if exchange == Exchange.SSE else f"sz{symbol}" + + +class _QmtExec: + """长生命周期 vnpy_qmt 执行器单例：EventEngine + QmtGateway + 行情/账本缓存。""" + + def __init__(self) -> None: + from vnpy.event import EventEngine + from vnpy.trader.event import EVENT_ACCOUNT, EVENT_POSITION, EVENT_ORDER + from vnpy_qmt import QmtGateway + + self.account: Any = None # 最新 AccountData + self.positions: dict[str, Any] = {} # vt_symbol -> PositionData + self.orders: dict[str, Any] = {} # vt_orderid -> OrderData + + self.ee = EventEngine() + self.ee.register(EVENT_ACCOUNT, self._on_account) + self.ee.register(EVENT_POSITION, self._on_position) + self.ee.register(EVENT_ORDER, self._on_order) + self.ee.start() + + self.gateway = QmtGateway(self.ee) + self.gateway.connect(self._setting()) + logger.info("QmtGatewayClient 连接 miniQMT: %s", self._setting().get("交易账号")) + + @staticmethod + def _setting() -> dict[str, str]: + account = os.environ.get("SANGUO_QMT_ACCOUNT", "66639661") + path = os.environ.get("SANGUO_QMT_PATH") or _find_userdata_mini() + return {"交易账号": account, "mini路径": path} + + def _on_account(self, event: Any) -> None: + self.account = event.data + + def _on_position(self, event: Any) -> None: + p = event.data + self.positions[p.vt_symbol] = p + + def _on_order(self, event: Any) -> None: + o = event.data + self.orders[o.vt_orderid] = o + + +def _find_userdata_mini() -> str: + """扫 C:\\ 找 userdata_mini 目录（避免中文路径字面量编码问题）。""" + try: + for name in os.listdir("C:\\"): + cand = os.path.join("C:\\", name, "userdata_mini") + if os.path.isdir(cand): + return cand + except OSError: + pass + return "" + + +def get_qmt_exec() -> _QmtExec: + """懒加载单例（线程安全）。首次调用连 miniQMT，之后复用。""" + global _singleton + if _singleton is None: + with _lock: + if _singleton is None: + _singleton = _QmtExec() + return _singleton + + +class QmtGatewayClient: + """进程内 vnpy_qmt 执行客户端，接口对齐 BridgeClient。 + + 构造签名兼容 BridgeClient(url, token)——url/token 被忽略（进程内无需）， + 便于 live_orchestrator 零改替换。 + """ + + def __init__(self, url: str = "", token: str = "") -> None: + self._exec = get_qmt_exec() + + def place_order(self, code: str, action: str, price: float, volume: int, + price_type: str = "limit", reason: str = "") -> dict | None: + """下单 → {ok: True, order_id: vt_orderid} 或 None。""" + try: + from vnpy.trader.constant import Direction, OrderType, Offset + from vnpy.trader.object import OrderRequest + symbol, exchange = _parse_code(code) + direction = Direction.LONG if action == "buy" else Direction.SHORT + otype = OrderType.LIMIT if price_type == "limit" else OrderType.MARKET + req = OrderRequest(symbol=symbol, exchange=exchange, direction=direction, + type=otype, volume=volume, price=price, + offset=Offset.NONE, reference=reason or "") + vt_orderid = self._exec.gateway.send_order(req) + return {"ok": True, "order_id": vt_orderid} + except Exception as e: # noqa: BLE001 影子下单绝不阻断 live_step + logger.warning("QmtGatewayClient.place_order 失败 code=%s: %s", code, e) + return None + + def get_account(self) -> dict | None: + a = self._exec.account + if a is None: + return None + mv = sum(p.volume * p.price for p in self._exec.positions.values()) + return {"ok": True, "cash": a.balance - mv, "frozen": a.frozen, + "market_value": mv, "total": a.balance} + + def get_positions(self) -> list | None: + return [{"code": _to_bridge_code(p.symbol, p.exchange), "volume": int(p.volume), + "can_use": int(p.yd_volume), "avg_price": float(p.price)} + for p in self._exec.positions.values() if p.volume > 0] + + def get_orders(self) -> list | None: + from vnpy.trader.constant import Status + return [{"vt_orderid": o.vt_orderid, "code": _to_bridge_code(o.symbol, o.exchange), + "direction": "buy" if o.direction.value == "多" else "sell", + "price": o.price, "volume": o.volume, "traded": o.traded, + "status": o.status.name, "reference": o.reference} + for o in self._exec.orders.values()] + + def cancel_order(self, order_id) -> dict | None: + try: + from vnpy.trader.object import CancelRequest + oid = str(order_id) + if oid.startswith("QMT."): # vt_orderid(QMT.184246#1)→ remark(184246#1) + oid = oid[4:] # vnpy_qmt td.orders 按 remark 建 key + self._exec.gateway.cancel_order(CancelRequest(orderid=oid)) + return {"ok": True, "order_id": order_id} + except Exception as e: # noqa: BLE001 + logger.warning("QmtGatewayClient.cancel_order 失败 %s: %s", order_id, e) + return {"ok": False, "error": str(e)}