fix(strategy): 数据取数失败≠策略信号——momentum假熊市清仓根治+small_cap同型误判纠正 [vps]

前后端session 2026-08-19巡检实锤(VPS shadow_47/48,8-18/8-19连续两日09:30):
momentum _cal_buy_sign 取数失败被静默当成熊市信号,handle_data if not buy_sign 全部
清仓;有持仓时任何一次数据抖动=全仓卖出。

①momentum/_ex: _cal_buy_sign→Optional[bool](取数异常/空panel→None,handle_data
  据此跳过当日调仓保持仓);_cal_rps/_select_stocks/_stock_pool 吞异常改上抛,牛市
  计算段统一捕获→跳过当日(不清仓);False只留给真实数据算出的熊市,空index_list
  维持False(确定性配置态)
②small_cap/_ex(新发现,前后端记忆判'语义安全'系误判——8-18无交易只因空仓):
  _pick_stocks任一取数失败→[]→_rebalance空名单分支全清仓,与momentum同型事故;
  改三态:None=数据失败跳过本次调仓(持仓不动),[]=合法空名单仍清仓(原策略语义),
  list=正常目标;_stock_pool/_cal_momentum_score同步Optional化
③+7回归测试(取数失败零下单保持仓×4/直查None×2/合法空仍清仓守卫);portfolio 324绿,
  全量889绿(4失败=Mac缺bullet_trade/vnpy环境性import,与本次无关)

sanguo_portfolio/strategies/momentum_timing.py

@@ -147,49 +147,63 @@ class MomentumTimingStrategy:
         # 1) 牛熊分界(当日预取:先只拉指数点位;熊市日不付股票池全量 IO,同原行为)
         self._ensure_day_panel(cfg.index_list, cur_date)
         buy_sign = self._cal_buy_sign(cfg.index_list, cfg.past_day, cur_date)
+        if buy_sign is None:
+            # 数据失败 ≠ 熊市(2026-08-19 假熊市清仓事故):跳过当日调仓保住持仓
+            logger.warning(
+                "[%s] 牛熊分界数据不可用,跳过当日调仓(持仓不动,不清仓)", cur_date,
+            )
+            return
         logger.info("[%s] buy_sign=%s", cur_date, buy_sign)
 
         positions = _get_positions(context)
 
         if not buy_sign:
-            # 熊市:全部清仓(原策略语义)
+            # 熊市:全部清仓(原策略语义,只对"用真实数据算出的熊市")
             logger.info("[%s] 熊市信号,清仓 %d 只", cur_date, len(positions))
             for stock in list(positions.keys()):
                 self._close_position(stock)
             return
 
-        # 2) 牛市:当日预取十行业成份股池 close 宽表(层2向量化,每日 1 次批量 SQL)
-        union_stocks: List[str] = []
-        for each_index in cfg.index_list:
-            union_stocks.extend(self._stock_pool_cached(each_index, cur_date))
-        self._ensure_day_panel(union_stocks, cur_date)
+        # 2~5) 牛市选股:任一取数失败 → 跳过当日调仓(持仓不动),
+        #     不吞异常退化成"空目标→全清仓"(2026-08-19 假熊市同型事故)
+        try:
+            # 牛市:当日预取十行业成份股池 close 宽表(层2向量化,每日 1 次批量 SQL)
+            union_stocks: List[str] = []
+            for each_index in cfg.index_list:
+                union_stocks.extend(self._stock_pool_cached(each_index, cur_date))
+            self._ensure_day_panel(union_stocks, cur_date)
 
-        # 3) 取强舍弱(每行业 RPS top_k 并集) → 候选池
-        candidates = self._find_stock_pool(cfg.index_list, cur_date, pre_date)
+            # 取强舍弱(每行业 RPS top_k 并集) → 候选池
+            candidates = self._find_stock_pool(cfg.index_list, cur_date, pre_date)
 
-        # 3) 均线动量过滤(close > MA_short > MA_long)
-        stocks = self._select_stocks(candidates, cur_date)
+            # 均线动量过滤(close > MA_short > MA_long)
+            stocks = self._select_stocks(candidates, cur_date)
 
-        # 4) 候选过多时再按 RPS 取前 top_k (原策略 handle_data 第 171-175 行)
-        if len(stocks) > cfg.top_k:
-            rps_df = self._cal_rps(stocks, cur_date, pre_date)
-            stocks = list(rps_df["code"])[: cfg.top_k]
+            # 候选过多时再按 RPS 取前 top_k (原策略 handle_data 第 171-175 行)
+            if len(stocks) > cfg.top_k:
+                rps_df = self._cal_rps(stocks, cur_date, pre_date)
+                stocks = list(rps_df["code"])[: cfg.top_k]
 
