diff --git a/scripts/factor_research/xcorr_family.py b/scripts/factor_research/xcorr_family.py
new file mode 100644
index 0000000..a52c634
--- /dev/null
+++ b/scripts/factor_research/xcorr_family.py
@@ -0,0 +1,234 @@
+#!/usr/bin/env python
+"""因子族分析 CLI(monthly_ic 两两 Pearson + |r| 阈值连通分量成族).
+
+固化 2026-08-31 对 62 量价 effective 因子的 ad-hoc 族分析(12 独立源/冗余 81%):
+各因子 monthly_ic 序列(eval_results.metrics_json[period]["monthly_ic"],
+sanguo_factor/metrics.py 生成)→ 对齐月份 → 两两 Pearson → |r|>=threshold 连边 →
+连通分量成族 → 族代表=|ICIR| 最大者;单因子(无连接)单列.
+多 run 传入时为跨批模式:族分析在并集上做,并额外输出跨批配对清单(防暗相关).
+
+用法示例:
+  列出可用 run:
+    venv310/bin/python scripts/factor_research/xcorr_family.py --list-runs
+  单批族分析(仅 effective,周期1,阈值0.7):
+    venv310/bin/python scripts/factor_research/xcorr_family.py \
+      --run-ids ev_20260831_120000_ab12 --effective-only
+  跨批暗相关(财务批 vs 量价批,导出完整 JSON):
+    venv310/bin/python scripts/factor_research/xcorr_family.py \
+      --run-ids ev_20260905_xxxx,ev_20260831_yyyy --effective-only --json family.json
+"""
+import argparse
+import json
+import os
+import sys
+
+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 pandas as pd
+
+
+def load_series(db_path, run_ids, period, effective_only):
+    """读各 run 因子 monthly_ic → (宽表 month×key, key→meta, run 摘要).
+
+    多 run 时 key=f"{factor}@{label}" 消歧(同名因子可跨批对比);
+    monthly_ic 不足 2 个月的因子跳过.
+    """
+    from sanguo_factor import eval_store
+
+    label_of = {r["run_id"]: r["label"] for r in eval_store.list_runs(db_path)}
+    single = len(run_ids) == 1
+    cols: dict[str, dict[str, float]] = {}
+    meta: dict[str, dict] = {}
+    seen: dict[str, int] = {}
+    for run_id in run_ids:
+        label = label_of.get(run_id, run_id)
+        for row in eval_store.get_rows(db_path, run_id):
+            m = (row.get("metrics") or {}).get(period) or {}
+            if effective_only and m.get("conclusion") != "effective":
+                continue
+            mi = m.get("monthly_ic") or []
+            if len(mi) < 2:
+                continue
+            base = row["factor"] if single else f"{row['factor']}@{label}"
+            n = seen.get(base, 0)
+            seen[base] = n + 1
+            key = base if n == 0 else f"{base}#{n + 1}"
+            cols[key] = {d["month"]: float(d["ic"]) for d in mi}
+            meta[key] = {"factor": row["factor"], "run_id": run_id, "label": label,
+                         "icir": m.get("icir")}
+    wide = pd.DataFrame({k: pd.Series(v) for k, v in cols.items()}).sort_index()
+    runs = [{"run_id": rid, "label": label_of.get(rid, rid)} for rid in run_ids]
+    return wide, meta, runs
+
+
+def build_families(corr: pd.DataFrame, threshold: float) -> list[list[str]]:
+    """|r|>=threshold 连边 → 连通分量;按成员数降序、首成员名升序."""
+    keys = list(corr.columns)
+    parent = list(range(len(keys)))
+
+    def find(x: int) -> int:
+        while parent[x] != x:
+            parent[x] = parent[parent[x]]
+            x = parent[x]
+        return x
+
+    for i in range(len(keys)):
+        for j in range(i + 1, len(keys)):
+            r = corr.iat[i, j]
+            if pd.notna(r) and abs(r) >= threshold:
+                ri, rj = find(i), find(j)
+                if ri != rj:
+                    parent[rj] = ri
+    groups: dict[int, list[str]] = {}
+    for i in range(len(keys)):
+        groups.setdefault(find(i), []).append(keys[i])
+    return sorted(groups.values(), key=lambda g: (-len(g), g[0]))
+
+
+def analyze(db_path, run_ids, period="1", threshold=0.7, min_months=6,
+            effective_only=False) -> dict:
+    """族分析主流程,返回 report dict(可 json 序列化)."""
+    wide, meta, runs = load_series(db_path, run_ids, period, effective_only)
+    if wide.empty:
+        raise ValueError("无可用 monthly_ic 序列(检查 run-ids/period/effective-only)")
+
+    corr = wide.corr(min_periods=max(min_months, 2))
+    groups = build_families(corr, threshold)
+
+    def member(key: str, is_rep: bool = False) -> dict:
+        m = meta[key]
+        return {"factor": m["factor"], "run_id": m["run_id"], "label": m["label"],
+                "icir": m["icir"], "is_rep": is_rep}
+
+    families, singles = [], []
+    for g in groups:
+        if len(g) == 1:
+            singles.append(member(g[0]))
+            continue
+        ordered = sorted(g, key=lambda k: -abs(meta[k]["icir"] or 0.0))
+        families.append({
+            "size": len(g),
+            "representative": meta[ordered[0]]["factor"],
+            "rep_icir": meta[ordered[0]]["icir"],
+            "members": [member(k, is_rep=(k == ordered[0])) for k in ordered],
+        })
+    singles.sort(key=lambda m: -abs(m["icir"] or 0.0))
+
+    cross_pairs = []
+    if len(run_ids) > 1:
+        keys = list(corr.columns)
+        for i in range(len(keys)):
+            for j in range(i + 1, len(keys)):
+                r = corr.iat[i, j]
+                if (pd.notna(r) and abs(r) >= threshold
+                        and meta[keys[i]]["run_id"] != meta[keys[j]]["run_id"]):
+                    a, b = member(keys[i]), member(keys[j])
+                    cross_pairs.append({
+                        "factor_a": a["factor"], "label_a": a["label"], "run_a": a["run_id"],
+                        "factor_b": b["factor"], "label_b": b["label"], "run_b": b["run_id"],
+                        "r": round(float(r), 6),
+                    })
+        cross_pairs.sort(key=lambda p: -abs(p["r"]))
+
+    n_total = len(meta)
+    n_comp = len(families) + len(singles)
+    return {
+        "runs": runs,
+        "period": period,
+        "threshold": threshold,
+        "min_months": min_months,
+        "effective_only": effective_only,
+        "months_aligned": int(len(wide.index)),
+        "factors_total": n_total,
+        "families": families,
+        "singles": singles,
+        "cross_pairs": cross_pairs,
+        "summary": {
+            "families_count": n_comp,
+            "multi_family_count": len(families),
+            "singles_count": len(singles),
+            "redundancy": 1.0 - n_comp / n_total if n_total else 0.0,
+        },
+    }
+
+
+def print_report(rep: dict) -> None:
+    single = len(rep["runs"]) == 1
+
+    def disp(m: dict) -> str:
+        return m["factor"] if single else f"{m['factor']}@{m['label']}"
+
+    print(f"[xcorr] {rep['factors_total']} 因子 × {rep['months_aligned']} 月, "
+          f"period={rep['period']}, threshold={rep['threshold']}, "
+          f"min_months={rep['min_months']}, effective_only={rep['effective_only']}")
+
+    if rep["families"]:
+        print(f"\n{'族#':<5s}{'成员数':>5s}  {'代表':<26s}{'代表ICIR':>9s}  成员(|ICIR|降序)")
+        for i, fam in enumerate(rep["families"], 1):
+            members = fam["members"]
+            head = ", ".join(f"{disp(m)}({(m['icir'] or 0):.2f})" for m in members[:6])
+            tail = f" …+{len(members) - 6}" if len(members) > 6 else ""
+            rep_m = next(m for m in members if m["is_rep"])
+            print(f"{i:<5d}{len(members):>5d}  {disp(rep_m):<26s}{fam['rep_icir']:>9.3f}"
+                  f"  {head}{tail}")
+
+    if rep["singles"]:
+        names = ", ".join(f"{disp(m)}({(m['icir'] or 0):.2f})" for m in rep["singles"])
+        print(f"\n单因子(无连接): {names}")
+
+    s = rep["summary"]
+    print(f"\n[xcorr] 族数 {s['families_count']}(多成员 {s['multi_family_count']} + "
+          f"单因子 {s['singles_count']}) / 因子 {rep['factors_total']} "
+          f"→ 冗余度 {s['redundancy']:.1%}")
+
+    if rep["cross_pairs"]:
+        print(f"\n跨批配对 |r|>={rep['threshold']} 共 {len(rep['cross_pairs'])} 对(按|r|降序,前20):")
+        for p in rep["cross_pairs"][:20]:
+            print(f"  {p['factor_a']}@{p['label_a']} × {p['factor_b']}@{p['label_b']}"
+                  f"  r={p['r']:.3f}")
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--db", default=None, help="factor_eval.db 路径;默认 default_eval_db_path()")
+    ap.add_argument("--run-ids", default="", help="逗号分隔一个或多个 run;多个=跨批模式")
+    ap.add_argument("--threshold", type=float, default=0.7)
+    ap.add_argument("--period", default="1", help="IC 周期键:1/5/10")
+    ap.add_argument("--min-months", type=int, default=6, help="配对最少对齐月数")
+    ap.add_argument("--effective-only", action="store_true", help="仅本周期 conclusion=effective 的因子")
+    ap.add_argument("--json", default=None, help="导出完整结果 JSON(含全成员清单)")
+    ap.add_argument("--list-runs", action="store_true", help="列出库中 run 后退出")
+    args = ap.parse_args()
+
+    from sanguo_factor.eval_store import default_eval_db_path, list_runs
+    db_path = args.db or default_eval_db_path()
+
+    if args.list_runs:
+        for r in list_runs(db_path):
+            print(f"{r['run_id']}  {r['created_at']}  [{r['status']:>7s}] {r['label']} "
+                  f"({r['factors_done']}/{r['factors_total']})")
+        return 0
+
+    run_ids = [s.strip() for s in args.run_ids.split(",") if s.strip()]
+    if not run_ids:
+        print("未指定 --run-ids(可用 --list-runs 查)", file=sys.stderr)
+        return 1
+
+    try:
+        report = analyze(db_path, run_ids, period=args.period, threshold=args.threshold,
+                         min_months=args.min_months, effective_only=args.effective_only)
+    except ValueError as e:
+        print(f"[xcorr] {e}", file=sys.stderr)
+        return 1
+
+    print_report(report)
+    if args.json:
+        with open(args.json, "w", encoding="utf-8") as f:
+            json.dump(report, f, ensure_ascii=False, indent=2)
+        print(f"\n[xcorr] JSON → {args.json}")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/tests/factor/test_xcorr_family.py b/tests/factor/test_xcorr_family.py
new file mode 100644
index 0000000..6c444f2
--- /dev/null
+++ b/tests/factor/test_xcorr_family.py
@@ -0,0 +1,102 @@
+# tests/factor/test_xcorr_family.py
+"""xcorr_family:合成 monthly_ic 验证分族/族代表/单因子/跨批暗相关."""
+import json
+import os
+import sys
+
+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__), "..", "..", "scripts", "factor_research")))
+
+import pytest
+
+from sanguo_factor import eval_store
+import xcorr_family
+
+MONTHS = [f"2024-{m:02d}" for m in range(1, 13)]
+RAMP = [round(0.01 * (i + 1), 4) for i in range(12)]         # A: 线性斜坡
+NEG_RAMP = [round(-v, 4) for v in RAMP]                      # r(A) = -1(负相关须同族)
+LIN_RAMP = [round(3 * v - 0.5, 4) for v in RAMP]             # r(A) = +1
+ZIGZAG = [0.05 if i % 2 == 0 else -0.05 for i in range(12)]  # |r(A)| ≈ 0.145 独立
+
+
+def _metrics(series, icir, conclusion="effective"):
+    return {"1": {"icir": icir, "conclusion": conclusion,
+                  "monthly_ic": [{"month": m, "ic": v} for m, v in zip(MONTHS, series)]}}
+
+
+def _mkdb(tmp_path, runs):
+    """runs={label: [(factor, series, icir[, conclusion]), ...]} → (db_path, {label: run_id})."""
+    db = str(tmp_path / "factor_eval.db")
+    eval_store.init_db(db)
+    ids = {}
+    for label, rows in runs.items():
+        run_id = eval_store.create_run(db, label=label, universe="all_a", symbols_count=10,
+                                       factors_total=len(rows), start="2024-01-01",
+                                       end="2024-12-31", params={})
+        eval_store.save_results(db, run_id, [
+            {"factor": f, "category": "t", "expression": f,
+             "metrics": _metrics(s, icir, concl)}
+            for f, s, icir, *rest in rows for concl in [rest[0] if rest else "effective"]
+        ])
+        ids[label] = run_id
+    return db, ids
+
+
+def test_family_and_representative(tmp_path):
+    """A/B(+1)/D(-1) 同族,代表=|ICIR|最大的A;C 独立单列;冗余度=1-2/4."""
+    db, ids = _mkdb(tmp_path, {"pv": [
+        ("A", RAMP, 0.5), ("B", LIN_RAMP, 0.3),
+        ("C", ZIGZAG, 0.1), ("D", NEG_RAMP, -0.4),
+        ("G", RAMP, 0.9, "eliminated"),  # 与A完全同步但已淘汰
+    ]})
+    rep = xcorr_family.analyze(db, [ids["pv"]], effective_only=True)
+
+    assert rep["factors_total"] == 4 and rep["months_aligned"] == 12
+    assert len(rep["families"]) == 1
+    fam = rep["families"][0]
+    assert {m["factor"] for m in fam["members"]} == {"A", "B", "D"}
+    assert fam["representative"] == "A" and fam["rep_icir"] == pytest.approx(0.5)
+    assert [m["factor"] for m in fam["members"]] == ["A", "D", "B"]  # |ICIR| 降序
+    assert [s["factor"] for s in rep["singles"]] == ["C"]
+    assert rep["summary"]["families_count"] == 2
+    assert rep["summary"]["redundancy"] == pytest.approx(0.5)
+    assert rep["cross_pairs"] == []  # 单 run 无跨批
+
+
+def test_without_effective_filter(tmp_path):
+    """不开 --effective-only:淘汰因子 G 纳入,|ICIR|=0.9 成为族代表."""
+    db, ids = _mkdb(tmp_path, {"pv": [
+        ("A", RAMP, 0.5), ("B", LIN_RAMP, 0.3), ("G", RAMP, 0.9, "eliminated"),
+    ]})
+    rep = xcorr_family.analyze(db, [ids["pv"]])
+    assert rep["factors_total"] == 3
+    fam = rep["families"][0]
+    assert fam["representative"] == "G" and fam["size"] == 3
+
+
+def test_cross_batch_pairs(tmp_path):
+    """批1 A/C × 批2 E(=A)/F(=C):跨批清单只含异 run 对,A/E 池化同族."""
+    db, ids = _mkdb(tmp_path, {
+        "pv": [("A", RAMP, 0.5), ("C", ZIGZAG, 0.1)],
+        "fund": [("E", RAMP, 0.4), ("F", ZIGZAG, 0.2)],
+    })
+    rep = xcorr_family.analyze(db, [ids["pv"], ids["fund"]])
+
+    pairs = rep["cross_pairs"]
+    assert pairs and all(p["run_a"] != p["run_b"] for p in pairs)
+    assert pairs[0]["r"] == pytest.approx(1.0, abs=1e-9)  # |r| 降序,最强在前
+    names = {frozenset((p["factor_a"], p["factor_b"])) for p in pairs}
+    assert frozenset(("A", "E")) in names
+    assert frozenset(("C", "F")) in names
+    assert frozenset(("A", "C")) not in names  # 同批对不入跨批清单
+
+    fam = next(f for f in rep["families"]
+               if "A" in {m["factor"] for m in f["members"]})
+    assert {"A", "E"} <= {m["factor"] for m in fam["members"]}  # 跨批同步→池化同族
+
+
+def test_report_json_serializable(tmp_path):
+    db, ids = _mkdb(tmp_path, {"pv": [("A", RAMP, 0.5), ("B", LIN_RAMP, 0.3),
+                                      ("C", ZIGZAG, 0.1)]})
+    rep = xcorr_family.analyze(db, [ids["pv"]])
+    assert json.loads(json.dumps(rep))["factors_total"] == 3