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)