Merge: 财务因子批P0全链——32因子+族分析工具+前端并列类别(factor-fundamental) [nas]

frontend/src/styles/tokens.css

@@ -23,6 +23,9 @@
   /* —— 次强调：紫（alpha158）—— */
   --purple-dim: #b48cff;  /* alpha158 类别色 */
 
+  /* —— 类别色：teal（fundamental 财务因子）—— */
+  --teal: #2dd4bf;        /* fundamental 类别色 */
+
   /* —— 数据高亮：琥珀 —— */
   --amber: #ffb000;       /* 数据高亮/警告 */
   --warn: #ffb000;        /* 语义别名 */

frontend/src/views/factor/FactorPicker.vue

@@ -166,6 +166,7 @@ defineExpose({ hydrate })
 .cat-alpha158 { color: var(--purple-dim); background: rgba(180,140,255,.1); border-color: rgba(180,140,255,.3); }
 .cat-builtin { color: var(--amber); background: rgba(255,176,0,.1); border-color: rgba(255,176,0,.3); }
 .cat-custom { color: var(--text-2); background: rgba(122,138,154,.1); border-color: rgba(122,138,154,.3); }
+.cat-fundamental { color: var(--teal); background: rgba(45,212,191,.1); border-color: rgba(45,212,191,.3); }
 .grid { display: flex; flex-wrap: wrap; gap: 5px; }
 .fxp { font-family: var(--mono); font-size: 11px; padding: 2px 9px; border-radius: var(--r-sm); border: 1px solid var(--border); background: var(--panel); color: var(--text-2); cursor: pointer; transition: all .12s ease; user-select: none; }
 .fxp:hover { color: var(--text); border-color: var(--text-3); }
@@ -173,6 +174,7 @@ defineExpose({ hydrate })
 .fxp.sel.sel-alpha158 { color: var(--purple-dim); border-color: rgba(180,140,255,.45); background: rgba(180,140,255,.12); }
 .fxp.sel.sel-builtin { color: var(--amber); border-color: rgba(255,176,0,.45); background: var(--amber-soft); }
 .fxp.sel.sel-custom { color: var(--text); border-color: var(--text-3); background: var(--bg-hover); }
+.fxp.sel.sel-fundamental { color: var(--teal); border-color: rgba(45,212,191,.45); background: rgba(45,212,191,.12); }
 .picked { display: flex; align-items: center; gap: 8px; margin-top: 12px; border: 1px dashed rgba(0,229,255,.3); background: rgba(0,229,255,.04); border-radius: var(--r-md); padding: 8px 12px; flex-wrap: wrap; min-height: 38px; }
 .pk { font-family: var(--mono); font-size: 10.5px; color: var(--brand); letter-spacing: .1em; white-space: nowrap; }
 .chips { display: flex; gap: 5px; flex-wrap: wrap; flex: 1; }

frontend/src/views/factor/Leaderboard.vue

@@ -22,6 +22,7 @@ const CATEGORY_CHIPS = [
   { key: 'alpha101', label: 'Alpha101' },
   { key: 'alpha158', label: 'Alpha158' },
   { key: 'builtin', label: '内置' },
+  { key: 'fundamental', label: '财务基本面' },
 ]
 
 async function load() {
@@ -87,6 +88,7 @@ function getCategoryClass(cat: string): string {
     alpha101: 'cat-alpha101',
     alpha158: 'cat-alpha158',
     builtin: 'cat-builtin',
+    fundamental: 'cat-fundamental',
   }
   return map[cat] || 'cat-builtin'
 }
@@ -608,6 +610,12 @@ td .fexpr {
   border-color: rgba(255, 176, 0, 0.3);
 }
 
+.cat-fundamental {
+  color: var(--teal);
+  background: rgba(45, 212, 191, 0.1);
+  border-color: rgba(45, 212, 191, 0.3);
+}
+
 .pos {
   color: var(--up);
 }

frontend/src/views/factor/LeaderboardDetail.vue

@@ -326,6 +326,7 @@ const categoryClass = computed(() => {
     alpha101: 'cat-alpha101',
     alpha158: 'cat-alpha158',
     builtin: 'cat-builtin',
+    fundamental: 'cat-fundamental',
   }
   return map[detail.value.category] || 'cat-builtin'
 })
@@ -750,6 +751,12 @@ function pct(val: number | null | undefined): string {
   border-color: rgba(255, 176, 0, 0.3);
 }
 
