# 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])