fix(data): ak-stock top_holders KeyError 'sdltgd' 根治(_safe 捕 KeyError+匹配sdltgd,跳北交所920/83/87/43)

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