-        # 5) 过滤涨停/跌停/停牌(复用 sanguo_portfolio.filters)
-        # 批量预取当日涨跌停/停牌状态(数据 session 判断好),三个 filter 共享一次查询
-        status_map = self._get_limit_status(stocks, cur_date)
-        stocks = filters.filter_limitup_stock(
-            stocks, self.provider,
-            positions=list(positions.keys()), status_map=status_map,
-        )
-        stocks = filters.filter_limitdown_stock(
-            stocks, self.provider,
-            positions=list(positions.keys()), status_map=status_map,
-        )
-        stocks = filters.filter_paused_stock(
-            stocks, self.provider, status_map=status_map,
-        )
-        stocks = _dedup(stocks)
+            # 过滤涨停/跌停/停牌(复用 sanguo_portfolio.filters)
+            # 批量预取当日涨跌停/停牌状态(数据 session 判断好),三个 filter 共享一次查询
+            status_map = self._get_limit_status(stocks, cur_date)
+            stocks = filters.filter_limitup_stock(
+                stocks, self.provider,
+                positions=list(positions.keys()), status_map=status_map,
+            )
+            stocks = filters.filter_limitdown_stock(
+                stocks, self.provider,
+                positions=list(positions.keys()), status_map=status_map,
+            )
+            stocks = filters.filter_paused_stock(
+                stocks, self.provider, status_map=status_map,
+            )
+            stocks = _dedup(stocks)
+        except Exception as exc:
+            logger.warning(
+                "[%s] 选股数据失败,跳过当日调仓(持仓不动,不清仓): %s", cur_date, exc,
+            )
+            return
 
         # 6) 调仓:先清掉不在 stocks 的
         for stock in list(positions.keys()):
@@ -309,13 +323,11 @@ class MomentumTimingStrategy:
             if panel.empty or len(panel) < 2:
                 return pd.DataFrame({"code": [], "rps_value": []})
         else:
-            try:
-                panel = self.provider.get_closes_panel(
-                    stocks, pre_date, cur_date, fq="raw",
-                )
-            except Exception as exc:
-                logger.warning("_cal_rps get_closes_panel 失败: %s", exc)
-                return pd.DataFrame({"code": [], "rps_value": []})
+            # 取数异常直接上抛(handle_data 捕获后跳过当日调仓):
+            # 吞掉返回空表会退化成"空目标→全清仓"(2026-08-19 假熊市同型事故)
+            panel = self.provider.get_closes_panel(
+                stocks, pre_date, cur_date, fq="raw",
+            )
             if panel is None or panel.empty or len(panel) < 2:
                 return pd.DataFrame({"code": [], "rps_value": []})
 
@@ -374,13 +386,10 @@ class MomentumTimingStrategy:
         start_date = _shift_date(cur_date, -cfg.ma_long * 2)
         panel = self._panel_slice(stocks, start_date, cur_date)
         if panel is None:
-            try:
-                panel = self.provider.get_closes_panel(
-                    stocks, start_date, cur_date, fq="raw",
-                )
-            except Exception as exc:
-                logger.warning("_select_stocks get_closes_panel 失败: %s", exc)
-                return []
+            # 取数异常直接上抛(handle_data 捕获后跳过当日调仓),不吞成空表
+            panel = self.provider.get_closes_panel(
+                stocks, start_date, cur_date, fq="raw",
+            )
         if panel is None or panel.empty:
             return []
         panel = panel.tail(cfg.ma_long)
@@ -408,7 +417,7 @@ class MomentumTimingStrategy:
         index_list: List[str],
         past_day: int,
         cur_date: str,
-    ) -> bool:
+    ) -> Optional[bool]:
         """统计 past_day 均线上方的指数占比 > index_thre → 牛市(True)。
 
         原策略 'index' 模式(第 110-115 行):对每个指数算 ``mavg(past_day,'close')``
@@ -421,6 +430,10 @@ class MomentumTimingStrategy:
 
         ⚠️ 原代码 ``float(count)/len(indexList)`` 在 py2 是浮点除法(因 float()强转),
         与 py3 一致。这里保留浮点除法语义。
+
+        ⚠️ 数据失败 ≠ 熊市(2026-08-19 假熊市清仓事故):取数异常/查无数据返回
+        ``None``,handle_data 据此跳过当日调仓;``False`` 只留给"用真实数据算出
+        的熊市"。空 ``index_list`` 是确定性配置态,维持 ``False`` 原语义。
         """
         cfg = self.config
         if not index_list:
@@ -434,12 +447,12 @@ class MomentumTimingStrategy:
                 )
             except Exception as exc:
                 logger.warning("_cal_buy_sign get_closes_panel 失败: %s", exc)
-                return False
+                return None
         if panel is None or panel.empty:
-            return False
+            return None
         panel = panel.tail(past_day)
         if panel.empty:
-            return False
+            return None
 
         # 向量化:对每列算 cur_close 与 past_day 均值,统计 close > ma_past 的占比
         valid_count = panel.notna().sum()
@@ -482,12 +495,12 @@ class MomentumTimingStrategy:
             return {}
 
     def _stock_pool(self, index_symbol: str, cur_date: str) -> List[str]:
