sanguo_vnpy_v2/sanguo_qmt_bridge/xt_gateway.py

"""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()