From cadc59e6dcb953908548660112358854ea43e524 Mon Sep 17 00:00:00 2001
From: claude_dev <claude-dev@sanguo.local>
Date: Sat, 11 Jul 2026 07:32:23 +0800
Subject: [PATCH] =?UTF-8?q?feat(bridge):=20bridge=E7=A8=B3=E5=AE=9A?=
 =?UTF-8?q?=E6=80=A7=E5=AE=8C=E5=96=84=E2=80=94=E8=87=AA=E5=8A=A8=E9=87=8D?=
 =?UTF-8?q?=E8=BF=9EminiQMT+health=E6=8E=A2=E6=B4=BB+=E4=BA=A4=E6=98=93?=
 =?UTF-8?q?=E6=97=A5=E5=88=A4=E6=96=AD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- xt_gateway: _open_session抽离, reconnect(stop旧trader+重建), is_alive(query探活), _retry_with_reconnect(query/order失败重连重试一次)
- query_account/positions/place_order包重试: 断线(异常/None)→reconnect→重试, broker拒单(order_id<=0)不重连
- bridge /health: is_alive真实探活(5s缓存)+断线后台reconnect(不阻塞), 不再假阳性
- trade_calendar: is_trading_day(周一-周五), /order非交易日加warning(120141提示)
- test_gateway15+test_trade_calendar6=NAS21passed, 回归bridge_client/d4a 10绿
- 修复Issue#4运维发现: miniQMT重启后bridge自动重连(无需手动重启)
---
 sanguo_qmt_bridge/bridge.py         |  44 +++++-
 sanguo_qmt_bridge/trade_calendar.py |  24 +++
 sanguo_qmt_bridge/xt_gateway.py     | 231 +++++++++++++++++++---------
 tests/trader/test_gateway.py        | 229 +++++++++++++++++++++++++++
 tests/trader/test_trade_calendar.py |  37 +++++
 5 files changed, 488 insertions(+), 77 deletions(-)
 create mode 100644 sanguo_qmt_bridge/trade_calendar.py
 create mode 100644 tests/trader/test_gateway.py
 create mode 100644 tests/trader/test_trade_calendar.py

diff --git a/sanguo_qmt_bridge/bridge.py b/sanguo_qmt_bridge/bridge.py
index 72dae07..0229503 100644
--- a/sanguo_qmt_bridge/bridge.py
+++ b/sanguo_qmt_bridge/bridge.py
@@ -7,6 +7,8 @@ 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
 