-        """成分股 + 过滤 ST/科创北交/次新。"""
-        try:
-            stocks = self.provider.get_index_stocks(index_symbol, cur_date)
-        except Exception as exc:
-            logger.warning("get_index_stocks(%s) 失败: %s", index_symbol, exc)
-            return []
+        """成分股 + 过滤 ST/科创北交/次新。
+
+        取数异常直接上抛(handle_data 捕获后跳过当日调仓):吞成空表会让
+        全部行业候选为空 → 空目标 → 全清仓(2026-08-19 假熊市同型事故)。
+        """
+        stocks = self.provider.get_index_stocks(index_symbol, cur_date)
         stocks = filters.filter_kcbj_stock(stocks)
         if self.config.max_pool > 0:
             stocks = stocks[: self.config.max_pool]

sanguo_portfolio/strategies/momentum_timing_ex.py

@@ -143,43 +143,57 @@ class MomentumTimingExStrategy:
 
         # 1) 牛熊分界
         buy_sign = self._cal_buy_sign(cfg.index_list, cfg.past_day, cur_date)
+        if buy_sign is None:
+            # 数据失败 ≠ 熊市(2026-08-19 假熊市清仓事故):跳过当日调仓保住持仓
+            logger.warning(
+                "[%s] 牛熊分界数据不可用,跳过当日调仓(持仓不动,不清仓)", cur_date,
+            )
+            return
         logger.info("[%s] buy_sign=%s", cur_date, buy_sign)
 
         positions = _get_positions(context)
 
         if not buy_sign:
-            # 熊市:全部清仓(原策略语义)
+            # 熊市:全部清仓(原策略语义,只对"用真实数据算出的熊市")
             logger.info("[%s] 熊市信号,清仓 %d 只", cur_date, len(positions))
             for stock in list(positions.keys()):
                 self._close_position(stock)
             return
 
-        # 2) 牛市:取强舍弱(每行业 RPS top_k 并集) → 候选池
-        candidates = self._find_stock_pool(cfg.index_list, cur_date, pre_date)
+        # 2~5) 牛市选股:任一取数失败 → 跳过当日调仓(持仓不动),
+        #     不吞异常退化成"空目标→全清仓"(2026-08-19 假熊市同型事故)
+        try:
+            # 牛市:取强舍弱(每行业 RPS top_k 并集) → 候选池
+            candidates = self._find_stock_pool(cfg.index_list, cur_date, pre_date)
 
-        # 3) 均线动量过滤(close > MA_short > MA_long)
-        stocks = self._select_stocks(candidates, cur_date)
+            # 均线动量过滤(close > MA_short > MA_long)
+            stocks = self._select_stocks(candidates, cur_date)
 
-        # 4) 候选过多时再按 RPS 取前 top_k (原策略 handle_data 第 171-175 行)
-        if len(stocks) > cfg.top_k:
-            rps_df = self._cal_rps(stocks, cur_date, pre_date)
-            stocks = list(rps_df["code"])[: cfg.top_k]
+            # 候选过多时再按 RPS 取前 top_k (原策略 handle_data 第 171-175 行)
+            if len(stocks) > cfg.top_k:
+                rps_df = self._cal_rps(stocks, cur_date, pre_date)
+                stocks = list(rps_df["code"])[: cfg.top_k]
 
-        # 5) 过滤涨停/跌停/停牌(复用 sanguo_portfolio.filters)
-        # 批量预取当日涨跌停/停牌状态(数据 session 判断好),三个 filter 共享一次查询
-        status_map = self._get_limit_status(stocks, cur_date)
-        stocks = filters.filter_limitup_stock(
-            stocks, self.provider,
-            positions=list(positions.keys()), status_map=status_map,
-        )
-        stocks = filters.filter_limitdown_stock(
-            stocks, self.provider,
-            positions=list(positions.keys()), status_map=status_map,
-        )
-        stocks = filters.filter_paused_stock(
-            stocks, self.provider, status_map=status_map,
-        )
-        stocks = _dedup(stocks)
+            # 过滤涨停/跌停/停牌(复用 sanguo_portfolio.filters)
+            # 批量预取当日涨跌停/停牌状态(数据 session 判断好),三个 filter 共享一次查询
+            status_map = self._get_limit_status(stocks, cur_date)
+            stocks = filters.filter_limitup_stock(
+                stocks, self.provider,
+                positions=list(positions.keys()), status_map=status_map,
+            )
+            stocks = filters.filter_limitdown_stock(
+                stocks, self.provider,
+                positions=list(positions.keys()), status_map=status_map,
+            )
+            stocks = filters.filter_paused_stock(
+                stocks, self.provider, status_map=status_map,
+            )
+            stocks = _dedup(stocks)
+        except Exception as exc:
+            logger.warning(
+                "[%s] 选股数据失败,跳过当日调仓(持仓不动,不清仓): %s", cur_date, exc,
+            )
+            return
 
         # 6) 调仓:先清掉不在 stocks 的
         for stock in list(positions.keys()):
@@ -228,13 +242,11 @@ class MomentumTimingExStrategy:
         n = len(stocks)
         if n == 0:
             return pd.DataFrame({"code": [], "rps_value": []})
