feat(factor): F族PIT生效日偏移开关——SANGUO_FORECAST_PIT_OFFSET_DAYS环境变量/进程内可设,4密闭测试双环境态绿+conftest session级钉死防环境渗透 [nas]
This commit is contained in:
@@ -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"),
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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))
|
||||
Reference in New Issue
Block a user