fix(data): 静态表下载两隐患根治——①北交所白烧重试跳过(东财三报表端点SH920xxx必败'NoneType'错,09-01季频洗库实测340股×3表×3retry+退避白烧≈5h;对齐top_holders BJ_PREFIXES同款跳过,valuation/financial_abstract实测BSE有数据不跳)②fetch失败语义修正:15个fetcher不再经_df_or_empty把重试耗尽(None)转空df→空parquet覆盖旧好数据+标done永不重试(09-01实证340空文件全BSE属侥幸,主板瞬时故障会真毁数据);改返None交download_one_unit既有failed分支=不写不标保旧数据复跑重试,真空df才写空+done;test_top_holders一处旧契约断言同步(失败返None);+8测试钉死 [vps]
This commit is contained in:
@@ -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]
|
||||
|
||||
Reference in New Issue
Block a user