From 3c24a96fa56e4f750912767fc7a2369b24cc4178 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 8 Sep 2026 00:22:21 +0800 Subject: [PATCH 01/10] =?UTF-8?q?feat(factor):=20=E5=9B=BA=E5=8C=96?= =?UTF-8?q?=E5=9B=A0=E5=AD=90=E6=97=8F=E5=88=86=E6=9E=90=E4=B8=BA=E5=8F=AF?= =?UTF-8?q?=E5=A4=8D=E7=94=A8=E8=84=9A=E6=9C=AC=20xcorr=5Ffamily.py=20[nas?= =?UTF-8?q?]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 08-31 对 62 量价 effective 因子的 ad-hoc 族分析(12 独立源/冗余81%)固化为 CLI: monthly_ic 两两 Pearson + |r|>=阈值连通分量成族 + 族代表=|ICIR|最大; 多 run 跨批模式额外输出异 run 配对清单(防财务批与量价批暗相关); 合成 monthly_ic 测试 4 例(含负相关同族/代表选择/跨批池化)。 Co-Authored-By: Claude Code [nas] --- scripts/factor_research/xcorr_family.py | 234 ++++++++++++++++++++++++ tests/factor/test_xcorr_family.py | 102 +++++++++++ 2 files changed, 336 insertions(+) create mode 100644 scripts/factor_research/xcorr_family.py create mode 100644 tests/factor/test_xcorr_family.py 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 From 70619ad0642a53d980775890f54aa99a80184e3d Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 8 Sep 2026 00:37:49 +0800 Subject: [PATCH 02/10] =?UTF-8?q?feat(factor):=20=E8=B4=A2=E5=8A=A1?= =?UTF-8?q?=E5=9B=A0=E5=AD=90=E9=80=82=E9=85=8D=E5=B1=82=E2=80=94=E2=80=94?= =?UTF-8?q?=E4=B8=89=E8=A1=A8+forecast=20parquet=E2=86=92PIT=E6=97=A5?= =?UTF-8?q?=E9=A2=91=E7=89=B9=E5=BE=81=E5=88=97(=E8=B4=A2=E5=8A=A1?= =?UTF-8?q?=E6=89=B9P0=20WP1)=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - fundamental_adapter.build_fundamental_features: NAS 静态域三表(按股文件)+ forecast(按报告期全市场文件) → vt_symbol×datetime×33 特征列 - 口径红线全落地: 单季差分缺上期NaN不填0 / TTM连续4季 / PIT=NOTICE_DATE≤决策日 (三表最晚,重述取UPDATE_DATE终值) / NOTICE_DATE缺失报告期整期跳过 / 金融股(无营业成本模板)盈利质量+成长族置NaN - SUE=Foster标准化(diff4/std过去8期ddof=1); forecast归母净利润行优先+fallback - 容错: 零行文件(北交920xxx)/缺文件(退市355只)/缺列(银行模板)全跳过不炸 - 合成数据端到端15测全绿: PIT边界/前向填充/NaN传染与恢复/SUE/forecast事件 [nas] Co-Authored-By: Claude Code --- sanguo_factor/fundamental_adapter.py | 466 +++++++++++++++++++++++ tests/factor/conftest.py | 127 ++++++ tests/factor/test_fundamental_adapter.py | 197 ++++++++++ 3 files changed, 790 insertions(+) create mode 100644 sanguo_factor/fundamental_adapter.py create mode 100644 tests/factor/test_fundamental_adapter.py diff --git a/sanguo_factor/fundamental_adapter.py b/sanguo_factor/fundamental_adapter.py new file mode 100644 index 0000000..efe6c84 --- /dev/null +++ b/sanguo_factor/fundamental_adapter.py @@ -0,0 +1,466 @@ +# 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 +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) + frames.append(df.with_columns(pl.lit(vt).alias("vt_symbol"))) + if not frames: + continue + merged = pl.concat(frames).select(["vt_symbol", *_DATE_COLS, *raw_cols]) + 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"), + ) + + +# ==================== 对外主入口 ==================== + +def _build_grid(codes: list[str], start: str, end: str, + trading_dates) -> pl.DataFrame: + """日频 grid: vt_symbol × datetime(交易日子集或日历日).""" + if trading_dates is not None: + days = pl.Series("datetime", trading_dates).cast(pl.Datetime("us")) + days = days.unique().sort() + else: + s = datetime.strptime(start, "%Y-%m-%d") + e = datetime.strptime(end, "%Y-%m-%d") + days = pl.datetime_range(s, e, interval="1d", eager=True).cast(pl.Datetime("us")) + day_list = days.to_list() + 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 build_fundamental_features( + codes: list[str], + start: str, + end: str, + data_dir: str = DEFAULT_STATIC_DIR, + trading_dates: pl.Series | list | None = None, +) -> 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 免造非交易日行) + + Returns: + 每行 = 决策日可见的最新报告期特征(NOTICE_DATE ≤ 决策日,asof 前向填充)。 + """ + stmt_cols = [c for c in FEATURE_COLUMNS if c not in _FORECAST_COLS] + if not codes: + return pl.DataFrame(schema={"vt_symbol": pl.Utf8, "datetime": pl.Datetime("us"), + **{c: pl.Float64 for c in FEATURE_COLUMNS}}) + + grid = _build_grid(codes, start, end, trading_dates) + + 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") + + # asof 前向填充: 决策日可见的最新披露 + out = grid.sort("datetime").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") + return out.sort(["vt_symbol", "datetime"]).select( + ["vt_symbol", "datetime", *FEATURE_COLUMNS]) diff --git a/tests/factor/conftest.py b/tests/factor/conftest.py index 8ce958e..e466590 100644 --- a/tests/factor/conftest.py +++ b/tests/factor/conftest.py @@ -6,3 +6,130 @@ 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": []}, +} + + +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) diff --git a/tests/factor/test_fundamental_adapter.py b/tests/factor/test_fundamental_adapter.py new file mode 100644 index 0000000..6f23f0f --- /dev/null +++ b/tests/factor/test_fundamental_adapter.py @@ -0,0 +1,197 @@ +# 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) From d1d9dd50ffff537e0518a803c13bcc464b8db4f0 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 8 Sep 2026 00:38:51 +0800 Subject: [PATCH 03/10] =?UTF-8?q?feat(factor):=20=E8=B4=A2=E5=8A=A1?= =?UTF-8?q?=E5=9B=A0=E5=AD=90P0=E8=A1=A8=E8=BE=BE=E5=BC=8F=E5=BA=93?= =?UTF-8?q?=E2=80=94=E2=80=9432=E5=9B=A0=E5=AD=90=E6=B3=A8=E5=86=8Ccs=5Fra?= =?UTF-8?q?nk=E4=B8=80=E5=B1=82(=E8=B4=A2=E5=8A=A1=E6=89=B9P0=20WP2)=20[na?= =?UTF-8?q?s]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - fundamental_library: 32因子(盈利7/质量7/成长6/估值4/资本4/预期4),全名单注释在档 - 因子=cs_rank(基础指标),负IC因子取负统一高=好;估值族市值=close×share_capital自算 - 契约测试锁定: 表达式裸标识符⊆adapter特征列∪close(防拼写漏加列)/cs_rank一层/ 负向因子翻转/幂等注册 [nas] Co-Authored-By: Claude Code --- sanguo_factor/fundamental_library.py | 80 ++++++++++++++++++++++++ tests/factor/test_fundamental_library.py | 77 +++++++++++++++++++++++ 2 files changed, 157 insertions(+) create mode 100644 sanguo_factor/fundamental_library.py create mode 100644 tests/factor/test_fundamental_library.py diff --git a/sanguo_factor/fundamental_library.py b/sanguo_factor/fundamental_library.py new file mode 100644 index 0000000..8fb57b7 --- /dev/null +++ b/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() diff --git a/tests/factor/test_fundamental_library.py b/tests/factor/test_fundamental_library.py new file mode 100644 index 0000000..0b310d3 --- /dev/null +++ b/tests/factor/test_fundamental_library.py @@ -0,0 +1,77 @@ +# 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 + +# 表达式可引用的列 = 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 From caa72d34262bb44e31014bbd09722156d94d7c03 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 8 Sep 2026 00:41:46 +0800 Subject: [PATCH 04/10] =?UTF-8?q?feat(factor):=20batch=5Feval=E6=8E=A5?= =?UTF-8?q?=E5=85=A5=E8=B4=A2=E5=8A=A1=E5=9B=A0=E5=AD=90=E2=80=94=E2=80=94?= =?UTF-8?q?=E7=89=B9=E5=BE=81=E5=88=97join+=E7=AB=AF=E5=88=B0=E7=AB=AFIC(?= =?UTF-8?q?=E8=B4=A2=E5=8A=A1=E6=89=B9P0=20WP3)=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - run_batch_eval 增 fund_data_dir 参数;仅因子列表含 category=fundamental 时 build_fundamental_features(bars 的 codes+trading_dates) join 进 alpha_df (列白名单经 join 扩展,量价批零开销零改动路径) - __init__ 补挂 fundamental_library(import sanguo_factor 全家可见) - 端到端3测: 财务4因子IC落库/量价+财务混批/纯量价无静态域回归 - 修顺序依赖: 其它测试清 _REGISTRY 只重挂alpha → 本批两模块 autouse 幂等重注册 [nas] Co-Authored-By: Claude Code --- sanguo_factor/__init__.py | 1 + sanguo_factor/batch_eval.py | 21 ++++- tests/factor/test_fundamental_batch.py | 101 +++++++++++++++++++++++++ 3 files changed, 122 insertions(+), 1 deletion(-) create mode 100644 tests/factor/test_fundamental_batch.py diff --git a/sanguo_factor/__init__.py b/sanguo_factor/__init__.py index d8bb735..b1909aa 100644 --- a/sanguo_factor/__init__.py +++ b/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 批(导入即注册) diff --git a/sanguo_factor/batch_eval.py b/sanguo_factor/batch_eval.py index b578031..5187a05 100644 --- a/sanguo_factor/batch_eval.py +++ b/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 build_fundamental_features, 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,17 @@ def run_batch_eval( alpha_df = bars.select(["vt_symbol", "datetime", "open", "high", "low", "close", "volume", "turnover", "vwap"]) + # 财务因子批: 特征列 join 进 alpha_df(表达式引擎按列名直接消费;bars 释放前完成) + fund_names = [n for n in factor_names + if (get_factor(n) or {}).get("category") == "fundamental"] + if fund_names: + static_dir = fund_data_dir or cfg.data_paths.get("static_dir") or DEFAULT_STATIC_DIR + feat_df = build_fundamental_features( + codes=bars["vt_symbol"].unique().to_list(), + start=start, end=end, data_dir=static_dir, + trading_dates=bars["datetime"].unique().sort(), + ) + alpha_df = alpha_df.join(feat_df, on=["vt_symbol", "datetime"], how="left") # Pre-compute per-symbol warmup cutoff dates (bar_idx >= WARMUP_BARS 的首日) # 用于替代 per-factor hash join，改为 pivot 后 pandas 广播掩码置 NaN cutoffs = ( diff --git a/tests/factor/test_fundamental_batch.py b/tests/factor/test_fundamental_batch.py new file mode 100644 index 0000000..03a0364 --- /dev/null +++ b/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 From 4f7f5c6f5df567820431ffc10b9b3a645af64322 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 8 Sep 2026 00:50:29 +0800 Subject: [PATCH 05/10] =?UTF-8?q?fix(factor):=20=E4=B8=89=E8=A1=A8?= =?UTF-8?q?=E8=B7=A8=E8=82=A1concat=E5=88=97=E5=BA=8F=E5=AF=B9=E9=BD=90?= =?UTF-8?q?=E2=80=94=E2=80=94=E5=90=84=E8=82=A1schema=E5=AD=90=E9=9B=86?= =?UTF-8?q?=E4=B8=8D=E5=90=8C=E8=87=B4vstack=E9=94=99=E9=85=8D(=E7=9C=9F?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=86=92=E7=83=9F=E5=AE=9E=E9=94=A4)=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit NAS容器只读冒烟(3真股×2023)暴露: 合成数据列全同未触发,真数据各股文件 schema子集不同(银行模板缺列补尾部),pl.concat在select统一列序之前执行 → ShapeError。修: frames append前先select统一列序/列集。 真数据冒烟同时锁定: 茅台ROE_TTM 34%/银行tacc置NaN(红线生效)/32表达式 全通过引擎/33特征31个覆盖健康;两断供列(ASSET_IMPAIRMENT_LOSS post-2020-09、 OTHER_RECE post-~2021)为数据层采集模板问题,adapter忠实保null,遗留清单在档。 [nas] Co-Authored-By: Claude Code --- sanguo_factor/fundamental_adapter.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/sanguo_factor/fundamental_adapter.py b/sanguo_factor/fundamental_adapter.py index efe6c84..d0b23ec 100644 --- a/sanguo_factor/fundamental_adapter.py +++ b/sanguo_factor/fundamental_adapter.py @@ -169,10 +169,13 @@ def _load_statements(codes: list[str], data_dir: str) -> pl.DataFrame: if df is None: continue df = _norm_dates(df, _DATE_COLS) - frames.append(df.with_columns(pl.lit(vt).alias("vt_symbol"))) + # 先统一列序/列集再入列(各股文件 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).select(["vt_symbol", *_DATE_COLS, *raw_cols]) + merged = pl.concat(frames) if table == "balance": short = {c: c for c in raw_cols} else: From f7235d61947b06d18368acf51a48a8679292a2e2 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 8 Sep 2026 00:50:41 +0800 Subject: [PATCH 06/10] =?UTF-8?q?test(factor):=20fundamental=E5=BA=93?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E8=A1=A5autouse=E5=B9=82=E7=AD=89=E9=87=8D?= =?UTF-8?q?=E6=B3=A8=E5=86=8C(WP3=E4=BF=AE=E5=A4=8D=E6=96=87=E4=BB=B6?= =?UTF-8?q?=E6=BC=8F=E6=8F=90=E4=BA=A4)=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit [nas] Co-Authored-By: Claude Code --- tests/factor/test_fundamental_library.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/factor/test_fundamental_library.py b/tests/factor/test_fundamental_library.py index 0b310d3..1838ed3 100644 --- a/tests/factor/test_fundamental_library.py +++ b/tests/factor/test_fundamental_library.py @@ -11,6 +11,13 @@ 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"} From e5ed91d755b65d69143f337f6304d919bd7e27cf Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 8 Sep 2026 00:54:08 +0800 Subject: [PATCH 07/10] =?UTF-8?q?feat(factor):=20run=5Feval=20CLI=E6=9A=B4?= =?UTF-8?q?=E9=9C=B2--fund-data-dir=E2=80=94=E2=80=94=E5=AE=B9=E5=99=A8?= =?UTF-8?q?=E8=B4=A2=E5=8A=A1=E6=89=B9=E5=8F=AF=E4=BC=A0=E9=9D=99=E6=80=81?= =?UTF-8?q?=E5=9F=9F(WP1-3=E6=94=B6=E5=B0=BE=E8=A1=A5=E5=B7=A5)=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - --fund-data-dir default=None 透传 batch_eval;默认语义已有三层fallback: 参数 → cfg.data_paths['static_dir'] → /volume1/stock/sanguo_vnpy_v2/data/static (None≠不join,CLI不传参容器内外同路径可跑财务批;量价批不含fundamental因子零开销) - docstring补财务批用法示例(--categories fundamental) - 验证: pytest tests/factor/ -x -q 145绿; --list-factors fundamental 列出32个 [nas] Co-Authored-By: Claude Code --- scripts/factor_research/run_eval.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/scripts/factor_research/run_eval.py b/scripts/factor_research/run_eval.py index 9ee08fd..4135d88 100644 --- a/scripts/factor_research/run_eval.py +++ b/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']}") From 67fae2fef2554d27c524a3008f960b9381ff8224 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 8 Sep 2026 03:52:23 +0800 Subject: [PATCH 08/10] =?UTF-8?q?perf(factor):=20=E8=B4=A2=E5=8A=A1?= =?UTF-8?q?=E7=89=B9=E5=BE=81join=5Fasof=E5=88=86=E5=9D=97=E5=8C=96?= =?UTF-8?q?=E2=80=94=E2=80=94NAS=E5=85=A8=E9=87=8F1480=E4=B8=87=E8=A1=8Cgr?= =?UTF-8?q?id=20OOM=E6=A0=B9=E6=B2=BB=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 根因: 全市场grid(5555股×2670日×33列≈4G)单次join_asof + batch_eval侧 特征帧+alpha_df双全量副本,峰值10G+,7.9G NAS必爆(合成3股测不出)。 - adapter重构: iter_fundamental_feature_chunks生成器——grid按股分批构建 (BATCH_CODES=500,批间无全量grid副本),事件右表(报告期+forecast)全批共用一份; build_fundamental_features改为chunks concat(单一代码路径) - batch_eval: 特征join移到del bars之后(省1G bars常驻),逐块filter→join alpha_df分片→concat,不再持有特征帧全量副本;断点续跑已完成的财务因子不再触发join - 消两处join_asof UserWarning: 显式按键sort后抑制polars 1.42 by分组无法 校验sortedness的无信息提示(sort即正确性保险;set_sorted实测压不住) - 等值测试: 6股合成域 batch=1/2/6 逐值等值(分块不改变结果) - NAS真数据探针(600真股×2018-2026×batch500): roe_ttm覆盖0.904, 峰值RSS 1017MB(含全量statement加载),零OOM;生产规模外推~2G内 [nas] Co-Authored-By: Claude Code --- sanguo_factor/batch_eval.py | 33 ++++-- sanguo_factor/fundamental_adapter.py | 128 +++++++++++++++-------- tests/factor/conftest.py | 7 ++ tests/factor/test_fundamental_adapter.py | 21 ++++ 4 files changed, 137 insertions(+), 52 deletions(-) diff --git a/sanguo_factor/batch_eval.py b/sanguo_factor/batch_eval.py index 5187a05..8d92d49 100644 --- a/sanguo_factor/batch_eval.py +++ b/sanguo_factor/batch_eval.py @@ -23,7 +23,7 @@ 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 build_fundamental_features, DEFAULT_STATIC_DIR +from .fundamental_adapter import DEFAULT_STATIC_DIR def _forward_return_matrices(close_wide: pd.DataFrame, periods=(1, 5, 10)) -> dict[int, pd.DataFrame]: @@ -71,17 +71,14 @@ def run_batch_eval( alpha_df = bars.select(["vt_symbol", "datetime", "open", "high", "low", "close", "volume", "turnover", "vwap"]) - # 财务因子批: 特征列 join 进 alpha_df(表达式引擎按列名直接消费;bars 释放前完成) + # 财务因子批: 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: - static_dir = fund_data_dir or cfg.data_paths.get("static_dir") or DEFAULT_STATIC_DIR - feat_df = build_fundamental_features( - codes=bars["vt_symbol"].unique().to_list(), - start=start, end=end, data_dir=static_dir, - trading_dates=bars["datetime"].unique().sort(), - ) - alpha_df = alpha_df.join(feat_df, on=["vt_symbol", "datetime"], how="left") + 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 = ( @@ -136,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"]: diff --git a/sanguo_factor/fundamental_adapter.py b/sanguo_factor/fundamental_adapter.py index d0b23ec..0c7a318 100644 --- a/sanguo_factor/fundamental_adapter.py +++ b/sanguo_factor/fundamental_adapter.py @@ -22,6 +22,7 @@ from __future__ import annotations import os +import warnings from datetime import datetime import polars as pl @@ -396,48 +397,33 @@ def _load_forecast_events(codes: list[str], data_dir: str) -> pl.DataFrame: # ==================== 对外主入口 ==================== -def _build_grid(codes: list[str], start: str, end: str, - trading_dates) -> pl.DataFrame: - """日频 grid: vt_symbol × datetime(交易日子集或日历日).""" +# 分块大小: 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")) - days = days.unique().sort() - else: - s = datetime.strptime(start, "%Y-%m-%d") - e = datetime.strptime(end, "%Y-%m-%d") - days = pl.datetime_range(s, e, interval="1d", eager=True).cast(pl.Datetime("us")) - day_list = days.to_list() + 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 build_fundamental_features( - codes: list[str], - start: str, - end: str, - data_dir: str = DEFAULT_STATIC_DIR, - trading_dates: pl.Series | list | None = None, -) -> 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 免造非交易日行) - - Returns: - 每行 = 决策日可见的最新报告期特征(NOTICE_DATE ≤ 决策日,asof 前向填充)。 - """ +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] - if not codes: - return pl.DataFrame(schema={"vt_symbol": pl.Utf8, "datetime": pl.Datetime("us"), - **{c: pl.Float64 for c in FEATURE_COLUMNS}}) - - grid = _build_grid(codes, start, end, trading_dates) - reports = _load_statements(codes, data_dir) if reports.height == 0: stmt_events = pl.DataFrame(schema={ @@ -451,19 +437,75 @@ def build_fundamental_features( 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") - # asof 前向填充: 决策日可见的最新披露 - out = grid.sort("datetime").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") - return out.sort(["vt_symbol", "datetime"]).select( - ["vt_symbol", "datetime", *FEATURE_COLUMNS]) diff --git a/tests/factor/conftest.py b/tests/factor/conftest.py index e466590..805f4c0 100644 --- a/tests/factor/conftest.py +++ b/tests/factor/conftest.py @@ -39,6 +39,13 @@ SYN_STOCKS = { "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": []}, } diff --git a/tests/factor/test_fundamental_adapter.py b/tests/factor/test_fundamental_adapter.py index 6f23f0f..c0ac3c3 100644 --- a/tests/factor/test_fundamental_adapter.py +++ b/tests/factor/test_fundamental_adapter.py @@ -195,3 +195,24 @@ 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)) From 977cebb2aa32399cdf349886b699b1b1d5822a85 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 8 Sep 2026 09:27:40 +0800 Subject: [PATCH 09/10] =?UTF-8?q?feat(factor):=20=E8=B4=A2=E5=8A=A1?= =?UTF-8?q?=E5=9B=A0=E5=AD=90=E5=89=8D=E7=AB=AF=E5=B9=B6=E5=88=97=E7=B1=BB?= =?UTF-8?q?=E5=88=AB+run=E5=AF=BC=E5=85=A5=E4=B8=BB=E5=BA=93=E8=84=9A?= =?UTF-8?q?=E6=9C=AC=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Leaderboard.vue 类别 chips/样式映射加 fundamental(财务基本面),teal 色系 - tokens.css 新增 --teal 类别色变量(#2dd4bf,未占用色) - import_runs_to_main_db.py:独立评估库指定 run_ids 幂等导入主库 (INSERT OR REPLACE,列序按 dst PRAGMA 实取,写前只读预检, 本地临时双库自测:幂等/零沾染/列序不匹配/缺失run rc=1 全过) Co-Authored-By: Claude Code --- frontend/src/styles/tokens.css | 3 + frontend/src/views/factor/Leaderboard.vue | 8 + .../factor_research/import_runs_to_main_db.py | 156 ++++++++++++++++++ 3 files changed, 167 insertions(+) create mode 100644 scripts/factor_research/import_runs_to_main_db.py diff --git a/frontend/src/styles/tokens.css b/frontend/src/styles/tokens.css index 015cb07..ae00d03 100644 --- a/frontend/src/styles/tokens.css +++ b/frontend/src/styles/tokens.css @@ -23,6 +23,9 @@ /* —— 次强调：紫（alpha158）—— */ --purple-dim: #b48cff; /* alpha158 类别色 */ + /* —— 类别色：teal（fundamental 财务因子）—— */ + --teal: #2dd4bf; /* fundamental 类别色 */ + /* —— 数据高亮：琥珀 —— */ --amber: #ffb000; /* 数据高亮/警告 */ --warn: #ffb000; /* 语义别名 */ diff --git a/frontend/src/views/factor/Leaderboard.vue b/frontend/src/views/factor/Leaderboard.vue index a3ca7e5..b773ba2 100644 --- a/frontend/src/views/factor/Leaderboard.vue +++ b/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); } diff --git a/scripts/factor_research/import_runs_to_main_db.py b/scripts/factor_research/import_runs_to_main_db.py new file mode 100644 index 0000000..bab62e0 --- /dev/null +++ b/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()) From 42c8240dc7ae265d99d75dab40465bcaf901d069 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 8 Sep 2026 09:30:14 +0800 Subject: [PATCH 10/10] =?UTF-8?q?fix(factor):=20=E8=AF=A6=E6=83=85?= =?UTF-8?q?=E9=A1=B5/=E5=9B=A0=E5=AD=90=E9=80=89=E6=8B=A9=E5=99=A8?= =?UTF-8?q?=E8=A1=A5=20fundamental=20teal=20=E7=B1=BB=E5=88=AB=E6=A0=B7?= =?UTF-8?q?=E5=BC=8F=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - LeaderboardDetail.vue categoryClass 映射+cat-fundamental chip 样式 - FactorPicker.vue cat-fundamental chip+sel-fundamental 选中态 (该页无裸 .fxp.sel 兜底,不补则选中财务因子无高亮) - npm run build(vue-tsc -b && vite build)全绿 Co-Authored-By: Claude Code --- frontend/src/views/factor/FactorPicker.vue | 2 ++ frontend/src/views/factor/LeaderboardDetail.vue | 7 +++++++ 2 files changed, 9 insertions(+) diff --git a/frontend/src/views/factor/FactorPicker.vue b/frontend/src/views/factor/FactorPicker.vue index c07b0bf..5b39bfe 100644 --- a/frontend/src/views/factor/FactorPicker.vue +++ b/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; } diff --git a/frontend/src/views/factor/LeaderboardDetail.vue b/frontend/src/views/factor/LeaderboardDetail.vue index c57f40e..7ad579d 100644 --- a/frontend/src/views/factor/LeaderboardDetail.vue +++ b/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;