From 6e719de0c6204ed00b263fb549ab843cec04f7cd Mon Sep 17 00:00:00 2001
From: claude_dev <claude-dev@sanguo.local>
Date: Tue, 8 Sep 2026 18:53:53 +0800
Subject: [PATCH] =?UTF-8?q?feat(factor):=20top=5Fholders=E9=A2=84=E8=81=9A?=
 =?UTF-8?q?=E5=90=88=E4=B8=80=E6=AC=A1=E6=80=A7=E8=84=9A=E6=9C=AC=E2=80=94?=
 =?UTF-8?q?=E2=80=94108,610=E5=B0=8F=E6=96=87=E4=BB=B6=E2=86=92factor=5Fca?=
 =?UTF-8?q?che=E8=81=9A=E5=90=88=E4=BA=A7=E7=89=A9=20[nas]?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

E13 前置: adapter 逐文件读全市场不可行,先聚合成单 parquet
(file_code/period/hold_pct/n_holders;null占比按0,坏文件名跳过,幂等--force)。
NAS 真跑已验证: 449.7s/108,610文件 → 97,836 行(skipped=4370 empty=6404)。
---
 .../preaggregate_top_holders.py               | 111 ++++++++++++++++++
 tests/factor/test_preaggregate_top_holders.py | 101 ++++++++++++++++
 2 files changed, 212 insertions(+)
 create mode 100644 scripts/factor_research/preaggregate_top_holders.py
 create mode 100644 tests/factor/test_preaggregate_top_holders.py

