feat(portfolio): P1.3涨跌停/停牌过滤批量化双端—①all_weather monthly_adjustment三filter接线_get_limit_status批量预取(原不传status_map/date走无数据保留死分支,照买涨停照交易停牌;同small_cap等三策略) ②ShadowBroker涨跌停拒单(买撞涨停/卖撞跌停/停牌双向拒;limit_getter注入,未注入/异常/无状态降级放行等价旧行为) ③runner build_limit_getter(优先miniQMT实时tick lastPrice vs UpStop/DownStop与实盘同源,回退get_limit_status_batch日线口径);9新测试 [vps]

2026-08-14 23:04:15 +08:00
parent fbd0c39c6c
commit 61cc01efbe
5 changed files with 231 additions and 9 deletions
@@ -244,14 +244,19 @@ class AllWeatherStrategy:
            logger.info("开外盘(海外 ETF)")
            target = list(cfg.foreign_etf)

-        # 4) 过滤涨停/跌停/停牌
+        # 4) 过滤涨停/跌停/停牌(P1.3 批量预取:一次查三 filter 共享,原逐 filter
+        #    不传 status_map/date → 全部走"无数据保留"死分支,照买涨停照交易停牌)
+        pos_keys = list(_get_positions(context).keys())
+        status_map = self._get_limit_status(target, current_dt)
        target = filters.filter_limitup_stock(
-            target, self.provider, positions=list(_get_positions(context).keys())
+            target, self.provider, positions=pos_keys, status_map=status_map
        )
        target = filters.filter_limitdown_stock(
-            target, self.provider, positions=list(_get_positions(context).keys())
+            target, self.provider, positions=pos_keys, status_map=status_map
+        )
+        target = filters.filter_paused_stock(
+            target, self.provider, status_map=status_map
        )
-        target = filters.filter_paused_stock(target, self.provider)
        target = _dedup(target)

        # 5) 调仓:先清掉不在 target 的(除昨日涨停)
