refactor(strategy): value空名单守卫打磨——指标层/过滤清空分流日志+Ex副本parametrize同测+死代码清 [vps] [no-doc]

10-01 二次审计 P2-4/P2-5 处置:守卫日志两支分流(指标层空=铁定数据故障
vs 6条过滤后全被涨跌停/停牌滤除=状态数据可疑或极端行情待人工研判),
行为一致均跳过调仓保仓;正身+Ex 副本 parametrize 同测(test_sizing 同款
工法, 3 用例×2 类);step5 的 target_num==0 死代码双文件删(守卫早退后
不可达)。

Co-Authored-By: Claude Code <noreply@anthropic.com>

sanguo_portfolio/strategies/value_selection.py

@@ -173,6 +173,7 @@ class ValueSelectionStrategy:
         # 3) 过滤涨停/跌停/停牌(复用 sanguo_portfolio.filters)
         # 批量预取当日涨跌停/停牌状态(数据 session 判断好),三个 filter 共享一次查询
         positions = _get_positions(context, self.broker)
+        raw_list = list(buy_list)  # 指标过滤后快照(涨跌停/停牌过滤前, 守卫分流用)
         status_map = self._get_limit_status(buy_list, previous_date)
         buy_list = filters.filter_limitup_stock(
             buy_list, self.provider,
@@ -188,14 +189,21 @@ class ValueSelectionStrategy:
         buy_list = _dedup(buy_list)
 
         # 空名单守卫(09-01 全清仓案根修;spec 宪法:全清仓=人选的档 6 防御模式,
-        # 不被空名单隐式触发):候选池非空而过滤后归零=数据面故障形态(健康数据下
-        # 6 条过滤交集不可能为空),原流程会走进「卖出不在名单」循环清光全部持仓
-        # ——跳过本轮调仓保持现仓,logger.error 落监控 ERROR 口径可见。
+        # 不被空名单隐式触发)。两支分流日志(10-01 二次审计 P2-5):指标层空=铁定
+        # 数据故障;被涨跌停/停牌过滤清空=状态数据可疑或极端行情(可能真信号,待
+        # 人工研判)——行为一致:跳过本轮调仓保持现仓,logger.error 落监控口径可见。
         if not buy_list:
-            logger.error(
-                "[%s] 空名单守卫: 过滤后 buy_list=0(候选 %d)——判数据异常,"
-                "本轮调仓跳过, 保持 %d 只持仓不清仓",
-                previous_date, len(candidates), len(positions))
+            if raw_list:
+                logger.error(
+                    "[%s] 空名单守卫(过滤清空): 6条过滤后 %d 只全被涨跌停/停牌"
+                    "滤除——状态数据可疑或极端行情待人工研判, 跳过本轮调仓, "
+                    "保持 %d 只持仓不清仓", previous_date, len(raw_list),
+                    len(positions))
+            else:
+                logger.error(
+                    "[%s] 空名单守卫(指标层空): 6条过滤后归零(候选 %d)——判数据"
+                    "异常, 跳过本轮调仓, 保持 %d 只持仓不清仓", previous_date,
+                    len(candidates), len(positions))
             return
 
         # 4) 调仓:卖出不在 buy_list 的(原策略 sell 函数)
@@ -206,9 +214,7 @@ class ValueSelectionStrategy:
 
         # 5) 等额买入 buy_list 里的新股(原策略 buy 函数, cash/countStocks 语义)
         positions = _get_positions(context, self.broker)  # 卖出后刷新
-        target_num = len(buy_list)
-        if target_num == 0:
-            return
+        target_num = len(buy_list)  # 守卫保证非空(空名单早退)
         total = _total_value(context)
         if total <= 0:
             return

sanguo_portfolio/strategies/value_selection_ex.py

@@ -175,6 +175,7 @@ class ValueSelectionExStrategy:
         # 3) 过滤涨停/跌停/停牌(复用 sanguo_portfolio.filters)
         # 批量预取当日涨跌停/停牌状态(数据 session 判断好),三个 filter 共享一次查询
         positions = _get_positions(context, self.broker)
+        raw_list = list(buy_list)  # 指标过滤后快照(涨跌停/停牌过滤前, 守卫分流用)
         status_map = self._get_limit_status(buy_list, previous_date)
         buy_list = filters.filter_limitup_stock(
             buy_list, self.provider,
@@ -190,14 +191,21 @@ class ValueSelectionExStrategy:
         buy_list = _dedup(buy_list)
 
         # 空名单守卫(09-01 全清仓案根修;spec 宪法:全清仓=人选的档 6 防御模式,
-        # 不被空名单隐式触发):候选池非空而过滤后归零=数据面故障形态(健康数据下
-        # 6 条过滤交集不可能为空),原流程会走进「卖出不在名单」循环清光全部持仓
-        # ——跳过本轮调仓保持现仓,logger.error 落监控 ERROR 口径可见。
+        # 不被空名单隐式触发)。两支分流日志(10-01 二次审计 P2-5):指标层空=铁定
+        # 数据故障;被涨跌停/停牌过滤清空=状态数据可疑或极端行情(可能真信号,待
+        # 人工研判)——行为一致:跳过本轮调仓保持现仓,logger.error 落监控口径可见。
         if not buy_list:
-            logger.error(
-                "[%s] 空名单守卫: 过滤后 buy_list=0(候选 %d)——判数据异常,"
-                "本轮调仓跳过, 保持 %d 只持仓不清仓",
-                previous_date, len(candidates), len(positions))
+            if raw_list:
+                logger.error(
+                    "[%s] 空名单守卫(过滤清空): 6条过滤后 %d 只全被涨跌停/停牌"
+                    "滤除——状态数据可疑或极端行情待人工研判, 跳过本轮调仓, "
+                    "保持 %d 只持仓不清仓", previous_date, len(raw_list),
+                    len(positions))
+            else:
+                logger.error(
+                    "[%s] 空名单守卫(指标层空): 6条过滤后归零(候选 %d)——判数据"
+                    "异常, 跳过本轮调仓, 保持 %d 只持仓不清仓", previous_date,
+                    len(candidates), len(positions))
             return
 
         # 4) 调仓:卖出不在 buy_list 的(原策略 sell 函数)
@@ -210,9 +218,7 @@ class ValueSelectionExStrategy:
         #    2026-08-25 A 修法:对「卖出已成交、台账归因未跑」窗口免疫,
         #    守卫同步 total<=0)
         positions = _get_positions(context, self.broker)  # 卖出后刷新
-        target_num = len(buy_list)
-        if target_num == 0:
-            return
+        target_num = len(buy_list)  # 守卫保证非空(空名单早退)
         total = _total_value(context)
         if total <= 0:
             return

tests/portfolio/test_value_selection.py

@@ -25,6 +25,9 @@ from sanguo_portfolio.strategies.value_selection import (
     ValueSelectionConfig,
     ValueSelectionStrategy,
 )
+from sanguo_portfolio.strategies.value_selection_ex import (
+    ValueSelectionExStrategy,
+)
 from tests.portfolio.conftest import FakeContext, FakePosition
 
 
@@ -553,16 +556,24 @@ class TestConfigDefaults:
 
 # =================== 空名单守卫(09-01 全清仓案根修) ===================
 class TestEmptyBuyListGuard:
-    """候选池非空+过滤后归零=数据故障形态 → 不清仓不买入,保持现仓。
+    """候选池非空+过滤后归零 → 不清仓不买入,保持现仓。
 
-    spec 宪法:全清仓=人选的档 6 防御模式,不被空名单隐式触发。
+    spec 宪法:全清仓=人选的档 6 防御模式,不被空名单隐式触发;
+    正身+Ex 副本 parametrize 同测(10-01 二次审计 P2-4,test_sizing 同款工法)。
     """
 
-    @staticmethod
-    def _strategy(stock_pool, stock_list):
-        s = make_strategy()
-        s._stock_pool = lambda *a, **k: stock_pool
-        s._get_stock_list = lambda *a, **k: stock_list
+    CLASSES = [ValueSelectionStrategy, ValueSelectionExStrategy]
+
+    @classmethod
+    def _make(cls, strategy_cls, limit_up=False):
+        provider = MagicMock(name="provider")
+        provider.get_limit_status_batch.side_effect = lambda codes, date=None: {
+            c: {"is_limit_up": limit_up, "is_limit_down": False,
+                "is_paused": False} for c in codes}
+        broker = BrokerFacade()
+        broker.order_target_value = MagicMock(return_value=MagicMock(filled=100))
+        broker.order_value = MagicMock(return_value=MagicMock(filled=100))
+        s = strategy_cls(provider=provider, broker=broker)
         return s
 
     @staticmethod
@@ -572,18 +583,36 @@ class TestEmptyBuyListGuard:
             positions={"600519.XSHG": FakePosition("600519.XSHG", 1600.0, 1500.0),
                        "000661.XSHE": FakePosition("000661.XSHE", 30.0, 28.0)})
 
-    def test_empty_buy_list_keeps_positions_and_logs_error(self, caplog):
-        """过滤后空名单:零下单零卖出,ERROR 落监控口径。"""
-        s = self._strategy(["600519.XSHG", "000661.XSHE"], [])
+    @pytest.mark.parametrize("cls", CLASSES, ids=["main", "ex"])
+    def test_metrics_empty_keeps_positions_and_logs_error(self, cls, caplog):
+        """指标层空=铁定数据故障:零下单零卖出,ERROR 落监控口径。"""
+        s = self._make(cls)
+        s._stock_pool = lambda *a, **k: ["600519.XSHG", "000661.XSHE"]
+        s._get_stock_list = lambda *a, **k: []
         with caplog.at_level("ERROR"):
             s.monthly_adjustment(self._ctx())
         s.broker.order_target_value.assert_not_called()
         s.broker.order_value.assert_not_called()
-        assert any("空名单守卫" in r.message for r in caplog.records)
+        assert any("空名单守卫(指标层空)" in r.message for r in caplog.records)
 
-    def test_nonempty_buy_list_still_rebalances(self):
+    @pytest.mark.parametrize("cls", CLASSES, ids=["main", "ex"])
+    def test_filters_empty_keeps_positions_logs_triage(self, cls, caplog):
+        """指标健康但被涨跌停滤空=分流日志待人工研判(P2-5);行为同守卫。
+        候选取未持有股(持仓中的涨停股不过滤,filters 既有语义)。"""
+        s = self._make(cls, limit_up=True)
+        s._stock_pool = lambda *a, **k: ["300750.XSHE"]
+        s._get_stock_list = lambda *a, **k: ["300750.XSHE"]
+        with caplog.at_level("ERROR"):
+            s.monthly_adjustment(self._ctx())
+        s.broker.order_target_value.assert_not_called()
+        assert any("空名单守卫(过滤清空)" in r.message for r in caplog.records)
+
+    @pytest.mark.parametrize("cls", CLASSES, ids=["main", "ex"])
+    def test_nonempty_buy_list_still_rebalances(self, cls):
         """对照:名单正常时守卫不拦——不在名单的持仓照卖(正常轮换不变)。"""
-        s = self._strategy(["000661.XSHE"], ["000661.XSHE"])
+        s = self._make(cls)
+        s._stock_pool = lambda *a, **k: ["000661.XSHE"]
+        s._get_stock_list = lambda *a, **k: ["000661.XSHE"]
         s.monthly_adjustment(self._ctx())
         sells = [c for c in s.broker.order_target_value.call_args_list
                  if c.args[1] == 0]