fix(strategy): 实盘选股挂死根治——证券信息批量委托本地库替600次xtdata RPC + 选股阶段耗时日志 [vps]

2026-08-21 实锤(live_19/shadow_60 同卡):momentum 9:30:06 buy_sign=True 后
47min 零输出被重启;small_cap 同型(fundamentals 超时120s后沉默)。channel_test
正常(不走走filters重取数)。value_selection/all_weather 月度未到期=零成交正常。

根因(VPS离场时段探针27s跑完全路径反推):filters.filter_st/filter_new 逐只
get_security_info,每只2次xtdata RPC(_detect_instrument_type+get_instrument_detail)
且无超时——momentum 10指数×30只×2过滤=600次RPC,开盘时段QMT忙时47min+不归;
provider无get_security_info_batch→filters._batch_security_info回退逐只(风暴路径)。

修法:
- SanguoMiniQmtProvider.get_security_info_batch:委托内部LocalUnifiedProvider
  (2条本地SQL替N×2次RPC,与78d35ba get_closes_panel/get_constituent_ex同款
  委托模式);失败/非dict返None→filters自动回退逐只(不崩策略);全部策略
  (aw/mom/sc/vs含_ex)经共享filters即时受益,影子/回测同速
- momentum/_ex+small_cap/_ex 选股段加阶段耗时日志(池预取/宽表/过滤3段,
  下个9:30若仍有卡点日志直接钉死位置)
- 遗留已报issue:SanguoMiniQmtProvider无get_limit_status_batch→实盘涨跌停/
  停牌filter静默失效(涨停股照买),归数据session按本地公式+单次批量tick根治

测试:+5(委托/失败回退None/非dict回退/filters消费批量零逐只RPC×2,__new__
绕过Mac无bullet_trade的构造raise);portfolio 380绿(13失败全=Mac缺bullet_trade
环境性,NAS CI过)

[vps]

sanguo_portfolio/providers/sanguo_fundamentals.py

@@ -162,6 +162,30 @@ class SanguoMiniQmtProvider(MiniQMTProvider):  # type: ignore[misc]
         """TET 别名:与 ``get_index_stocks`` 同一实现(_ex 策略副本调用名)。"""
         return self.get_index_stocks(index, date)
 
+    def get_security_info_batch(
+        self,
+        securities: List[str],
+        date: Optional[Union[str, datetime]] = None,
+    ) -> Optional[Dict[str, Dict[str, Any]]]:
+        """批量证券信息:委托本地库 2 条 SQL,替 N×2 次逐只 xtdata RPC。
+
+        2026-08-21 实盘选股挂死根因:filters.filter_st/filter_new 逐只
+        ``get_security_info``,每次 2 个 xtdata RPC(_detect_instrument_type +
+        get_instrument_detail)且无超时——开盘时段 QMT 忙,momentum 600 次 RPC
+        47min+ 不归(live_19/shadow_60 同卡,离场时段同路径全程 27s)。
+        filters._batch_security_info 探测到本方法即一次批量预取。
+
+        委托失败/返非 dict → 返回 None,调用方(filters)自动回退逐只旧路径。
+        ⚠️ display_name 取 constituent_unified 历史名,极端滞后(近期刚戴帽
+        ST)可能漏过滤——沿用 filters「宁错过不误杀」口径,可接受。
+        """
+        try:
+            out = self._unified.get_security_info_batch(securities, date)
+        except Exception as exc:  # noqa: BLE001 - 委托失败回退逐只,不崩策略
+            logger.warning("get_security_info_batch 委托本地库失败,回退逐只: %s", exc)
+            return None
+        return out if isinstance(out, dict) else None
+
     # ------------------------ 主入口 ------------------------
     def get_fundamentals(
         self,

sanguo_portfolio/strategies/momentum_timing.py

@@ -24,6 +24,7 @@ from __future__ import annotations
 
 import datetime
 import logging
+import time
 from dataclasses import dataclass, field
 from typing import Any, List, Optional
 
@@ -168,10 +169,17 @@ class MomentumTimingStrategy:
         #     不吞异常退化成"空目标→全清仓"(2026-08-19 假熊市同型事故)
         try:
             # 牛市:当日预取十行业成份股池 close 宽表(层2向量化,每日 1 次批量 SQL)
+            # 阶段耗时日志(2026-08-21 实盘选股挂死定位用:9:30 buy_sign 后 47min
+            # 无下文,靠分段日志下个开盘直接钉死卡点)
+            _t0 = time.monotonic()
             union_stocks: List[str] = []
             for each_index in cfg.index_list:
                 union_stocks.extend(self._stock_pool_cached(each_index, cur_date))
+            logger.info("[%s] 选股1/3 池预取完成: %d 只 %.1fs",
+                        cur_date, len(union_stocks), time.monotonic() - _t0)
             self._ensure_day_panel(union_stocks, cur_date)
+            logger.info("[%s] 选股2/3 close宽表预取完成 %.1fs",
+                        cur_date, time.monotonic() - _t0)
 
             # 取强舍弱(每行业 RPS top_k 并集) → 候选池
             candidates = self._find_stock_pool(cfg.index_list, cur_date, pre_date)
@@ -199,6 +207,8 @@ class MomentumTimingStrategy:
                 stocks, self.provider, status_map=status_map,
             )
             stocks = _dedup(stocks)