-        try:
-            panel = self.provider.get_closes_panel_ex(
-                stocks, pre_date, cur_date, fq="raw",
-            )
-        except Exception as exc:
-            logger.warning("_cal_rps get_closes_panel 失败: %s", exc)
-            return pd.DataFrame({"code": [], "rps_value": []})
+        # 取数异常直接上抛(handle_data 捕获后跳过当日调仓):
+        # 吞掉返回空表会退化成"空目标→全清仓"(2026-08-19 假熊市同型事故)
+        panel = self.provider.get_closes_panel_ex(
+            stocks, pre_date, cur_date, fq="raw",
+        )
         if panel is None or panel.empty or len(panel) < 2:
             return pd.DataFrame({"code": [], "rps_value": []})
 
@@ -291,13 +303,10 @@ class MomentumTimingExStrategy:
         if not stocks:
             return []
         start_date = _shift_date(cur_date, -cfg.ma_long * 2)
-        try:
-            panel = self.provider.get_closes_panel_ex(
-                stocks, start_date, cur_date, fq="raw",
-            )
-        except Exception as exc:
-            logger.warning("_select_stocks get_closes_panel 失败: %s", exc)
-            return []
+        # 取数异常直接上抛(handle_data 捕获后跳过当日调仓),不吞成空表
+        panel = self.provider.get_closes_panel_ex(
+            stocks, start_date, cur_date, fq="raw",
+        )
         if panel is None or panel.empty:
             return []
         panel = panel.tail(cfg.ma_long)
@@ -325,7 +334,7 @@ class MomentumTimingExStrategy:
         index_list: List[str],
         past_day: int,
         cur_date: str,
-    ) -> bool:
+    ) -> Optional[bool]:
         """统计 past_day 均线上方的指数占比 > index_thre → 牛市(True)。
 
         原策略 'index' 模式(第 110-115 行):对每个指数算 ``mavg(past_day,'close')``
@@ -338,6 +347,10 @@ class MomentumTimingExStrategy:
 
         ⚠️ 原代码 ``float(count)/len(indexList)`` 在 py2 是浮点除法(因 float()强转),
         与 py3 一致。这里保留浮点除法语义。
+
+        ⚠️ 数据失败 ≠ 熊市(2026-08-19 假熊市清仓事故):取数异常/查无数据返回
+        ``None``,handle_data 据此跳过当日调仓;``False`` 只留给"用真实数据算出
+        的熊市"。空 ``index_list`` 是确定性配置态,维持 ``False`` 原语义。
         """
         cfg = self.config
         if not index_list:
@@ -349,12 +362,12 @@ class MomentumTimingExStrategy:
             )
         except Exception as exc:
             logger.warning("_cal_buy_sign get_closes_panel 失败: %s", exc)
-            return False
+            return None
         if panel is None or panel.empty:
-            return False
+            return None
         panel = panel.tail(past_day)
         if panel.empty:
-            return False
+            return None
 
         # 向量化:对每列算 cur_close 与 past_day 均值,统计 close > ma_past 的占比
         valid_count = panel.notna().sum()
@@ -397,12 +410,12 @@ class MomentumTimingExStrategy:
             return {}
 
     def _stock_pool(self, index_symbol: str, cur_date: str) -> List[str]:
-        """成分股 + 过滤 ST/科创北交/次新。"""
-        try:
-            stocks = self.provider.get_constituent_ex(index_symbol, cur_date)
-        except Exception as exc:
-            logger.warning("get_index_stocks(%s) 失败: %s", index_symbol, exc)
-            return []
+        """成分股 + 过滤 ST/科创北交/次新。
+
+        取数异常直接上抛(handle_data 捕获后跳过当日调仓):吞成空表会让
+        全部行业候选为空 → 空目标 → 全清仓(2026-08-19 假熊市同型事故)。
+        """
+        stocks = self.provider.get_constituent_ex(index_symbol, cur_date)
         stocks = filters.filter_kcbj_stock(stocks)
         if self.config.max_pool > 0:
             stocks = stocks[: self.config.max_pool]

sanguo_portfolio/strategies/small_cap.py

@@ -156,21 +156,29 @@ class SmallCapStrategy:
         )
 
         if is_rebalance_day:
-            # 2) 选股
+            # 2) 选股(None=数据失败跳过,[]/list=正常语义,见 _pick_stocks docstring)
             new_picks = self._pick_stocks(context)
-            self.in_position_stocks = new_picks
-            logger.info(
-                "[day=%d] picked %d stocks: %s",
-                self.day_count, len(new_picks), new_picks,
-            )
-            # 3) 调仓(仅股票部分,去掉对冲)
-            self._rebalance(context)
+            if new_picks is None:
+                # 数据失败 ≠ 空名单(2026-08-19 假熊市清仓事故同型):
+                # 跳过本次调仓保住持仓;合法空名单仍走 _rebalance 清仓 = 原语义
+                logger.warning(
+                    "[day=%d] 选股数据失败,跳过本次调仓(持仓不动,不清仓)",
+                    self.day_count,
+                )
+            else:
+                self.in_position_stocks = new_picks
+                logger.info(
+                    "[day=%d] picked %d stocks: %s",
+                    self.day_count, len(new_picks), new_picks,
+                )
+                # 3) 调仓(仅股票部分,去掉对冲)
+                self._rebalance(context)
 
         # 4) 天数加一(对齐原策略 g.t += 1)
         self.day_count += 1
 
     # =================== pick_stocks (选股) ===================
