claude_dev
dfa72e1440
vnpy 4.4.0 的 CancelRequest 必填 (orderid,symbol,exchange)，原代码只传
orderid → TypeError → 撤单从不工作。改为从 _exec.orders 缓存查 OrderData
拿 symbol/exchange 一起传。E2E 实证 {ok:True} 无 TypeError（盘后 miniQMT
非交易时段不实际撤单，是另一回事）。
179 lines
7.5 KiB
Python
179 lines
7.5 KiB
Python
"""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)
|
||
vt = oid if oid.startswith("QMT.") else f"QMT.{oid}"
|
||
o = self._exec.orders.get(vt)
|
||
if o is None:
|
||
return {"ok": False, "error": f"order {oid} 不在缓存（gateway 未收到该单事件）"}
|
||
remark = vt[4:] # vt_orderid(QMT.184246#1)→ remark(184246#1)，vnpy_qmt 按 remark 建 key
|
||
# vnpy 4.4.0 的 CancelRequest 必填 (orderid, symbol, exchange)，缺 symbol/exchange 会 TypeError
|
||
self._exec.gateway.cancel_order(
|
||
CancelRequest(orderid=remark, symbol=o.symbol, exchange=o.exchange))
|
||
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)}
|