@@ -460,15 +465,25 @@ class AllWeatherStrategy:
        幅度精确算涨跌停价,与 filters/其他策略同源)。provider 未实现/异常
        → None(跳过该股,等价原失效行为)。
        """
+        return self._get_limit_status([stock], now_time).get(stock)
+
+    def _get_limit_status(self, stocks: List[str], date: Any) -> Dict[str, Any]:
+        """批量预取涨跌停/停牌状态(P1.3:monthly_adjustment 三 filter 共享一次查询)。
+
+        与 small_cap/momentum_timing/value_selection 同款:provider 未实现
+        get_limit_status_batch / 异常 → 返空 dict(filter 见 None 走"无数据
+        保留所有"分支,等价原失效行为)。
+        """
+        if not stocks:
+            return {}
        fn = getattr(self.provider, "get_limit_status_batch", None)
        if fn is None:
-            return None
+            return {}
        try:
-            result = fn([stock], now_time) or {}
+            return fn(stocks, date) or {}
        except Exception as exc:
-            logger.warning("get_limit_status_batch 失败 %s: %s", stock, exc)
-            return None
-        return result.get(stock)
+            logger.warning("get_limit_status_batch 失败: %s", exc)
+            return {}


 # ======================== context 适配 ========================
@@ -34,6 +34,7 @@ class ShadowBroker:  # noqa: R0903 - 仅实现 BrokerBase 协议(bullet_trade du
        min_commission: float = 5.0,
        slippage: float = 0.0,
        price_getter: Optional[Callable[[str], Optional[float]]] = None,
+        limit_getter: Optional[Callable[[str], Optional[Dict[str, Any]]]] = None,
        on_trade: Optional[Callable[[Dict[str, Any]], None]] = None,
        now_provider: Optional[Callable[[], datetime]] = None,
    ) -> None:
@@ -44,6 +45,7 @@ class ShadowBroker:  # noqa: R0903 - 仅实现 BrokerBase 协议(bullet_trade du
        self.min_commission = float(min_commission)
        self.slippage = float(slippage)
        self.price_getter = price_getter
+        self.limit_getter = limit_getter
        self.on_trade = on_trade
        self._now = now_provider or datetime.now
        self._connected = True  # 本地柜台永远"在线"
@@ -104,6 +106,29 @@ class ShadowBroker:  # noqa: R0903 - 仅实现 BrokerBase 协议(bullet_trade du
                ref = None
        return ref

+    def _limit_blocked(self, security: str, side: str) -> Optional[str]:
+        """涨跌停/停牌拒单原因(P1.3,双轨对账与实盘约束对齐)。
+
+        buy 撞涨停买不进、sell 撞跌停卖不出、停牌双向拒。
+        limit_getter 未注入/异常/无状态 → None(放行,等价旧行为)。
+        """
+        if self.limit_getter is None:
+            return None
+        try:
+            status = self.limit_getter(security)
+        except Exception as exc:  # noqa: BLE001 - 状态查询失败降级放行
+            logger.warning("[shadow] 涨跌停状态查询失败 %s: %s", security, exc)
+            return None
+        if not isinstance(status, dict):
+            return None
+        if status.get("is_paused") is True:
+            return "停牌不可交易"
+        if side == "buy" and status.get("is_limit_up") is True:
+            return "涨停拒买"
+        if side == "sell" and status.get("is_limit_down") is True:
+            return "跌停拒卖"
+        return None
+
    # ===== 下单(即时全额成交) =====
    async def buy(self, security: str, amount: int, price: Optional[float] = None,
                  wait_timeout: Optional[float] = None, remark: Optional[str] = None,
@@ -112,6 +137,9 @@ class ShadowBroker:  # noqa: R0903 - 仅实现 BrokerBase 协议(bullet_trade du
        ref = self._ref_price(security, price)
        if ref is None or ref <= 0:
            return self._reject(order_id, "无参考价")
+        blocked = self._limit_blocked(security, "buy")
+        if blocked:
+            return self._reject(order_id, blocked)
        amount = int(amount)
        if amount <= 0:
            return self._reject(order_id, "数量非法")
@@ -139,6 +167,9 @@ class ShadowBroker:  # noqa: R0903 - 仅实现 BrokerBase 协议(bullet_trade du
        ref = self._ref_price(security, price)
        if ref is None or ref <= 0:
            return self._reject(order_id, "无参考价")
+        blocked = self._limit_blocked(security, "sell")
+        if blocked:
+            return self._reject(order_id, blocked)
        amount = int(amount)
        pos = self.positions.get(security)
        held = int(pos["amount"]) if pos else 0
@@ -73,6 +73,44 @@ def build_price_getter(provider: Any) -> Any:
    return get_price


+def build_limit_getter(provider: Any) -> Any:
+    """从数据 provider 取标的实时涨跌停/停牌状态(P1.3,ShadowBroker 拒单用)。
+
+    优先 ``get_live_current``(miniQMT 实时 tick:lastPrice vs UpStop/DownStop,
+    与实盘同源同刻);无实时能力(回测 provider)→ 回退 ``get_limit_status_batch``
+    (日线 prev_close×幅度口径)。查不到 → None(ShadowBroker 放行,等价旧行为)。
+    """
+    from datetime import datetime
+    from typing import Optional as _Opt
+
+    def get_limit(security: str) -> _Opt[dict]:
+        live_fn = getattr(provider, "get_live_current", None)
+        if live_fn is not None:
+            try:
+                cur = live_fn(security) or {}
+                last = cur.get("last_price")
+                high = cur.get("high_limit") or 0.0
+                low = cur.get("low_limit") or 0.0
+                if last:
+                    return {
+                        "is_limit_up": bool(high and float(last) >= float(high)),
+                        "is_limit_down": bool(low and float(last) <= float(low)),
+                        "is_paused": bool(cur.get("paused")),
+                    }
+            except Exception:  # noqa: BLE001 - 实时源失败试批量源
+                pass
+        batch_fn = getattr(provider, "get_limit_status_batch", None)
+        if batch_fn is not None:
+            try:
+                return (batch_fn([security], datetime.now().strftime("%Y-%m-%d"))
+                        or {}).get(security)
+            except Exception:  # noqa: BLE001
+                return None
+        return None
+
+    return get_limit
+
+
 def _paper_on_trade(db: str, account_id: int, strategy_id: str):
    """成交回调:落 paper_trades(与组合实走 EOD 同表,前端模拟盘页直接可见)。"""
    from sanguo_trader.persistence import save_trade
@@ -161,6 +199,7 @@ def run_shadow(provider_config: Optional[Dict[str, Any]] = None) -> None:
        min_commission=float(cfg["min_comm"]),
        slippage=float(cfg["slippage"]),
        price_getter=build_price_getter(provider),
+        limit_getter=build_limit_getter(provider),
        on_trade=_paper_on_trade(cfg["db"], int(cfg["account_id"]), le["strategy"])
        if cfg["db"] and cfg["account_id"] else None,
    )
@@ -270,6 +270,57 @@ class TestMonthlyAdjustmentDecision:
        for etf in s.config.foreign_etf:
            assert etf in called_codes, f"未触发海外 ETF 下单: {etf}"

+    def test_foreign_etf_branch_skips_limitup_and_paused(self):
+        """P1.3:涨停(未持有)与停牌的 ETF 不买入——filter 批量预取接线。"""
+        s = make_strategy(
+            index_stocks_map={
+                "000300.XSHG": ["600519.XSHG"],
+                "399101.XSHE": ["000001.XSHE"],
+            },
+            price_df_map={
+                ("['600519.XSHG']", ("close",), 10): pd.DataFrame({
+                    "time": pd.to_datetime(["2024-09-20", "2024-09-30"]),
+                    "code": ["600519.XSHG"] * 2,
+                    "close": [15.0, 10.0],  # 跌
+                }),
+                ("['000001.XSHE']", ("close",), 10): pd.DataFrame({
+                    "time": pd.to_datetime(["2024-09-20", "2024-09-30"]),
+                    "code": ["000001.XSHE"] * 2,
+                    "close": [15.0, 10.0],  # 跌
+                }),
+            },
+        )
+        s.provider.get_fundamentals_df.return_value = make_fund_df([
+            {"code": "600519.XSHG", "circulating_market_cap": 20000, "market_cap": 20000},
+            {"code": "000001.XSHE", "circulating_market_cap": 500, "market_cap": 500},
+        ])
+        # 518880 涨停(未持有不买)、513030 停牌(不交易),其余正常
+        def _glbs(codes, date):
+            out = {}
+            for c in codes:
+                if c == "518880.XSHG":
+                    out[c] = {"is_limit_up": True, "is_limit_down": False, "is_paused": False}
+                elif c == "513030.XSHG":
+                    out[c] = {"is_limit_up": False, "is_limit_down": False, "is_paused": True}
+                else:
+                    out[c] = {"is_limit_up": False, "is_limit_down": False, "is_paused": False}
+            return out
+        s.provider.get_limit_status_batch.side_effect = _glbs
+
+        ctx = MagicMock()
+        ctx.current_dt = datetime(2024, 10, 8, 9, 30)
+        ctx.previous_date = "2024-09-30"
+        ctx.portfolio.positions = {}
+        ctx.portfolio.available_cash = 1_000_000
+
+        s.monthly_adjustment(ctx)
+
+        called_codes = [c.args[0] for c in s.broker.order_target_value.call_args_list]
+        assert "518880.XSHG" not in called_codes, "涨停 ETF 不应买入"
+        assert "513030.XSHG" not in called_codes, "停牌 ETF 不应交易"
+        for etf in ("513100.XSHG", "164824.XSHE", "159866.XSHE"):
+            assert etf in called_codes, f"正常 ETF 应下单: {etf}"
+
    def test_big_market_branch_when_b_trend_dominant(self):
        """b_mean > s_mean 且 b_mean > 0 → 开大(选 B_stocks)。"""
        s = make_strategy(
@@ -97,6 +97,92 @@ def test_no_price_rejects():
    assert "无参考价" in b.orders[oid]["reject_reason"]


+# ---- P1.3 涨跌停/停牌拒单(双轨对账:与实盘 QMT 约束对齐,减少对账噪音) ----
+
+def _limit_map_getter(status: dict):
+    return lambda sec: status.get(sec)
+
+
+def test_buy_rejected_when_limit_up():
+    b = _mk_broker(limit_getter=_limit_map_getter({
+        "600519.SH": {"is_limit_up": True, "is_limit_down": False, "is_paused": False},
+    }))
+    oid = _buy(b, "600519.SH", 100)
+    assert b.orders[oid]["status"] == "rejected"
+    assert "涨停" in b.orders[oid]["reject_reason"]
+    assert "600519.SH" not in b.positions
+
+
+def test_sell_rejected_when_limit_down():
+    b = _mk_broker(limit_getter=_limit_map_getter({
+        "600519.SH": {"is_limit_up": False, "is_limit_down": True, "is_paused": False},
+    }))
+    _buy(b, "600519.SH", 100, px=100.0)  # 买入时非跌停
+    oid = _sell(b, "600519.SH", 100)
+    assert b.orders[oid]["status"] == "rejected"
+    assert "跌停" in b.orders[oid]["reject_reason"]
+    assert b.positions["600519.SH"]["amount"] == 100
+
+
+def test_buy_sell_rejected_when_paused():
+    b = _mk_broker(limit_getter=_limit_map_getter({
+        "600519.SH": {"is_limit_up": False, "is_limit_down": False, "is_paused": True},
+    }))
+    oid = _buy(b, "600519.SH", 100)
+    assert b.orders[oid]["status"] == "rejected"
+    assert "停牌" in b.orders[oid]["reject_reason"]
+
+
+def test_limit_getter_failure_degrades_to_fill():
+    """limit_getter 抛异常 → 降级放行(等价无涨跌停数据的旧行为),不崩柜台。"""
+    def boom(sec):
+        raise RuntimeError("boom")
+    b = _mk_broker(limit_getter=boom)
+    oid = _buy(b, "600519.SH", 100)
+    assert b.orders[oid]["status"] == "filled"
+
+
+def test_no_limit_getter_keeps_old_behavior():
+    b = _mk_broker()  # 不注入 limit_getter
+    oid = _buy(b, "600519.SH", 100)
+    assert b.orders[oid]["status"] == "filled"
+
+
+# ---- build_limit_getter:provider 状态 → broker 语义映射 ----
+
+def test_limit_getter_maps_live_current_limit_up():
+    from sanguo_trader.shadow.runner import build_limit_getter
+
+    class P:
+        def get_live_current(self, sec):
+            # last_price == high_limit → 涨停
+            return {"last_price": 11.0, "high_limit": 11.0,
+                    "low_limit": 9.0, "paused": False}
+
+    g = build_limit_getter(P())
+    assert g("600519.SH") == {"is_limit_up": True, "is_limit_down": False,
+                              "is_paused": False}
+
+
+def test_limit_getter_falls_back_to_batch_when_no_live_current():
+    from sanguo_trader.shadow.runner import build_limit_getter
+
+    class P:
+        def get_limit_status_batch(self, codes, date):
+            return {c: {"is_limit_up": False, "is_limit_down": True,
+                        "is_paused": False} for c in codes}
+
+    g = build_limit_getter(P())
+    assert g("600519.SH")["is_limit_down"] is True
+
+
+def test_limit_getter_returns_none_without_any_source():
+    from sanguo_trader.shadow.runner import build_limit_getter
+
+    g = build_limit_getter(object())  # 两接口都没有
+    assert g("600519.SH") is None
+
+
 def test_on_trade_callback_receives_fills():
    seen: list[dict] = []
    b = ShadowBroker(