diff --git a/scripts/factor_research/preaggregate_top_holders.py b/scripts/factor_research/preaggregate_top_holders.py
new file mode 100644
index 0000000..cdf65a1
--- /dev/null
+++ b/scripts/factor_research/preaggregate_top_holders.py
@@ -0,0 +1,111 @@
+#!/usr/bin/env python
+# scripts/factor_research/preaggregate_top_holders.py
+"""top_holders 十大流通股东 → (code, 期) 合计占比一次性预聚合.
+
+背景(P1-B 批 E13): static/top_holders/ 按股×期 108,610 个小文件
+(每文件 10 行),adapter 逐文件读全市场不可行——先聚合成单 parquet,
+adapter 只读聚合产物(期→法定披露截止的 PIT 映射在 adapter 做)。
+
+契约:
+- 输入文件名 {code}.{SH|SZ}_{YYYYMMDD}_top_holders.parquet(北交无文件,
+  非契约文件名/坏文件跳过并计数)
+- hold_pct = 单文件「占总流通股本持股比例」(%,54.07=54%)求和,null 按 0
+- 输出列: file_code("600519.SH") / period(Date) / hold_pct(Float64) /
+  n_holders(原始行数,sanity 用)
+- 幂等: dst 存在默认跳过返回既有内容(--force 重算)
+- 绝不写 static 树,输出到独立 factor_cache 目录
+
+用法(NAS 容器内真跑):
+  python scripts/factor_research/preaggregate_top_holders.py \
+      --src /volume1/stock/sanguo_vnpy_v2/data/static/top_holders \
+      --dst /volume1/stock/sanguo_vnpy_v2/data/factor_cache/top_holders_agg.parquet
+预计耗时 ~5-15 分钟(108,610 个小文件顺序读)。
+"""
+from __future__ import annotations
+
+import argparse
+import os
+import re
+import sys
+import time
+from datetime import datetime
+
+import polars as pl
+
+DEFAULT_SRC = "/volume1/stock/sanguo_vnpy_v2/data/static/top_holders"
+DEFAULT_DST = "/volume1/stock/sanguo_vnpy_v2/data/factor_cache/top_holders_agg.parquet"
+_RATIO_COL = "占总流通股本持股比例"
+# 契约文件名: 6 位代码 + .SH/.SZ + 8 位期 + _top_holders.parquet(北交排除)
+_FNAME_RE = re.compile(r"^(\d{6}\.(?:SH|SZ))_(\d{8})_top_holders\.parquet$")
+_LOG_EVERY = 10000
+
+
+def _parse_filename(fname: str) -> tuple[str, object] | tuple[None, None]:
+    """文件名 → (file_code, period);非契约名返回 (None, None)."""
+    m = _FNAME_RE.match(fname)
+    if not m:
+        return None, None
+    try:
+        period = datetime.strptime(m.group(2), "%Y%m%d").date()
+    except ValueError:
+        return None, None
+    return m.group(1), period
+
+
+def aggregate(src: str, dst: str, force: bool = False,
+              log_every: int = _LOG_EVERY) -> pl.DataFrame:
+    """扫描 src 全量文件按 (file_code, period) 求和占比,写出/返回聚合表."""
+    if os.path.exists(dst) and not force:
+        print(f"[preaggregate_top_holders] dst 已存在,跳过重算(--force 重算): {dst}")
+        return pl.read_parquet(dst)
+    if not os.path.isdir(src):
+        raise SystemExit(f"[preaggregate_top_holders] src 不存在: {src}")
+
+    files = sorted(f for f in os.listdir(src) if f.endswith(".parquet"))
+    rows: list[dict] = []
+    skipped = empty = 0
+    t0 = time.time()
+    for i, fname in enumerate(files, 1):
+        file_code, period = _parse_filename(fname)
+        if file_code is None:
+            skipped += 1
+            continue
+        try:
+            df = pl.read_parquet(os.path.join(src, fname), columns=[_RATIO_COL])
+        except Exception:
+            skipped += 1
+            continue
+        if df.height == 0:
+            empty += 1
+            continue
+        # null 占比按 0(缺披露的席位不贡献,也不清零整期)
+        hold_pct = df[_RATIO_COL].cast(pl.Float64, strict=False).fill_null(0.0).sum()
+        rows.append({"file_code": file_code, "period": period,
+                     "hold_pct": float(hold_pct), "n_holders": df.height})
+        if i % log_every == 0:
+            el = time.time() - t0
+            print(f"[preaggregate_top_holders] {i}/{len(files)} "
+                  f"({el:.0f}s, 均 {el / i * 1000:.1f}ms/文件)", file=sys.stderr)
+
+    if not rows:
+        raise SystemExit(f"[preaggregate_top_holders] src 无可用文件: {src}")
+    out = pl.DataFrame(rows).sort(["file_code", "period"])
+    os.makedirs(os.path.dirname(dst) or ".", exist_ok=True)
+    out.write_parquet(dst)
+    print(f"[preaggregate_top_holders] rows={out.height} files={len(files)} "
+          f"skipped={skipped} empty={empty} → {dst} "
+          f"({time.time() - t0:.1f}s)")
+    return out
+
+
+def main(argv: list[str] | None = None) -> None:
+    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    ap.add_argument("--src", default=DEFAULT_SRC, help="top_holders 原始目录")
+    ap.add_argument("--dst", default=DEFAULT_DST, help="聚合产物 parquet 路径")
+    ap.add_argument("--force", action="store_true", help="dst 存在也重算")
+    args = ap.parse_args(argv)
+    aggregate(args.src, args.dst, force=args.force)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/factor/test_preaggregate_top_holders.py b/tests/factor/test_preaggregate_top_holders.py
new file mode 100644
index 0000000..7622108
--- /dev/null
+++ b/tests/factor/test_preaggregate_top_holders.py
@@ -0,0 +1,101 @@
+# tests/factor/test_preaggregate_top_holders.py
+"""top_holders 预聚合一次性脚本: 108,610 小文件 → 单 parquet 聚合产物.
+
+契约(P1-B 任务书):
+- 输出 (file_code, period, hold_pct, n_holders),hold_pct = 单文件 10 行
+  「占总流通股本持股比例」(%)求和;null 占比按 0
+- 幂等: dst 存在默认跳过(--force 重算);绝不写 static 树
+- 增减列混合类型(「不变」字符串)不参与求和,读入 coerce 不炸
+"""
+import os
+import sys
+from datetime import date
+
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
+sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vnpy_v4.4.0")))
+
+import polars as pl
+import pytest
+
+from scripts.factor_research.preaggregate_top_holders import aggregate, main
+
+
+def _write_file(root: str, file_code: str, period: str, ratios, with_null=False):
+    os.makedirs(root, exist_ok=True)
+    rows = []
+    for r, ratio in enumerate(ratios):
+        rows.append({
+            "名次": r + 1, "股东名称": f"股东{r}", "股东性质": "基金",
+            "股份类型": "流通A股", "持股数": 1000 + r,
+            "占总流通股本持股比例": ratio,
+            "增减": "不变" if r % 2 else f"{1000 + r}",  # 混合类型
+            "变动比率": 0.0,
+        })
+    if with_null:
+        rows.append({"名次": 11, "股东名称": "股东10", "股东性质": "基金",
+                     "股份类型": "流通A股", "持股数": 1,
+                     "占总流通股本持股比例": None, "增减": "不变", "变动比率": 0.0})
+    pl.DataFrame(rows).write_parquet(
+        os.path.join(root, f"{file_code}_{period}_top_holders.parquet"))
+
+
+@pytest.fixture()
+def th_tree(tmp_path):
+    root = str(tmp_path / "top_holders")
+    _write_file(root, "600000.SH", "20221231", [6.0] * 10)
+    # null 占比行(9 行有值 + 1 行 null → sum=45,不因 null 炸/漏)
+    _write_file(root, "000001.SZ", "20221231", [5.0] * 9, with_null=True)
+    _write_file(root, "600000.SH", "20230331", [6.5] * 10)
+    return root
+
+
+def test_aggregate_sums_per_code_period(th_tree, tmp_path):
+    dst = str(tmp_path / "factor_cache" / "agg.parquet")
+    df = aggregate(th_tree, dst)
+    assert set(df.columns) == {"file_code", "period", "hold_pct", "n_holders"}
+    got = {(r["file_code"], r["period"]): r["hold_pct"] for r in df.iter_rows(named=True)}
+    assert got == {
+        ("600000.SH", date(2022, 12, 31)): 60.0,
+        ("600000.SH", date(2023, 3, 31)): 65.0,
+        ("000001.SZ", date(2022, 12, 31)): 45.0,  # null 行按 0,9×5.0
+    }
+    # n_holders = 原始行数(null 行也计入)
+    n = {(r["file_code"], r["period"]): r["n_holders"] for r in df.iter_rows(named=True)}
+    assert n[("000001.SZ", date(2022, 12, 31))] == 10
+
+
+def test_aggregate_idempotent_and_force(th_tree, tmp_path, capsys):
+    dst = str(tmp_path / "factor_cache" / "agg.parquet")
+    aggregate(th_tree, dst)
+    first_mtime = os.path.getmtime(dst)
+    # 已存在 → 跳过重算(mtime 不变)
+    df2 = aggregate(th_tree, dst)
+    assert os.path.getmtime(dst) == first_mtime
+    assert df2.height == 3
+    out = capsys.readouterr().out
+    assert "跳过" in out
+    # --force 重算(mtime 更新;新文件落盘触发)
+    import time
+    time.sleep(0.01)
+    _write_file(th_tree, "600000.SH", "20230630", [6.2] * 10)
+    aggregate(th_tree, dst, force=True)
+    assert os.path.getmtime(dst) > first_mtime
+    df3 = pl.read_parquet(dst)
+    assert df3.height == 4
+
+
+def test_aggregate_skips_bad_filenames(th_tree, tmp_path):
+    # 非契约文件名(北交/杂名)跳过不炸
+    _write_file(th_tree, "920001.BJ", "20221231", [1.0] * 10)
+    (tmp_path / "top_holders" / "notes.txt").write_text("junk")
+    dst = str(tmp_path / "agg2.parquet")
+    df = aggregate(th_tree, dst)
+    assert df.filter(pl.col("file_code") == "920001.BJ").height == 0
+
+
+def test_main_cli(th_tree, tmp_path, capsys):
+    dst = str(tmp_path / "cli" / "agg.parquet")
+    main(["--src", th_tree, "--dst", dst])
+    df = pl.read_parquet(dst)
+    assert df.height == 3
+    assert "rows" in capsys.readouterr().out