diff --git a/scripts/data_platform/akshare_static_download.py b/scripts/data_platform/akshare_static_download.py index d3a28a2..fcc49c5 100644 --- a/scripts/data_platform/akshare_static_download.py +++ b/scripts/data_platform/akshare_static_download.py @@ -21,8 +21,11 @@ 4. 重试退避: 东财 ConnectionError 常见, 3 次重试, 指数退避 (2s/4s/8s) 5. 断路器: 连续 30 个单位 (stock/date/period) failed → exit 2 6. empty vs failed 区分: - - 空 df (无北向持仓 / 退市 / 节假日无龙虎榜) → status='empty' 中性 - - 异常 / 超时 → status='failed' 计断路器 + - 空 df (端点成功但 0 行: 无北向持仓 / 退市 / 节假日无龙虎榜) + → status='empty' 中性, 写空 parquet + marker ("查过了确实无数据") + - 异常 / 超时 (重试耗尽) → status='failed' 计断路器, 不写 parquet + 不标 done (保住旧好数据, 复跑重试; 2026-09-01 洗库事故排查发现的 + 暗坑: 旧版 _df_or_empty 把失败转空 df → 空文件覆盖好数据+标 done) 7. marker 断点续传 (per-unit): 只在成功写 parquet 后写 marker 8. 开头 unset proxy (akshare 底层 requests 读 proxy 环境变量) @@ -162,6 +165,11 @@ 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") +# 三大报表同理: 东财 stock_*_sheet_by_report_em 也不支持北交所 (SH920xxx 必败 +# 'NoneType' object is not subscriptable), 09-01 季频洗库实测 340 股 × 3 表 +# 白烧 3 retry + 2/4/8s 退避 ≈ 5h。valuation/financial_abstract 实测 BSE 有 +# 数据, 不跳。 +STATEMENT_TYPES = ("balance", "income", "cashflow") ALL_TYPES = ( PER_STOCK_TYPES @@ -446,65 +454,66 @@ def write_parquet_and_marker( return False -def _df_or_empty(result: Tuple[Optional[pd.DataFrame], str]) -> pd.DataFrame: - """把 call_ak_with_retry 的返回 (df_or_None, status) 转成非 None df。 - - 必须 None 显式判断 (不能 `df or pd.DataFrame()`, DataFrame 的 truth value - ambiguous, 会抛 "The truth value of a DataFrame is ambiguous"。 - """ - df = result[0] - return df if df is not None else pd.DataFrame() - - # ======================== per-stock fetch 函数 (8 类) ======================== +# 约定 (2026-09-01 事故后): fetch_xxx 失败(重试耗尽)返 None, 由 +# download_one_unit 的 df-is-None 分支按 failed 处理 — 不写 parquet 不标 +# done, 保住旧好数据; 只有端点成功返回的真·空 df 才写空 parquet + marker。 +# (旧版经 _df_or_empty 把失败转空 df → 空文件覆盖好数据+标 done 永不重试) -def fetch_valuation(symbol: str) -> pd.DataFrame: +def fetch_valuation(symbol: str) -> Optional[pd.DataFrame]: """stock_value_em(symbol='600519') — 估值 (PE/PB/市值等13列, ~2000行/股)。""" - return _df_or_empty(call_ak_with_retry( + df, _status = call_ak_with_retry( ak.stock_value_em, f"valuation/{symbol}", symbol=symbol, - )) + ) + return df -def fetch_northbound(symbol: str) -> pd.DataFrame: +def fetch_northbound(symbol: str) -> Optional[pd.DataFrame]: """stock_hsgt_individual_em(symbol='600519') — 北向持股 (~1700行/股)。""" - return _df_or_empty(call_ak_with_retry( + df, _status = call_ak_with_retry( ak.stock_hsgt_individual_em, f"northbound/{symbol}", symbol=symbol, - )) + ) + return df -def fetch_share_capital(symbol: str) -> pd.DataFrame: +def fetch_share_capital(symbol: str) -> Optional[pd.DataFrame]: """stock_share_change_cninfo(symbol='600519') — 股本变动 (44列)。""" - return _df_or_empty(call_ak_with_retry( + df, _status = call_ak_with_retry( ak.stock_share_change_cninfo, f"share_capital/{symbol}", symbol=symbol, - )) + ) + return df -def fetch_balance_sheet(symbol: str) -> pd.DataFrame: +def fetch_balance_sheet(symbol: str) -> Optional[pd.DataFrame]: """stock_balance_sheet_by_report_em(symbol='SH600519') — 资产负债表 (319列)。""" - return _df_or_empty(call_ak_with_retry( + df, _status = call_ak_with_retry( ak.stock_balance_sheet_by_report_em, f"balance/{symbol}", symbol=symbol, - )) + ) + return df -def fetch_income_sheet(symbol: str) -> pd.DataFrame: +def fetch_income_sheet(symbol: str) -> Optional[pd.DataFrame]: """stock_profit_sheet_by_report_em(symbol='SH600519') — 利润表 (203列)。""" - return _df_or_empty(call_ak_with_retry( + df, _status = call_ak_with_retry( ak.stock_profit_sheet_by_report_em, f"income/{symbol}", symbol=symbol, - )) + ) + return df -def fetch_cashflow_sheet(symbol: str) -> pd.DataFrame: +def fetch_cashflow_sheet(symbol: str) -> Optional[pd.DataFrame]: """stock_cash_flow_sheet_by_report_em(symbol='SH600519') — 现金流量表 (254列)。""" - return _df_or_empty(call_ak_with_retry( + df, _status = call_ak_with_retry( ak.stock_cash_flow_sheet_by_report_em, f"cashflow/{symbol}", symbol=symbol, - )) + ) + return df -def fetch_financial_abstract(symbol: str) -> pd.DataFrame: +def fetch_financial_abstract(symbol: str) -> Optional[pd.DataFrame]: """stock_financial_abstract(symbol='600519') — 财务摘要 (80指标)。""" - return _df_or_empty(call_ak_with_retry( + df, _status = call_ak_with_retry( ak.stock_financial_abstract, f"financial_abstract/{symbol}", symbol=symbol, - )) + ) + return df # ======================== top_holders (per-stock × per-period) ======================== @@ -557,60 +566,67 @@ def fetch_top_holders_one_period( 通过 _safe_top_10_em 包装: 报告期未披露 (sdltgd=[]) 时 akshare 抛 Length mismatch ValueError, 这里捕获返空 df (避免 3 次无意义重试, 确定性 无数据不应消耗断路器配额)。""" - return _df_or_empty(call_ak_with_retry( + df, _status = call_ak_with_retry( _safe_top_10_em, f"top_holders/{symbol}/{period}", symbol=symbol, date=period, - )) + ) + return df # ======================== per-date fetch 函数 (4 类) ======================== -def fetch_dragon_tiger(date: str) -> pd.DataFrame: +def fetch_dragon_tiger(date: str) -> Optional[pd.DataFrame]: """stock_lhb_detail_em(start_date=end_date=date) — 单日龙虎榜。""" - return _df_or_empty(call_ak_with_retry( + df, _status = call_ak_with_retry( ak.stock_lhb_detail_em, f"dragon_tiger/{date}", start_date=date, end_date=date, - )) + ) + return df -def fetch_block_trade(date: str) -> pd.DataFrame: +def fetch_block_trade(date: str) -> Optional[pd.DataFrame]: """stock_dzjy_mrmx(symbol='A股', start_date=end_date=date) — 大宗交易明细。""" - return _df_or_empty(call_ak_with_retry( + df, _status = call_ak_with_retry( ak.stock_dzjy_mrmx, f"block_trade/{date}", symbol="A股", start_date=date, end_date=date, - )) + ) + return df -def fetch_margin_sse(date: str) -> pd.DataFrame: +def fetch_margin_sse(date: str) -> Optional[pd.DataFrame]: """stock_margin_detail_sse(date=date) — 沪市融资融券明细。""" - return _df_or_empty(call_ak_with_retry( + df, _status = call_ak_with_retry( ak.stock_margin_detail_sse, f"margin_sse/{date}", date=date, - )) + ) + return df -def fetch_restricted(date: str) -> pd.DataFrame: +def fetch_restricted(date: str) -> Optional[pd.DataFrame]: """stock_restricted_release_detail_em(start_date=end_date=date) — 解禁明细。""" - return _df_or_empty(call_ak_with_retry( + df, _status = call_ak_with_retry( ak.stock_restricted_release_detail_em, f"restricted/{date}", start_date=date, end_date=date, - )) + ) + return df # ======================== per-period fetch 函数 (2 类) ======================== -def fetch_forecast(period: str) -> pd.DataFrame: +def fetch_forecast(period: str) -> Optional[pd.DataFrame]: """stock_yjyg_em(date=period) — 业绩预告 (全市场, 单期单调用)。""" - return _df_or_empty(call_ak_with_retry( + df, _status = call_ak_with_retry( ak.stock_yjyg_em, f"forecast/{period}", date=period, - )) + ) + return df -def fetch_express(period: str) -> pd.DataFrame: +def fetch_express(period: str) -> Optional[pd.DataFrame]: """stock_yjkb_em(date=period) — 业绩快报 (全市场, 单期单调用)。""" - return _df_or_empty(call_ak_with_retry( + df, _status = call_ak_with_retry( ak.stock_yjkb_em, f"express/{period}", date=period, - )) + ) + return df # ======================== one-shot fetch 函数 (2 类) ======================== @@ -641,11 +657,12 @@ def fetch_index_const() -> pd.DataFrame: return pd.concat(frames, ignore_index=True) -def fetch_industry() -> pd.DataFrame: +def fetch_industry() -> Optional[pd.DataFrame]: """sw_index_first_info() — 申万一级行业列表 (东财接口 ConnectionError, 申万替代)。""" - return _df_or_empty(call_ak_with_retry( + df, _status = call_ak_with_retry( ak.sw_index_first_info, "industry", - )) + ) + return df # ======================== 通用下载单元 (写 parquet + marker) ======================== @@ -659,8 +676,8 @@ def download_one_unit( ) -> Tuple[str, int]: """通用单 unit 下载: 拉 df → 写 parquet + marker。 - fetch_fn() → df (可能空) 或 raise (call_ak_with_retry 已吞异常返 None, - 各 fetch_xxx 已把 None 转空 df; 这里 df 永远非 None 但可能空)。 + fetch_fn() → df / None(重试耗尽, 见 per-stock fetch 函数区注释) / raise。 + None 走 failed: 不写 parquet 不标 done (保旧好数据, 复跑重试)。 返 (status, rows), status ∈ {'ok', 'skipped', 'empty', 'failed'}。 @@ -691,7 +708,7 @@ def download_one_unit( return "failed", 0 if df is None: - # fetch_xxx 保证返非 None, 但保险 + # fetch_xxx 重试耗尽返 None (2026-09-01 起语义): 不写不标, 保旧数据 return "failed", 0 # 写 parquet + marker (空 df 也写, 静态语义: "查过了确实无数据") @@ -821,6 +838,20 @@ def build_per_stock_units( todo_codes = [(c, guess_exchange_by_code(c)) for c in codes_set] else: todo_codes = list(all_codes) + # 北交所跳过 (仅三大报表类): 东财端点必败, 不烧重试配额不拖马拉松 + # (09-01 实测 340 股 × 3 表白烧 ≈5h; valuation/financial_abstract BSE 有数据不跳) + if data_type in STATEMENT_TYPES: + n_before = len(todo_codes) + todo_codes = [ + (c, e) for c, e in todo_codes + if not c.startswith(BJ_PREFIXES) + ] + n_skipped = n_before - len(todo_codes) + if n_skipped: + logger.info( + "[%s] 跳过北交所 %d 票 (东财报表端点不支持)", + data_type, n_skipped, + ) # --limit 截断 if args.limit > 0: todo_codes = todo_codes[: args.limit] diff --git a/tests/data_platform/test_akshare_static_download.py b/tests/data_platform/test_akshare_static_download.py new file mode 100644 index 0000000..1a16e73 --- /dev/null +++ b/tests/data_platform/test_akshare_static_download.py @@ -0,0 +1,101 @@ +# -*- coding: utf-8 -*- +"""静态表下载两大隐患修复测试 (2026-09-01 洗库事故, 修正①②)。 + +①: 三大报表类跳过北交所前缀 (920/43/83/87) — 东财端点必败白烧重试 + (09-01 实测 340 股 × 3 表 × 3 retry ≈ 5h), valuation 类实测 BSE 有数据不跳。 +②: fetch 失败(重试耗尽)返 None → download_one_unit 按 failed 处理, + 不写空 parquet 覆盖好数据、不标 done (旧版 _df_or_empty 转空 df 的暗坑)。 +""" +import argparse + +import pandas as pd + +import akshare_static_download as mod # noqa: E402 (conftest 已加 sys.path) + + +def _args(**kw): + base = dict(codes=None, limit=0) + base.update(kw) + return argparse.Namespace(**base) + + +# ======================== ① 北交所跳过 ======================== + +class TestStatementBseSkip: + ALL_CODES = [ + ("600519", "SH"), ("000001", "SZ"), + ("920985", "SH"), ("430047", "SZ"), # 北交所新/旧代码段 + ] + + def test_statements_skip_bse(self): + """balance/income/cashflow 三类跳北交所, 主板保留。""" + for t, endpoint in (("balance", "balance_sheet"), + ("income", "income_sheet"), + ("cashflow", "cashflow_sheet")): + units = mod.build_per_stock_units( + t, endpoint, mod.fetch_balance_sheet, self.ALL_CODES, _args()) + codes = {uid.split(".")[0] for uid, _ in units} + assert codes == {"600519", "000001"}, t + + def test_valuation_keeps_bse(self): + """valuation/financial_abstract 实测 BSE 有数据, 不跳。""" + for t in ("valuation", "financial_abstract"): + units = mod.build_per_stock_units( + t, t, mod.fetch_valuation, self.ALL_CODES, _args()) + assert len(units) == 4, t + + def test_codes_filter_bse_only_yields_zero(self): + """--codes 显式指定北交所 + 报表类 → 0 unit (语义明确)。""" + units = mod.build_per_stock_units( + "balance", "balance_sheet", mod.fetch_balance_sheet, + [("920985", "SH")], _args(codes="920985")) + assert units == [] + + +# ======================== ② 失败不覆盖好数据 ======================== + +class TestFailedNoOverwrite: + + def test_failed_fetch_writes_nothing(self, tmp_path, monkeypatch): + """fetch 返 None (重试耗尽) → failed, parquet/marker 都不落。""" + monkeypatch.setattr(mod, "OUT_DIR", tmp_path) + status, rows = mod.download_one_unit( + "balance", "600519.SH_balance", lambda: None, force=True) + assert status == "failed" + assert not (tmp_path / "balance" / "600519.SH_balance.parquet").exists() + assert not (tmp_path / "balance" / ".600519.SH_balance.akshare").exists() + + def test_genuine_empty_writes_and_marks(self, tmp_path, monkeypatch): + """端点成功返真空 df → 写空 parquet + marker ("查过了确实无数据")。""" + monkeypatch.setattr(mod, "OUT_DIR", tmp_path) + status, _ = mod.download_one_unit( + "balance", "600519.SH_balance", + lambda: pd.DataFrame(), force=True) + assert status == "empty" + assert (tmp_path / "balance" / "600519.SH_balance.parquet").exists() + assert (tmp_path / "balance" / ".600519.SH_balance.akshare").exists() + + def test_ok_fetch_writes_data(self, tmp_path, monkeypatch): + monkeypatch.setattr(mod, "OUT_DIR", tmp_path) + status, rows = mod.download_one_unit( + "balance", "600519.SH_balance", + lambda: pd.DataFrame({"v": [1, 2]}), force=True) + assert status == "ok" and rows == 2 + assert len(pd.read_parquet( + tmp_path / "balance" / "600519.SH_balance.parquet")) == 2 + + def test_fetchers_return_none_when_retry_exhausted(self, monkeypatch): + """三类模式 (per-stock/per-date/per-period) 的 fetcher 失败统一返 None。""" + monkeypatch.setattr( + mod, "call_ak_with_retry", + lambda *a, **k: (None, "failed")) + assert mod.fetch_balance_sheet("SH600519") is None + assert mod.fetch_dragon_tiger("20260901") is None + assert mod.fetch_forecast("20260630") is None + + def test_fetchers_return_df_on_ok(self, monkeypatch): + df = pd.DataFrame({"a": [1]}) + monkeypatch.setattr(mod, "call_ak_with_retry", + lambda *a, **k: (df, "ok")) + assert mod.fetch_income_sheet("SH600519") is df + assert mod.fetch_express("20260630") is df diff --git a/tests/data_platform/test_top_holders_parse.py b/tests/data_platform/test_top_holders_parse.py index 455559c..62fbe60 100644 --- a/tests/data_platform/test_top_holders_parse.py +++ b/tests/data_platform/test_top_holders_parse.py @@ -194,8 +194,9 @@ class TestFetchTopHoldersOnePeriod: assert call_count["n"] == mod.AK_MAX_RETRIES, ( f"网络异常应重试 {mod.AK_MAX_RETRIES} 次, 实际 {call_count['n']} 次" ) - # 重试耗尽返空 df (无 schema, 因为确实是失败) - assert isinstance(df, pd.DataFrame) + # 重试耗尽返 None (2026-09-01 语义修正: 失败≠空数据, download_one_unit + # 按 failed 处理, 不写空 parquet 覆盖好数据/不标 done) + assert df is None class TestSafeTop10EmKeyErrorSdltgd: