diff --git a/scripts/data_platform/akshare_static_download.py b/scripts/data_platform/akshare_static_download.py index e3d901c..97989c8 100644 --- a/scripts/data_platform/akshare_static_download.py +++ b/scripts/data_platform/akshare_static_download.py @@ -159,6 +159,9 @@ ONE_SHOT_TYPES = ( ) # top_holders 特殊: per-stock × per-period TOP_HOLDERS = "top_holders" +# 北交所代码段 (920新段 + 83/87/43历史段): akshare 东财 stock_gdfx_free_top_10_em +# 不支持北交所, build_top_holders_units 阶段直接跳过 (避免每只×20期×3retry 失败风暴). +BJ_PREFIXES = ("920", "83", "87", "43") ALL_TYPES = ( PER_STOCK_TYPES @@ -474,25 +477,28 @@ TOP_HOLDERS_COLUMNS = [ def _safe_top_10_em(symbol: str, date: str) -> pd.DataFrame: - """akshare stock_gdfx_free_top_10_em 包装: 容忍空 sdltgd 响应。 + """akshare stock_gdfx_free_top_10_em 包装: 容忍空/缺字段响应。 bug 根因 (实证 akshare 1.18.x stock_gdfx_em.py): - 报告期未披露 (如当年 Q2 季报未发) 时东财 PageSDLTGD 接口返 sdltgd=[], - akshare pd.DataFrame([]).reset_index() 得 1 列 df, 然后 columns=[12 列] - 抛 ValueError("Length mismatch: Expected axis has 1 elements, new values have 12"). - 这是确定性无数据 (非瞬时故障), 但 call_ak_with_retry 会当网络错重试 3 次 - (14s 退避) + 噪声 ERROR 日志。 + - 旧版本 sdltgd=[] (报告期未披露) 时 pd.DataFrame([]).reset_index() 得 1 列 df, + columns=[12 列] 抛 ValueError("Length mismatch: ..."). 子串匹配稳定。 + - 新版本部分标的不支持 (如北交所 920xxx) 时返缺 sdltgd 字段, 抛 KeyError('sdltgd')。 + KeyError 不是 ValueError 子类, 旧版只 except ValueError 捕不到。 + 两者都是确定性无数据 (非瞬时故障), 不应消耗重试配额 (call_ak_with_retry 会 + 当网络错重试 3 次 14s 退避 + 噪声 ERROR 日志)。 - 本包装预判该特定 ValueError (子串匹配 pandas 错误信息, 稳定): - - Length mismatch → 返空 df (带 TOP_HOLDERS_COLUMNS schema), 不抛 - - 其他 ValueError / ConnectionError → 透传给 call_ak_with_retry 走重试 + 本包装预判这两类确定性无数据 (子串匹配错误消息, 稳定): + - Length mismatch (旧版 ValueError) → 返空 df (带 TOP_HOLDERS_COLUMNS schema), 不抛 + - 'sdltgd' (新版 KeyError, str(KeyError('sdltgd')) == "'sdltgd'" 含引号) → 同上 + - 其他 ValueError/KeyError/ConnectionError → 透传给 call_ak_with_retry 走重试 """ try: return ak.stock_gdfx_free_top_10_em(symbol=symbol, date=date) - except ValueError as e: - if "Length mismatch" in str(e): + except (ValueError, KeyError) as e: + s = str(e) + if "Length mismatch" in s or "sdltgd" in s: logger.debug( - "stock_gdfx_free_top_10_em(%s, %s) Length mismatch → sdltgd 空 (报告期未披露), 返空 df", + "stock_gdfx_free_top_10_em(%s, %s) → 空 (报告期未披露或标的不支持), 返空 df", symbol, date, ) return pd.DataFrame(columns=TOP_HOLDERS_COLUMNS) @@ -781,15 +787,22 @@ def build_top_holders_units( periods = REPORT_PERIODS[-20:] if len(REPORT_PERIODS) >= 20 else REPORT_PERIODS units: List[Tuple[str, Callable[[], pd.DataFrame]]] = [] + skipped_bj = 0 for code, exc in todo_codes: + if code.startswith(BJ_PREFIXES): + skipped_bj += 1 + continue # 跳北交所 (akshare 东财 stock_gdfx_free_top_10_em 不支持, + # 避免每只×20期×3retry 失败风暴, _safe_top_10_em 是双保险) symbol = code_to_symbol(code, exc, "top_holders") for period in periods: unit_id = f"{code}.{exc}_{period}_{TOP_HOLDERS}" fn = partial(fetch_top_holders_one_period, symbol, period) units.append((unit_id, fn)) + if skipped_bj: + logger.info("[top_holders] 跳过北交所 %d 票 (akshare 东财不支持)", skipped_bj) logger.info( "[top_holders] %d 票 × %d 期 = %d units", - len(todo_codes), len(periods), len(units), + len(todo_codes) - skipped_bj, len(periods), len(units), ) return units diff --git a/tests/data_platform/test_top_holders_parse.py b/tests/data_platform/test_top_holders_parse.py index e97f124..455559c 100644 --- a/tests/data_platform/test_top_holders_parse.py +++ b/tests/data_platform/test_top_holders_parse.py @@ -196,3 +196,117 @@ class TestFetchTopHoldersOnePeriod: ) # 重试耗尽返空 df (无 schema, 因为确实是失败) assert isinstance(df, pd.DataFrame) + + +class TestSafeTop10EmKeyErrorSdltgd: + """_safe_top_10_em 同样捕获 KeyError('sdltgd') → 返空 df。 + + bug 根因 (实证 akshare 新版): + 北交所等不支持的标的, akshare 内部返缺 'sdltgd' 字段 → 抛 KeyError('sdltgd')。 + KeyError 不是 ValueError 子类, 旧版只 except ValueError 捕不到, 透传到 + call_ak_with_retry 走 3 次重试 (14s 退避) + 噪声日志。 + 本测试集覆盖: KeyError('sdltgd') 被吞返空 df / 其他 KeyError 不被吞 / 集成路径无重试。 + """ + + def test_keyerror_sdltgd_returns_empty_df_with_schema(self): + """akshare 抛 KeyError('sdltgd') → 返空 df 带 8 列 schema。""" + from akshare_static_download import _safe_top_10_em + + # akshare 实际抛 KeyError('sdltgd'), str(err) == "'sdltgd'" (含引号) + err = KeyError("sdltgd") + with patch.object(mod.ak, "stock_gdfx_free_top_10_em", side_effect=err): + df = _safe_top_10_em(symbol="sh920510", date="20250930") + + assert isinstance(df, pd.DataFrame) + assert len(df) == 0 + assert df.columns.tolist() == EXPECTED_COLS + + def test_keyerror_sdltgd_does_not_raise(self): + """KeyError('sdltgd') 必须被吞掉 (call_ak_with_retry 才不会重试)。""" + from akshare_static_download import _safe_top_10_em + + err = KeyError("sdltgd") + with patch.object(mod.ak, "stock_gdfx_free_top_10_em", side_effect=err): + # 不应抛 + df = _safe_top_10_em(symbol="sh920510", date="20250930") + assert df.empty + + def test_other_keyerror_reraises(self): + """非 'sdltgd' 的 KeyError 不应被吞 (让重试机制处理)。""" + from akshare_static_download import _safe_top_10_em + + # 任意其他 KeyError (网络层错误等), 不应被吞 + other_err = KeyError("some_other_field") + with patch.object(mod.ak, "stock_gdfx_free_top_10_em", side_effect=other_err): + with pytest.raises(KeyError, match="some_other_field"): + _safe_top_10_em(symbol="sh920510", date="20250930") + + def test_keyerror_sdltgd_no_retry_in_fetch(self, monkeypatch): + """KeyError('sdltgd') 走集成层也不重试 (call_ak_with_retry 只调一次)。""" + monkeypatch.setattr(mod, "RETRY_BACKOFF", [0, 0, 0]) + + call_count = {"n": 0} + + def side_effect(*args, **kwargs): + call_count["n"] += 1 + raise KeyError("sdltgd") + + with patch.object(mod.ak, "stock_gdfx_free_top_10_em", side_effect=side_effect): + df = mod.fetch_top_holders_one_period(symbol="sh920510", period="20250930") + + # 关键: 只调 1 次 (KeyError 被吞, 不重试) + assert call_count["n"] == 1, ( + f"KeyError('sdltgd') 应该被 _safe_top_10_em 一次吞掉, 但调了 {call_count['n']} 次" + ) + assert df.empty + assert df.columns.tolist() == EXPECTED_COLS + + +class TestBuildTopHoldersUnitsSkipBJ: + """build_top_holders_units 跳过北交所 920xxx / 83xxx / 87xxx / 43xxx。 + + 根因: akshare 东财 stock_gdfx_free_top_10_em 不支持北交所。每只北交所 × 20 期 × 3 retry + ≈ 60 次失败调用/股 → weekly 跑不完。在 build 阶段直接跳过, 双保险。 + """ + + def _make_args(self, codes: str = ""): + import argparse + return argparse.Namespace(codes=codes, limit=0) + + def test_skips_920_prefix(self): + """920xxx (北交所新段) 不进 units。""" + codes = [("920510", "BJ"), ("600519", "SH")] + units = mod.build_top_holders_units(codes, self._make_args()) + + unit_ids = [uid for uid, _ in units] + # 920510 应该被跳, 不出现 + assert not any("920510" in uid for uid in unit_ids), ( + f"北交所 920510 不该进 units, 但出现了: {unit_ids}" + ) + # 600519 应该在 (每报告期一个 unit) + assert any("600519" in uid for uid in unit_ids) + + def test_skips_all_bj_prefixes(self): + """83xx/87xx/43xx (北交所历史段) 全跳。""" + codes = [ + ("830789", "BJ"), + ("870866", "BJ"), + ("430139", "BJ"), + ("000001", "SZ"), + ] + units = mod.build_top_holders_units(codes, self._make_args()) + unit_ids = [uid for uid, _ in units] + + for bj_code in ("830789", "870866", "430139"): + assert not any(bj_code in uid for uid in unit_ids), ( + f"北交所 {bj_code} 不该进 units" + ) + # 000001 应该在 + assert any("000001" in uid for uid in unit_ids) + + def test_no_bj_codes_arg_mixed(self): + """--codes "920001,600519" 时 920001 跳, 600519 拉。""" + units = mod.build_top_holders_units([], self._make_args(codes="920001,600519")) + unit_ids = [uid for uid, _ in units] + assert not any("920001" in uid for uid in unit_ids) + assert any("600519" in uid for uid in unit_ids)