sanguo_vnpy_v2/sanguo_qmt_bridge/bridge.py

"""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
import threading
import time
from contextlib import asynccontextmanager
from typing import Literal

from fastapi import Depends, FastAPI
from pydantic import BaseModel, Field

from auth import verify_token
from trade_calendar import is_trading_day
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)


# ===== /health 探活缓存（避免每次 health 都 query_stock_asset）=====
_PROBE_CACHE_SECONDS = 5.0
_last_probe_time: float = 0.0
_last_probe_alive: bool = False


def _probe_alive() -> bool:
    """探活 miniQMT 真实连接（5 秒缓存）。

    缓存过期时调 gateway.is_alive()（query_stock_asset 探活）；
    探活失败且 gateway 自认已连接 → 后台触发 reconnect（不阻塞 health 响应）。
    """
    global _last_probe_time, _last_probe_alive
    now = time.monotonic()
    if now - _last_probe_time < _PROBE_CACHE_SECONDS:
        return _last_probe_alive

    alive = gateway.is_alive()
    _last_probe_time = now
    _last_probe_alive = alive

    if not alive and gateway.connected:
        # _connected 假阳性（miniQMT 可能重启），后台重连不阻塞响应
        threading.Thread(target=gateway.reconnect, daemon=True).start()

    return alive


# ===== 接口 =====

@app.get("/health")
async def health() -> dict:
    """健康检查（无需鉴权，供 frpc/Caddy 探活）。

    miniqmt_connected 调 gateway.is_alive() 真实探活（5 秒缓存），
    不再依赖可能假阳性的 _connected 标志。
    """
    return {"status": "ok", "miniqmt_connected": _probe_alive()}


@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)
    result: dict = {"ok": True, "order_id": order_id}
    if not is_trading_day():
        result["warning"] = "非交易日，miniQMT 可能拒绝(120141)"
    return result


@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}