"""Test configuration for factor module tests.""" import sys import os # Add vnpy source to path _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) # ==================== 合成财务静态域(财务因子批测试共用) ==================== # 6 只股 × 24 报告期(2018Q1~2023Q4;2018-2019 为 16 季滚动窗/5 年 CAGR 预热史), # 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 (2018, 2019, 2020, 2021, 2022, 2023) for m in ("03-31", "06-30", "09-30", "12-31") ] # 单季 REV/NP 序列(累计=年内前缀和): 报告期索引 i=0..23(2018Q1..2023Q4) # 2020 起的段与 P0 期完全一致(旧断言零回归);2018-2019 为新增预热史 REV_Q = ([70, 100, 110, 120] + [80, 110, 120, 130] # 2018 / 2019 + [90, 130, 140, 150] + [100, 120, 150, 160] # 2020 / 2021 + [110, 140, 150, 180] + [130, 140, 170, 180]) # 2022 / 2023 NP_Q = ([7, 10, 11, 12] + [8, 11, 12, 13] # 2018 / 2019 + [9, 13, 14, 15] + [10, 12, 15, 16] # 2020 / 2021 + [11, 14, 15, 18] + [13, 14, 17, 18]) # 2022 / 2023 # 资产负债表(时点存量,随报告期线性演化;2021Q1 起算,idx=i-12) TA_OF = lambda i: 1000 + 50 * max(i - 12, 0) EQ_OF = lambda i: 500 + 20 * max(i - 12, 0) AR_OF = lambda i: 100 + 10 * max(i - 12, 0) SC_OF = lambda i: 100 if i < 20 else 110 # 2023 起股本扩张 10%(NSI=0.1) SYN_STOCKS = { "600000.SH": {"vt": "600000.SSE", "scale": 1.0, "drop_periods": [], "null_notice_periods": []}, "000001.SZ": {"vt": "000001.SZSE", "scale": 1.0, "drop_periods": ["2022-03-31"], # 缺 2022Q1 → 单季差分/TTM 链 NaN "null_notice_periods": ["2022-09-30"]}, # NOTICE_DATE 缺失 → 该报告期跳过 "300001.SZ": {"vt": "300001.SZSE", "scale": 2.0, "drop_periods": [], "null_notice_periods": []}, # 三只分块等值测试扩容股(全史无残缺,不同 scale 增截面多样性) "600004.SH": {"vt": "600004.SSE", "scale": 0.5, "drop_periods": [], "null_notice_periods": []}, "000333.SZ": {"vt": "000333.SZSE", "scale": 1.7, "drop_periods": [], "null_notice_periods": []}, "300124.SZ": {"vt": "300124.SZSE", "scale": 3.0, "drop_periods": [], "null_notice_periods": []}, # 银行模板股(OPERATE_COST 缺失 → _is_fin;INVENTORY 缺失测 null 传播): # 盈利质量/成长/费用类特征应置 NaN,盈利能力/估值族保留(§7 红线 5) "601398.SH": {"vt": "601398.SSE", "scale": 4.0, "bank": True, "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"]) is_bank = spec.get("bank", False) 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 缺失 → 金融股判定) "OPERATE_COST": None if is_bank else 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, # P1 新原料(比率设计成可手算: τ=0.25 / INT=0.2·NP / 研发5% / 销售8%) "RESEARCH_EXPENSE": 0.05 * rev_c, "SALE_EXPENSE": 0.08 * rev_c, "FE_INTEREST_EXPENSE": 0.2 * np_c, "INCOME_TAX": 0.275 * np_c, "BASIC_EPS": _cum_in_year(NP_Q, i) / SC_OF(i)}) # EPS 与 scale 无关 # 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 - 12, 0)), "GOODWILL": 50.0, "SHARE_CAPITAL": float(SC_OF(i)), "SHORT_LOAN": s * (100 + max(i - 12, 0)), # P1 新原料: 存货/货币资金(银行无存货 → null 传播) "INVENTORY": None if is_bank else s * (200 + 5 * max(i - 12, 0)), "MONETARYFUNDS": s * (150 + 10 * max(i - 12, 0))}) cashflow_rows.append({**common, "NETCASH_OPERATE": 1.2 * np_c, "SALES_SERVICES": 1.05 * rev_c, "ACCEPT_INVEST_CASH": 0.1 * rev_c, # P1 新原料(DA 合计 = 0.10·REV / capex 15% / 债务净发行 = +0.10·REV) "FA_IR_DEPR": 0.06 * rev_c, "IA_AMORTIZE": 0.02 * rev_c, "LPE_AMORTIZE": 0.01 * rev_c, "USERIGHT_ASSET_AMORTIZE": 0.01 * rev_c, "CONSTRUCT_LONG_ASSET": 0.15 * rev_c, "RECEIVE_LOAN_CASH": 0.3 * rev_c, "ISSUE_BOND": 0.05 * rev_c, "PAY_DEBT_CASH": 0.25 * 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) # 预测数值 = 预告中值(元): A=30(实际 27 → beat=-0.1) / B=40(beat=-0.325) os.makedirs(os.path.join(static_dir, "forecast"), exist_ok=True) pl.DataFrame([ {"股票代码": "600000", "预测指标": "归属于上市公司股东的净利润", "预测数值": 30.0, "业绩变动幅度": 56.79, "预告类型": "预增", "公告日期": date(2023, 7, 15)}, {"股票代码": "000001", "预测指标": "归属于上市公司股东的净利润", "预测数值": 40.0, "业绩变动幅度": -30.0, "预告类型": "预减", "公告日期": date(2023, 7, 20)}, {"股票代码": "300001", "预测指标": "营业收入", "预测数值": 999.0, "业绩变动幅度": 5.0, "预告类型": "略增", "公告日期": date(2023, 7, 10)}, # 北交所两段前缀(92x/43x)→ .BJSE 映射用例 {"股票代码": "920001", "预测指标": "归属于上市公司股东的净利润", "预测数值": 50.0, "业绩变动幅度": 20.0, "预告类型": "预增", "公告日期": date(2023, 7, 18)}, {"股票代码": "430047", "预测指标": "净利润", "预测数值": 20.0, "业绩变动幅度": 10.0, "预告类型": "预增", "公告日期": date(2023, 7, 19)}, ]).write_parquet(os.path.join(static_dir, "forecast", "20230630_forecast.parquet")) pl.DataFrame([ {"股票代码": "600000", "预测指标": "净利润", "预测数值": 65.0, "业绩变动幅度": 100.0, "预告类型": "扭亏", "公告日期": date(2023, 10, 15)}, ]).write_parquet(os.path.join(static_dir, "forecast", "20230930_forecast.parquet")) # 迟到预告用例(F07 锚语义): 2022 年报披露 2023-04-25,预告公告 2023-05-10 # 晚于披露日 → 兑现差锚 = max(披露日, 公告日) = 05-10(实际 NP 58,中值 55) pl.DataFrame([ {"股票代码": "600000", "预测指标": "归属于上市公司股东的净利润", "预测数值": 55.0, "业绩变动幅度": 15.0, "预告类型": "预增", "公告日期": date(2023, 5, 10)}, ]).write_parquet(os.path.join(static_dir, "forecast", "20221231_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)