From 4ab83097eccd7001e89ec3519e1f596b4794713d Mon Sep 17 00:00:00 2001 From: claude_dev Date: Sun, 30 Aug 2026 12:49:13 +0800 Subject: [PATCH] =?UTF-8?q?fix(data):=20constituent=5Funified=E4=B8=AD?= =?UTF-8?q?=E8=AF=811000(000852)=E5=85=A80=E4=BF=AE=E5=A4=8D+current?= =?UTF-8?q?=E9=94=9A=E7=82=B9=E4=B8=89=E9=87=8D=E9=98=B2=E7=BA=BF=E2=80=94?= =?UTF-8?q?=E2=80=94000852=E7=9A=841233=E6=9D=A1=E6=88=90=E5=88=86?= =?UTF-8?q?=E5=85=A8in=5Fcurrent=3D0(=E5=9B=A0=E5=AD=90=E6=B1=A0=E7=AB=AF?= =?UTF-8?q?=E7=82=B9=E6=98=BE=E7=A4=BA0=E5=8F=AA);=E6=A0=B9=E5=9B=A0=3D08-?= =?UTF-8?q?16=20idx-monthly=20STEP0=E7=9A=84fetch=5Fakshare=5Fcurrent(0008?= =?UTF-8?q?52)=E7=9E=AC=E6=97=B6=E7=BD=91=E7=BB=9C=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E8=A2=AB=E3=80=8C=E9=9D=9E=E8=87=B4=E5=91=BDwarning=E3=80=8D?= =?UTF-8?q?=E5=90=9E=E2=86=92announce=5Funion=E9=87=8D=E5=86=99=E4=B8=A2cu?= =?UTF-8?q?rrent=E9=94=9A=E7=82=B9=E2=86=92migrate=E5=85=A8=E6=A0=87was=5F?= =?UTF-8?q?removed,=E8=80=8Csnapshot=E5=85=9C=E5=BA=95=E5=AF=B9=E4=B8=A4?= =?UTF-8?q?=E6=8C=87=E6=95=B0=E5=9D=87=E4=B8=BAcsindex=E8=B6=85=E6=97=B6pl?= =?UTF-8?q?aceholder=E4=BB=8E=E6=9C=AA=E8=B5=B7=E4=BD=9C=E7=94=A8;?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E4=BF=AE=E5=A4=8D(=E5=B7=B2=E6=89=A7?= =?UTF-8?q?=E8=A1=8C)=3Dakshare=E7=8E=B0=E6=8B=891000=E5=8F=AAappend?= =?UTF-8?q?=E8=BF=9Bannounce=5Funion+=E6=AD=A3=E8=A7=84migrate/merge?= =?UTF-8?q?=E9=87=8D=E5=BB=BA,VPS=E4=B8=BB=E5=BA=931673=E6=9D=A1cur=3D1000?= =?UTF-8?q?=E7=B2=BE=E7=A1=AE=E8=BE=BE=E6=A0=87,=E5=85=A821=E6=8C=87?= =?UTF-8?q?=E6=95=B0diff=E4=BB=85000852,NAS=20sync=5Ftables=E5=B7=B2?= =?UTF-8?q?=E5=90=8C=E6=AD=A5=E5=90=8C=E6=80=81,=E5=85=A80=E6=89=AB?= =?UTF-8?q?=E6=8F=8F=E4=B8=BA=E7=A9=BA;=E4=BB=A3=E7=A0=81=E9=98=B2?= =?UTF-8?q?=E7=BA=BF=3D=E2=91=A0fetch=E9=87=8D=E8=AF=953=E6=AC=A1=E2=91=A1?= =?UTF-8?q?=E7=A9=BA=E8=A7=A3=E6=9E=90raise=E5=A4=A7=E5=A3=B0(=E5=88=97?= =?UTF-8?q?=E5=AE=9A=E4=BD=8D=E5=A4=B1=E8=B4=A5=E4=B8=8D=E5=86=8D=E9=9D=99?= =?UTF-8?q?=E9=BB=98=E8=BF=94=E7=A9=BA)=E2=91=A2=5Fcarry=5Fforward=5Fcurre?= =?UTF-8?q?nt=E5=A4=B1=E8=B4=A5=E6=97=B6=E6=B2=BF=E7=94=A8=E6=97=A7parquet?= =?UTF-8?q?=20current=E8=A1=8C=E9=94=9A=E7=82=B9=E5=8F=AA=E5=A2=9E?= =?UTF-8?q?=E4=B8=8D=E4=B8=A2(000852/932000=E4=B8=A4=E5=9D=97=E6=8E=A5?= =?UTF-8?q?=E7=BA=BF);6=E6=96=B0=E6=B5=8B=E8=AF=95=E9=92=89=E6=AD=BB?= =?UTF-8?q?=E4=B8=89=E9=98=B2=E7=BA=BF=20[vps]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../data_platform/parse_csindex_announce.py | 54 ++++++++- .../test_parse_csindex_current_anchor.py | 113 ++++++++++++++++++ 2 files changed, 164 insertions(+), 3 deletions(-) create mode 100644 tests/data_platform/test_parse_csindex_current_anchor.py diff --git a/scripts/data_platform/parse_csindex_announce.py b/scripts/data_platform/parse_csindex_announce.py index bda5201..65cf213 100644 --- a/scripts/data_platform/parse_csindex_announce.py +++ b/scripts/data_platform/parse_csindex_announce.py @@ -577,15 +577,39 @@ def process_notice(nid: int, detail_cache: Path, file_cache: Path, return records -def fetch_akshare_current(index_code: str) -> List[dict]: +def fetch_akshare_current(index_code: str, attempts: int = 3) -> List[dict]: """拉 akshare 当前快照 -> records(adjust_type='current') akshare 返列: 日期/指数代码/指数名称/指数英文名称/成分券代码/成分券名称/... 成分券代码 = 倒数第 2 不是 iloc[0](=日期), 用 header 名定位稳健。 + + 2026-08-30 加固(000852 全 0 事故根因): 网络抖动瞬时失败曾以「非致命 + warning」被吞 -> announce_union 重写丢 current 锚点 -> migrate 全指数标 + removed。改: 3 次重试 + 0 行也 raise(空结果=列定位失败等, 必须大声), + 由调用方 _carry_forward_current 沿用旧锚点兜底。 """ import akshare as ak log.info(f"拉 akshare 当前快照 index={index_code}...") - df = ak.index_stock_cons_csindex(symbol=index_code) + last_exc: Exception = RuntimeError("unreachable") + for attempt in range(1, attempts + 1): + try: + df = ak.index_stock_cons_csindex(symbol=index_code) + out = _parse_csindex_current_df(df, index_code) + if not out: + raise RuntimeError( + f"akshare current {index_code} 解析得 0 只(列定位失败?)") + log.info(f" akshare current {index_code}: {len(out)} stocks") + return out + except Exception as e: # noqa: BLE001 - 重试后交上层 carry-forward + last_exc = e + log.warning(f" akshare current {index_code} 第{attempt}/{attempts}次失败: {e}") + if attempt < attempts: + time.sleep(3) + raise last_exc + + +def _parse_csindex_current_df(df: "pd.DataFrame", index_code: str) -> List[dict]: + """akshare csindex 返 DataFrame -> current records(header 定位, 原 :588-612 原样搬)。""" # header 定位 cols = list(df.columns) code_col = next((c for c in cols if "成分券代码" in str(c) or "股票代码" in str(c) or "证券代码" in str(c)), None) @@ -610,10 +634,28 @@ def fetch_akshare_current(index_code: str) -> List[dict]: "code": code, "code_name": name, "adjust_type": "current", "notice_id": 0, "source": "akshare.index_stock_cons_csindex", }) - log.info(f" akshare current {index_code}: {len(out)} stocks") return out +def _carry_forward_current(out_path: Path, records: List[dict], index_code: str) -> List[dict]: + """current 锚点只增不丢: 本次拉取失败时沿用旧 parquet 的 current 行。 + + 2026-08-30: 000852 曾因当次 akshare 失败 -> 重写 parquet 无 current 行 -> + 下游 migrate 把 1233 只全标 was_removed(因子池端点显示 0 只)。拉取失败 + 不该抹掉上一期的当前成员事实; 真无旧锚点才放行(大声 error 提示)。""" + if any(r.get("adjust_type") == "current" for r in records): + return records + if out_path.exists(): + old = pd.read_parquet(out_path) + old_cur = old[old["adjust_type"] == "current"] + if len(old_cur): + log.warning( + f"{index_code} 本次 current 拉取失败, 沿用旧 parquet current {len(old_cur)} 行") + return old_cur.to_dict("records") + log.error(f"{index_code} current 锚点缺失(本次失败且无旧锚点), 该指数将全标 removed") + return records + + def parse_launch_xlsx(path: Path, index_code: str, publish_date: str, notice_id: int) -> List[dict]: """解析 launch xlsx (单 sheet, header 6 列: 指数代码/指数简称/指数英文简称/证券代码/证券中文简称/证券英文名称) @@ -860,6 +902,9 @@ def main(): df_1000 = pd.DataFrame(records_1000, columns=[ "updateDate", "index_code", "code", "code_name", "adjust_type", "notice_id", "source"]) out_1000 = out_dir / "000852_announce_union.parquet" + records_1000 = _carry_forward_current(out_1000, records_1000, "000852") + df_1000 = pd.DataFrame(records_1000, columns=[ + "updateDate", "index_code", "code", "code_name", "adjust_type", "notice_id", "source"]) df_1000.to_parquet(out_1000, index=False) distinct_1000 = df_1000["code"].nunique() n_add = (df_1000["adjust_type"] == "add").sum() @@ -905,6 +950,9 @@ def main(): df_2000 = pd.DataFrame(records_2000, columns=[ "updateDate", "index_code", "code", "code_name", "adjust_type", "notice_id", "source"]) out_2000 = out_dir / "932000_announce_union.parquet" + records_2000 = _carry_forward_current(out_2000, records_2000, "932000") + df_2000 = pd.DataFrame(records_2000, columns=[ + "updateDate", "index_code", "code", "code_name", "adjust_type", "notice_id", "source"]) df_2000.to_parquet(out_2000, index=False) distinct_2000 = df_2000["code"].nunique() n_init = (df_2000["adjust_type"] == "initial").sum() diff --git a/tests/data_platform/test_parse_csindex_current_anchor.py b/tests/data_platform/test_parse_csindex_current_anchor.py new file mode 100644 index 0000000..727f16b --- /dev/null +++ b/tests/data_platform/test_parse_csindex_current_anchor.py @@ -0,0 +1,113 @@ +# -*- coding: utf-8 -*- +"""current 锚点防线测试(2026-08-30 数据session, 000852 全 0 事故加固)。 + +事故链: 08-16 idx-monthly STEP0 的 fetch_akshare_current("000852") 瞬时网络 +失败 -> 「非致命 warning」被吞 -> announce_union 重写无 current 行 -> migrate +把 1233 只全标 was_removed -> 因子池端点中证1000 显示 0 只。 + +三重防线: ①fetch 重试 3 次 ②空解析结果 raise(大声) ③carry-forward 沿用旧 +parquet current 行(锚点只增不丢)。""" +import sys +from unittest.mock import MagicMock + +import pandas as pd +import pytest + +if "parse_csindex_announce" not in sys.modules: + # 模块顶层 import pdfplumber/akshare(仅 VPS 生产装了), Mac 测试 stub 掉 + for _mod in ("pdfplumber", "akshare"): + sys.modules.setdefault(_mod, MagicMock()) + from scripts.data_platform import parse_csindex_announce as pc +else: + from scripts.data_platform import parse_csindex_announce as pc + + +def _valid_df(): + return pd.DataFrame({ + "日期": ["2026-08-30"] * 2, + "成分券代码": ["000012", "600519"], + "成分券名称": ["南玻A", "贵州茅台"], + }) + + +@pytest.fixture(autouse=True) +def _no_sleep(monkeypatch): + monkeypatch.setattr("time.sleep", lambda s: None) + + +# ---------- 防线①: 重试 ---------- + +def test_fetch_retries_then_succeeds(monkeypatch): + """瞬时抖动: 前两次 raise 第三次成功 -> 正常返回, 不抛。""" + import akshare as ak + calls = {"n": 0} + + def flaky(symbol): + calls["n"] += 1 + if calls["n"] < 3: + raise ConnectionError("csindex hiccup") + return _valid_df() + + monkeypatch.setattr(ak, "index_stock_cons_csindex", flaky) + rows = pc.fetch_akshare_current("000852") + assert calls["n"] == 3 + assert len(rows) == 2 + assert rows[0]["adjust_type"] == "current" + assert rows[0]["code"] == "000012" + + +def test_fetch_raises_after_all_attempts(monkeypatch): + import akshare as ak + ak.index_stock_cons_csindex = MagicMock( + side_effect=ConnectionError("down")) + with pytest.raises(ConnectionError): + pc.fetch_akshare_current("000852", attempts=3) + + +# ---------- 防线②: 空结果大声 ---------- + +def test_fetch_empty_parse_raises(monkeypatch): + """返表结构异常(定位不到代码列)->0 行: 必须 raise 交 carry-forward, 不得静默返空。""" + import akshare as ak + ak.index_stock_cons_csindex = MagicMock( + return_value=pd.DataFrame({"日期": ["2026-08-30"]})) + with pytest.raises(RuntimeError, match="0 只"): + pc.fetch_akshare_current("000852") + + +# ---------- 防线③: carry-forward ---------- + +def _old_parquet_with_current(tmp_path): + df = pd.DataFrame([ + {"updateDate": "current", "index_code": "000852", "code": "000012", + "code_name": "南玻A", "adjust_type": "current", "notice_id": 0, + "source": "akshare.index_stock_cons_csindex"}, + ]) + p = tmp_path / "000852_announce_union.parquet" + df.to_parquet(p, index=False) + return p + + +def test_carry_forward_when_current_missing(tmp_path): + """本次失败(records 无 current) + 旧 parquet 有 -> 沿用旧行, 锚点不丢。""" + p = _old_parquet_with_current(tmp_path) + records = [{"code": "002766", "adjust_type": "remove"}] + out = pc._carry_forward_current(p, records, "000852") + assert any(r["adjust_type"] == "current" for r in out) + assert out[-1]["code"] == "000012" + + +def test_no_carry_when_current_present(tmp_path): + """本次成功 -> 原样返回, 不叠加旧行。""" + p = _old_parquet_with_current(tmp_path) + records = [{"code": "000012", "adjust_type": "current"}] + out = pc._carry_forward_current(p, records, "000852") + assert out == records + + +def test_no_carry_when_no_old_file(tmp_path): + """无旧锚点也不编造: 原样放行(error 提示由 migrate 侧可见)。""" + p = tmp_path / "never_exists.parquet" + records = [{"code": "002766", "adjust_type": "add"}] + out = pc._carry_forward_current(p, records, "000852") + assert out == records