diff --git a/sanguo_factor/__init__.py b/sanguo_factor/__init__.py index b1909aa..e26f172 100644 --- a/sanguo_factor/__init__.py +++ b/sanguo_factor/__init__.py @@ -2,3 +2,4 @@ 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 批(导入即注册) +from . import composite_library # noqa: F401 合成层 v1(导入即注册,源依赖前三个) diff --git a/sanguo_factor/batch_eval.py b/sanguo_factor/batch_eval.py index 4d9418b..d157e58 100644 --- a/sanguo_factor/batch_eval.py +++ b/sanguo_factor/batch_eval.py @@ -52,7 +52,8 @@ def run_batch_eval( fund_data_dir: 财务静态域根目录(None → cfg.data_paths["static_dir"] → NAS 默认 /volume1/stock/sanguo_vnpy_v2/data/static);仅当因子列表含 - category="fundamental" 时才读取并 join(量价批零开销)。 + category="fundamental" 或 "composite"(合成因子内嵌财务源)时才读取并 + join(量价批零开销)。 """ from vnpy.alpha.dataset.utility import calculate_by_expression @@ -74,7 +75,7 @@ def run_batch_eval( # 财务因子批: 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 (get_factor(n) or {}).get("category") in ("fundamental", "composite")] 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() @@ -139,7 +140,7 @@ def run_batch_eval( # 引用瘦身: 只 join 本批 fundamental 表达式实际引用的特征列—— # FEATURE_COLUMNS 全量 68 列 × 1480 万行 ≈ 8G 超 NAS 7.9G 内存, # 按引用裁列后单族批 ≈ 1-2G(P0 33 列全量 4G 基线的必要减负) - has_fund = any((get_factor(n) or {}).get("category") == "fundamental" + has_fund = any((get_factor(n) or {}).get("category") in ("fundamental", "composite") for n in factor_names) if has_fund: import re as _re diff --git a/sanguo_factor/composite_library.py b/sanguo_factor/composite_library.py new file mode 100644 index 0000000..f7e1a76 --- /dev/null +++ b/sanguo_factor/composite_library.py @@ -0,0 +1,109 @@ +"""合成层 v1(Combiner): 量价 12 独立源 + 财务 7 源 → 等权 rank 合成总分因子. + +「因子战法八环」第④环首个落地. 因子 = 表达式字符串注册(category="composite"), +表达式由 build 函数从源清单常量表循环拼接(19 源数千字符,引擎 eval 解析无碍; +可读性/可维护性由源常量表承载,不手写长表达式). + +cs_rank 语义(vnpy_v4.4.0/vnpy/alpha/dataset/cs_function.py:15): + pl.col("data").rank().over("datetime") —— 按 datetime 分组、组内全池 + 截面 rank(batch_eval alpha_df 为全池长表,即每日全市场截面排名,正确口径; + polars rank 默认 average 法,值域 1..N 非归一化,N=当日非空源值数). + +方向口径(钉死,二者 IC 等价——每日仅差截面常数,不影响秩相关): +- 量价源: 注册表达式为原始连续值(无方向) → build 统一包 cs_rank 截面化; + 负向源项 = `(-1) * cs_rank(())`,负号在 rank 外(与 alpha16 引擎先例 + `-1 * cs_rank(...)` 同族);备选 `cs_rank((-1) * (...))`(rank 内取负=秩反转) + 弃用,全文统一 rank 外. +- 财务源: 注册表达式本身已是 `cs_rank(指标)`(负 IC 已在表达式内取负定向, + 高=好) → 直接内嵌、不再套 cs_rank;源表方向列 = 内嵌后净方向,登记 "-" + 时才前置 `(-1) * `(当前 7 源全为已定向正向,不触发). + +Null 传播: polars 算术 null 传染 → 合成值任一源 null 即 null(覆盖 = 19 源 +交集;gdhs/topholder 等事件流源在上市早期覆盖受限,合成因子覆盖期短于 +单源,属已知限制非缺陷). + +warmup: 量价源最大窗 vma_60=60 bar / alpha83 内层 ts_sum(ts_mean(volume,10),50) +≈ 59 bar < batch_eval WARMUP_BARS=120,统一预热掩码覆盖. +""" +from .registry import register_factor, get_factor, _REGISTRY + +# 量价 12 独立源(08-31 族分析固化;方向 = 该源 ICIR 符号) +QUANT_SOURCES: list[tuple[str, str]] = [ + ("vma_60", "+"), # alpha158 量能收缩(族1 代表族) + ("alpha16", "+"), # alpha101 量价背离(族2 代表,全池最强单源) + ("alpha83", "+"), # alpha101 + ("alpha12", "+"), # alpha101 + ("alpha2", "+"), # alpha101 + ("alpha42", "+"), # alpha101 + ("vol_ma5", "-"), # builtin + ("wvma_20", "-"), # alpha158 + ("klow", "-"), # alpha158 + ("cord_5", "-"), # alpha158 + ("kup", "-"), # alpha158 + ("alpha81", "-"), # alpha101 +] + +# 财务 6 源(P0+P1 IC 结论 + 族分析独立性选定;注册表达式已 cs_rank 定向). +# v1.1: 剔 fund_topholder_chg —— top_holders 源数据 2021Q3 起(NAS 镜像实测,2020Q1 +# 契约口径与实际不符),首期无差分 → 特征 2022 年中才生效;null 传染使合成因子在 +# h1(2016-2021)整窗空序列(实测 months=0).引擎无 fill_null 类算子,无法表达式内 +# null-safe 平均,故 v1 剔除;v2 数据补全或特征层合成时回填. +FUND_SOURCES: list[tuple[str, str]] = [ + ("fund_bp", "+"), # h2 ICIR 0.319 唯一 effective + ("fund_nsi", "+"), # h2 0.233 + ("fund_gp_over_m", "+"), # h2 0.212 新晋 + ("fund_gdhs_chg", "+"), # 独立源 0.15/0.11 + ("fund_growth_scissors", "+"), # h1 0.236(h2 0.076 衰减但方向稳) + ("fund_sue_np", "+"), # h1 0.200 +] + +# (合成名, 源清单)——项形态由源注册 category 决定: fundamental=内嵌,其余=cs_rank 包裹 +COMPOSITE_FACTORS: list[tuple[str, list[tuple[str, str]]]] = [ + ("composite_quant12", QUANT_SOURCES), + ("composite_fund6", FUND_SOURCES), + ("composite_all18", QUANT_SOURCES + FUND_SOURCES), +] + + +def rank_term(expression: str, direction: str) -> str: + """原始值源贡献项: cs_rank 截面化;负向 = rank 外前置 (-1) *(口径钉死).""" + term = f"cs_rank(({expression}))" + return f"(-1) * {term}" if direction == "-" else term + + +def _embed_term(expression: str, direction: str) -> str: + """已定向源(财务)贡献项: 原样内嵌;净方向 "-" 时才前置负号.""" + return f"({expression})" if direction == "+" else f"(-1) * ({expression})" + + +def build_composite_expression(terms: list[str]) -> str: + """等权平均拼接: (项1 + 项2 + ...) / N.""" + return "(" + " + ".join(terms) + f") / {len(terms)}" + + +def _register_all() -> None: + """注册全部合成因子(幂等);先幂等补挂源(测试清空 _REGISTRY 后可独立重建).""" + from . import library as _builtin_lib + from . import alpha_datasets as _alpha_ds + from . import fundamental_library as _fund_lib + _builtin_lib._register_all() + _alpha_ds.mount_all() + _fund_lib._register_all() + + for name, sources in COMPOSITE_FACTORS: + if name in _REGISTRY: + continue + terms = [] + for src_name, direction in sources: + src = get_factor(src_name) + if src is None: + raise ValueError(f"合成源未注册: {src_name}") + if src["category"] == "fundamental": + terms.append(_embed_term(src["expression"], direction)) + else: + terms.append(rank_term(src["expression"], direction)) + register_factor(name, build_composite_expression(terms), category="composite") + + +# 模块导入时自动注册(与 library.py/fundamental_library.py 同模式) +_register_all() diff --git a/tests/factor/test_composite_library.py b/tests/factor/test_composite_library.py new file mode 100644 index 0000000..dd879aa --- /dev/null +++ b/tests/factor/test_composite_library.py @@ -0,0 +1,260 @@ +# tests/factor/test_composite_library.py +"""合成层 v1(Combiner): 3 合成因子注册 + 引擎解析 + 方向传播 + 数值域 + 端到端.""" +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 numpy as np +import pandas as pd +import polars as pl +import pytest + +from sanguo_factor import composite_library +from sanguo_factor.composite_library import ( + QUANT_SOURCES, FUND_SOURCES, build_composite_expression, rank_term, +) +from sanguo_factor.registry import list_factors, get_factor +from vnpy.alpha.dataset.utility import calculate_by_expression + + +@pytest.fixture(autouse=True) +def _ensure_composite_registered(): + """其它测试模块清空 _REGISTRY 后,这里幂等重挂源+合成,保证本模块与顺序无关.""" + composite_library._register_all() + + +# ==================== 注册与源表契约 ==================== + +def test_three_composites_registered(): + facs = list_factors("composite") + assert {f["name"] for f in facs} == {"composite_quant12", "composite_fund6", + "composite_all18"} + + +def test_source_tables_cover_18_independent_sources(): + assert len(QUANT_SOURCES) == 12 + assert len(FUND_SOURCES) == 6 + names = [s[0] for s in QUANT_SOURCES] + [s[0] for s in FUND_SOURCES] + assert len(set(names)) == 18 # 无重复源 + for name, _dir in QUANT_SOURCES: + f = get_factor(name) + assert f is not None and f["category"] in ("alpha101", "alpha158", "builtin"), name + for name, _dir in FUND_SOURCES: + f = get_factor(name) + assert f is not None and f["category"] == "fundamental", name + # 方向只允许 +/- + assert {d for _, d in QUANT_SOURCES + FUND_SOURCES} <= {"+", "-"} + + +def test_quant_sources_direction_table(): + """负向源 6 个(vol_ma5/wvma_20/klow/cord_5/kup/alpha81),其余正向.""" + dirs = dict(QUANT_SOURCES) + neg = {n for n, d in dirs.items() if d == "-"} + assert neg == {"vol_ma5", "wvma_20", "klow", "cord_5", "kup", "alpha81"} + + +def test_registration_idempotent(): + import importlib + importlib.reload(composite_library) + assert len(list_factors("composite")) == 3 + + +# ==================== 表达式构造契约 ==================== + +def test_quant12_wraps_every_source_in_cs_rank_with_direction(): + """负向源形态钉死: (-1) * cs_rank((expr))(负号在 rank 外,口径见 library 注释).""" + expr = get_factor("composite_quant12")["expression"] + assert expr.startswith("(") and expr.endswith("/ 12") + for name, direction in QUANT_SOURCES: + assert rank_term(get_factor(name)["expression"], direction) in expr, \ + f"{name} 项缺失或形态不符" + assert expr.count("(-1) * cs_rank((") == 6 # 6 个负向源 + + +def test_fund6_embeds_registered_expressions_without_double_rank(): + """财务源注册表达式已 cs_rank 定向 → 直接内嵌,不得再包一层 cs_rank.""" + expr = get_factor("composite_fund6")["expression"] + assert expr.endswith("/ 6") + for name, _direction in FUND_SOURCES: + assert f"({get_factor(name)['expression']})" in expr, f"{name} 未原样内嵌" + assert "cs_rank((cs_rank(" not in expr # 无双重 rank 包裹 + + +def test_all18_mixes_both_kinds(): + expr = get_factor("composite_all18")["expression"] + assert expr.endswith("/ 18") + # 12 量价项包 cs_rank + 6 财务项内嵌 + for name, direction in QUANT_SOURCES: + assert rank_term(get_factor(name)["expression"], direction) in expr, name + for name, _direction in FUND_SOURCES: + assert f"({get_factor(name)['expression']})" in expr, name + + +# ==================== cs_rank 截面语义锁定 ==================== + +def test_cs_rank_is_per_day_cross_section(): + """引擎 cs_rank 语义 = 按 datetime 分组截面 rank(非全表 rank)——合成口径的前提.""" + df = pl.DataFrame({ + "datetime": ["2023-01-02"] * 3 + ["2023-01-03"] * 3, + "vt_symbol": ["A", "B", "C"] * 2, + "close": [10.0, 5.0, 20.0, 30.0, 5.0, 10.0], + }).with_columns(pl.col("datetime").str.to_datetime()) + out = calculate_by_expression(df, "cs_rank(close)") + vals = out["data"].to_list() + # day1: 10<20 → rank(10)=2, rank(5)=1, rank(20)=3;day2: 30=最大 → 3 + assert vals == [2.0, 1.0, 3.0, 3.0, 1.0, 2.0] + + +# ==================== 方向传播与数值域(真实引擎) ==================== + +_STOCKS5 = ["600000.SSE", "000001.SZSE", "300001.SZSE", "600004.SSE", "000333.SZSE"] +_FUND_COLS = ["equity", "share_capital", "nsi", "gp_ttm", "gdhs_chg", + "growth_scissors", "sue_np"] + + +def _engine_df(n_days: int = 130) -> pl.DataFrame: + """5 股 × n_days 合成长表: 覆盖 12 量价源全部引用列 + 7 财务特征列.""" + rng = np.random.default_rng(42) + days = pd.bdate_range("2023-01-02", periods=n_days) + rows = [] + px = {s: 8.0 * (i + 1) for i, s in enumerate(_STOCKS5)} + bias = {s: rng.normal(0, 0.5) for s in _STOCKS5} + for i, day in enumerate(days): + for j, s in enumerate(_STOCKS5): + ret = rng.normal(0, 0.02) + px[s] = max(px[s] * (1 + ret), 0.5) + open_ = px[s] * (1 + rng.normal(0, 0.005)) + high = max(open_, px[s]) * (1 + abs(rng.normal(0, 0.004))) + low = min(open_, px[s]) * (1 - abs(rng.normal(0, 0.004))) + volume = float(np.exp(rng.normal(10, 0.4))) + vwap = (high + low + px[s]) / 3.0 + row = {"datetime": day, "vt_symbol": s, "open": open_, "high": high, + "low": low, "close": px[s], "volume": volume, "vwap": vwap} + for c in _FUND_COLS: + row[c] = bias[s] + rng.normal(0, 1.0) + rows.append(row) + return pl.DataFrame(rows).with_columns(pl.col("datetime").cast(pl.Datetime("us"))) + + +def _eval_sorted(df: pl.DataFrame, expression: str) -> pd.DataFrame: + out = calculate_by_expression(df, expression).to_pandas() + return out.sort_values(["datetime", "vt_symbol"]).reset_index(drop=True) + + +def test_negative_sign_cancels_same_source(): + """同源一正一负 → 合成恒 ≈ 0(负号经 rank 外前置 (-1) * 正确传递).""" + df = _engine_df(40) + expr = build_composite_expression([rank_term("close", "+"), rank_term("close", "-")]) + out = _eval_sorted(df, expr) + assert np.allclose(out["data"].dropna(), 0.0, atol=1e-9) + assert out["data"].notna().all() + + +def test_same_direction_equals_single_source(): + """同源同向 → 合成 = 单源 rank(等权均值退化).""" + df = _engine_df(40) + expr = build_composite_expression([rank_term("close", "+"), rank_term("close", "+")]) + got = _eval_sorted(df, expr)["data"] + want = _eval_sorted(df, "cs_rank(close)")["data"] + np.testing.assert_allclose(got, want) + + +@pytest.mark.parametrize("composite,quant_dirs,fund_dirs", [ + ("composite_quant12", dict(QUANT_SOURCES), {}), + ("composite_fund6", {}, dict(FUND_SOURCES)), + ("composite_all18", dict(QUANT_SOURCES), dict(FUND_SOURCES)), +]) +def test_composite_equals_mean_of_directed_terms(composite, quant_dirs, fund_dirs): + """数值域: 合成值 == 逐源定向求均值(引擎逐项评估 vs 整体表达式全等).""" + df = _engine_df(130) + got = _eval_sorted(df, get_factor(composite)["expression"])["data"] + + parts = [] + for name, direction in quant_dirs.items(): + # 量价源先截面化(与 build 同口径)再定向 + s = _eval_sorted(df, f"cs_rank(({get_factor(name)['expression']}))")["data"] + parts.append(s if direction == "+" else -s) + for name, direction in fund_dirs.items(): + # 财务源注册表达式已定向,原样评估 + s = _eval_sorted(df, get_factor(name)["expression"])["data"] + parts.append(s if direction == "+" else -s) + want = sum(parts) / len(parts) + + assert (got.isna() == want.isna()).all(), "null 位置应一致(逐源 null 传染)" + valid = ~got.isna() + assert valid.sum() > 0, "长窗合成应有非空值(预热期外)" + np.testing.assert_allclose(got[valid], want[valid], atol=1e-9) + # 数值域: 合成 ∈ [min(项均值), max(项均值)] 逐日截面收紧到 [−N, N] 秩域 + n_stocks = len(_STOCKS5) + assert got[valid].min() >= -n_stocks - 1e-9 + assert got[valid].max() <= n_stocks + 1e-9 + + +# ==================== 端到端(batch_eval 复合门) ==================== + +_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) +""" + +_EOD_STOCKS = [("600000", "SSE"), ("000001", "SZSE"), ("300001", "SZSE"), + ("600004", "SSE"), ("000333", "SZSE")] + + +@pytest.fixture(scope="module") +def db(tmp_path_factory): + """5 只 × 2022-06~2024-01 合成日线.快振荡价(相位错开 → 截面排序频繁交叉), + OHLCV 逐日扰动,turnover=wap×volume(vwap≠close).退化形态会杀源因子: + O=C → alpha2 全并列 NaN;慢振荡/趋势 → 5 bar 窗内 cs_rank(high) 恒定 + → alpha16 的 ts_cov 恒 null(3 只慢振荡实测 0/243 天可算).""" + composite_library._register_all() + rng = np.random.default_rng(7) + p = tmp_path_factory.mktemp("cb") + db = str(p / "qt.db") + conn = sqlite3.connect(db) + conn.execute(_DDL) + days = pd.bdate_range("2022-06-01", "2024-01-05") + phase = {s: 1.3 * j for j, (s, _ex) in enumerate(_EOD_STOCKS)} + for i, day in enumerate(days): + d = day.strftime("%Y-%m-%d") + for sym, ex in _EOD_STOCKS: + px = 10.0 + 3.0 * np.sin(i / 2.2 + phase[sym]) + rng.normal(0, 0.25) + open_ = px * (1 + rng.normal(0, 0.005)) + high = max(open_, px) * (1 + abs(rng.normal(0, 0.004))) + low = min(open_, px) * (1 - abs(rng.normal(0, 0.004))) + volume = float(np.exp(rng.normal(4.6, 0.4))) + wap = (high + low + px) / 3.0 + conn.execute("INSERT INTO dbbardata VALUES(?,?,?,?,?,?,?,?,?,?,?)", + (sym, ex, f"{d} 00:00:00", "d", volume, wap * volume, 0, + open_, high, low, px)) + conn.commit() + conn.close() + return db + + +def test_composite_end_to_end(db, synthetic_static, tmp_path): + """category=composite 同样触发财务特征 join(复合门),三因子无错误评估.""" + from sanguo_factor.batch_eval import run_batch_eval + from sanguo_factor import eval_store + eval_db = str(tmp_path / "comp_eval.db") + out = run_batch_eval( + factor_names=["composite_quant12", "composite_fund6", "composite_all18"], + start="2023-02-01", end="2023-12-31", + eval_db=eval_db, label="comp_t", cfg=None, vnpy_db_override=db, + fund_data_dir=synthetic_static, + ) + assert out["factors_done"] == 3 + assert out["errors"] == [] + assert {r["factor"] for r in eval_store.get_rows(eval_db, out["run_id"], + category="composite")} == \ + {"composite_quant12", "composite_fund6", "composite_all18"} + for name in ("composite_quant12", "composite_fund6", "composite_all18"): + m = eval_store.get_detail(eval_db, out["run_id"], name)["metrics"] + assert "error" not in m, f"{name}: {m.get('error')}" + # 量价源全天候覆盖 → quant12 截面 IC 样本点非零; + # fund7/all19 在合成静态域 topholder_chg 仅 1/3 股覆盖 → count 可为 0(非错误) + m = eval_store.get_detail(eval_db, out["run_id"], "composite_quant12")["metrics"] + assert m["1"]["count"] > 0