@@ -14,6 +16,7 @@ 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(
@@ -51,12 +54,44 @@ async def lifespan(_app: FastAPI):
 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 探活）。"""
-    return {"status": "ok", "miniqmt_connected": gateway.connected}
+    """健康检查（无需鉴权，供 frpc/Caddy 探活）。
+
+    miniqmt_connected 调 gateway.is_alive() 真实探活（5 秒缓存），
+    不再依赖可能假阳性的 _connected 标志。
+    """
+    return {"status": "ok", "miniqmt_connected": _probe_alive()}
 
 
 @app.post("/order", dependencies=[Depends(verify_token)])
@@ -86,7 +121,10 @@ async def place_order(req: OrderRequest) -> dict:
     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}
+    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)])
diff --git a/sanguo_qmt_bridge/trade_calendar.py b/sanguo_qmt_bridge/trade_calendar.py
new file mode 100644
index 0000000..c65d6a0
--- /dev/null
+++ b/sanguo_qmt_bridge/trade_calendar.py
@@ -0,0 +1,24 @@
+"""交易日判断（首版仅周末判断，节假日后续扩展）。
+
+TODO: 后续接入 A 股法定节假日列表（交易所日历数据源或手动配置文件）。
+"""
+from datetime import date
+
+
+def is_trading_day(day: date | None = None) -> bool:
+    """判断是否为 A 股交易日。
+
+    首版：周一至周五 True，周六周日 False。
+    法定节假日（国庆/春节/清明等）暂不处理——miniQMT broker 会自行拒绝，
+    调用方可根据返回值在响应中加 warning 提示。
+
+    Args:
+        day: 待判断日期，None = 今天。
+
+    Returns:
+        True = 工作日（大概率交易日），False = 周末。
+    """
+    if day is None:
+        day = date.today()
+    # TODO: 接入 A 股法定节假日列表（目前仅排除周末）
+    return day.weekday() < 5  # 0=Mon..4=Fri, 5=Sat, 6=Sun
diff --git a/sanguo_qmt_bridge/xt_gateway.py b/sanguo_qmt_bridge/xt_gateway.py
index 607f91f..b5b6fc7 100644
--- a/sanguo_qmt_bridge/xt_gateway.py
+++ b/sanguo_qmt_bridge/xt_gateway.py
@@ -58,6 +58,7 @@ class XtGateway:
         self._xt: Any = None
         self._account: Any = None
         self._connected: bool = False
+        self._last_query_ok: bool = False  # 上次 query 是否成功（供 /health 探活参考）
 
     @property
     def connected(self) -> bool:
@@ -67,68 +68,143 @@ class XtGateway:
     def connect(self) -> bool:
         """连接 miniQMT 客户端。失败记日志不崩溃，返回 False。"""
         try:
-            from xtquant import xttrader
-            from xtquant.xttype import StockAccount
+            return self._open_session()
         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 _open_session(self) -> bool:
+        """建 XtQuantTrader + start + connect + subscribe（延迟 import xtquant）。
+
+        从 connect() 抽出，供 reconnect() 复用。raise 异常由调用方捕获。
+        """
+        from xtquant import xttrader
+        from xtquant.xttype import StockAccount
+
+        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
+
+    def reconnect(self) -> bool:
+        """重连：stop 旧 trader（若有），再 _open_session 重建连接。
+
+        miniQMT 客户端重启后旧 XtQuantTrader 连接失效，必须重建。
+        """
+        if self._xt is not None and hasattr(self._xt, "stop"):
+            try:
+                self._xt.stop()
+            except Exception as e:
+                logger.warning("旧 trader stop 异常（忽略，继续重建）: %s", e)
+        self._xt = None
+        self._account = None
+        self._connected = False
+        logger.info("开始重连 miniQMT ...")
+        return self.connect()
+
+    def is_alive(self) -> bool:
+        """轻量探活：query_stock_asset 返回非空 = True。
+
+        供 /health 调用检测真实连接状态（_connected 标志可能假阳性）。
+        不触发 reconnect（保持轻量），由调用方决定是否重连。
+        """
+        try:
+            if not self._connected:
+                return False
+            asset = self._xt.query_stock_asset(self._account)
+            alive = asset is not None
+            self._last_query_ok = alive
+            return alive
+        except Exception as e:
+            logger.debug("is_alive 探活异常: %s", e)
+            self._last_query_ok = False
+            return False
+
+    def _retry_with_reconnect(self, fn: Any, fail_msg: str) -> Any:
+        """执行 fn()，失败（异常）时 reconnect 一次再重试。
+
+        重连仍失败则 raise（由调用方按原逻辑降级返回空/False）。
+
+        Args:
+            fn: 无参可调用，执行实际 query/order。
+            fail_msg: 日志标识（如 "query_account"）。
+        """
+        try:
+            result = fn()
+            self._last_query_ok = True
+            return result
+        except Exception as e:
+            self._last_query_ok = False
+            logger.warning("%s 首次失败，尝试重连: %s", fail_msg, e)
+            if not self.reconnect():
+                raise RuntimeError(f"重连失败，放弃 {fail_msg}: {e}") from e
+            result = fn()  # 重试一次（不再重连）
+            self._last_query_ok = True
+            return result
+
     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),
-        }
+        """查资金：{cash, frozen, market_value, total}（断线自动重连重试一次）。"""
+
+        def _do() -> dict[str, float]:
+            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),
+            }
+
+        return self._retry_with_reconnect(_do, "query_account")
 
     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
-        ]
+        """查持仓：[{code, volume, can_use, avg_price}]，code 已转 sanguo 格式。
+
+        None（断线）触发重连重试，空列表 []（真没持仓）是正常结果。
+        """
+
+        def _do() -> list[dict[str, Any]]:
+            if not self._connected:
+                raise RuntimeError("xtquant 未连接")
+            positions = self._xt.query_stock_positions(self._account)
+            if positions is None:
+                raise RuntimeError("query_stock_positions 返回 None（疑似断线）")
+            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
+            ]
+
+        return self._retry_with_reconnect(_do, "query_positions")
 
     def place_order(
         self,
@@ -140,6 +216,9 @@ class XtGateway:
     ) -> int:
         """下单，返回 order_id（>0 = 报单成功）。
 
+        断线（调用抛异常）自动重连重试一次；order_id<=0 是 broker 拒单
+        （如非交易日/资金不足），不重试直接返回。
+
         Args:
             code: sanguo 格式(sh600000/sz000001)，内部转 xtquant 格式。
             action: "buy" / "sell"。
@@ -149,33 +228,37 @@ class XtGateway:
 
         Raises:
             ValueError: action/price_type/code 不合法。
-            RuntimeError: xtquant 未连接。
+            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)
+        def _do() -> int:
+            if not self._connected:
+                raise RuntimeError("xtquant 未连接")
+            from xtquant import xtconstant
 