-    def _pick_stocks(self, context: Any) -> List[str]:
+    def _pick_stocks(self, context: Any) -> Optional[List[str]]:
         """选股:全市场市值最小 100 只 → 过滤 → 动量评分取前 20。
 
         对齐原策略 ``pick_stocks`` (source.py 第 113-155 行):
@@ -178,15 +186,22 @@ class SmallCapStrategy:
         2. 过滤上市<120 天 / 停牌 / ST / 涨跌停
         3. 动量评分 = (现价-130日低) + (现价-130日高) + (现价-15日均线),升序
         4. 取前 buy_stock_count 只
+
+        ⚠️ 返回值三态(2026-08-19 假熊市清仓事故同型修复):
+        - ``None`` = 数据取数失败 → handle_data 跳过本次调仓(持仓不动);
+        - ``[]`` = 合法空名单(过滤后真空)→ 清仓 = 原策略语义;
+        - 非空 list = 正常目标。
         """
         cfg = self.config
         previous_date = _previous_date_str(context)
         if previous_date is None:
-            logger.warning("pick_stocks: previous_date 为 None,返回空列表")
-            return []
+            logger.warning("pick_stocks: previous_date 为 None,跳过本次选股")
+            return None
 
         # 1) 全市场候选池(universe 成份股)
         candidates = self._stock_pool(cfg.universe, previous_date)
+        if candidates is None:
+            return None
         if not candidates:
             logger.info("[%s] 候选池为空", previous_date)
             return []
@@ -202,10 +217,10 @@ class SmallCapStrategy:
             )
         except Exception as exc:
             logger.warning("get_fundamentals_df 失败: %s", exc)
-            return []
+            return None
         if df is None or df.empty:
             logger.warning("[%s] fundamentals 为空", previous_date)
-            return []
+            return None
 
         # 3) 过滤 eps > 0(原策略 indicator.eps > 0)
         eps_col = "eps" if "eps" in df.columns else None
@@ -219,7 +234,7 @@ class SmallCapStrategy:
         # 4) 按 market_cap 升序(原策略 valuation.market_cap.asc()),取前 pick_stock_count
         if "market_cap" not in df.columns:
             logger.warning("fundamentals 缺 market_cap 列")
-            return []
+            return None
         df = df.sort_values("market_cap", ascending=True, na_position="last")
         top_candidates = list(df.index)[: cfg.pick_stock_count]
         if not top_candidates:
@@ -252,6 +267,8 @@ class SmallCapStrategy:
 
         # 7) 动量评分(130 日高低 + 15 日均线),升序
         scored = self._cal_momentum_score(top_candidates, previous_date)
+        if scored is None:
+            return None
         if scored.empty:
             return []
 
@@ -262,7 +279,7 @@ class SmallCapStrategy:
     # =================== 动量评分 ===================
     def _cal_momentum_score(
         self, stocks: List[str], end_date: str,
-    ) -> pd.DataFrame:
+    ) -> Optional[pd.DataFrame]:
         """动量评分:score = (cur-low_130) + (cur-high_130) + (cur-ma15),升序。
 
         对齐原策略 ``pick_stocks`` 评分逻辑(source.py 第 140-153 行):
@@ -282,7 +299,8 @@ class SmallCapStrategy:
         py2→py3:``df.sort(columns=)`` → ``df.sort_values(by=)``。
 
         Returns:
-            DataFrame(index=code, column=['score']),按 score 升序。
+            DataFrame(index=code, column=['score']),按 score 升序;
+            ``None`` = 取数失败/查无数据(调用方跳过本次调仓,不退化成空名单清仓)。
         """
         cfg = self.config
         if not stocks:
@@ -296,12 +314,12 @@ class SmallCapStrategy:
             )
         except Exception as exc:
             logger.warning("_cal_momentum_score get_closes_panel 失败: %s", exc)
-            return pd.DataFrame(columns=["score"])
+            return None
         if panel is None or panel.empty:
-            return pd.DataFrame(columns=["score"])
+            return None
         panel = panel.tail(cfg.ma_window)
         if panel.empty:
-            return pd.DataFrame(columns=["score"])
+            return None
 
         # 向量化算 score = (cur-low) + (cur-high) + (cur-ma15)
         # 低/高用 close 序列代理(原 high.max()/low.min())
@@ -316,7 +334,8 @@ class SmallCapStrategy:
         mask = (valid_count >= 1) & cur_price.notna() & np.isfinite(cur_price)
         score = score[mask].dropna()
         if score.empty:
-            return pd.DataFrame(columns=["score"])
+            # 全部候选现价无效 = 数据态而非市场态(停牌已在前面被过滤)
+            return None
         out = score.to_frame("score")
         # 升序:分数越低越靠前(原策略 df.sort(columns='score', ascending=True))
         out = out.sort_values("score", ascending=True)
@@ -401,12 +420,14 @@ class SmallCapStrategy:
         对齐原策略 ``~valuation.code.like('300%')`` 剔除创业板。
         ``filters.filter_kcbj_stock`` 会一并剔除创业板(3)、科创(68)、北交(4/8),
         比原策略更严但符合"剔除非主板"意图(spec 要求)。