+            logger.info("[%s] 选股3/3 过滤完成: %d 只候选 %.1fs",
+                        cur_date, len(stocks), time.monotonic() - _t0)
         except Exception as exc:
             logger.warning(
                 "[%s] 选股数据失败,跳过当日调仓(持仓不动,不清仓): %s", cur_date, exc,

sanguo_portfolio/strategies/momentum_timing_ex.py

@@ -27,6 +27,7 @@ from __future__ import annotations
 
 import datetime
 import logging
+import time
 from dataclasses import dataclass, field
 from typing import Any, List, Optional
 
@@ -163,8 +164,12 @@ class MomentumTimingExStrategy:
         # 2~5) 牛市选股:任一取数失败 → 跳过当日调仓(持仓不动),
         #     不吞异常退化成"空目标→全清仓"(2026-08-19 假熊市同型事故)
         try:
+            # 阶段耗时日志(2026-08-21 实盘选股挂死定位用,与主文件同款)
+            _t0 = time.monotonic()
             # 牛市:取强舍弱(每行业 RPS top_k 并集) → 候选池
             candidates = self._find_stock_pool(cfg.index_list, cur_date, pre_date)
+            logger.info("[%s] 选股1/2 候选池(RPS)完成: %d 只 %.1fs",
+                        cur_date, len(candidates), time.monotonic() - _t0)
 
             # 均线动量过滤(close > MA_short > MA_long)
             stocks = self._select_stocks(candidates, cur_date)
@@ -189,6 +194,8 @@ class MomentumTimingExStrategy:
                 stocks, self.provider, status_map=status_map,
             )
             stocks = _dedup(stocks)
