fix(data): constituent_unified中证1000(000852)全0修复+current锚点三重防线——000852的1233条成分全in_current=0(因子池端点显示0只);根因=08-16 idx-monthly STEP0的fetch_akshare_current(000852)瞬时网络失败被「非致命warning」吞→announce_union重写丢current锚点→migrate全标was_removed,而snapshot兜底对两指数均为csindex超时placeholder从未起作用;数据修复(已执行)=akshare现拉1000只append进announce_union+正规migrate/merge重建,VPS主库1673条cur=1000精确达标,全21指数diff仅000852,NAS sync_tables已同步同态,全0扫描为空;代码防线=①fetch重试3次②空解析raise大声(列定位失败不再静默返空)③_carry_forward_current失败时沿用旧parquet current行锚点只增不丢(000852/932000两块接线);6新测试钉死三防线 [vps]

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()

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