+
+        Returns: ``None`` = 取数失败(调用方跳过本次调仓);``[]`` = 合法空池。
         """
         try:
             stocks = self.provider.get_index_stocks(index_symbol, previous_date)
         except Exception as exc:
             logger.warning("get_index_stocks(%s) 失败: %s", index_symbol, exc)
-            return []
+            return None
         stocks = filters.filter_kcbj_stock(stocks)  # 剔除创业板/科创北交
         if self.config.max_pool > 0:
             stocks = stocks[: self.config.max_pool]

sanguo_portfolio/strategies/small_cap_ex.py

@@ -159,21 +159,29 @@ class SmallCapExStrategy:
         )
 
         if is_rebalance_day:
-            # 2) 选股
+            # 2) 选股(None=数据失败跳过,[]/list=正常语义,见 _pick_stocks docstring)
             new_picks = self._pick_stocks(context)
-            self.in_position_stocks = new_picks
-            logger.info(
-                "[day=%d] picked %d stocks: %s",
-                self.day_count, len(new_picks), new_picks,
-            )
-            # 3) 调仓(仅股票部分,去掉对冲)
-            self._rebalance(context)
+            if new_picks is None:
+                # 数据失败 ≠ 空名单(2026-08-19 假熊市清仓事故同型):
+                # 跳过本次调仓保住持仓;合法空名单仍走 _rebalance 清仓 = 原语义
+                logger.warning(
+                    "[day=%d] 选股数据失败,跳过本次调仓(持仓不动,不清仓)",
+                    self.day_count,
+                )
+            else:
+                self.in_position_stocks = new_picks
+                logger.info(
+                    "[day=%d] picked %d stocks: %s",
+                    self.day_count, len(new_picks), new_picks,
+                )
+                # 3) 调仓(仅股票部分,去掉对冲)
+                self._rebalance(context)
 
         # 4) 天数加一(对齐原策略 g.t += 1)
         self.day_count += 1
 
     # =================== pick_stocks (选股) ===================
-    def _pick_stocks(self, context: Any) -> List[str]:
+    def _pick_stocks(self, context: Any) -> Optional[List[str]]:
         """选股:全市场市值最小 100 只 → 过滤 → 动量评分取前 20。
 
         对齐原策略 ``pick_stocks`` (source.py 第 113-155 行):
@@ -181,15 +189,22 @@ class SmallCapExStrategy:
         2. 过滤上市<120 天 / 停牌 / ST / 涨跌停
         3. 动量评分 = (现价-130日低) + (现价-130日高) + (现价-15日均线),升序
         4. 取前 buy_stock_count 只
+
+        ⚠️ 返回值三态(2026-08-19 假熊市清仓事故同型修复):
+        - ``None`` = 数据取数失败 → handle_data 跳过本次调仓(持仓不动);
+        - ``[]`` = 合法空名单(过滤后真空)→ 清仓 = 原策略语义;
+        - 非空 list = 正常目标。
         """
         cfg = self.config
         previous_date = _previous_date_str(context)
         if previous_date is None:
-            logger.warning("pick_stocks: previous_date 为 None,返回空列表")
-            return []
+            logger.warning("pick_stocks: previous_date 为 None,跳过本次选股")
+            return None
 
         # 1) 全市场候选池(universe 成份股)
         candidates = self._stock_pool(cfg.universe, previous_date)
+        if candidates is None:
+            return None
         if not candidates:
             logger.info("[%s] 候选池为空", previous_date)
             return []
@@ -205,10 +220,10 @@ class SmallCapExStrategy:
             )
         except Exception as exc:
             logger.warning("get_fundamentals_df 失败: %s", exc)
-            return []
+            return None
         if df is None or df.empty:
             logger.warning("[%s] fundamentals 为空", previous_date)
-            return []
+            return None
 
         # 3) 过滤 eps > 0(原策略 indicator.eps > 0)
         eps_col = "eps" if "eps" in df.columns else None
@@ -222,7 +237,7 @@ class SmallCapExStrategy:
         # 4) 按 market_cap 升序(原策略 valuation.market_cap.asc()),取前 pick_stock_count
         if "market_cap" not in df.columns:
             logger.warning("fundamentals 缺 market_cap 列")
-            return []
+            return None
         df = df.sort_values("market_cap", ascending=True, na_position="last")
         top_candidates = list(df.index)[: cfg.pick_stock_count]
         if not top_candidates:
@@ -255,6 +270,8 @@ class SmallCapExStrategy:
 
         # 7) 动量评分(130 日高低 + 15 日均线),升序
         scored = self._cal_momentum_score(top_candidates, previous_date)
+        if scored is None:
+            return None
         if scored.empty:
             return []
 
@@ -265,7 +282,7 @@ class SmallCapExStrategy:
     # =================== 动量评分 ===================
     def _cal_momentum_score(
         self, stocks: List[str], end_date: str,
-    ) -> pd.DataFrame:
+    ) -> Optional[pd.DataFrame]:
         """动量评分:score = (cur-low_130) + (cur-high_130) + (cur-ma15),升序。
 
         对齐原策略 ``pick_stocks`` 评分逻辑(source.py 第 140-153 行):