+            logger.info("[%s] 选股2/2 过滤完成: %d 只候选 %.1fs",
+                        cur_date, len(stocks), time.monotonic() - _t0)
         except Exception as exc:
             logger.warning(
                 "[%s] 选股数据失败,跳过当日调仓(持仓不动,不清仓): %s", cur_date, exc,

sanguo_portfolio/strategies/small_cap.py

@@ -30,6 +30,7 @@ from __future__ import annotations
 
 import datetime
 import logging
+import time
 from dataclasses import dataclass
 from typing import Any, List, Optional
 
@@ -198,8 +199,12 @@ class SmallCapStrategy:
             logger.warning("pick_stocks: previous_date 为 None,跳过本次选股")
             return None
 
+        # 阶段耗时日志(2026-08-21 实盘选股挂死定位用)
+        _t0 = time.monotonic()
         # 1) 全市场候选池(universe 成份股)
         candidates = self._stock_pool(cfg.universe, previous_date)
+        logger.info("[%s] 选股1/3 候选池完成: %d 只 %.1fs",
+                    previous_date, len(candidates or []), time.monotonic() - _t0)
         if candidates is None:
             return None
         if not candidates:
@@ -240,6 +245,9 @@ class SmallCapStrategy:
         if not top_candidates:
             return []
 
+        logger.info("[%s] 选股2/3 fundamentals 完成: top %d 只 %.1fs",
+                    previous_date, len(top_candidates), time.monotonic() - _t0)
+
         # 5) 过滤次新股(原策略上市 > 120 天)
         top_candidates = filters.filter_new_stock(
             top_candidates, self.provider, previous_date, cfg.new_stock_days,
@@ -274,6 +282,8 @@ class SmallCapStrategy:
 
         # 8) 取前 buy_stock_count
         out = list(scored.index)[: cfg.buy_stock_count]
+        logger.info("[%s] 选股3/3 动量评分完成: %d 只 %.1fs",
+                    previous_date, len(out), time.monotonic() - _t0)
         return out
 
     # =================== 动量评分 ===================

sanguo_portfolio/strategies/small_cap_ex.py

@@ -33,6 +33,7 @@ from __future__ import annotations
 
 import datetime
 import logging
+import time
 from dataclasses import dataclass
 from typing import Any, List, Optional
 
@@ -201,8 +202,12 @@ class SmallCapExStrategy:
             logger.warning("pick_stocks: previous_date 为 None,跳过本次选股")
             return None
 
+        # 阶段耗时日志(2026-08-21 实盘选股挂死定位用)
+        _t0 = time.monotonic()
         # 1) 全市场候选池(universe 成份股)
         candidates = self._stock_pool(cfg.universe, previous_date)
+        logger.info("[%s] 选股1/3 候选池完成: %d 只 %.1fs",
+                    previous_date, len(candidates or []), time.monotonic() - _t0)
         if candidates is None:
             return None
         if not candidates:
@@ -243,6 +248,9 @@ class SmallCapExStrategy:
         if not top_candidates:
             return []
 
+        logger.info("[%s] 选股2/3 fundamentals 完成: top %d 只 %.1fs",
+                    previous_date, len(top_candidates), time.monotonic() - _t0)
+
         # 5) 过滤次新股(原策略上市 > 120 天)
         top_candidates = filters.filter_new_stock(
             top_candidates, self.provider, previous_date, cfg.new_stock_days,
@@ -277,6 +285,8 @@ class SmallCapExStrategy:
 
         # 8) 取前 buy_stock_count
         out = list(scored.index)[: cfg.buy_stock_count]
+        logger.info("[%s] 选股3/3 动量评分完成: %d 只 %.1fs",
+                    previous_date, len(out), time.monotonic() - _t0)
         return out
 
     # =================== 动量评分 ===================

tests/portfolio/test_provider_secinfo_batch.py

@@ -0,0 +1,79 @@
+"""SanguoMiniQmtProvider.get_security_info_batch 委托测试(2026-08-21 挂死根治)。
+
+背景:实盘 momentum/small_cap 9:30 选股挂死——filters.filter_st/filter_new 逐只
+get_security_info,每只 2 次 xtdata RPC 无超时,开盘时段 600 次 RPC 47min+ 不归。
+本方法委托内部 LocalUnifiedProvider 批量(2 条本地 SQL),filters 探测到即走批量。
+
+Mac 无 bullet_trade:类定义处 MiniQMTProvider=object 兜底,可直接实例化;
+_unified 用 MagicMock 替换,只测委托语义(不连本地库)。
+"""
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+from sanguo_portfolio import SanguoMiniQmtProvider
+from sanguo_portfolio import filters
+
+
+def _provider(unified: MagicMock) -> SanguoMiniQmtProvider:
+    # __new__ 绕过 __init__:Mac 无 bullet_trade 时构造直接 raise
+    # (RuntimeError);被测方法只用 self._unified,委托语义不依赖基类初始化。
+    p = object.__new__(SanguoMiniQmtProvider)
+    p._unified = unified
+    return p
+
+
+class TestGetSecurityInfoBatch:
+    def test_delegates_to_unified(self):
+        unified = MagicMock(name="unified")
+        expected = {
+            "600519.XSHG": {"display_name": "贵州茅台", "start_date": "2001-08-27"},
+            "600036.XSHG": {"display_name": "招商银行", "start_date": "2002-04-09"},
+        }
+        unified.get_security_info_batch.return_value = expected
+        p = _provider(unified)
+        out = p.get_security_info_batch(["600519.XSHG", "600036.XSHG"])
+        assert out == expected
+        unified.get_security_info_batch.assert_called_once_with(
+            ["600519.XSHG", "600036.XSHG"], None)
+
+    def test_unified_failure_returns_none_not_raise(self):
+        """委托失败 → None(不崩策略),filters 自动回退逐只旧路径。"""
+        unified = MagicMock(name="unified")
+        unified.get_security_info_batch.side_effect = RuntimeError("db locked")
+        p = _provider(unified)
+        assert p.get_security_info_batch(["600519.XSHG"]) is None
+
+    def test_non_dict_result_returns_none(self):
+        unified = MagicMock(name="unified")
+        unified.get_security_info_batch.return_value = ["not", "a", "dict"]
+        p = _provider(unified)
+        assert p.get_security_info_batch(["600519.XSHG"]) is None
+
+
+class TestFiltersConsumeBatch:
+    def test_filter_st_uses_batch_and_skips_per_stock_rpc(self):
+        """filters 探测到批量方法 → 一次预取,绝不再逐只 get_security_info
+        (即挂死根因的 RPC 风暴路径)。"""
+        provider = MagicMock(name="provider")
+        provider.get_security_info_batch.return_value = {
+            "600519.XSHG": {"display_name": "贵州茅台", "start_date": "2001-08-27"},
+            "600000.XSHG": {"display_name": "ST浦发", "start_date": "1999-11-10"},
+        }
+        out = filters.filter_st_stock(["600519.XSHG", "600000.XSHG"], provider)
+        assert out == ["600519.XSHG"]  # ST 被滤掉
+        provider.get_security_info_batch.assert_called_once()
+        provider.get_security_info.assert_not_called()  # 零逐只 RPC
+
+    def test_filter_new_stock_uses_batch(self):
+        provider = MagicMock(name="provider")
+        provider.get_security_info_batch.return_value = {
+            "600519.XSHG": {"display_name": "贵州茅台",
+                            "start_date": "2001-08-27"},
+            "301999.XSHE": {"display_name": "次新股",
+                            "start_date": "2026-08-01"},
+        }
+        out = filters.filter_new_stock(
+            ["600519.XSHG", "301999.XSHE"], provider, "2026-08-21", days=375)
+        assert out == ["600519.XSHG"]  # 上市 20 天的次新被滤掉
+        provider.get_security_info.assert_not_called()