From eff9ed2ae9d9ab962bbcba509a3933bee23f7985 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Fri, 10 Jul 2026 23:48:59 +0800 Subject: [PATCH] =?UTF-8?q?feat(bridge):=20D-1=20bridge=20MVP=E2=80=94Fast?= =?UTF-8?q?API=204=E6=8E=A5=E5=8F=A3(xtquant=E5=B0=81=E8=A3=85+token?= =?UTF-8?q?=E9=89=B4=E6=9D=83+sh/sz=E4=BB=A3=E7=A0=81=E8=BD=AC=E6=8D=A2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - bridge.py: lifespan连miniQMT, health/order/account/positions, 连不上不崩 - xt_gateway.py: xtquant单例封装(延迟import), 照搬check_xtquant验证模式 - auth.py: X-Bridge-Token校验(hmac防时序攻击), 未配token返回503不裸奔 - requirements.txt(fastapi+uvicorn) + README.md(Windows部署步骤) 安全: 无硬编码secret, token/userdata/account均走环境变量(grep验证CLEAN) --- sanguo_qmt_bridge/README.md | 126 ++++++++++++++++++++ sanguo_qmt_bridge/auth.py | 44 +++++++ sanguo_qmt_bridge/bridge.py | 115 ++++++++++++++++++ sanguo_qmt_bridge/requirements.txt | 2 + sanguo_qmt_bridge/xt_gateway.py | 182 +++++++++++++++++++++++++++++ 5 files changed, 469 insertions(+) create mode 100644 sanguo_qmt_bridge/README.md create mode 100644 sanguo_qmt_bridge/auth.py create mode 100644 sanguo_qmt_bridge/bridge.py create mode 100644 sanguo_qmt_bridge/requirements.txt create mode 100644 sanguo_qmt_bridge/xt_gateway.py diff --git a/sanguo_qmt_bridge/README.md b/sanguo_qmt_bridge/README.md new file mode 100644 index 0000000..357626f --- /dev/null +++ b/sanguo_qmt_bridge/README.md @@ -0,0 +1,126 @@ +# sanguo QMT bridge (D-1 MVP) + +Windows 端 FastAPI 服务，封装 xtquant，供 sanguo(NAS) 跨网调 miniQMT 下单/查询。 + +## 前提 + +1. **miniQMT 客户端已登录并保持运行**（极简模式即可） +2. **xtquant 可 import**（miniQMT 安装目录自带，或 `pip install xtquant`） +3. Python 3.10+ + +## 安装 + +```powershell +cd C:\sanguo_qmt_bridge # bridge 代码所在目录 +pip install -r requirements.txt +``` + +> xtquant 来自 miniQMT 安装目录（`bin.x64\Lib\site-packages`），无需 pip install。 +> 若 import 失败，确认 miniQMT 的 site-packages 在 PYTHONPATH，或 `pip install xtquant`。 + +## 配置（环境变量） + +| 变量 | 说明 | 默认值 | +|------|------|--------| +| `BRIDGE_TOKEN` | **必填**。共享密钥，sanguo 端同值。不设则受保护接口返回 503 | — | +| `MINIQMT_USERDATA` | miniQMT userdata_mini 路径 | `D:\国金QMT交易端模拟\userdata_mini` | +| `ACCOUNT_ID` | miniQMT 账户 ID | `66639661` | +| `BRIDGE_SESSION_ID` | xtquant session ID（int） | `20260710` | + +**设置 BRIDGE_TOKEN**（PowerShell）： + +```powershell +# 临时（当前终端） +$env:BRIDGE_TOKEN = "你的随机密钥" + +# 永久（系统环境变量） +[Environment]::SetEnvironmentVariable("BRIDGE_TOKEN", "你的随机密钥", "User") +``` + +> 生成密钥：`python -c "import secrets; print(secrets.token_urlsafe(32))"` + +## 启动 + +```powershell +cd C:\sanguo_qmt_bridge +uvicorn bridge:app --host 127.0.0.1 --port 8765 +``` + +启动日志看到 `xtquant 连接成功` 即就绪。若 miniQMT 未连上，服务仍运行（`/health` 报 `disconnected`）。 + +## 接口 + +所有受保护接口需 header `X-Bridge-Token: `，`/health` 豁免。 + +### GET /health（无需鉴权） + +```json +{"status": "ok", "miniqmt_connected": true} +``` + +### POST /order + +请求： +```json +{ + "code": "sh600000", + "action": "buy", + "price": 10.50, + "volume": 100, + "price_type": "limit", + "reason": "策略信号" +} +``` + +成功： +```json +{"ok": true, "order_id": 12345} +``` + +失败： +```json +{"ok": false, "error": "报单失败 order_id=-1"} +``` + +### GET /account + +```json +{"ok": true, "cash": 100000.0, "frozen": 0.0, "market_value": 50000.0, "total": 150000.0} +``` + +### GET /positions + +```json +{ + "ok": true, + "positions": [ + {"code": "sh600000", "volume": 200, "can_use": 200, "avg_price": 10.30} + ] +} +``` + +## 代码格式 + +- 入参（/order）：sanguo 格式 `sh600000` / `sz000001` +- 返回（/positions）：sanguo 格式 `sh600000` / `sz000001` +- bridge 内部自动转 xtquant 格式 `600000.SH` / `000001.SZ` + +## 验证 + +跑过 `check_xtquant.py` 确认 xtquant 可用后，启动 bridge 再验证： + +```powershell +# 健康检查（无需 token） +curl http://127.0.0.1:8765/health + +# 查资金（需 token） +curl -H "X-Bridge-Token: 你的密钥" http://127.0.0.1:8765/account +``` + +## 自启（可选，D-2） + +任务计划程序创建 `sanguo-bridge`，onstart，命令： +``` +uvicorn bridge:app --host 127.0.0.1 --port 8765 +``` +工作目录设为 bridge 代码目录。依赖 miniQMT 客户端已先启动。 diff --git a/sanguo_qmt_bridge/auth.py b/sanguo_qmt_bridge/auth.py new file mode 100644 index 0000000..d5c33ef --- /dev/null +++ b/sanguo_qmt_bridge/auth.py @@ -0,0 +1,44 @@ +"""Token 鉴权：每个请求带 X-Bridge-Token header，bridge 校验。 + +设计（spec §6）： +- bridge 从环境变量 BRIDGE_TOKEN 读期望值，不进 git。 +- 不符/缺失 -> 401。 +- /health 豁免（该路由不挂 Depends(verify_token)）。 +- 使用 hmac.compare_digest 防时序攻击。 +""" +import hmac +import logging +import os + +from fastapi import HTTPException, Request + +logger = logging.getLogger(__name__) + + +def get_expected_token() -> str | None: + """从环境变量读 BRIDGE_TOKEN。""" + return os.environ.get("BRIDGE_TOKEN") + + +def verify_token(request: Request) -> None: + """FastAPI Depends：校验 X-Bridge-Token header。 + + Raises: + HTTPException 503: BRIDGE_TOKEN 未配置（安全默认：拒绝所有受保护请求）。 + HTTPException 401: token 缺失或不匹配。 + """ + expected = get_expected_token() + if not expected: + # 未配置 token -> 拒绝所有受保护请求（不裸奔） + raise HTTPException( + status_code=503, + detail="BRIDGE_TOKEN 环境变量未配置，拒绝服务", + ) + + token = request.headers.get("X-Bridge-Token", "") + if not hmac.compare_digest(token, expected): + logger.warning( + "token 校验失败 source=%s", + request.client.host if request.client else "unknown", + ) + raise HTTPException(status_code=401, detail="token 无效") diff --git a/sanguo_qmt_bridge/bridge.py b/sanguo_qmt_bridge/bridge.py new file mode 100644 index 0000000..72dae07 --- /dev/null +++ b/sanguo_qmt_bridge/bridge.py @@ -0,0 +1,115 @@ +"""D-1 实盘 bridge MVP：FastAPI 4 接口（health/order/account/positions）。 + +监听 127.0.0.1:8765，启动时连 miniQMT（连不上不崩溃，/health 报 disconnected）。 +xtquant 调用照搬 check_xtquant.py 验证过的模式（封装在 xt_gateway.py）。 +token 鉴权（auth.py），/health 豁免。 + +启动：uvicorn bridge:app --host 127.0.0.1 --port 8765 +""" +import logging +from contextlib import asynccontextmanager +from typing import Literal + +from fastapi import Depends, FastAPI +from pydantic import BaseModel, Field + +from auth import verify_token +from xt_gateway import gateway + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s %(message)s", +) +logger = logging.getLogger(__name__) + + +# ===== 请求模型 ===== + +class OrderRequest(BaseModel): + """下单请求（sanguo 格式代码 sh600000/sz000001）。""" + + code: str = Field(..., examples=["sh600000"]) + action: Literal["buy", "sell"] + price: float = Field(..., gt=0, examples=[10.50]) + volume: int = Field(..., gt=0, examples=[100]) + price_type: Literal["limit", "market"] = "limit" + reason: str | None = None + + +# ===== 生命周期 ===== + +@asynccontextmanager +async def lifespan(_app: FastAPI): + """启动时连接 miniQMT；连不上不阻塞服务。""" + ok = gateway.connect() + if not ok: + logger.warning("miniQMT 连接失败，服务继续运行（/health 报 disconnected）") + yield + # shutdown: 无资源需释放（xtquant.stop 由 miniQMT 管控） + + +app = FastAPI(title="sanguo QMT bridge", version="0.1.0", lifespan=lifespan) + + +# ===== 接口 ===== + +@app.get("/health") +async def health() -> dict: + """健康检查（无需鉴权，供 frpc/Caddy 探活）。""" + return {"status": "ok", "miniqmt_connected": gateway.connected} + + +@app.post("/order", dependencies=[Depends(verify_token)]) +async def place_order(req: OrderRequest) -> dict: + """下单 -> {ok, order_id?} 或 {ok:false, error}。""" + if not gateway.connected: + return {"ok": False, "error": "miniQMT 未连接"} + + logger.info( + "下单请求 code=%s action=%s price=%s volume=%s type=%s reason=%s", + req.code, req.action, req.price, req.volume, req.price_type, req.reason, + ) + try: + order_id = gateway.place_order( + code=req.code, + action=req.action, + price=req.price, + volume=req.volume, + price_type=req.price_type, + ) + except (ValueError, RuntimeError) as e: + return {"ok": False, "error": str(e)} + except Exception as e: + logger.error("下单异常: %s", e) + return {"ok": False, "error": f"下单异常: {e}"} + + if order_id <= 0: + return {"ok": False, "error": f"报单失败 order_id={order_id}"} + logger.info("下单成功 order_id=%s code=%s action=%s", order_id, req.code, req.action) + return {"ok": True, "order_id": order_id} + + +@app.get("/account", dependencies=[Depends(verify_token)]) +async def query_account() -> dict: + """查资金 -> {ok, cash, frozen, market_value, total} 或 {ok:false, error}。""" + if not gateway.connected: + return {"ok": False, "error": "miniQMT 未连接"} + try: + data = gateway.query_account() + except Exception as e: + logger.error("查资金异常: %s", e) + return {"ok": False, "error": f"查询异常: {e}"} + return {"ok": True, **data} + + +@app.get("/positions", dependencies=[Depends(verify_token)]) +async def query_positions() -> dict: + """查持仓 -> {ok, positions:[...]} 或 {ok:false, error}。""" + if not gateway.connected: + return {"ok": False, "error": "miniQMT 未连接"} + try: + positions = gateway.query_positions() + except Exception as e: + logger.error("查持仓异常: %s", e) + return {"ok": False, "error": f"查询异常: {e}"} + return {"ok": True, "positions": positions} diff --git a/sanguo_qmt_bridge/requirements.txt b/sanguo_qmt_bridge/requirements.txt new file mode 100644 index 0000000..fc7b393 --- /dev/null +++ b/sanguo_qmt_bridge/requirements.txt @@ -0,0 +1,2 @@ +fastapi>=0.100.0 +uvicorn[standard]>=0.23.0 diff --git a/sanguo_qmt_bridge/xt_gateway.py b/sanguo_qmt_bridge/xt_gateway.py new file mode 100644 index 0000000..607f91f --- /dev/null +++ b/sanguo_qmt_bridge/xt_gateway.py @@ -0,0 +1,182 @@ +"""xtquant 网关封装：单例 trader，连接 miniQMT 客户端。 + +照搬 check_xtquant.py 验证过的 xtquant 调用模式： +- XtQuantTrader(userdata, session_id) -> start() -> connect() +- StockAccount(account_id) -> subscribe(account) +- query_stock_asset / query_stock_positions / order_stock + +启动时连接 miniQMT；连不上不崩溃，bridge /health 报 disconnected。 +所有 xtquant import 延迟到 connect() 内部（Mac/NAS 无 xtquant 时模块仍可加载）。 +""" +import logging +import os +from typing import Any + +logger = logging.getLogger(__name__) + +# ===== 配置（环境变量优先，fallback 到 check_xtquant.py 默认值）===== +MINIQMT_USERDATA = os.environ.get( + "MINIQMT_USERDATA", + r"D:\国金QMT交易端模拟\userdata_mini", +) +ACCOUNT_ID = os.environ.get("ACCOUNT_ID", "66639661") +SESSION_ID = int(os.environ.get("BRIDGE_SESSION_ID", "20260710")) + + +# ===== 代码格式转换 ===== + +def to_xtquant_code(code: str) -> str: + """sanguo 格式(sh600000/sz000001) -> xtquant 格式(600000.SH/000001.SZ)。""" + code = code.strip().lower() + if "." in code: + return code.upper() + if code.startswith("sh"): + return f"{code[2:]}.SH" + if code.startswith("sz"): + return f"{code[2:]}.SZ" + raise ValueError(f"无法识别的股票代码格式: {code}") + + +def to_sanguo_code(code: str) -> str: + """xtquant 格式(600000.SH/000001.SZ) -> sanguo 格式(sh600000/sz000001)。""" + code = code.strip() + if "." not in code: + raise ValueError(f"无法识别的 xtquant 代码格式: {code}") + symbol, market = code.split(".", 1) + return f"{market.lower()}{symbol}" + + +# ===== 网关 ===== + +class XtGateway: + """xtquant 单例网关，封装 connect/query/place_order。 + + connected 属性供 bridge /health 查询；未连接时 query/place_order 抛 RuntimeError。 + """ + + def __init__(self) -> None: + self._xt: Any = None + self._account: Any = None + self._connected: bool = False + + @property + def connected(self) -> bool: + """miniQMT 是否已连接。""" + return self._connected + + def connect(self) -> bool: + """连接 miniQMT 客户端。失败记日志不崩溃，返回 False。""" + try: + from xtquant import xttrader + from xtquant.xttype import StockAccount + except ImportError as e: + logger.error("xtquant import 失败（检查 site-packages）: %s", e) + self._connected = False + return False + + try: + xt = xttrader.XtQuantTrader(MINIQMT_USERDATA, SESSION_ID) + xt.start() + ret = xt.connect() + if ret != 0: + logger.error( + "xtquant connect 返回 %s（miniQMT 未登录或路径错误）", ret, + ) + self._connected = False + return False + + account = StockAccount(ACCOUNT_ID) + try: + xt.subscribe(account) + except Exception as e: + logger.warning("subscribe 异常（可忽略，继续）: %s", e) + + self._xt = xt + self._account = account + self._connected = True + logger.info("xtquant 连接成功 account=%s", ACCOUNT_ID) + return True + except Exception as e: + logger.error("xtquant 连接异常: %s", e) + self._connected = False + return False + + def query_account(self) -> dict[str, float]: + """查资金：{cash, frozen, market_value, total}。""" + if not self._connected: + raise RuntimeError("xtquant 未连接") + asset = self._xt.query_stock_asset(self._account) + if asset is None: + raise RuntimeError("query_stock_asset 返回空（账户ID/权限问题）") + return { + "cash": float(asset.cash), + "frozen": float(asset.frozen_cash), + "market_value": float(asset.market_value), + "total": float(asset.total_asset), + } + + def query_positions(self) -> list[dict[str, Any]]: + """查持仓：[{code, volume, can_use, avg_price}]，code 已转 sanguo 格式。""" + if not self._connected: + raise RuntimeError("xtquant 未连接") + positions = self._xt.query_stock_positions(self._account) or [] + return [ + { + "code": to_sanguo_code(p.stock_code), + "volume": int(p.volume), + "can_use": int(p.can_use_volume), + "avg_price": float(p.avg_price), + } + for p in positions + ] + + def place_order( + self, + code: str, + action: str, + price: float, + volume: int, + price_type: str = "limit", + ) -> int: + """下单，返回 order_id（>0 = 报单成功）。 + + Args: + code: sanguo 格式(sh600000/sz000001)，内部转 xtquant 格式。 + action: "buy" / "sell"。 + price: 委托价格（市价单忽略）。 + volume: 委托数量（股）。 + price_type: "limit"(限价 FIX_PRICE) / "market"(市价最新 LATEST_PRICE)。 + + Raises: + ValueError: action/price_type/code 不合法。 + RuntimeError: xtquant 未连接。 + """ + if not self._connected: + raise RuntimeError("xtquant 未连接") + from xtquant import xtconstant + + order_type = ( + xtconstant.STOCK_BUY if action == "buy" else xtconstant.STOCK_SELL + ) + xt_price_type = ( + xtconstant.FIX_PRICE + if price_type == "limit" + else xtconstant.LATEST_PRICE + ) + xt_code = to_xtquant_code(code) + + order_id = self._xt.order_stock( + self._account, + xt_code, + order_type, + volume, + xt_price_type, + price, + "sanguo_bridge", + "", + ) + return int(order_id) + + +# 模块级单例（bridge.py 启动时调 connect） +gateway = XtGateway()