@@ -285,7 +302,8 @@ class SmallCapExStrategy:
         py2→py3:``df.sort(columns=)`` → ``df.sort_values(by=)``。
 
         Returns:
-            DataFrame(index=code, column=['score']),按 score 升序。
+            DataFrame(index=code, column=['score']),按 score 升序;
+            ``None`` = 取数失败/查无数据(调用方跳过本次调仓,不退化成空名单清仓)。
         """
         cfg = self.config
         if not stocks:
@@ -299,12 +317,12 @@ class SmallCapExStrategy:
             )
         except Exception as exc:
             logger.warning("_cal_momentum_score get_closes_panel 失败: %s", exc)
-            return pd.DataFrame(columns=["score"])
+            return None
         if panel is None or panel.empty:
-            return pd.DataFrame(columns=["score"])
+            return None
         panel = panel.tail(cfg.ma_window)
         if panel.empty:
-            return pd.DataFrame(columns=["score"])
+            return None
 
         # 向量化算 score = (cur-low) + (cur-high) + (cur-ma15)
         # 低/高用 close 序列代理(原 high.max()/low.min())
@@ -319,7 +337,8 @@ class SmallCapExStrategy:
         mask = (valid_count >= 1) & cur_price.notna() & np.isfinite(cur_price)
         score = score[mask].dropna()
         if score.empty:
-            return pd.DataFrame(columns=["score"])
+            # 全部候选现价无效 = 数据态而非市场态(停牌已在前面被过滤)
+            return None
         out = score.to_frame("score")
         # 升序:分数越低越靠前(原策略 df.sort(columns='score', ascending=True))
         out = out.sort_values("score", ascending=True)
@@ -404,12 +423,14 @@ class SmallCapExStrategy:
         对齐原策略 ``~valuation.code.like('300%')`` 剔除创业板。
         ``filters.filter_kcbj_stock`` 会一并剔除创业板(3)、科创(68)、北交(4/8),
         比原策略更严但符合"剔除非主板"意图(spec 要求)。
+
+        Returns: ``None`` = 取数失败(调用方跳过本次调仓);``[]`` = 合法空池。
         """
         try:
             stocks = self.provider.get_constituent_ex(index_symbol, previous_date)
         except Exception as exc:
             logger.warning("get_index_stocks(%s) 失败: %s", index_symbol, exc)
-            return []
+            return None
         stocks = filters.filter_kcbj_stock(stocks)  # 剔除创业板/科创北交
         if self.config.max_pool > 0:
             stocks = stocks[: self.config.max_pool]

tests/portfolio/test_momentum_timing.py

@@ -531,3 +531,72 @@ class TestConfigDefaults:
         assert cfg.top_k == 6         # g.topK
         assert cfg.ma_short == 5      # mavg(5)
         assert cfg.ma_long == 15      # mavg(15)
+
+
+# =================== 数据失败安全(2026-08-19 假熊市事故回归) ===================
+class TestDataFailureSafety:
+    """数据取数失败 ≠ 策略信号:失败跳过当日调仓,绝不退化成清仓。
+
+    事故实锤(VPS shadow_47/48, 8-18/8-19 09:30):miniQMT 未实现
+    get_closes_panel → _cal_buy_sign 吞异常 return False → handle_data
+    当熊市全清仓。有持仓时任何一次数据抖动 = 全仓卖出。
+    """
+
+    def test_cal_buy_sign_fetch_failure_returns_none(self):
+        """取数异常 → None(数据不可用),不是 False(熊市)。"""
+        s = make_strategy()
+        s.provider.get_closes_panel.side_effect = RuntimeError(
+            "sanguo_miniqmt 未实现 get_closes_panel"
+        )
+        assert s._cal_buy_sign(["IDX.XSHG"], past_day=30, cur_date="2024-09-30") is None
+
+    def test_cal_buy_sign_empty_panel_returns_none(self):
+        """空 panel(查无数据)→ None,不是 False。"""
+        s = make_strategy()
+        s.provider.get_closes_panel.side_effect = None
+        s.provider.get_closes_panel.return_value = pd.DataFrame(
+            index=pd.DatetimeIndex([])
+        )
+        assert s._cal_buy_sign(["IDX.XSHG"], past_day=30, cur_date="2024-09-30") is None
+
+    def test_buy_sign_failure_skips_day_keeps_positions(self):
+        """牛熊分界取数失败 + 已有持仓 → 跳过当日调仓,零下单(不清仓)。"""
+        cfg = MomentumTimingConfig(index_list=["IDX.XSHG"])
+        s = make_strategy(config=cfg)
+        s.provider.get_closes_panel.side_effect = RuntimeError("fetch down")
+        ctx = FakeContext(
+            current_dt=datetime(2024, 10, 8, 9, 30),
+            previous_date="2024-09-30",
+            positions={
+                "600519.XSHG": FakePosition("600519.XSHG", avg_cost=1600, price=1500),
+                "000001.XSHE": FakePosition("000001.XSHE", avg_cost=10, price=9),
+            },
+        )
+        s.handle_data(ctx)
+        s.broker.order_target_value.assert_not_called()
+
+    def test_bull_branch_fetch_failure_skips_day_keeps_positions(self):
+        """牛市分支选股取数失败(指数 panel 正常,股票 panel 抛错)→ 跳过当日,不清仓。"""
+        cfg = MomentumTimingConfig(
+            index_list=["IDX.XSHG"], top_k=6, ma_short=5, ma_long=15,
+        )
+        s = make_strategy(
+            index_stocks_map={"IDX.XSHG": ["CAND.XSHG"]}, config=cfg,
+        )
+
+        def _gcp(symbols, start=None, end=None, interval="d", fq="raw"):
+            syms = list(symbols)
+            if all(x.startswith("IDX") for x in syms):
+                return _make_close_wide(syms, [[10.0 + i for i in range(30)]], days=30)
+            raise RuntimeError("stock panel down")
+
+        s.provider.get_closes_panel.side_effect = _gcp
+        ctx = FakeContext(
+            current_dt=datetime(2024, 10, 8, 9, 30),
+            previous_date="2024-09-30",
+            positions={
+                "600519.XSHG": FakePosition("600519.XSHG", avg_cost=1600, price=1500),
+            },
+        )
+        s.handle_data(ctx)
+        s.broker.order_target_value.assert_not_called()