+.cat-fundamental {
+  color: var(--teal);
+  background: rgba(45, 212, 191, 0.1);
+  border-color: rgba(45, 212, 191, 0.3);
+}
+
 .st {
   display: inline-flex;
   align-items: center;

sanguo_factor/__init__.py

@@ -1,3 +1,4 @@
 """Sanguo factor module for vnpy alpha strategies."""
 from . import library  # noqa: F401  (triggers _register_all to register built-in factors)
 from . import alpha_datasets  # noqa: F401  挂载 Alpha101/158(导入即注册)
+from . import fundamental_library  # noqa: F401  财务因子 P0 批(导入即注册)

sanguo_factor/batch_eval.py

@@ -22,6 +22,8 @@ from .registry import get_factor
 from . import eval_store
 from .metrics import summarize_factor
 from .fast_ops import register_fast_ops
+from . import fundamental_library  # noqa: F401  财务因子 import 即注册(alpha_datasets 同模式)
+from .fundamental_adapter import DEFAULT_STATIC_DIR
 
 
 def _forward_return_matrices(close_wide: pd.DataFrame, periods=(1, 5, 10)) -> dict[int, pd.DataFrame]:
@@ -43,9 +45,15 @@ def run_batch_eval(
     cfg=None,
     progress_cb=None,
     vnpy_db_override: str | None = None,
+    fund_data_dir: str | None = None,
     run_id: str | None = None,
 ) -> dict:
-    """跑一轮批量评估,结果增量写入 eval_db,返回摘要."""
+    """跑一轮批量评估,结果增量写入 eval_db,返回摘要.
+
+    fund_data_dir: 财务静态域根目录(None → cfg.data_paths["static_dir"] →
+    NAS 默认 /volume1/stock/sanguo_vnpy_v2/data/static);仅当因子列表含
+    category="fundamental" 时才读取并 join(量价批零开销)。
+    """
     from vnpy.alpha.dataset.utility import calculate_by_expression
 
     # Register fast polars operators (idempotent)
@@ -63,6 +71,14 @@ def run_batch_eval(
 
     alpha_df = bars.select(["vt_symbol", "datetime", "open", "high", "low", "close",
                             "volume", "turnover", "vwap"])
+    # 财务因子批: bars 释放前仅抽取小体量 codes/dates(特征 join 移到 del bars 后,
+    # 分块进行——全量特征帧+alpha_df 双全量副本在 7.9G NAS 必 OOM)
+    fund_names = [n for n in factor_names
+                  if (get_factor(n) or {}).get("category") == "fundamental"]
+    if fund_names:
+        fund_static_dir = fund_data_dir or cfg.data_paths.get("static_dir") or DEFAULT_STATIC_DIR
+        fund_codes = bars["vt_symbol"].unique().to_list()
+        fund_days = bars["datetime"].unique().sort()
     # Pre-compute per-symbol warmup cutoff dates (bar_idx >= WARMUP_BARS 的首日)
     # 用于替代 per-factor hash join，改为 pivot 后 pandas 广播掩码置 NaN
     cutoffs = (
@@ -117,6 +133,24 @@ def run_batch_eval(
     errors: list[str] = []
     done = 0
     buffer: list[dict] = []
+
+    # 财务特征分块 join: 逐块 filter→join alpha_df 分片→concat(bars 已释放;
+    # 单块特征帧用完即弃,峰值 ≈ alpha_df + 已 join 分片累积,无全量特征副本)
+    has_fund = any((get_factor(n) or {}).get("category") == "fundamental"
+                   for n in factor_names)
+    if has_fund:
+        from .fundamental_adapter import iter_fundamental_feature_chunks
+        parts = []
+        for feat_chunk in iter_fundamental_feature_chunks(
+                fund_codes, start, end, data_dir=fund_static_dir,
+                trading_dates=fund_days):
+            syms = feat_chunk["vt_symbol"].unique().to_list()
+            parts.append(
+                alpha_df.filter(pl.col("vt_symbol").is_in(syms))
+                .join(feat_chunk, on=["vt_symbol", "datetime"], how="left"))
+        alpha_df = pl.concat(parts, how="vertical")
+        parts.clear()
+
     for i, name in enumerate(factor_names):
         row = _eval_one(name, alpha_df, cutoff_map, start_dt, end_dt, rets, calculate_by_expression)
         if "error" in row["metrics"]:

sanguo_factor/fundamental_adapter.py

@@ -0,0 +1,511 @@
+# sanguo_factor/fundamental_adapter.py
+"""财务因子适配层: NAS 静态域三表+forecast parquet → PIT 日频特征列.
+
+口径红线(docs/fundamental_factor_survey_20260907.md §7):
+- R1 单季差分: Q1 直接取累计,其余 = 本期累计 − 年内上期累计;缺上期置 NaN 不填 0
+- R2 TTM: 连续 4 个单季之和,不足 4 期 NaN
+- R3 PIT: NOTICE_DATE ≤ 决策日才可见(有效披露日 = 三表 NOTICE_DATE 最大值,保守);
+  NOTICE_DATE 缺失的报告期整期跳过(宁缺毋假)
+- 金融股(OPERATE_COST 缺失/为 0 的银行模板): 盈利质量/成长族特征置 NaN(估值族保留)
+
+数据形态(NAS 实测 2026-09-07):
+- 三表按股: static/{income,balance,cashflow}/{code}.{SH|SZ}_{table}.parquet
+  (北交 920xxx 残留零行文件、沪深退市 355 只文件不存在 → 读取容错跳过)
+- forecast 按报告期全市场: static/forecast/{YYYYMMDD}_forecast.parquet(中文列,
+  一股多行=按「预测指标」,归母净利润行优先)
+- 日期列为 "YYYY-MM-DD 00:00:00" 字符串;东财金额单位=元
+- valuation(中文列)不读: 估值类因子市值 = close × SHARE_CAPITAL 自算(表达式层)
+
+输出: vt_symbol × datetime(日频) × FEATURE_COLUMNS,join 到 alpha_df 后
+供 cs_rank(列) 表达式直接消费。
+"""
+from __future__ import annotations
+
+import os
+import warnings
+from datetime import datetime
+
+import polars as pl
+
+DEFAULT_STATIC_DIR = "/volume1/stock/sanguo_vnpy_v2/data/static"
+_SYM = "vt_symbol"  # 时序分组键
+
+# vt_symbol 后缀 → NAS 文件名后缀(600000.SSE → 600000.SH)
+_VT_TO_FILE_SUFFIX = {"SSE": "SH", "SZSE": "SZ"}
+
+# forecast 预告类型 → 有序分(§3.6 F04)
+FORECAST_TYPE_SCORE: dict[str, int] = {
+    "预增": 3, "略增": 2, "扭亏": 2, "续盈": 1, "减亏": 1, "不确定": 0,
+    "略减": -1, "增亏": -2, "续亏": -2, "预减": -3, "首亏": -3,
+}
+
+# 累计口径列映射: 源列 → 短名(需单季化+TTM)
+_CUM_MAP = {
+    "TOTAL_OPERATE_INCOME": "rev",
+    "OPERATE_COST": "cogs",
+    "PARENT_NETPROFIT": "np",
+    "DEDUCT_PARENT_NETPROFIT": "dnp",
+    "TOTAL_PROFIT": "tp",
+    "INVEST_INCOME": "invest_inc",
+    "FAIRVALUE_CHANGE_INCOME": "fv_inc",
+    "ASSET_IMPAIRMENT_LOSS": "asset_imp",
+    "CREDIT_IMPAIRMENT_LOSS": "credit_imp",
+    "NETCASH_OPERATE": "cfo",
+    "SALES_SERVICES": "sales_cash",
+    "ACCEPT_INVEST_CASH": "acc_inv_cash",
+}
+# 存量列(balance,时点值直接用;缺列 → null)
+_BALANCE_COLS = ["TOTAL_ASSETS", "TOTAL_PARENT_EQUITY", "ACCOUNTS_RECE",
+                 "OTHER_RECE", "GOODWILL", "SHARE_CAPITAL", "SHORT_LOAN",
+                 "SHORT_FIN_PAYABLE", "NONCURRENT_LIAB_1YEAR", "LONG_LOAN",
+                 "BOND_PAYABLE", "LEASE_LIAB"]
+_IBD_PARTS = ["SHORT_LOAN", "SHORT_FIN_PAYABLE", "NONCURRENT_LIAB_1YEAR",
+              "LONG_LOAN", "BOND_PAYABLE", "LEASE_LIAB"]
+_DATE_COLS = ["REPORT_DATE", "NOTICE_DATE", "UPDATE_DATE"]
+
+# 输出特征列(32 因子的全部原料;契约由 test_fundamental_library 锁定)
+FEATURE_COLUMNS: list[str] = [
+    # 报告期级比率(盈利能力 A / 盈利质量 B / 成长 C / 资本结构 E / 预期事件 F)
+    "roe_ttm", "roe_deduct_ttm", "roa_ttm", "gp_over_assets", "gross_margin",
+    "net_margin", "cfo_over_assets",
+    "tacc", "nonrec_ratio", "impairment_ratio", "invest_income_dep",
+    "receivables_anomaly", "sales_cash_ratio", "other_rece_ratio",
+    "rev_q_yoy", "np_q_yoy", "growth_scissors", "gm_delta", "roe_delta",
+    "asset_growth", "nsi", "ibd_ratio", "goodwill_ratio",
+    "sue_np", "sue_rev",
+    "forecast_type_score", "forecast_change_pct",
+    # 估值/资本行为因子的日频原料(表达式层 ÷ close×share_capital)
+    "np_ttm", "dnp_ttm", "cfo_ttm", "equity", "share_capital",
+    "acc_invest_cash_ttm",
+]
+# 金融股置 NaN 的特征(盈利质量 B 族 + 成长 C 族,§7 红线 5)
+_FIN_NULL_COLS = ["tacc", "nonrec_ratio", "impairment_ratio",
+                  "invest_income_dep", "receivables_anomaly",
+                  "sales_cash_ratio", "other_rece_ratio",
+                  "rev_q_yoy", "np_q_yoy", "growth_scissors", "gm_delta",
+                  "roe_delta", "asset_growth"]
+_FORECAST_COLS = ("forecast_type_score", "forecast_change_pct")
+
+
+# ==================== 读取层 ====================
+
+def _vt_to_file_code(vt_symbol: str) -> str | None:
+    """``600000.SSE`` → ``600000.SH``;无法映射的(ETF/北交)返回 None."""
+    code, _, suffix = vt_symbol.partition(".")
+    file_suffix = _VT_TO_FILE_SUFFIX.get(suffix.upper())
+    return f"{code}.{file_suffix}" if file_suffix else None
+
+
+def _read_static(path: str, want_cols: list[str]) -> pl.DataFrame | None:
+    """读单股单表 parquet,缺列补 null;零行/缺文件/坏文件 → None(容错跳过)."""
+    if not os.path.exists(path):
+        return None
+    try:
+        schema = pl.read_parquet_schema(path)
+        if "REPORT_DATE" not in schema:
+            return None
+        cols = [c for c in _DATE_COLS + want_cols if c in schema]
+        df = pl.read_parquet(path, columns=cols)
+    except Exception:
+        return None
+    if df.height == 0:
+        return None
+    missing = [c for c in _DATE_COLS + want_cols if c not in df.columns]
+    if missing:
+        df = df.with_columns([pl.lit(None, dtype=pl.Utf8).alias(c) for c in missing])
+    for c in want_cols:
+        df = df.with_columns(pl.col(c).cast(pl.Float64, strict=False))
+    return df
+
+
+def _norm_dates(df: pl.DataFrame, cols: list[str]) -> pl.DataFrame:
+    """日期列归一为 pl.Date:字符串截前 10 位解析,Date/Datetime 直接 cast."""
+    exprs = []
+    for c in cols:
+        dtype = df.schema[c]
+        if dtype == pl.Utf8:
+            exprs.append(pl.col(c).str.slice(0, 10).str.to_date("%Y-%m-%d", strict=False).alias(c))
+        elif dtype != pl.Date:
+            exprs.append(pl.col(c).cast(pl.Date, strict=False).alias(c))
+    return df.with_columns(exprs) if exprs else df
+
+
+def _dedupe_reports(df: pl.DataFrame, value_cols: list[str]) -> pl.DataFrame:
+    """按 (vt_symbol, REPORT_DATE) 去重:值取 (UPDATE_DATE, NOTICE_DATE) 排序末行
+    (重述取终值),有效披露日取组内 NOTICE_DATE 最大值(保守,不提前)."""
+    df = df.sort(["vt_symbol", "REPORT_DATE", "UPDATE_DATE", "NOTICE_DATE"],
+                 nulls_last=False)
+    aggs = [pl.col(c).last().alias(c) for c in value_cols]
+    aggs.append(pl.col("NOTICE_DATE").max().alias("notice_eff"))
+    return df.group_by(["vt_symbol", "REPORT_DATE"]).agg(aggs)
+
+
+# 各表显式读取清单(income/cashflow 列集不相交,join 不加后缀)
+_TABLE_RAW = {
+    "income": ["TOTAL_OPERATE_INCOME", "OPERATE_COST", "PARENT_NETPROFIT",
+               "DEDUCT_PARENT_NETPROFIT", "TOTAL_PROFIT", "INVEST_INCOME",
+               "FAIRVALUE_CHANGE_INCOME", "ASSET_IMPAIRMENT_LOSS",
+               "CREDIT_IMPAIRMENT_LOSS"],
+    "balance": _BALANCE_COLS,
+    "cashflow": ["NETCASH_OPERATE", "SALES_SERVICES", "ACCEPT_INVEST_CASH"],
+}
+
+
+def _load_statements(codes: list[str], data_dir: str) -> pl.DataFrame:
+    """全 codes 三表 → 报告期宽表(每 vt_symbol × REPORT_DATE 一行).
+
+    income/cashflow 值列重命名为短名(_CUM_MAP),balance 保持源列名;
+    anchor = 三表报告期并集,left join 保证单表缺期不拖垮其它表特征。
+    """
+    table_cols = _TABLE_RAW
+    per_table: dict[str, pl.DataFrame] = {}
+    for table, raw_cols in table_cols.items():
+        frames = []
+        for vt in codes:
+            file_code = _vt_to_file_code(vt)
+            if file_code is None:
+                continue
+            df = _read_static(
+                os.path.join(data_dir, table, f"{file_code}_{table}.parquet"), raw_cols)
+            if df is None:
+                continue
+            df = _norm_dates(df, _DATE_COLS)
+            # 先统一列序/列集再入列(各股文件 schema 子集不同,concat 前必须对齐)
+            frames.append(
+                df.with_columns(pl.lit(vt).alias("vt_symbol"))
+                .select(["vt_symbol", *_DATE_COLS, *raw_cols]))
+        if not frames:
+            continue
+        merged = pl.concat(frames)
+        if table == "balance":
+            short = {c: c for c in raw_cols}
+        else:
+            short = {k: v for k, v in _CUM_MAP.items() if k in raw_cols}
+        dedup = _dedupe_reports(merged, list(raw_cols)).rename(short)
+        per_table[table] = dedup
+
+    empty = pl.DataFrame(schema={"vt_symbol": pl.Utf8, "REPORT_DATE": pl.Date})
+    if not per_table:
+        return empty
+
+    out = pl.concat([t.select(["vt_symbol", "REPORT_DATE"])
+                     for t in per_table.values()]).unique()
+    notice_cols = []
+    for table, dedup in per_table.items():
+        renamed = dedup.rename({"notice_eff": f"_notice_{table}"})
+        notice_cols.append(f"_notice_{table}")
+        out = out.join(renamed, on=["vt_symbol", "REPORT_DATE"], how="left")
+    # 有效披露日 = 三表 NOTICE_DATE 行最大(最晚可见,保守不提前)
+    return out.with_columns(pl.max_horizontal(notice_cols).alias("notice_eff"))
+
+
+# ==================== 报告期级指标预计算 ====================
+# 所有 shift/rolling 必须在 .over(_SYM) 组内执行(跨股串行 = 致命错误)。
+
+def _qidx() -> pl.Expr:
+    """连续季度索引: year*4 + quarter(月 3/6/9/12 → 1/2/3/4)."""
+    return (pl.col("REPORT_DATE").dt.year() * 4
+            + (pl.col("REPORT_DATE").dt.month() - 1) // 3 + 1)
+
+
+def _single_quarter(col: str) -> pl.Expr:
+    """R1 单季化: Q1 直接取累计;其余要求上期恰为上一季度(同年)做差,否则 NaN."""
+    cur = pl.col(col)
+    prev = cur.shift(1)
+    prev_ok = (pl.col("_qidx") - pl.col("_qidx").shift(1)) == 1
+    return (
+        pl.when(cur.is_null()).then(None)
+        .when(pl.col("REPORT_DATE").dt.month() == 3).then(cur)
+        .when(prev_ok & prev.is_not_null()).then(cur - prev)
+        .otherwise(None)
+    ).over(_SYM)
+
+
+def _ttm_of(col: str) -> pl.Expr:
+    """R2 TTM = 连续 4 个报告期单季之和(窗内任一单季 NaN → NaN)."""
+    q = pl.col(col)
+    window_ok = (pl.col("_qidx") - pl.col("_qidx").shift(3)) == 3
+    s = q + q.shift(1) + q.shift(2) + q.shift(3)
+    return pl.when(window_ok).then(s).otherwise(None).over(_SYM)
+
+
+def _yoy4(col: str) -> pl.Expr:
+    """yoy = X_t / X_{t−4季} − 1;基期缺失/≤0 → NaN(负基数 yoy 无意义)."""
+    cur, base = pl.col(col), pl.col(col).shift(4)
+    ok = (pl.col("_qidx") - pl.col("_qidx").shift(4)) == 4
+    return (
+        pl.when(ok & base.is_not_null() & (base > 0) & cur.is_not_null())
+        .then(cur / base - 1.0).otherwise(None)
+    ).over(_SYM)
+
+
+def _delta4(col: str) -> pl.Expr:
+    """ΔX = X_t − X_{t−4季}(要求恰好隔 4 个季度)."""
+    ok = (pl.col("_qidx") - pl.col("_qidx").shift(4)) == 4
+    return pl.when(ok).then(pl.col(col) - pl.col(col).shift(4)).otherwise(None).over(_SYM)
+
+
+def _safe_ratio(num: pl.Expr, den: pl.Expr) -> pl.Expr:
+    """分母缺失/为 0 → NaN(比率类通用守卫)."""
+    return pl.when(den.is_not_null() & (den != 0)).then(num / den).otherwise(None)
+
+
+def _compute_report_features(reports: pl.DataFrame) -> pl.DataFrame:
+    """报告期宽表 → 全部报告期级特征(逐级 with_columns,over 组内时序)."""
+    df = reports.sort([_SYM, "REPORT_DATE"]).with_columns(_qidx().alias("_qidx"))
+
+    # 第一级: 单季化 + TTM(累计列)
+    df = df.with_columns(
+        [_single_quarter(c).alias(f"q_{c}") for c in _CUM_MAP.values()]
+    ).with_columns(
+        [_ttm_of(f"q_{c}").alias(f"ttm_{c}") for c in _CUM_MAP.values()]
+    )
+
+    # 第二级: IBD(缺组件按 0) + 金融股判定(银行模板无营业成本) + 毛利
+    ibd = pl.sum_horizontal([pl.col(p).fill_null(0.0) for p in _IBD_PARTS])
+    is_fin = pl.col("cogs").is_null() | (pl.col("cogs") == 0)
+    df = df.with_columns(
+        ibd.alias("_ibd"),
+        is_fin.alias("_is_fin"),
+        (pl.col("ttm_rev") - pl.col("ttm_cogs")).alias("_gp_ttm"),
+    )
+
+    # 第三级: 行本地比率(无时序,无需 over)
+    eq, ta = pl.col("TOTAL_PARENT_EQUITY"), pl.col("TOTAL_ASSETS")
+    df = df.with_columns(
+        # 盈利能力 A
+        _safe_ratio(pl.col("ttm_np"), eq).alias("roe_ttm"),
+        _safe_ratio(pl.col("ttm_dnp"), eq).alias("roe_deduct_ttm"),
+        _safe_ratio(pl.col("ttm_np"), ta).alias("roa_ttm"),
+        _safe_ratio(pl.col("_gp_ttm"), ta).alias("gp_over_assets"),
+        _safe_ratio(pl.col("_gp_ttm"), pl.col("ttm_rev")).alias("gross_margin"),
+        _safe_ratio(pl.col("ttm_np"), pl.col("ttm_rev")).alias("net_margin"),
+        _safe_ratio(pl.col("ttm_cfo"), ta).alias("cfo_over_assets"),
+        # 盈利质量 B
+        _safe_ratio(pl.col("ttm_np") - pl.col("ttm_cfo"), ta).alias("tacc"),
+        _safe_ratio(pl.col("ttm_np") - pl.col("ttm_dnp"),
+                    pl.col("ttm_np").abs()).alias("nonrec_ratio"),
+        _safe_ratio((pl.col("ttm_asset_imp") + pl.col("ttm_credit_imp")).abs(),
+                    ta).alias("impairment_ratio"),
+        _safe_ratio(pl.col("ttm_invest_inc") + pl.col("ttm_fv_inc"),
+                    pl.col("ttm_tp").abs()).alias("invest_income_dep"),
+        _safe_ratio(pl.col("ttm_sales_cash"), pl.col("ttm_rev")).alias("sales_cash_ratio"),
+        _safe_ratio(pl.col("OTHER_RECE"), ta).alias("other_rece_ratio"),
+        # 成长 C / 资本结构 E
+        _safe_ratio(pl.col("_ibd"), ta).alias("ibd_ratio"),
+        _safe_ratio(pl.col("GOODWILL"), ta).alias("goodwill_ratio"),
+    )
+
+    # 第四级: 跨期差分/同比/剪刀差(over 组内时序)
+    df = df.with_columns(
+        _yoy4("q_rev").alias("rev_q_yoy"),
+        _yoy4("q_np").alias("np_q_yoy"),
+        _yoy4("ACCOUNTS_RECE").alias("_ar_yoy"),
+        _yoy4("ttm_rev").alias("_rev_ttm_yoy"),
+        _yoy4("SHARE_CAPITAL").alias("nsi"),
+        _yoy4("TOTAL_ASSETS").alias("asset_growth"),
+        _delta4("gross_margin").alias("gm_delta"),
+        _delta4("roe_ttm").alias("roe_delta"),
+        _delta4("q_np").alias("_diff4_np"),
+        _delta4("q_rev").alias("_diff4_rev"),
+    ).with_columns(
+        (pl.col("np_q_yoy") - pl.col("rev_q_yoy")).alias("growth_scissors"),
+        (pl.col("_ar_yoy") - pl.col("_rev_ttm_yoy")).alias("receivables_anomaly"),
+    )
+
+    # 第五级: SUE(Foster 标准化)= diff4 / std(过去 8 期 diff4, ddof=1)
+    for src, out in (("_diff4_np", "sue_np"), ("_diff4_rev", "sue_rev")):
+        sd = pl.col(src).rolling_std(window_size=8, ddof=1).over(_SYM)
+        df = df.with_columns(
+            pl.when(sd.is_not_null() & (sd > 0) & pl.col(src).is_not_null())
+            .then(pl.col(src) / sd).otherwise(None).alias(out)
+        )
+
+    # 金融股: 盈利质量/成长族特征置 NaN(§7 红线 5)
+    df = df.with_columns([
+        pl.when(pl.col("_is_fin")).then(None).otherwise(pl.col(c)).alias(c)
+        for c in _FIN_NULL_COLS
+    ])
+
+    # 输出别名(估值/资本行为因子的日频原料)
+    df = df.with_columns(
+        pl.col("ttm_np").alias("np_ttm"),
+        pl.col("ttm_dnp").alias("dnp_ttm"),
+        pl.col("ttm_cfo").alias("cfo_ttm"),
+        pl.col("ttm_acc_inv_cash").alias("acc_invest_cash_ttm"),
+        pl.col("TOTAL_PARENT_EQUITY").alias("equity"),
+        pl.col("SHARE_CAPITAL").alias("share_capital"),
+    )
+    return df
+
+
+# ==================== forecast 事件层 ====================
+
+def _load_forecast_events(codes: list[str], data_dir: str) -> pl.DataFrame:
+    """forecast 按期文件 → (vt_symbol, eff=公告日期, type_score, change_pct) 事件行.
+
+    一股一公告日多行(按预测指标): 归母净利润行优先(含"净利润"且不含"扣"),
+    无净利润行 fallback 任意行。同股多公告日全保留(asof 取最新)。
+    """
+    schema = {"vt_symbol": pl.Utf8, "eff": pl.Date,
+              "forecast_type_score": pl.Float64, "forecast_change_pct": pl.Float64}
+    fc_dir = os.path.join(data_dir, "forecast")
+    if not os.path.isdir(fc_dir):
+        return pl.DataFrame(schema=schema)
+    code_set = set(codes)
+    frames = []
+    for fname in sorted(os.listdir(fc_dir)):
+        if not fname.endswith(".parquet"):
+            continue
+        try:
+            f = pl.read_parquet(os.path.join(fc_dir, fname))
+        except Exception:
+            continue
+        if f.height == 0 or not all(c in f.columns for c in
+                                    ("股票代码", "预告类型", "公告日期")):
+            continue
+        code = pl.col("股票代码").cast(pl.Utf8).str.strip_chars().str.zfill(6)
+        vt = (pl.when(code.str.starts_with("60")).then(code + pl.lit(".SSE"))
+              .otherwise(code + pl.lit(".SZSE")).alias("vt_symbol"))
+        if "预测指标" in f.columns:
+            ind = pl.col("预测指标").cast(pl.Utf8)
+            pref = (ind.str.contains("净利润") & ~ind.str.contains("扣")).cast(pl.Int32)
+        else:
+            pref = pl.lit(0, pl.Int32)
+        pct = (pl.col("业绩变动幅度").cast(pl.Float64, strict=False)
+               if "业绩变动幅度" in f.columns else pl.lit(None, pl.Float64))
+        f = f.with_columns(
+            vt,
+            pref.alias("_pref"),
+            pct.alias("_pct"),
+            pl.col("公告日期").cast(pl.Date, strict=False).alias("eff"),
+            pl.col("预告类型").cast(pl.Utf8).replace(
+                FORECAST_TYPE_SCORE, default=None, return_dtype=pl.Float64
+            ).alias("_score"),
+        ).filter(pl.col("vt_symbol").is_in(code_set) & pl.col("eff").is_not_null())
+        if f.height:
+            frames.append(f.select(["vt_symbol", "eff", "_pref", "_score", "_pct"]))
+    if not frames:
+        return pl.DataFrame(schema=schema)
+    fc = pl.concat(frames).sort(["vt_symbol", "eff", "_pref"])
+    # 同 (vt_symbol, eff) 取优先级最高行(_pref 大者排序在后 → last)
+    return fc.group_by(["vt_symbol", "eff"]).agg(
+        pl.col("_score").last().alias("forecast_type_score"),
+        pl.col("_pct").last().alias("forecast_change_pct"),
+    )
+
+
+# ==================== 对外主入口 ====================
+
+# 分块大小: NAS 7.9G 物理内存下,全市场 grid(5555股×2670日×33列≈4G)单次
+# join_asof + 全量副本必 OOM;按股分批使峰值 ≈ 事件表(常驻) + 单批 grid + 输出累积
+BATCH_CODES = 500
+
+
+def _prepare_days(start: str, end: str, trading_dates) -> list:
+    """交易日/日历日序列(一次解析,各批共用)."""
+    if trading_dates is not None:
+        days = pl.Series("datetime", trading_dates).cast(pl.Datetime("us"))
+        return days.unique().sort().to_list()
+    s = datetime.strptime(start, "%Y-%m-%d")
+    e = datetime.strptime(end, "%Y-%m-%d")
+    return pl.datetime_range(s, e, interval="1d", eager=True).cast(
+        pl.Datetime("us")).to_list()
+
+
+def _build_grid(codes: list[str], day_list: list) -> pl.DataFrame:
+    """单批日频 grid: codes × day_list(批内全量,批间无全量 grid 副本)."""
+    return pl.DataFrame({
+        "vt_symbol": pl.Series([c for c in codes for _ in day_list], dtype=pl.Utf8),
+        "datetime": pl.Series(day_list * len(codes), dtype=pl.Datetime("us")),
+    }, schema={"vt_symbol": pl.Utf8, "datetime": pl.Datetime("us")})
+
+
+def _load_feature_events(codes: list[str], data_dir: str) -> tuple[pl.DataFrame, pl.DataFrame]:
+    """报告期特征事件 + forecast 事件(全 codes 一次加载,分块 join 共用右表)."""
+    stmt_cols = [c for c in FEATURE_COLUMNS if c not in _FORECAST_COLS]
+    reports = _load_statements(codes, data_dir)
+    if reports.height == 0:
+        stmt_events = pl.DataFrame(schema={
+            "vt_symbol": pl.Utf8, "eff": pl.Datetime("us"), **{c: pl.Float64 for c in stmt_cols}})
+    else:
+        feat = _compute_report_features(reports)
+        # NOTICE_DATE 缺失报告期整期跳过(红线: 宁缺毋假)
+        feat = feat.filter(pl.col("notice_eff").is_not_null())
+        stmt_events = feat.select(
+            pl.col("vt_symbol"),
+            pl.col("notice_eff").cast(pl.Datetime("us")).alias("eff"),
+            *stmt_cols,
+        ).sort("eff")
+    fc_events = _load_forecast_events(codes, data_dir).select(
+        pl.col("vt_symbol"),
+        pl.col("eff").cast(pl.Datetime("us")),
+        *_FORECAST_COLS,
+    ).sort("eff")
+    return stmt_events, fc_events
+
+
+def iter_fundamental_feature_chunks(
+    codes: list[str],
+    start: str,
+    end: str,
+    data_dir: str = DEFAULT_STATIC_DIR,
+    trading_dates: pl.Series | list | None = None,
+    batch_codes: int = BATCH_CODES,
+):
+    """按 vt_symbol 分批产出 PIT 日频特征块(生成器,NAS 全量防 OOM 主入口).
+
+    每块 = 一批 codes × 全部日期 × FEATURE_COLUMNS,顺序即 codes 列表顺序;
+    批内 grid 用完即弃,事件右表(报告期+forecast)全批共用仅此一份。
+    batch_eval 侧应逐块 join alpha_df 分片后 concat,避免持有本帧全量副本。
+    """
+    stmt_events, fc_events = _load_feature_events(codes, data_dir)
+    day_list = _prepare_days(start, end, trading_dates)
+    for i in range(0, len(codes), batch_codes):
+        chunk_codes = codes[i:i + batch_codes]
+        grid = _build_grid(chunk_codes, day_list).sort("datetime")
+        # join_asof 前已显式按键排序;polars 1.42 用 by 分组时无法校验 sortedness,
+        # 该提示无信息量,就地抑制(sort 即正确性保险)
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", UserWarning)
+            out = grid.join_asof(
+                stmt_events, left_on="datetime", right_on="eff",
+                by="vt_symbol", strategy="backward")
+            out = out.sort("datetime").join_asof(
+                fc_events, left_on="datetime", right_on="eff",
+                by="vt_symbol", strategy="backward")
+        yield out.sort(["vt_symbol", "datetime"]).select(
+            ["vt_symbol", "datetime", *FEATURE_COLUMNS])
+        grid = out = None  # 批间释放(下一批重绑定)
+
+
+def build_fundamental_features(
+    codes: list[str],
+    start: str,
+    end: str,
+    data_dir: str = DEFAULT_STATIC_DIR,
+    trading_dates: pl.Series | list | None = None,
+    batch_codes: int = BATCH_CODES,
+) -> pl.DataFrame:
+    """构建 PIT 日频财务特征: vt_symbol × datetime × FEATURE_COLUMNS.
+
+    Args:
+        codes: vt_symbol 列表(如 "600000.SSE")
+        start/end: "YYYY-MM-DD" 窗口(trading_dates=None 时生成日历日 grid)
+        data_dir: 静态域根目录(NAS=/volume1/stock/sanguo_vnpy_v2/data/static)
+        trading_dates: 交易日子集(传 bars 的 unique datetime 免造非交易日行)
+        batch_codes: 按股分批大小(全量防 OOM;测试可调小验分块等值)
+
+    Returns:
+        每行 = 决策日可见的最新报告期特征(NOTICE_DATE ≤ 决策日,asof 前向填充)。
+    """
+    schema = {"vt_symbol": pl.Utf8, "datetime": pl.Datetime("us"),
+              **{c: pl.Float64 for c in FEATURE_COLUMNS}}
+    if not codes:
+        return pl.DataFrame(schema=schema)
+    return pl.concat(
+        iter_fundamental_feature_chunks(
+            codes, start, end, data_dir=data_dir,
+            trading_dates=trading_dates, batch_codes=batch_codes),
+        how="vertical")
+

sanguo_factor/fundamental_library.py

@@ -0,0 +1,80 @@
+"""财务因子批 P0 表达式库: 32 个因子注册(category="fundamental").
+
+选型来源 docs/fundamental_factor_survey_20260907.md(§3 候选池 96 个 + §4.1 去重),
+首批 32 = P0 50 个按族去重后的主代表;全部「因子 = cs_rank(基础指标)」一层截面,
+负 IC 因子表达式取负(统一高=好,与量价批 IC 口径一致;原始方向见下表)。
+
+基础指标列由 fundamental_adapter.build_fundamental_features 产出并 join 到
+alpha_df;估值族市值 = close × share_capital 自算(三表股本,规避 valuation
+中文列名表,PIT 口径与报表一致)。
+
+P0 32 因子名单(编号=调研文档 §3):
+  A 盈利能力(7): A01 ROE / A02 扣非ROE / A04 ROA / A05 GP/A(族主代表) /
+      A06 毛利率 / A07 净利率 / A13 CFO/TA
+  B 盈利质量(7): B01 TACC(Sloan,族核心) / B04 非经常占比 / B05 减值冲击 /
+      B07 投资收益依赖 / B08 应收异常 / B10 销售收现率 / B11 其他应收占比
+  C 成长(6): C01 营收单季同比 / C02 净利单季同比 / C09 增速剪刀差 /
+      C10 ΔGM(副代表) / C12 ΔROE / C15 资产增速(投资-融资群代表)
+  D 估值(4): D01 EP_TTM / D03 扣非EP(A股特色) / D04 BP / D06 CP(CNE5 CETOP)
+  E 资本结构/行为(4): E01 NSI 净股票发行 / E02 股权融资强度 / E05 有息负债率 /
+      E09 商誉占比
+  F 预期事件(4): F01 SUE(净利,Foster) / F02 SUE(营收) / F04 预告类型分 /
+      F05 预告幅度
+同族被裁变体(A03/A08/A09/A10/A12/B02/B06/C03/C04/C05/C06/D02/D05/E04 等)
+在 P1 或互证后再议,见文档 §4。
+"""
+from .registry import register_factor, _REGISTRY
+
+# (name, expression, 文档编号, 原始 IC 方向)
+FUNDAMENTAL_FACTORS: list[tuple[str, str, str, str]] = [
+    # ---- A 盈利能力(7) ----
+    ("fund_roe_ttm", "cs_rank(roe_ttm)", "A01", "+"),
+    ("fund_roe_deduct_ttm", "cs_rank(roe_deduct_ttm)", "A02", "+"),
+    ("fund_roa_ttm", "cs_rank(roa_ttm)", "A04", "+"),
+    ("fund_gp_over_assets", "cs_rank(gp_over_assets)", "A05", "+"),
+    ("fund_gross_margin", "cs_rank(gross_margin)", "A06", "+"),
+    ("fund_net_margin", "cs_rank(net_margin)", "A07", "+"),
+    ("fund_cfo_over_assets", "cs_rank(cfo_over_assets)", "A13", "+"),
+    # ---- B 盈利质量(7) ----
+    ("fund_tacc", "cs_rank(-tacc)", "B01", "-"),
+    ("fund_nonrec_ratio", "cs_rank(-nonrec_ratio)", "B04", "-"),
+    ("fund_impairment_ratio", "cs_rank(-impairment_ratio)", "B05", "-"),
+    ("fund_invest_income_dep", "cs_rank(-invest_income_dep)", "B07", "-"),
+    ("fund_receivables_anomaly", "cs_rank(-receivables_anomaly)", "B08", "-"),
+    ("fund_sales_cash_ratio", "cs_rank(sales_cash_ratio)", "B10", "+"),
+    ("fund_other_rece_ratio", "cs_rank(-other_rece_ratio)", "B11", "-"),
+    # ---- C 成长(6) ----
+    ("fund_rev_q_yoy", "cs_rank(rev_q_yoy)", "C01", "+"),
+    ("fund_np_q_yoy", "cs_rank(np_q_yoy)", "C02", "+"),
+    ("fund_growth_scissors", "cs_rank(growth_scissors)", "C09", "+"),
+    ("fund_gm_delta", "cs_rank(gm_delta)", "C10", "+"),
+    ("fund_roe_delta", "cs_rank(roe_delta)", "C12", "+"),
+    ("fund_asset_growth", "cs_rank(-asset_growth)", "C15", "-"),
+    # ---- D 估值(4) ----
+    ("fund_ep_ttm", "cs_rank(np_ttm / (close * share_capital))", "D01", "+"),
+    ("fund_ep_deduct_ttm", "cs_rank(dnp_ttm / (close * share_capital))", "D03", "+"),
+    ("fund_bp", "cs_rank(equity / (close * share_capital))", "D04", "+"),
+    ("fund_cp", "cs_rank(cfo_ttm / (close * share_capital))", "D06", "+"),
+    # ---- E 资本结构/行为(4) ----
+    ("fund_nsi", "cs_rank(-nsi)", "E01", "-"),
+    ("fund_equity_fin_intensity",
+     "cs_rank(-(acc_invest_cash_ttm / (close * share_capital)))", "E02", "-"),
+    ("fund_ibd_ratio", "cs_rank(-ibd_ratio)", "E05", "-"),
+    ("fund_goodwill_ratio", "cs_rank(-goodwill_ratio)", "E09", "-"),
+    # ---- F 预期事件(4) ----
+    ("fund_sue_np", "cs_rank(sue_np)", "F01", "+"),
+    ("fund_sue_rev", "cs_rank(sue_rev)", "F02", "+"),
+    ("fund_forecast_type", "cs_rank(forecast_type_score)", "F04", "+"),
+    ("fund_forecast_change", "cs_rank(forecast_change_pct)", "F05", "+"),
+]
+
+
+def _register_all() -> None:
+    """注册全部财务因子(已存在同名跳过,幂等;同 library.py 模式)."""
+    for name, expression, _doc_id, _ic in FUNDAMENTAL_FACTORS:
+        if name not in _REGISTRY:
+            register_factor(name, expression, category="fundamental")
+
+
+# 模块导入时自动注册(与 library.py/alpha_datasets.py 同一模式)
+_register_all()

scripts/factor_research/import_runs_to_main_db.py

@@ -0,0 +1,156 @@
+#!/usr/bin/env python
+"""把独立因子评估库的指定 run 导入主 factor_eval.db(幂等追加).
+
+背景: 财务批在 NAS 独立库 /volume1/stock/factor_eval_fundamental.db 产出
+(fund-p0-h1 / fund-p0-h2 / quant62_ref, category=fundamental),前端
+Leaderboard 只读主库(默认 factor_eval.db),导入后即可与量价类别并列展示。
+
+用法:
+  python scripts/factor_research/import_runs_to_main_db.py \
+    --src /volume1/stock/factor_eval_fundamental.db \
+    --dst /volume1/stock/sanguo_vnpy_v2/data_backup/factor_eval.db \
+    --run-ids fund-p0-h1,fund-p0-h2,quant62_ref
+  (--dst 省略时用 default_eval_db_path(),容器内即主库默认位)
+
+语义:
+  - 对 --run-ids 每个 run_id,src 的 eval_runs 行 + 该 run 全部 eval_results
+    行 INSERT OR REPLACE 进 dst;列取 dst 实际 PRAGMA table_info(不手写 schema),
+    src 只需含 dst 的列即可;
+  - dst 中不属于这些 run_id 的行一律不动(纯追加语义,重复执行幂等);
+  - 写前只读预检:逐 run 打印 src 将导入/dst 将被替换行数;任一 run_id 在
+    src 不存在则报错退出、零写入。
+"""
+import argparse
+import os
+import sqlite3
+import sys
+
+_REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+TABLES = ("eval_runs", "eval_results")
+
+
+def _connect_ro(path: str) -> sqlite3.Connection:
+    conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=30)
+    conn.execute("PRAGMA busy_timeout=30000")
+    return conn
+
+
+def _connect_rw(path: str) -> sqlite3.Connection:
+    conn = sqlite3.connect(path, timeout=30)
+    conn.execute("PRAGMA busy_timeout=30000")
+    return conn
+
+
+def _require_table(conn: sqlite3.Connection, table: str, db_tag: str) -> None:
+    row = conn.execute(
+        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,)
+    ).fetchone()
+    if row is None:
+        raise SystemExit(f"[错误] {db_tag} 缺表 {table}(不是 factor_eval 库?)")
+
+
+def _table_cols(conn: sqlite3.Connection, table: str) -> list[str]:
+    return [r[1] for r in conn.execute(f"PRAGMA table_info({table})")]
+
+
+def _count(conn: sqlite3.Connection, table: str, run_id: str) -> int:
+    return conn.execute(
+        f"SELECT COUNT(*) FROM {table} WHERE run_id=?", (run_id,)
+    ).fetchone()[0]
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(
+        description="因子评估 run 跨库导入(独立库→主库,幂等追加)")
+    ap.add_argument("--src", required=True, help="源 factor_eval 库(只读打开)")
+    ap.add_argument("--dst", default=None,
+                    help="主 factor_eval 库;默认 default_eval_db_path()")
+    ap.add_argument("--run-ids", required=True, help="逗号分隔 run_id 列表")
+    args = ap.parse_args()
+
+    run_ids = [s.strip() for s in args.run_ids.split(",") if s.strip()]
+    if not run_ids:
+        print("[错误] --run-ids 为空", file=sys.stderr)
+        return 1
+
+    dst = args.dst
+    if dst is None:
+        sys.path.insert(0, _REPO_ROOT)
+        from sanguo_factor.eval_store import default_eval_db_path
+        dst = default_eval_db_path()
+
+    src = _connect_ro(args.src)
+    dstc = _connect_rw(dst)
+    try:
+        for table in TABLES:
+            _require_table(src, table, f"src={args.src}")
+            _require_table(dstc, table, f"dst={dst}")
+
+        # —— 只读预检:src 将导入 / dst 将被替换 —— #
+        print(f"[预检] src={args.src}")
+        print(f"[预检] dst={dst}")
+        missing = []
+        plan: list[tuple[str, int, int]] = []  # (run_id, src_runs, src_results)
+        for rid in run_ids:
+            n_runs = _count(src, "eval_runs", rid)
+            n_res = _count(src, "eval_results", rid)
+            if n_runs == 0 and n_res == 0:
+                missing.append(rid)
+                continue
+            plan.append((rid, n_runs, n_res))
+            print(f"[预检] {rid}: src runs={n_runs} results={n_res} | "
+                  f"dst 已有 runs={_count(dstc, 'eval_runs', rid)} "
+                  f"results={_count(dstc, 'eval_results', rid)}(同主键将被替换)")
+        if missing:
+            print(f"[错误] 以下 run_id 在 src 不存在,零写入退出: {missing}",
+                  file=sys.stderr)
+            return 1
+
+        tot_runs = sum(p[1] for p in plan)
+        tot_res = sum(p[2] for p in plan)
+        print(f"[预检] 将导入 eval_runs {tot_runs} 行 + eval_results {tot_res} 行 "
+              f"(INSERT OR REPLACE,dst 其他行不动)")
+
+        # —— 写入:单事务包两表,任一失败整体回滚 —— #
+        with dstc:
+            for table in TABLES:
+                dst_cols = _table_cols(dstc, table)
+                src_cols = set(_table_cols(src, table))
+                absent = [c for c in dst_cols if c not in src_cols]
+                if absent:
+                    print(f"[错误] src.{table} 缺列 {absent}(src schema 过旧),"
+                          "零写入退出", file=sys.stderr)
+                    return 1
+                placeholders = ",".join("?" for _ in dst_cols)
+                col_list = ",".join(dst_cols)
+                rows = src.execute(
+                    f"SELECT {col_list} FROM {table} WHERE run_id "
+                    f"IN ({','.join('?' for _ in run_ids)})",
+                    run_ids,
+                ).fetchall()
+                dstc.executemany(
+                    f"INSERT OR REPLACE INTO {table}({col_list}) "
+                    f"VALUES({placeholders})", rows)
+                print(f"[写入] {table}: {len(rows)} 行")
+    finally:
+        src.close()
+        dstc.close()
+
+    # —— 核验(幂等口径:再跑一次行数不变) —— #
+    check = _connect_ro(dst)
+    try:
+        marks = ",".join("?" for _ in run_ids)
+        n_runs = check.execute(
+            f"SELECT COUNT(*) FROM eval_runs WHERE run_id IN ({marks})",
+            run_ids).fetchone()[0]
+        n_res = check.execute(
+            f"SELECT COUNT(*) FROM eval_results WHERE run_id IN ({marks})",
+            run_ids).fetchone()[0]
+    finally:
+        check.close()
+    print(f"[核验] dst 中这些 run 现有: eval_runs={n_runs} eval_results={n_res}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/factor_research/run_eval.py

@@ -8,6 +8,9 @@
       --factors alpha2,alpha6,alpha12,alpha18,kmid,klen,roc_5,ma_20,std_20,wvma_20
   全量(Alpha101+158 × 全A × 8.5年):
     venv310/bin/python scripts/factor_research/run_eval.py --label batch1-full
+  财务批(32因子;静态域默认 /volume1/stock/sanguo_vnpy_v2/data/static 容器内外同路径):
+    venv310/bin/python scripts/factor_research/run_eval.py \
+      --start 2024-01-01 --end 2026-06-30 --categories fundamental --label fund-p0
 """
 import argparse
 import os
@@ -28,6 +31,10 @@ def main() -> int:
     ap.add_argument("--label", default="batch1")
     ap.add_argument("--db", default=None, help="factor_eval.db 路径;默认 default_eval_db_path()")
     ap.add_argument("--run-id", default=None, help="断点续跑:复用既有 run_id,跳过已落库因子")
+    ap.add_argument("--fund-data-dir", default=None,
+                    help="财务静态域根目录;默认 cfg.data_paths['static_dir'] → "
+                         "/volume1/stock/sanguo_vnpy_v2/data/static(容器内外同路径),"
+                         "仅含 fundamental 因子时才读取")
     ap.add_argument("--list-factors", default="", metavar="CATEGORY", help="列出类目因子后退出")
     args = ap.parse_args()
 
@@ -58,7 +65,8 @@ def main() -> int:
         print(f"[eval] {done}/{total} {current}", flush=True)
 
     out = run_batch_eval(factor_names, args.start, args.end, db_path, label=args.label,
-                         symbols=symbols, limit=args.limit, cfg=None, progress_cb=_cb, run_id=args.run_id)
+                         symbols=symbols, limit=args.limit, cfg=None, progress_cb=_cb,
+                         run_id=args.run_id, fund_data_dir=args.fund_data_dir)
 
     print(f"[eval] run_id={out['run_id']} done={out['factors_done']} "
           f"errors={len(out['errors'])} elapsed={out['elapsed_sec']}s symbols={out['symbols_count']}")

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

tests/factor/conftest.py

@@ -6,3 +6,137 @@ import os
 _VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vnpy_v4.4.0"))
 if _VNPY_SRC not in sys.path:
     sys.path.insert(0, _VNPY_SRC)
+
+
+# ==================== 合成财务静态域(财务因子批测试共用) ====================
+# 3 只股 × 16 报告期(2020Q1~2023Q4,2020 为 SUE 滚动窗预热),NOTICE_DATE 错位:
+#   Q1→当年4-28 / H1→当年8-29 / Q3→当年10-27 / 年报→次年4-25
+# 数值全部手算可验证(断言用),公式见各 build 函数内注释。
+from datetime import date
+
+import polars as pl
+
+REPORT_DATES = [
+    f"{y}-{m}" for y in (2020, 2021, 2022, 2023) for m in ("03-31", "06-30", "09-30", "12-31")
+]
+
+# 单季 REV/NP 序列(累计=年内前缀和): 报告期索引 i=0..15(2020Q1..2023Q4)
+# 2020 为 SUE 滚动窗预热史;2021+ 的断言数值由 2021 段起算
+REV_Q = [90, 130, 140, 150] + [100, 120, 150, 160, 110, 140, 150, 180, 130, 140, 170, 180]
+NP_Q = [9, 13, 14, 15] + [10, 12, 15, 16, 11, 14, 15, 18, 13, 14, 17, 18]
+
+# 资产负债表(时点存量,随报告期线性演化;2021Q1 起算,idx=i-4)
+TA_OF = lambda i: 1000 + 50 * max(i - 4, 0)
+EQ_OF = lambda i: 500 + 20 * max(i - 4, 0)
+AR_OF = lambda i: 100 + 10 * max(i - 4, 0)
+SC_OF = lambda i: 100 if i < 12 else 110  # 2023 起股本扩张 10%(NSI=0.1)
+
+SYN_STOCKS = {
+    "600000.SH": {"vt": "600000.SSE", "scale": 1.0,
+                  "drop_periods": [], "null_notice_periods": []},
+    "000001.SZ": {"vt": "000001.SZSE", "scale": 1.0,
+                  "drop_periods": ["2022-03-31"],          # 缺 2022Q1 → 单季差分/TTM 链 NaN
+                  "null_notice_periods": ["2022-09-30"]},  # NOTICE_DATE 缺失 → 该报告期跳过
+    "300001.SZ": {"vt": "300001.SZSE", "scale": 2.0,
+                  "drop_periods": [], "null_notice_periods": []},
+    # 三只分块等值测试扩容股(全史无残缺,不同 scale 增截面多样性)
+    "600004.SH": {"vt": "600004.SSE", "scale": 0.5,
+                  "drop_periods": [], "null_notice_periods": []},
+    "000333.SZ": {"vt": "000333.SZSE", "scale": 1.7,
+                  "drop_periods": [], "null_notice_periods": []},
+    "300124.SZ": {"vt": "300124.SZSE", "scale": 3.0,
+                  "drop_periods": [], "null_notice_periods": []},
+}
+
+
+def _notice_date(report_date: str) -> str:
+    """A 股典型披露节奏(年报次年 4-25 / 一季报 4-28 / 中报 8-29 / 三季报 10-27)."""
+    y, m = int(report_date[:4]), int(report_date[5:7])
+    if m == 3:
+        return f"{y}-04-28"
+    if m == 6:
+        return f"{y}-08-29"
+    if m == 9:
+        return f"{y}-10-27"
+    return f"{y + 1}-04-25"
+
+
+def _cum_in_year(q_values: list[float], i: int) -> float:
+    """报告期 i 的年内累计值 = 当年前几季单季之和."""
+    year_start = (i // 4) * 4
+    return float(sum(q_values[year_start:i + 1]))
+
+
+def build_synthetic_static(root: str) -> str:
+    """写合成静态域 parquet 树(data_dir),返回 static 根目录路径.
+
+    目录结构与 NAS 一致: static/{income,balance,cashflow}/{code}.{SH|SZ}_{table}.parquet
+    forecast 按报告期全市场文件: static/forecast/{YYYYMMDD}_forecast.parquet
+    """
+    static_dir = os.path.join(str(root), "static")
+    for table in ("income", "balance", "cashflow"):
+        os.makedirs(os.path.join(static_dir, table), exist_ok=True)
+
+    for file_code, spec in SYN_STOCKS.items():
+        s, drop, null_notice = spec["scale"], set(spec["drop_periods"]), set(spec["null_notice_periods"])
+        income_rows, balance_rows, cashflow_rows = [], [], []
+        for i, rd in enumerate(REPORT_DATES):
+            if rd in drop:
+                continue
+            notice = None if rd in null_notice else _notice_date(rd)
+            rev_c, np_c = s * _cum_in_year(REV_Q, i), s * _cum_in_year(NP_Q, i)
+            common = {
+                "REPORT_DATE": f"{rd} 00:00:00",
+                "NOTICE_DATE": (f"{notice} 00:00:00" if notice else None),
+                "UPDATE_DATE": f"{notice} 00:00:00" if notice else None,
+            }
+            income_rows.append({**common,
+                "TOTAL_OPERATE_INCOME": rev_c, "OPERATE_COST": 0.6 * rev_c,
+                "PARENT_NETPROFIT": np_c, "DEDUCT_PARENT_NETPROFIT": 0.9 * np_c,
+                "TOTAL_PROFIT": 1.1 * np_c, "INVEST_INCOME": 0.05 * np_c,
+                "FAIRVALUE_CHANGE_INCOME": 0.01 * np_c,
+                "ASSET_IMPAIRMENT_LOSS": 0.02 * np_c, "CREDIT_IMPAIRMENT_LOSS": 0.01 * np_c})
+            # IBD 只给 SHORT_LOAN 一列(其余组件列缺失,测 schema 缺列容错)
+            balance_rows.append({**common,
+                "TOTAL_ASSETS": s * TA_OF(i), "TOTAL_PARENT_EQUITY": s * EQ_OF(i),
+                "ACCOUNTS_RECE": s * AR_OF(i), "OTHER_RECE": s * (5 + max(i - 4, 0)),
+                "GOODWILL": 50.0, "SHARE_CAPITAL": float(SC_OF(i)),
+                "SHORT_LOAN": s * (100 + max(i - 4, 0))})
+            cashflow_rows.append({**common,
+                "NETCASH_OPERATE": 1.2 * np_c, "SALES_SERVICES": 1.05 * rev_c,
+                "ACCEPT_INVEST_CASH": 0.1 * rev_c})
+
+        for table, rows in (("income", income_rows), ("balance", balance_rows), ("cashflow", cashflow_rows)):
+            pl.DataFrame(rows).write_parquet(
+                os.path.join(static_dir, table, f"{file_code}_{table}.parquet"))
+
+    # forecast: 按报告期全市场文件(中文列,归母净利润行优先 + 无净利润行 fallback)
+    os.makedirs(os.path.join(static_dir, "forecast"), exist_ok=True)
+    pl.DataFrame([
+        {"股票代码": "600000", "预测指标": "归属于上市公司股东的净利润",
+         "业绩变动幅度": 56.79, "预告类型": "预增", "公告日期": date(2023, 7, 15)},
+        {"股票代码": "000001", "预测指标": "归属于上市公司股东的净利润",
+         "业绩变动幅度": -30.0, "预告类型": "预减", "公告日期": date(2023, 7, 20)},
+        {"股票代码": "300001", "预测指标": "营业收入",
+         "业绩变动幅度": 5.0, "预告类型": "略增", "公告日期": date(2023, 7, 10)},
+    ]).write_parquet(os.path.join(static_dir, "forecast", "20230630_forecast.parquet"))
+    pl.DataFrame([
+        {"股票代码": "600000", "预测指标": "净利润",
+         "业绩变动幅度": 100.0, "预告类型": "扭亏", "公告日期": date(2023, 10, 15)},
+    ]).write_parquet(os.path.join(static_dir, "forecast", "20230930_forecast.parquet"))
+    return static_dir
+
+
+import pytest
+
+
+@pytest.fixture(scope="session")
+def synthetic_static(tmp_path_factory) -> str:
+    """session 级合成静态域根目录(test_fundamental_* 共用)."""
+    return build_synthetic_static(tmp_path_factory.mktemp("fund_static"))
+
+
+@pytest.fixture(scope="session")
+def np_q_series() -> list[float]:
+    """合成归母净利单季序列(SUE 期望值独立重算用)."""
+    return list(NP_Q)

tests/factor/test_fundamental_adapter.py

@@ -0,0 +1,218 @@
+# tests/factor/test_fundamental_adapter.py
+"""财务因子适配层:合成三表+forecast parquet → PIT 日频特征列.
+
+核心口径红线(docs/fundamental_factor_survey_20260907.md §7):
+- 单季差分缺上期 → NaN 不填 0
+- PIT = NOTICE_DATE ≤ 决策日,NOTICE_DATE 缺失行整报告期跳过
+- TTM 不足连续 4 季 → NaN
+"""
+import sys, os
+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 pytest
+
+from sanguo_factor.fundamental_adapter import build_fundamental_features, FEATURE_COLUMNS
+
+A, B, C = "600000.SSE", "000001.SZSE", "300001.SZSE"
+
+
+@pytest.fixture(scope="module")
+def feat(synthetic_static):
+    return build_fundamental_features(
+        [A, B, C], "2023-01-01", "2023-12-31", data_dir=synthetic_static)
+
+
+def _dt(day: str):
+    import datetime as _d
+    return _d.datetime.strptime(day, "%Y-%m-%d")
+
+
+def val(df, vt: str, day: str, col: str):
+    row = df.filter((df["vt_symbol"] == vt) & (df["datetime"] == _dt(day)))
+    assert row.height == 1, f"grid 缺行 {vt} {day}"
+    v = row[col][0]
+    return None if v is None else float(v)
+
+
+# ---------- 单季差分(R1) ----------
+
+def test_single_quarter_diff_and_q1_direct(feat):
+    # 报告期级断言走 equity 之外的特征:用 grid 在披露日后取值
+    # 2022H1 披露于 2022-08-29,2023 年窗口看不到;单季值通过 rev_q_yoy 间接锁
+    # 直接锁:2023Q3 (i=10) rev_q_yoy = 170/150 - 1(单季差分链正确才可得)
+    assert val(feat, A, "2023-10-27", "rev_q_yoy") == pytest.approx(170 / 150 - 1)
+
+
+def test_missing_prev_quarter_is_nan_not_zero(synthetic_static):
+    # B 缺 2022Q1 → 2022Q2 单季差分 NaN;传染:TTM 至 2023Q1、yoy 至 2023H1 均 NaN
+    df = build_fundamental_features([B], "2023-01-01", "2023-12-31", data_dir=synthetic_static)
+    # 2023Q1(i=8) 报告期 TTM 窗含 2022Q2(差分 NaN) → np_ttm NaN
+    assert val(df, B, "2023-04-28", "np_ttm") is None
+    # 2023H1(i=9) yoy 基期 = 2022H1 单季 NaN → NaN
+    assert val(df, B, "2023-08-29", "rev_q_yoy") is None
+    # 窗口滑出坏点后恢复:2023Q3(i=10) 基期 2022Q3 单季=15 有效
+    assert val(df, B, "2023-10-27", "rev_q_yoy") == pytest.approx(170 / 150 - 1)
+    assert val(df, B, "2023-08-29", "np_ttm") == pytest.approx(60.0)
+
+
+def test_ttm_rolling_four_quarters(feat):
+    # A 2023Q2 报告期(i=9): NP_TTM = 15+18+13+14 = 60(=上年年报580+H1−上年H1 交叉验证同值)
+    assert val(feat, A, "2023-08-29", "np_ttm") == pytest.approx(60.0)
+    # 年报 NOTICE=2024-04-25,2023 窗末 np_ttm 停留在 Q3 报告期(i=10): 18+13+14+17=62
+    assert val(feat, A, "2023-12-31", "np_ttm") == pytest.approx(62.0)
+
+
+# ---------- PIT(R3):NOTICE_DATE 前不可见 ----------
+
+def test_pit_no_lookahead(feat):
+    # A 2023H1(报告期 06-30) NOTICE=08-29;equity 平衡表时点值 i=9→680, i=8→660
+    assert val(feat, A, "2023-08-28", "equity") == pytest.approx(660.0)
+    assert val(feat, A, "2023-08-29", "equity") == pytest.approx(680.0)
+
+
+def test_notice_date_null_period_skipped(feat):
+    # B 2022Q3 NOTICE 缺失 → 整期跳过:2023-01-01 可见的最新披露 = 2022H1(i=5, EQ=600)
+    assert val(feat, B, "2023-01-01", "equity") == pytest.approx(600.0)
+    # 恢复:2022 年报(i=7) 2023-04-25 可见
+    assert val(feat, B, "2023-04-25", "equity") == pytest.approx(640.0)
+
+
+def test_forward_fill_between_notices(feat):
+    assert val(feat, A, "2023-09-15", "equity") == pytest.approx(680.0)
+    assert val(feat, A, "2023-10-26", "equity") == pytest.approx(680.0)
+    assert val(feat, A, "2023-10-27", "equity") == pytest.approx(700.0)
+    # 年报 NOTICE=次年 04-25,2023 窗末仍是 Q3 值
+    assert val(feat, A, "2023-12-31", "equity") == pytest.approx(700.0)
+
+
+# ---------- 报告期级指标公式 ----------
+
+def test_profitability_ratios(feat):
+    # A 2023H1(i=9): NP_TTM=60, EQ=680, TA=1450, CFO_TTM=72, GP_TTM=0.4*REV_TTM
+    assert val(feat, A, "2023-08-29", "roe_ttm") == pytest.approx(60 / 680)
+    assert val(feat, A, "2023-08-29", "roa_ttm") == pytest.approx(60 / 1450)
+    assert val(feat, A, "2023-08-29", "cfo_over_assets") == pytest.approx(72 / 1450)
+    assert val(feat, A, "2023-08-29", "gp_over_assets") == pytest.approx(240 / 1450)
+    assert val(feat, A, "2023-08-29", "gross_margin") == pytest.approx(0.4, abs=1e-9)
+    assert val(feat, A, "2023-08-29", "net_margin") == pytest.approx(60 / 600, rel=1e-6)
+    assert val(feat, A, "2023-08-29", "roe_deduct_ttm") == pytest.approx(0.9 * 60 / 680)
+
+
+def test_quality_ratios(feat):
+    assert val(feat, A, "2023-08-29", "tacc") == pytest.approx((60 - 72) / 1450)
+    assert val(feat, A, "2023-08-29", "nonrec_ratio") == pytest.approx(6 / 60)
+    # 减值 = abs(0.02+0.01)*NP_TTM / TA
+    assert val(feat, A, "2023-08-29", "impairment_ratio") == pytest.approx(1.8 / 1450)
+    # 投资收益依赖 = (0.05+0.01)*NP_TTM / abs(1.1*NP_TTM)
+    assert val(feat, A, "2023-08-29", "invest_income_dep") == pytest.approx(0.06 / 1.1)
+    assert val(feat, A, "2023-08-29", "sales_cash_ratio") == pytest.approx(1.05 * 600 / 600, rel=1e-6)
+    assert val(feat, A, "2023-08-29", "other_rece_ratio") == pytest.approx((5 + 9) / 1450)
+    # 应收异常 = AR同比 − REV_TTM 同比(2023H1 vs 2022H1; TTM 窗含 2022Q2..2023H1)
+    ar_yoy = (100 + 90) / (100 + 50) - 1
+    rev_ttm_yoy = 600 / (150 + 160 + 110 + 140) - 1
+    assert val(feat, A, "2023-08-29", "receivables_anomaly") == pytest.approx(ar_yoy - rev_ttm_yoy)
+
+
+def test_growth_and_capital_features(feat):
+    assert val(feat, A, "2023-08-29", "np_q_yoy") == pytest.approx(14 / 14 - 1)
+    assert val(feat, A, "2023-10-27", "np_q_yoy") == pytest.approx(17 / 15 - 1)
+    assert val(feat, A, "2023-08-29", "growth_scissors") == pytest.approx(0.0, abs=1e-9)
+    assert val(feat, A, "2023-08-29", "gm_delta") == pytest.approx(0.0, abs=1e-9)
+    # roe_delta = ROE(2023H1) − ROE(2022H1) = 60/680 − 56/600
+    assert val(feat, A, "2023-08-29", "roe_delta") == pytest.approx(60 / 680 - 56 / 600)
+    # 2023Q1(i=8): 资产增速 = 1400/1200−1; NSI = (110−100)/100
+    assert val(feat, A, "2023-04-28", "asset_growth") == pytest.approx(1400 / 1200 - 1)
+    assert val(feat, A, "2023-04-28", "nsi") == pytest.approx(0.1)
+    # IBD = SHORT_LOAN 唯一组件(其余列缺失按 0) = 109 (i=9)
+    assert val(feat, A, "2023-08-29", "ibd_ratio") == pytest.approx(109 / 1450)
+    assert val(feat, A, "2023-08-29", "goodwill_ratio") == pytest.approx(50 / 1450)
+
+
+def test_sue_foster_standardization(feat, np_q_series):
+    # 独立重算: diff4 = Q_t − Q_{t-4}, SUE = diff4 / std(过去 8 期 diff4, ddof=1)
+    import statistics
+    q = [float(x) for x in np_q_series]  # 16 期序列,2020 为 SUE 滚动窗预热
+    diff4 = [q[i] - q[i - 4] for i in range(4, 16)]  # diff4[r] ↔ 报告期 r+4
+
+    def sue_at(rep_idx: int):
+        r = rep_idx - 4
+        if r < 7:
+            return None
+        win = diff4[r - 7:r + 1]
+        return diff4[r] / statistics.stdev(win)
+
+    # 2023Q3(报告期 i=14, 披露 2023-10-27): 窗 diff4[3..10]
+    assert val(feat, A, "2023-10-27", "sue_np") == pytest.approx(sue_at(14))
+    # 前一日仍见 2023H1(i=13)
+    assert val(feat, A, "2023-10-26", "sue_np") == pytest.approx(sue_at(13))
+    # 2023Q4 披露在 2024-04-25,窗末仍是 Q3 值
+    assert val(feat, A, "2023-12-31", "sue_np") == pytest.approx(sue_at(14))
+    # 合成史 16 期 → i=11(2022Q4)起才有完整 8 期窗,更早报告期 SUE=NaN(间接受 PIT 保护)
+
+
+def test_forecast_event_features(feat):
+    # A: 2023-07-15 预增(+3, 56.79);2023-10-15 扭亏(+2, 100.0)覆盖
+    assert val(feat, A, "2023-07-14", "forecast_type_score") is None
+    assert val(feat, A, "2023-07-15", "forecast_type_score") == 3.0
+    assert val(feat, A, "2023-07-15", "forecast_change_pct") == pytest.approx(56.79)
+    assert val(feat, A, "2023-10-14", "forecast_type_score") == 3.0
+    assert val(feat, A, "2023-10-15", "forecast_type_score") == 2.0
+    # B 预减(−3); C 无净利润行 fallback 营业收入行 略增(+2)
+    assert val(feat, B, "2023-07-20", "forecast_type_score") == -3.0
+    assert val(feat, B, "2023-07-20", "forecast_change_pct") == pytest.approx(-30.0)
+    assert val(feat, C, "2023-07-10", "forecast_type_score") == 2.0
+
+
+# ---------- 结构与容错 ----------
+
+def test_feature_columns_complete(feat):
+    out_cols = set(feat.columns) - {"vt_symbol", "datetime"}
+    assert out_cols == set(FEATURE_COLUMNS)
+    assert feat.height > 0
+
+
+def test_missing_file_and_empty_rows_tolerated(synthetic_static):
+    # 600999 无文件(退市股形态) → 行存在但特征全 null,不炸
+    df = build_fundamental_features(["600999.SSE"], "2023-06-01", "2023-06-10",
+                                    data_dir=synthetic_static)
+    assert df.height > 0
+    for col in FEATURE_COLUMNS:
+        assert df[col].null_count() == df.height, f"{col} 应全 null"
+
+
+def test_trading_dates_grid(feat, synthetic_static):
+    # 传入交易日子集 → grid 只含这些日期
+    df = build_fundamental_features(
+        [A], "2023-08-01", "2023-08-31", data_dir=synthetic_static,
+        trading_dates=[_dt("2023-08-14"), _dt("2023-08-29")])
+    assert df.height == 2
+    assert val(df, A, "2023-08-14", "equity") == pytest.approx(660.0)
+    assert val(df, A, "2023-08-29", "equity") == pytest.approx(680.0)
+
+
+def test_scale_two_stock(feat):
+    # C(scale=2): TA=2*1450, NP_TTM=2*60 → roa 同 A(比率不变), np_ttm 翻倍
+    assert val(feat, C, "2023-08-29", "roa_ttm") == pytest.approx(60 / 1450)
+    assert val(feat, C, "2023-08-29", "np_ttm") == pytest.approx(120.0)
+
+
+# ---------- 分块等值(NAS 全量防 OOM 路径) ----------
+
+_SIX = ["600000.SSE", "000001.SZSE", "300001.SZSE",
+        "600004.SSE", "000333.SZSE", "300124.SZSE"]
+
+
+def test_chunked_equals_full(synthetic_static):
+    """batch_codes=2(3 批) 与 batch_codes=6(单批) 逐值等值——分块不改变结果."""
+    full = build_fundamental_features(
+        _SIX, "2023-01-01", "2023-12-31", data_dir=synthetic_static, batch_codes=6)
+    chunked = build_fundamental_features(
+        _SIX, "2023-01-01", "2023-12-31", data_dir=synthetic_static, batch_codes=2)
+    assert full.height == chunked.height == 6 * 365
+    key = ["vt_symbol", "datetime"]
+    assert full.sort(key).equals(chunked.sort(key))
+    # 批大小 1(极端) 与 trading_dates 路径同款等值
+    by_one = build_fundamental_features(
+        _SIX, "2023-01-01", "2023-12-31", data_dir=synthetic_static, batch_codes=1)
+    assert by_one.sort(key).equals(full.sort(key))

tests/factor/test_fundamental_batch.py

@@ -0,0 +1,101 @@
+# tests/factor/test_fundamental_batch.py
+"""财务因子批端到端: 合成行情库 + 合成静态域 → batch_eval 出 IC 指标."""
+import sqlite3
+import sys, os
+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 pytest
+
+from sanguo_factor import alpha_datasets  # 挂载量价因子
+from sanguo_factor import fundamental_library  # noqa: F401  挂载财务因子(import 即注册)
+from sanguo_factor.batch_eval import run_batch_eval
+from sanguo_factor import eval_store
+
+@pytest.fixture(autouse=True)
+def _ensure_fundamental_registered():
+    """其它测试模块清空 _REGISTRY 后只重挂 alpha/builtin(顺序依赖前科),
+    这里逐测试幂等重注册财务因子,保证本模块与顺序无关."""
+    fundamental_library._register_all()
+
+
+_DDL = """
+CREATE TABLE dbbardata(
+  symbol TEXT, exchange TEXT, datetime TEXT, interval TEXT,
+  volume REAL, turnover REAL, open_interest REAL,
+  open_price REAL, high_price REAL, low_price REAL, close_price REAL)
+"""
+
+_STOCKS = [("600000", "SSE"), ("000001", "SZSE"), ("300001", "SZSE")]
+
+
+@pytest.fixture(scope="module")
+def db(tmp_path_factory):
+    """3 只 × 2022-06~2024-01 合成日线(与合成财务域同代码;趋势构造截面差异)."""
+    alpha_datasets.mount_all()
+    p = tmp_path_factory.mktemp("fb")
+    db = str(p / "qt.db")
+    conn = sqlite3.connect(db)
+    conn.execute(_DDL)
+    import pandas as pd
+    days = pd.bdate_range("2022-06-01", "2024-01-05")
+    drift = {"600000": 0.0012, "000001": -0.0006, "300001": 0.0004}
+    base = {"600000": 10.0, "000001": 5.0, "300001": 20.0}
+    for i, day in enumerate(days):
+        d = day.strftime("%Y-%m-%d")
+        rows = []
+        for sym, ex in _STOCKS:
+            px = base[sym] * (1.0 + drift[sym]) ** i
+            rows.append((sym, ex, f"{d} 00:00:00", "d", 100.0, px * 100.0, 0,
+                         px, px, px, px))
+        conn.executemany("INSERT INTO dbbardata VALUES(?,?,?,?,?,?,?,?,?,?,?)", rows)
+    conn.commit()
+    conn.close()
+    return db
+
+
+def test_fundamental_end_to_end(db, synthetic_static, tmp_path):
+    eval_db = str(tmp_path / "fund_eval.db")
+    out = run_batch_eval(
+        factor_names=["fund_roe_ttm", "fund_tacc", "fund_ep_ttm",
+                      "fund_forecast_type"],
+        start="2023-02-01", end="2023-12-31",
+        eval_db=eval_db, label="fund_t", cfg=None, vnpy_db_override=db,
+        fund_data_dir=synthetic_static,
+    )
+    assert out["factors_done"] == 4
+    assert out["errors"] == []
+    for name in ("fund_roe_ttm", "fund_tacc", "fund_ep_ttm", "fund_forecast_type"):
+        m = eval_store.get_detail(eval_db, out["run_id"], name)["metrics"]
+        assert "error" not in m, f"{name}: {m.get('error')}"
+        # 截面 3 只 → 逐日 rank IC 可算(预告因子 2023-07-15 后才有效, count 少但非零)
+        assert isinstance(m["1"]["ic_mean"], float) or m["1"]["ic_mean"] is None
+        assert m["1"]["count"] >= 0
+
+
+def test_mixed_batch_price_plus_fundamental(db, synthetic_static, tmp_path):
+    """量价+财务同批混跑: 白名单 join 只在含财务因子时发生,互不干扰."""
+    eval_db = str(tmp_path / "mix_eval.db")
+    out = run_batch_eval(
+        factor_names=["ma_20", "fund_sue_np"],
+        start="2023-02-01", end="2023-06-30",
+        eval_db=eval_db, label="mix_t", cfg=None, vnpy_db_override=db,
+        fund_data_dir=synthetic_static,
+    )
+    assert out["errors"] == []
+    rows = eval_store.get_rows(eval_db, out["run_id"])
+    assert {r["factor"] for r in rows} == {"ma_20", "fund_sue_np"}
+    # SUE 在 2023-06-30 前可见的报告期均无完整 8 期窗 → 全 NaN → eliminated 非 error
+    m = eval_store.get_detail(eval_db, out["run_id"], "fund_sue_np")["metrics"]
+    assert "error" not in m
+
+
+def test_pure_price_batch_untouched(db, tmp_path):
+    """纯量价批不传静态域照常跑(回归: 未新增强依赖)."""
+    eval_db = str(tmp_path / "px_eval.db")
+    out = run_batch_eval(
+        factor_names=["roc_5"], start="2023-02-01", end="2023-03-31",
+        eval_db=eval_db, label="px_t", cfg=None, vnpy_db_override=db,
+    )
+    assert out["errors"] == []
+    assert out["factors_done"] == 1

tests/factor/test_fundamental_library.py

@@ -0,0 +1,84 @@
+# tests/factor/test_fundamental_library.py
+"""财务因子表达式库: 32 个 P0 因子注册 + 表达式↔adapter 特征列契约锁定."""
+import re
+import sys, os
+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 pytest
+
+from sanguo_factor import fundamental_library  # noqa: F401  import 即注册
+from sanguo_factor.fundamental_adapter import FEATURE_COLUMNS
+from sanguo_factor.registry import list_factors, get_factor
+
+@pytest.fixture(autouse=True)
+def _ensure_fundamental_registered():
+    """其它测试模块清空 _REGISTRY 后只重挂 alpha/builtin(顺序依赖前科),
+    这里逐测试幂等重注册财务因子,保证本模块与顺序无关."""
+    fundamental_library._register_all()
+
+
+# 表达式可引用的列 = adapter 特征列 + 行情列 close(估值类 ÷ close×share_capital)
+_ALLOWED = set(FEATURE_COLUMNS) | {"close", "cs_rank"}
+
+
+def _fundamental_factors() -> list[dict]:
+    return list_factors("fundamental")
+
+
+def test_p0_32_factors_registered():
+    facs = _fundamental_factors()
+    names = {f["name"] for f in facs}
+    assert len(facs) == 32, f"P0 首批应为 32 个,实际 {len(facs)}"
+    # 六族代表抽查(全部名单见 fundamental_library 注释)
+    expect = {
+        "fund_roe_ttm", "fund_gp_over_assets",          # A
+        "fund_tacc", "fund_nonrec_ratio",               # B
+        "fund_rev_q_yoy", "fund_np_q_yoy", "fund_asset_growth",  # C
+        "fund_ep_ttm", "fund_bp", "fund_cp",            # D
+        "fund_nsi", "fund_ibd_ratio",                   # E
+        "fund_sue_np", "fund_forecast_type",            # F
+    }
+    assert expect <= names
+
+
+def test_expressions_only_reference_feature_columns():
+    """契约: 表达式裸标识符 ⊆ adapter 特征列 + close/cs_rank(漏加列=拼写错)."""
+    for f in _fundamental_factors():
+        idents = set(re.findall(r"[A-Za-z_][A-Za-z0-9_]*", f["expression"]))
+        bad = idents - _ALLOWED
+        assert not bad, f"{f['name']} 引用了未产出列: {bad} in {f['expression']}"
+
+
+def test_all_factors_are_cross_sectional_rank():
+    """P0 设计: 因子 = cs_rank(基础指标) 一层(负 IC 因子取负定向,高=好)."""
+    for f in _fundamental_factors():
+        assert f["expression"].startswith("cs_rank("), f["name"]
+        assert f["expression"].endswith(")")
+
+
+def test_negative_ic_factors_flipped():
+    """文档负 IC 因子(B01/B04/B05/B07/B08/B11/C15/E01/E02/E05/E09)表达式含负号."""
+    flipped = {"fund_tacc", "fund_nonrec_ratio", "fund_impairment_ratio",
+               "fund_invest_income_dep", "fund_receivables_anomaly",
+               "fund_other_rece_ratio", "fund_asset_growth",
+               "fund_nsi", "fund_equity_fin_intensity", "fund_ibd_ratio",
+               "fund_goodwill_ratio"}
+    for name in flipped:
+        expr = get_factor(name)["expression"]
+        assert "(-" in expr or expr.startswith("cs_rank(-"), f"{name} 应翻转: {expr}"
+
+
+def test_valuation_factors_use_close_times_share_capital():
+    """估值族市值 = close × share_capital 自算(规避 valuation 中文列名表)."""
+    for name in ("fund_ep_ttm", "fund_ep_deduct_ttm", "fund_bp", "fund_cp",
+                 "fund_equity_fin_intensity"):
+        expr = get_factor(name)["expression"]
+        assert "close * share_capital" in expr, f"{name}: {expr}"
+
+
+def test_registration_idempotent():
+    """重复 import 不炸(注册表防重入,同 library.py 模式)."""
+    import importlib
+    importlib.reload(fundamental_library)
+    assert len(_fundamental_factors()) == 32

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