From 10b7ecf7a9de2a266d3b64aae8985b94c6902925 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Thu, 10 Sep 2026 18:59:49 +0800 Subject: [PATCH] =?UTF-8?q?feat(factor):=20F=E6=97=8FPIT=E7=94=9F=E6=95=88?= =?UTF-8?q?=E6=97=A5=E5=81=8F=E7=A7=BB=E5=BC=80=E5=85=B3=E2=80=94=E2=80=94?= =?UTF-8?q?SANGUO=5FFORECAST=5FPIT=5FOFFSET=5FDAYS=E7=8E=AF=E5=A2=83?= =?UTF-8?q?=E5=8F=98=E9=87=8F/=E8=BF=9B=E7=A8=8B=E5=86=85=E5=8F=AF?= =?UTF-8?q?=E8=AE=BE,4=E5=AF=86=E9=97=AD=E6=B5=8B=E8=AF=95=E5=8F=8C?= =?UTF-8?q?=E7=8E=AF=E5=A2=83=E6=80=81=E7=BB=BF+conftest=20session?= =?UTF-8?q?=E7=BA=A7=E9=92=89=E6=AD=BB=E9=98=B2=E7=8E=AF=E5=A2=83=E6=B8=97?= =?UTF-8?q?=E9=80=8F=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sanguo_factor/fundamental_forecast.py | 19 ++- tests/factor/conftest.py | 14 ++ .../test_fundamental_forecast_pit_offset.py | 122 ++++++++++++++++++ 3 files changed, 154 insertions(+), 1 deletion(-) create mode 100644 tests/factor/test_fundamental_forecast_pit_offset.py diff --git a/sanguo_factor/fundamental_forecast.py b/sanguo_factor/fundamental_forecast.py index a7e9539..44334ad 100644 --- a/sanguo_factor/fundamental_forecast.py +++ b/sanguo_factor/fundamental_forecast.py @@ -19,6 +19,17 @@ from .fundamental_schema import FORECAST_TYPE_SCORE # 预告净利年化系数(按报告期进度;D13 预期 EP): Q1×4 / H1×2 / Q3×4/3 / 年报×1 _ANNUALIZE_FACTOR = {3: 4.0, 6: 2.0, 9: 4.0 / 3.0, 12: 1.0} +# PIT 生效日偏移开关(eff+N 敏感性对照实验,调用时读取). 预告常在收盘后公告, +# 默认 0 = 公告日 D 当天即可用其信息;设 N 时所有预告事件生效日整体推后 N 个 +# 自然日(D 日不可见,D+N 起可见),同时作用于 fc_events 与 fc_pair(F07 兑现差 +# 锚的预告腿);三表/分红/户数/十大流通股东/vb 域不受此开关影响. +# 跑批侧两种设置方式(取其一): +# ①容器/进程环境变量: SANGUO_FORECAST_PIT_OFFSET_DAYS=1(docker run -e 即可) +# ②进程内直设(batch_eval 在调 iter_fundamental_feature_chunks 前): +# import sanguo_factor.fundamental_forecast as _ff +# _ff.FORECAST_PIT_OFFSET_DAYS = 1 +FORECAST_PIT_OFFSET_DAYS = int(os.environ.get("SANGUO_FORECAST_PIT_OFFSET_DAYS", "0") or 0) + def _load_forecast_events(codes: list[str], data_dir: str) -> tuple[pl.DataFrame, pl.DataFrame]: """forecast 按期文件 → (fc_events, fc_pair). @@ -31,6 +42,9 @@ def _load_forecast_events(codes: list[str], data_dir: str) -> tuple[pl.DataFrame fc_pair: (vt_symbol, REPORT_DATE, _fc_mid, eff) —— F07 预告兑现差的配对原料, 每 (股, 报告期) 取最新公告日的净利行中值(REPORT_DATE 取自文件名)。 + + 生效日 eff 受模块级 FORECAST_PIT_OFFSET_DAYS 偏移(默认 0=公告日本身), + 两出口(fc_events/fc_pair)同步生效。 """ schema = {"vt_symbol": pl.Utf8, "eff": pl.Date, "REPORT_DATE": pl.Date, "forecast_type_score": pl.Float64, "forecast_change_pct": pl.Float64, @@ -55,6 +69,9 @@ def _load_forecast_events(codes: list[str], data_dir: str) -> tuple[pl.DataFrame except ValueError: continue code = pl.col("股票代码").cast(pl.Utf8).str.strip_chars().str.zfill(6) + eff = pl.col("公告日期").cast(pl.Date, strict=False) + if FORECAST_PIT_OFFSET_DAYS: # eff+N 敏感性开关(见模块级常量注释) + eff = eff.dt.offset_by(f"{FORECAST_PIT_OFFSET_DAYS}d") # 交易所映射: 60→SSE;北交前缀白名单(92/43/82/83)→BJSE;其余→SZSE # (互评备注: 北交种类不得落入 SZSE——容器/实盘 universe 按后缀路由) is_bj = (code.str.starts_with("92") | code.str.starts_with("43") @@ -78,7 +95,7 @@ def _load_forecast_events(codes: list[str], data_dir: str) -> tuple[pl.DataFrame mid.alias("_mid"), pl.lit(report_date, dtype=pl.Date).alias("REPORT_DATE"), pl.lit(_ANNUALIZE_FACTOR.get(report_date.month), dtype=pl.Float64).alias("_annf"), - pl.col("公告日期").cast(pl.Date, strict=False).alias("eff"), + eff.alias("eff"), pl.col("预告类型").cast(pl.Utf8).replace_strict( FORECAST_TYPE_SCORE, default=None, return_dtype=pl.Float64 ).alias("_score"), diff --git a/tests/factor/conftest.py b/tests/factor/conftest.py index b4ac07f..932bf0b 100644 --- a/tests/factor/conftest.py +++ b/tests/factor/conftest.py @@ -2,12 +2,26 @@ import sys import os +import pytest + # 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) +@pytest.fixture(autouse=True, scope="session") +def _pin_forecast_pit_offset_env(): + # 密闭测试(session 级,须先于任何 module 级 feat fixture 构建): 防外部环境变量 + # SANGUO_FORECAST_PIT_OFFSET_DAYS 渗入改变预告域生效日语义。其余测试一律测 + # 默认口径 0;偏移行为由 test_fundamental_forecast_pit_offset.py 专门覆盖 + # (其内部用 monkeypatch 自行设值/还原,不受本钉死影响)。 + os.environ.pop("SANGUO_FORECAST_PIT_OFFSET_DAYS", None) + import sanguo_factor.fundamental_forecast as _ff + _ff.FORECAST_PIT_OFFSET_DAYS = 0 + yield + + # ==================== 合成财务静态域(财务因子批测试共用) ==================== # 6 只股 × 24 报告期(2018Q1~2023Q4;2018-2019 为 16 季滚动窗/5 年 CAGR 预热史), # NOTICE_DATE 错位: Q1→当年4-28 / H1→当年8-29 / Q3→当年10-27 / 年报→次年4-25 diff --git a/tests/factor/test_fundamental_forecast_pit_offset.py b/tests/factor/test_fundamental_forecast_pit_offset.py new file mode 100644 index 0000000..e0e96e0 --- /dev/null +++ b/tests/factor/test_fundamental_forecast_pit_offset.py @@ -0,0 +1,122 @@ +# tests/factor/test_fundamental_forecast_pit_offset.py +"""业绩预告域(F 族)PIT 生效日偏移开关: eff+N 敏感性对照实验. + +FORECAST_PIT_OFFSET_DAYS(fundamental_forecast 模块级开关,默认 0=现行为): +预告常在收盘后公告,默认口径公告日 D 当天即用其信息,存在「当天已无法 +交易却用了信息」的前视嫌疑;开关 =N 时所有预告事件生效日推后 N 个自然日 +(D 日不可见,D+N 起可见),作用于 fc_events 三列与 fc_pair(F07 兑现差 +锚的预告腿),三表/分红/户数/十大流通股东/vb 域一律不动. + +跑批侧设置方式(取其一;batch_eval 未来接入,本次不改): + ①环境变量: SANGUO_FORECAST_PIT_OFFSET_DAYS=1(容器 docker run -e 即可) + ②进程内直设: + import sanguo_factor.fundamental_forecast as _ff + _ff.FORECAST_PIT_OFFSET_DAYS = 1 # 在 iter_fundamental_feature_chunks 之前设一次 +""" +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"))) + +from datetime import date + +import polars as pl +import pytest + +import sanguo_factor.fundamental_forecast as ffc +from sanguo_factor.fundamental_adapter import build_fundamental_features, FEATURE_COLUMNS +from sanguo_factor.fundamental_schema import _BEAT_COLS, _FORECAST_COLS + +A, B, C = "600000.SSE", "000001.SZSE", "300001.SZSE" + +# 预告族列(F 族): 允许随开关变化的全部列;其余 FEATURE_COLUMNS 为守卫列 +_FC_FAMILY = ["vt_symbol", "datetime", *(_FORECAST_COLS + _BEAT_COLS)] +_GUARD_COLS = ["vt_symbol", "datetime"] + [c for c in FEATURE_COLUMNS + if c not in set(_FORECAST_COLS) | set(_BEAT_COLS)] + + +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) + + +# ---------- ① offset=0: 与现行为完全一致 ---------- + +def test_default_offset_is_zero(monkeypatch): + # 生产默认 = 0(环境变量未设时的开关基线);环境变量可注入 N(容器 -e 跑批) + import importlib + try: + monkeypatch.delenv("SANGUO_FORECAST_PIT_OFFSET_DAYS", raising=False) + assert importlib.reload(ffc).FORECAST_PIT_OFFSET_DAYS == 0 + monkeypatch.setenv("SANGUO_FORECAST_PIT_OFFSET_DAYS", "1") + assert importlib.reload(ffc).FORECAST_PIT_OFFSET_DAYS == 1 + finally: + # 还原模块默认 0,防 reload 残留污染后续测试 + monkeypatch.delenv("SANGUO_FORECAST_PIT_OFFSET_DAYS", raising=False) + importlib.reload(ffc) + + +def test_offset_zero_equals_current_behavior(synthetic_static, monkeypatch): + monkeypatch.setattr(ffc, "FORECAST_PIT_OFFSET_DAYS", 0) + ev_def, pair_def = ffc._load_forecast_events([A, B, C], synthetic_static) + ev_0, pair_0 = ffc._load_forecast_events([A, B, C], synthetic_static) + assert ev_0.sort(["vt_symbol", "eff"]).equals(ev_def.sort(["vt_symbol", "eff"])) + assert pair_0.sort(["vt_symbol", "REPORT_DATE"]).equals( + pair_def.sort(["vt_symbol", "REPORT_DATE"])) + # 生效日逐值 = fixture 公告日期(现行为锚): A 三条 / B / C + assert (ev_0.filter(pl.col("vt_symbol") == A).sort("eff")["eff"].to_list() + == [date(2023, 5, 10), date(2023, 7, 15), date(2023, 10, 15)]) + assert ev_0.filter(pl.col("vt_symbol") == B)["eff"].to_list() == [date(2023, 7, 20)] + assert ev_0.filter(pl.col("vt_symbol") == C)["eff"].to_list() == [date(2023, 7, 10)] + # PIT 可见性 = 现行为: 公告日当天即见(07-15 见 56.79,前一日仍见上一条 15.0) + df = build_fundamental_features([A], "2023-07-01", "2023-07-31", data_dir=synthetic_static) + assert val(df, A, "2023-07-14", "forecast_change_pct") == pytest.approx(15.0) + assert val(df, A, "2023-07-15", "forecast_change_pct") == pytest.approx(56.79) + + +# ---------- ② offset=1: 生效日 D → D+1(自然日) ---------- + +def test_offset_one_shifts_eff_one_calendar_day(synthetic_static, monkeypatch): + monkeypatch.setattr(ffc, "FORECAST_PIT_OFFSET_DAYS", 1) + ev, pair = ffc._load_forecast_events([A, B, C], synthetic_static) + # 事件层: 所有公告日 D 的生效日变 D+1 + assert (ev.filter(pl.col("vt_symbol") == A).sort("eff")["eff"].to_list() + == [date(2023, 5, 11), date(2023, 7, 16), date(2023, 10, 16)]) + assert ev.filter(pl.col("vt_symbol") == B)["eff"].to_list() == [date(2023, 7, 21)] + assert ev.filter(pl.col("vt_symbol") == C)["eff"].to_list() == [date(2023, 7, 11)] + # fc_pair(F07 配对原料)同步推后: 2022 年报预告腿 05-10 → 05-11 + late = pair.filter((pl.col("vt_symbol") == A) + & (pl.col("REPORT_DATE") == date(2022, 12, 31))) + assert late["eff"].to_list() == [date(2023, 5, 11)] + # 特征层(= batch_eval 同路径): D 日不可见(仍见上一条),D+1 起可见 + df = build_fundamental_features([A], "2023-07-01", "2023-07-31", data_dir=synthetic_static) + assert val(df, A, "2023-07-15", "forecast_change_pct") == pytest.approx(15.0) + assert val(df, A, "2023-07-15", "forecast_type_score") == 3.0 + assert val(df, A, "2023-07-16", "forecast_change_pct") == pytest.approx(56.79) + # F07 兑现差锚 = max(实际披露 04-25, 预告腿 05-11) = 05-11: + # 05-10(默认锚日)不可见,05-11 起可见 + df2 = build_fundamental_features([A], "2023-05-01", "2023-05-31", data_dir=synthetic_static) + assert val(df2, A, "2023-05-10", "forecast_beat") is None + assert val(df2, A, "2023-05-11", "forecast_beat") == pytest.approx((58 - 55) / 55) + + +# ---------- ③ 守卫: 其他域不受开关影响 ---------- + +def test_offset_one_leaves_other_domains_untouched(synthetic_static, monkeypatch): + monkeypatch.setattr(ffc, "FORECAST_PIT_OFFSET_DAYS", 0) + df0 = build_fundamental_features( + [A, B, C], "2023-01-01", "2023-12-31", data_dir=synthetic_static) + monkeypatch.setattr(ffc, "FORECAST_PIT_OFFSET_DAYS", 1) + df1 = build_fundamental_features( + [A, B, C], "2023-01-01", "2023-12-31", data_dir=synthetic_static) + key = ["vt_symbol", "datetime"] + # 三表/分红/户数/十大流通股东/vb 全部输出列整帧逐值等值 + assert df0.select(_GUARD_COLS).sort(key).equals(df1.select(_GUARD_COLS).sort(key)) + # 非空守卫: 预告族列确有变化(开关生效,上面的等值不是恒真比较) + assert not df0.select(_FC_FAMILY).sort(key).equals(df1.select(_FC_FAMILY).sort(key))