tests/portfolio/test_small_cap.py

@@ -290,13 +290,13 @@ class TestCalMomentumScore:
         assert list(out.index) == ["DOWN.XSHG", "FLAT.XSHG", "UP.XSHG"]
 
     def test_insufficient_data_skipped(self):
-        """K 线序列不足/空 → 该股跳过(不在结果里)。"""
+        """K 线序列不足/空 → 数据不可用返回 None(跳过调仓,不退化成空名单清仓)。"""
         s = make_strategy()
         # 让 provider.get_closes_panel 返回空 DataFrame
         s.provider.get_closes_panel.side_effect = None
         s.provider.get_closes_panel.return_value = pd.DataFrame(index=pd.DatetimeIndex([]))
         out = s._cal_momentum_score(["EMPTY.XSHG"], end_date="2024-09-30")
-        assert out.empty
+        assert out is None
 
 
 # =================== _pick_stocks (主选股流程) ===================
@@ -562,3 +562,57 @@ class TestPortingDifferences:
         # 策略实例没有 _compute_hedge_ratio 方法
         assert not hasattr(s, "_compute_hedge_ratio")
         assert not hasattr(s, "_get_next_month_future")
+
+
+# =================== 数据失败安全(2026-08-19 同型事故回归) ===================
+class TestDataFailureSafety:
+    """数据取数失败 ≠ 策略信号:失败跳过本次调仓,绝不退化成清仓。
+
+    momentum 假熊市事故的同型路径:get_fundamentals_df 抛错 → _pick_stocks
+    吞异常 return [] → _rebalance 名单空分支全清仓。合法空名单(过滤后真空)
+    仍清仓 = 原策略语义,两者必须区分。
+    """
+
+    def test_pick_stocks_fundamentals_failure_returns_none(self):
+        """fundamentals 取数异常 → None(跳过),不是 [](清仓)。"""
+        s = make_strategy(universe_stocks=["600519.XSHG"])
+        s.provider.get_fundamentals_df.side_effect = RuntimeError(
+            "unexpected keyword argument 'fields'"
+        )
+        ctx = FakeContext(
+            current_dt=datetime(2024, 10, 8, 9, 30),
+            previous_date="2024-09-30",
+        )
+        assert s._pick_stocks(ctx) is None
+
+    def test_handle_data_pick_failure_skips_not_liquidates(self):
+        """选股数据失败 + 已有持仓 → 跳过本次调仓,零下单(不清仓)。"""
+        s = make_strategy(universe_stocks=["600519.XSHG"])
+        s.provider.get_fundamentals_df.side_effect = RuntimeError("fetch down")
+        ctx = FakeContext(
+            current_dt=datetime(2024, 10, 8, 9, 30),
+            previous_date="2024-09-30",
+            positions={
+                "600519.XSHG": FakePosition("600519.XSHG", avg_cost=1600, price=1500),
+                "000001.XSHE": FakePosition("000001.XSHE", avg_cost=10, price=9),
+            },
+        )
+        s.handle_data(ctx)
+        s.broker.order_target_value.assert_not_called()
+
+    def test_handle_data_legit_empty_picks_still_clears(self):
+        """合法空名单(候选池真空)→ 仍清仓(原策略语义,区别于数据失败)。"""
+        s = make_strategy(universe_stocks=[])
+        ctx = FakeContext(
+            current_dt=datetime(2024, 10, 8, 9, 30),
+            previous_date="2024-09-30",
+            positions={
+                "600519.XSHG": FakePosition("600519.XSHG", avg_cost=1600, price=1500),
+                "000001.XSHE": FakePosition("000001.XSHE", avg_cost=10, price=9),
+            },
+        )
+        s.handle_data(ctx)
+        sell_calls = [
+            c for c in s.broker.order_target_value.call_args_list if c.args[1] == 0
+        ]
+        assert len(sell_calls) == 2