diff --git a/sanguo_portfolio/strategies/momentum_timing.py b/sanguo_portfolio/strategies/momentum_timing.py index 6f4a9eb..1cd8eb4 100644 --- a/sanguo_portfolio/strategies/momentum_timing.py +++ b/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] diff --git a/sanguo_portfolio/strategies/momentum_timing_ex.py b/sanguo_portfolio/strategies/momentum_timing_ex.py index 0f374e7..a70ef92 100644 --- a/sanguo_portfolio/strategies/momentum_timing_ex.py +++ b/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] diff --git a/sanguo_portfolio/strategies/small_cap.py b/sanguo_portfolio/strategies/small_cap.py index e4ede43..ad3cdac 100644 --- a/sanguo_portfolio/strategies/small_cap.py +++ b/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] diff --git a/sanguo_portfolio/strategies/small_cap_ex.py b/sanguo_portfolio/strategies/small_cap_ex.py index f73b3f2..b8a982b 100644 --- a/sanguo_portfolio/strategies/small_cap_ex.py +++ b/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] diff --git a/tests/portfolio/test_momentum_timing.py b/tests/portfolio/test_momentum_timing.py index 3d9f96a..34b60df 100644 --- a/tests/portfolio/test_momentum_timing.py +++ b/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() diff --git a/tests/portfolio/test_small_cap.py b/tests/portfolio/test_small_cap.py index 7273994..5eb7c1e 100644 --- a/tests/portfolio/test_small_cap.py +++ b/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