-        order_id = self._xt.order_stock(
-            self._account,
-            xt_code,
-            order_type,
-            volume,
-            xt_price_type,
-            price,
-            "sanguo_bridge",
-            "",
-        )
-        return int(order_id)
+            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)
+
+        return self._retry_with_reconnect(_do, "place_order")
 
 
 # 模块级单例（bridge.py 启动时调 connect）
diff --git a/tests/trader/test_gateway.py b/tests/trader/test_gateway.py
new file mode 100644
index 0000000..2ce03e8
--- /dev/null
+++ b/tests/trader/test_gateway.py
@@ -0,0 +1,229 @@
+"""D-2 gateway 稳定性测试：自动重连 + 探活（mock xtquant，Mac 无 xtquant）。
+
+注入 mock xtquant 模块到 sys.modules，验证：
+- connect 成功/失败
+- query 失败 → reconnect → 重试一次（query_account / query_positions / place_order）
+- place_order broker 拒单(order_id<=0) 不触发重连重试
+- is_alive（query_stock_asset 非空=True，None/异常/未连接=False）
+- reconnect 显式调用 stop 旧 trader
+"""
+import sys
+import types
+from unittest.mock import MagicMock
+
+import pytest
+
+from sanguo_qmt_bridge.xt_gateway import XtGateway
+
+
+# ===== mock xtquant 注入/清理工具 =====
+
+def _install_mock_xtquant(*, connect_ret=0, asset=None, positions=None, order_id=1):
+    """注入 mock xtquant 到 sys.modules，返回 (mock_xt, patches)。"""
+    mock_xt = MagicMock()
+    mock_xt.connect.return_value = connect_ret
+    mock_xt.query_stock_asset.return_value = asset
+    mock_xt.query_stock_positions.return_value = positions
+    mock_xt.order_stock.return_value = order_id
+
+    xttrader_mod = types.ModuleType("xtquant.xttrader")
+    xttrader_mod.XtQuantTrader = MagicMock(return_value=mock_xt)
+
+    xttype_mod = types.ModuleType("xtquant.xttype")
+    xttype_mod.StockAccount = MagicMock()
+
+    xtconstant_mod = types.ModuleType("xtquant.xtconstant")
+    xtconstant_mod.STOCK_BUY = 23
+    xtconstant_mod.STOCK_SELL = 24
+    xtconstant_mod.FIX_PRICE = 11
+    xtconstant_mod.LATEST_PRICE = 5
+
+    xtquant_mod = types.ModuleType("xtquant")
+
+    patches: dict[str, object] = {}
+    for name, mod in [
+        ("xtquant", xtquant_mod),
+        ("xtquant.xttrader", xttrader_mod),
+        ("xtquant.xttype", xttype_mod),
+        ("xtquant.xtconstant", xtconstant_mod),
+    ]:
+        patches[name] = sys.modules.get(name)
+        sys.modules[name] = mod
+
+    return mock_xt, patches
+
+
+def _remove_mock_xtquant(patches):
+    """恢复 sys.modules 到注入前状态。"""
+    for name, mod in patches.items():
+        if mod is None:
+            sys.modules.pop(name, None)
+        else:
+            sys.modules[name] = mod
+
+
+@pytest.fixture
+def mock_xtquant():
+    """注入 mock xtquant，yield mock_xt 供测试调整返回值，结束后清理。"""
+    mock_asset = MagicMock(
+        cash=100000.0, frozen_cash=0.0, market_value=50000.0, total_asset=150000.0,
+    )
+    mock_xt, patches = _install_mock_xtquant(
+        connect_ret=0, asset=mock_asset, positions=[], order_id=12345,
+    )
+    yield mock_xt
+    _remove_mock_xtquant(patches)
+
+
+def _make_mock_position(code="600000.SH", volume=100, can_use=100, avg_price=10.5):
+    """构造 mock 持仓对象（属性与 xtquant XtPosition 一致）。"""
+    return MagicMock(
+        stock_code=code, volume=volume, can_use_volume=can_use, avg_price=avg_price,
+    )
+
+
+# ===== connect =====
+
+class TestConnect:
+    def test_connect_success(self, mock_xtquant):
+        gw = XtGateway()
+        assert gw.connect() is True
+        assert gw.connected is True
+
+    def test_connect_failure_returns_false(self, mock_xtquant):
+        mock_xtquant.connect.return_value = -1
+        gw = XtGateway()
+        assert gw.connect() is False
+        assert gw.connected is False
+
+
+# ===== reconnect =====
+
+class TestReconnect:
+    def test_reconnect_stops_old_trader(self, mock_xtquant):
+        gw = XtGateway()
+        gw.connect()
+        assert mock_xtquant.stop.call_count == 0
+        gw.reconnect()
+        assert mock_xtquant.stop.call_count == 1
+        assert gw.connected is True
+
+    def test_reconnect_returns_true_on_success(self, mock_xtquant):
+        gw = XtGateway()
+        gw.connect()
+        assert gw.reconnect() is True
+
+
+# ===== query 重连重试 =====
+
+class TestQueryRetry:
+    def test_query_account_reconnects_then_succeeds(self, mock_xtquant):
+        """首次 query_stock_asset 返回 None → 重连 → 重试成功。"""
+        mock_asset = MagicMock(
+            cash=80000.0, frozen_cash=0.0, market_value=40000.0, total_asset=120000.0,
+        )
+        mock_xtquant.query_stock_asset.side_effect = [None, mock_asset]
+
+        gw = XtGateway()
+        gw.connect()
+        data = gw.query_account()
+
+        assert data["cash"] == 80000.0
+        assert data["total"] == 120000.0
+        assert gw._last_query_ok is True
+        assert mock_xtquant.stop.call_count == 1  # 确实触发了 reconnect
+
+    def test_query_account_fails_after_retry(self, mock_xtquant):
+        """重连后仍返回 None → RuntimeError 透出。"""
+        mock_xtquant.query_stock_asset.return_value = None
+
+        gw = XtGateway()
+        gw.connect()
+        with pytest.raises(RuntimeError, match="query_stock_asset 返回空"):
+            gw.query_account()
+        assert gw._last_query_ok is False
+
+    def test_query_positions_reconnects_then_succeeds(self, mock_xtquant):
+        """首次 query_stock_positions 返回 None → 重连 → 重试返回持仓列表。"""
+        pos = _make_mock_position()
+        mock_xtquant.query_stock_positions.side_effect = [None, [pos]]
+
+        gw = XtGateway()
+        gw.connect()
+        positions = gw.query_positions()
+
+        assert len(positions) == 1
+        assert positions[0]["code"] == "sh600000"
+        assert positions[0]["volume"] == 100
+        assert mock_xtquant.stop.call_count == 1
+
+    def test_query_positions_empty_list_is_not_disconnection(self, mock_xtquant):
+        """空列表 []（真没持仓）不触发重连。"""
+        mock_xtquant.query_stock_positions.return_value = []
+
+        gw = XtGateway()
+        gw.connect()
+        positions = gw.query_positions()
+
+        assert positions == []
+        assert mock_xtquant.stop.call_count == 0  # 没重连
+        assert gw._last_query_ok is True
+
+    def test_place_order_reconnects_on_exception(self, mock_xtquant):
+        """order_stock 首次抛异常 → 重连 → 重试成功。"""
+        mock_xtquant.order_stock.side_effect = [Exception("连接断开"), 99999]
+
+        gw = XtGateway()
+        gw.connect()
+        order_id = gw.place_order("sh600000", "buy", 10.5, 100)
+
+        assert order_id == 99999
+        assert mock_xtquant.stop.call_count == 1
+        assert gw._last_query_ok is True
+
+    def test_place_order_no_retry_on_broker_reject(self, mock_xtquant):
+        """order_id<=0（broker 拒单）不触发重连，直接返回。"""
+        mock_xtquant.order_stock.return_value = -1
+
+        gw = XtGateway()
+        gw.connect()
+        order_id = gw.place_order("sh600000", "buy", 10.5, 100)
+
+        assert order_id == -1
+        assert mock_xtquant.stop.call_count == 0  # 没重连
+
+
+# ===== is_alive =====
+
+class TestIsAlive:
+    def test_is_alive_true_when_asset_returned(self, mock_xtquant):
+        gw = XtGateway()
+        gw.connect()
+        assert gw.is_alive() is True
+        assert gw._last_query_ok is True
+
+    def test_is_alive_false_when_none(self, mock_xtquant):
+        mock_xtquant.query_stock_asset.return_value = None
+        gw = XtGateway()
+        gw.connect()
+        assert gw.is_alive() is False
+        assert gw._last_query_ok is False
+
+    def test_is_alive_false_when_not_connected(self, mock_xtquant):
+        gw = XtGateway()  # 未 connect
+        assert gw.is_alive() is False
+
+    def test_is_alive_false_on_exception(self, mock_xtquant):
+        mock_xtquant.query_stock_asset.side_effect = RuntimeError("断线")
+        gw = XtGateway()
+        gw.connect()
+        assert gw.is_alive() is False  # 不崩溃
+        assert gw._last_query_ok is False
+
+    def test_is_alive_does_not_trigger_reconnect(self, mock_xtquant):
+        """is_alive 保持轻量，探活失败不自动重连。"""
+        mock_xtquant.query_stock_asset.return_value = None
+        gw = XtGateway()
+        gw.connect()
+        gw.is_alive()
+        assert mock_xtquant.stop.call_count == 0  # 不重连
diff --git a/tests/trader/test_trade_calendar.py b/tests/trader/test_trade_calendar.py
new file mode 100644
index 0000000..3255faf
--- /dev/null
+++ b/tests/trader/test_trade_calendar.py
@@ -0,0 +1,37 @@
+"""D-2 交易日判断单元测试（纯函数，无外部依赖）。"""
+from datetime import date
+
+from sanguo_qmt_bridge.trade_calendar import is_trading_day
+
+
+class TestIsTradingDay:
+    def test_weekday_is_trading_day(self):
+        assert is_trading_day(date(2026, 7, 13)) is True  # 周一
+
+    def test_friday_is_trading_day(self):
+        assert is_trading_day(date(2026, 7, 17)) is True  # 周五
+
+    def test_saturday_is_not_trading_day(self):
+        assert is_trading_day(date(2026, 7, 18)) is False  # 周六
+
+    def test_sunday_is_not_trading_day(self):
+        assert is_trading_day(date(2026, 7, 19)) is False  # 周日
+
+    def test_default_uses_today(self):
+        """无参数时用今天，至少返回 bool 不崩溃。"""
+        result = is_trading_day()
+        assert isinstance(result, bool)
+
+    def test_full_week(self):
+        """完整一周：周一到周五 True，周六周日 False。"""
+        week = [
+            (date(2026, 7, 13), True),   # Mon
+            (date(2026, 7, 14), True),   # Tue
+            (date(2026, 7, 15), True),   # Wed
+            (date(2026, 7, 16), True),   # Thu
+            (date(2026, 7, 17), True),   # Fri
+            (date(2026, 7, 18), False),  # Sat
+            (date(2026, 7, 19), False),  # Sun
+        ]
+        for day, expected in week:
+            assert is_trading_day(day) is expected, f"{day} expected {expected}"