"""TTM 股息率毛刺消除测试。 **背景**：A 股相邻两次除权的间隔经常不是 365 天。硬 365 天窗口于是在每年 除权日附近制造两种日历假象： - **重叠虚高**：间隔 < 365 时新旧分红同时在窗口内。实测招商银行 2015-07-03 股息率 0.620 → 1.290（+108%），10 天后回落到 0.670。 - **断档虚低**：间隔 > 365 时旧的已到期而新的未入场。实测中国神华 2016-07-04：0.740 → 0.320（−57%）。 两者都会污染「历史分位」这一核心信号，且筛选器用的也是同一个数 （直接决定选股），因此必须消除。 """ from __future__ import annotations import numpy as np import pandas as pd import pytest from hdiv.factor.dividend_yield import ( build_dps_events, ttm_dps_at, ttm_dps_series, ttm_params, ) TTM = 365 GRACE = 45 def _events(pairs: list[tuple[str, float]]) -> pd.DataFrame: """构造分红事件表：(除权日, 金额)。""" return pd.DataFrame({ "ex_date": pd.to_datetime([d for d, _ in pairs]), "imp_ann_date": pd.to_datetime([d for d, _ in pairs]), "cash_div_tax": [a for _, a in pairs], }) def _daily(start: str, end: str) -> pd.DatetimeIndex: return pd.date_range(start, end, freq="D") # --------------------------------------------------------------------------- # 核心：两种毛刺都要消除 # --------------------------------------------------------------------------- def test_overlap_spike_is_removed() -> None: """间隔 360 天：新分红入场时旧的不应再计入（消除 +100% 虚高）。""" d = _daily("2020-01-01", "2023-12-31") ev = _events([("2021-06-01", 1.0), ("2022-05-27", 1.2), ("2023-05-22", 1.4)]) raw = pd.Series(ttm_dps_series(d, ev, ttm_days=TTM, grace_days=GRACE, smooth_spikes=False), index=d) sm = pd.Series(ttm_dps_series(d, ev, ttm_days=TTM, grace_days=GRACE, smooth_spikes=True), index=d) # 未平滑时：2022-05-27 当刻涨到 1.0+1.2=2.2，5 天后旧的到期回落 assert raw.max() > 2.1, f"未平滑应出现重叠虚高，实际 max={raw.max()}" # 平滑后：不应出现两笔相加 assert sm.max() <= 1.45, f"平滑后不应双算，实际 max={sm.max()}" # 且切换当天不跳变 after = sm.loc[pd.Timestamp("2022-05-27"):].iloc[:5] assert after.max() / after.min() - 1 < 0.05, "接管当天不应有跳变" def test_gap_dip_is_filled() -> None: """间隔 370 天：旧的到期后应继续计到新的入场（消除断档虚低）。""" d = _daily("2020-01-01", "2023-12-31") ev = _events([("2021-06-01", 1.0), ("2022-06-06", 1.2)]) # 间隔 370 天 raw = pd.Series(ttm_dps_series(d, ev, ttm_days=TTM, grace_days=GRACE, smooth_spikes=False), index=d) sm = pd.Series(ttm_dps_series(d, ev, ttm_days=TTM, grace_days=GRACE, smooth_spikes=True), index=d) # 只看「第一笔到第二笔」这段（尾部无后继本就该归零，属正确行为） win = slice(pd.Timestamp("2021-06-01"), pd.Timestamp("2022-06-06")) raw_a, sm_a = raw.loc[win], sm.loc[win] # 未平滑：2022-06-01 旧的到期、新的还没来 → 归零 5 天 assert raw_a.min() == 0.0, "未平滑应出现断档归零" # 平滑后：同区间不应归零 assert sm_a.min() > 0.9, f"平滑后不应断档，实际 min={sm_a.min()}" def test_intra_year_multiple_payments_are_not_merged() -> None: """年内多次分红（间隔 180 天）必须都保留 —— 否则会把中期分红误删。""" d = _daily("2021-01-01", "2023-12-31") ev = _events([ ("2022-06-01", 0.3), ("2022-11-28", 0.7), ("2023-05-29", 0.3), ("2023-11-25", 0.7), ]) sm = pd.Series(ttm_dps_series(d, ev, ttm_days=TTM, grace_days=GRACE, smooth_spikes=True), index=d) # 年中确实应同时含两笔（0.3 + 0.7 = 1.0） peak = sm.loc[pd.Timestamp("2023-06-01"):pd.Timestamp("2023-11-20")] assert peak.max() > 0.95, f"年内两笔分红应同时计入，实际 max={peak.max()}" def test_true_cessation_still_goes_to_zero() -> None: """真停发必须如实归零，不能因为平滑就永远挂着旧分红。""" d = _daily("2020-01-01", "2025-12-31") ev = _events([("2021-06-01", 1.0), ("2023-06-01", 1.0)]) # 中间空了两年 sm = pd.Series(ttm_dps_series(d, ev, ttm_days=TTM, grace_days=GRACE, smooth_spikes=True), index=d) # 2021 那笔在 2022-06-01 + 45 天宽限后必须归零 gap = sm.loc[pd.Timestamp("2022-09-01"):pd.Timestamp("2023-05-31")] assert gap.max() == 0.0, f"停发期间应归零，实际 max={gap.max()}" def test_smoothing_can_be_disabled() -> None: """smooth_spikes=False 应精确复现旧的「硬窗口 + 归零才兜底」行为。""" d = _daily("2020-01-01", "2023-12-31") ev = _events([("2021-06-01", 1.0), ("2022-05-27", 1.2)]) off = ttm_dps_series(d, ev, ttm_days=TTM, grace_days=GRACE, smooth_spikes=False) # 旧行为：重叠期双算 assert off.max() >= 2.1 on = ttm_dps_series(d, ev, ttm_days=TTM, grace_days=GRACE, smooth_spikes=True) assert on.max() < off.max() def test_build_dps_events_matches_series_expectations() -> None: """事件表经 build_dps_events 规范化后仍可用。""" raw = pd.DataFrame({ "symbol": ["X"] * 3, "ex_date": ["2021-06-01", "2022-05-27", "2023-05-22"], "imp_ann_date": ["2021-05-25", "2022-05-20", "2023-05-15"], "cash_div_tax": [1.0, 1.2, 1.4], }) e = build_dps_events(raw)["X"] d = _daily("2021-01-01", "2023-12-31") out = ttm_dps_series(d, e, ttm_days=TTM, grace_days=GRACE, smooth_spikes=True) assert len(out) == len(d) and out.max() <= 1.45 # --------------------------------------------------------------------------- # 口径统一：筛选 / 画像 / 回测 / Web 必须用同一份参数 # --------------------------------------------------------------------------- def test_ttm_params_is_single_source_of_truth() -> None: """四处调用点必须都从 ttm_params() 取参，不得各自硬编码。""" import inspect from pathlib import Path root = Path(__file__).resolve().parents[1] / "src" / "hdiv" # 回测引擎曾硬编码 ttm_days=365, grace_days=45 eng = (root / "backtest" / "engine.py").read_text(encoding="utf-8") assert "ttm_days=365, grace_days=45" not in eng, "引擎仍在硬编码 TTM 参数" assert "ttm_params()" in eng # walk-forward 与 web 曾用函数默认值 for rel in ("backtest/walk_forward.py", "web/analysis.py"): text = (root / rel).read_text(encoding="utf-8") assert "ttm_params()" in text, f"{rel} 未使用统一参数" # 因子层自身 f = inspect.getsource(__import__( "hdiv.factor.dividend_yield", fromlist=["x"])) assert "def ttm_params" in f def test_ttm_dps_at_matches_series_right_endpoint() -> None: """单点求值（筛选器用）必须与序列右端点一致。""" d = _daily("2020-01-01", "2022-12-31") ev = _events([("2021-06-01", 1.0), ("2022-05-27", 1.2)]) asof = pd.Timestamp("2021-12-31").date() one = ttm_dps_at(asof, ev) ser = ttm_dps_series(pd.DatetimeIndex([pd.Timestamp(asof)]), ev, ttm_days=TTM, grace_days=GRACE, smooth_spikes=True) assert one is not None assert abs(one - float(ser[0])) < 1e-9 def test_ttm_params_reads_config() -> None: from hdiv.core.config import load_config w, g, sm = ttm_params() c = load_config("profile").ttm_dividend assert (w, g, sm) == (c.window_days, c.grace_days, c.smooth_spikes) def test_config_exposes_smooth_spikes_switch() -> None: """开关必须暴露在 YAML 里，用户可自行关闭。""" from hdiv.core.config import load_config assert hasattr(load_config("profile").ttm_dividend, "smooth_spikes") @pytest.mark.parametrize("grace", [0, 10, 45, 90]) def test_smoothing_never_produces_negative_or_nan(grace: int) -> None: d = _daily("2020-01-01", "2023-12-31") ev = _events([("2021-06-01", 1.0), ("2022-06-06", 1.2), ("2023-06-01", 1.4)]) out = ttm_dps_series(d, ev, ttm_days=TTM, grace_days=grace, smooth_spikes=True) assert np.isfinite(out).all() assert (out >= 0).all() # --------------------------------------------------------------------------- # 透传一致性：包装函数必须接收并转发所有参数 # --------------------------------------------------------------------------- def test_wrapper_signature_forwards_all_params() -> None: """回归：`dividend_yield_series` 是 `ttm_dps_series` 的包装。 曾经只给**调用方**加了 `smooth_spikes`，却忘了在包装函数签名里声明， 于是 walk-forward 直接 `TypeError` 崩在第一个窗口 —— 而测试全绿， 因为测试没走 walk-forward 那条路径。 """ import inspect from hdiv.factor import dividend_yield as dy inner = set(inspect.signature(dy.ttm_dps_series).parameters) - {"dates", "events"} outer = set(inspect.signature(dy.dividend_yield_series).parameters) - {"close", "events"} missing = inner - outer assert not missing, ( f"dividend_yield_series 未转发参数 {sorted(missing)}；" "调用方传了就会 TypeError" ) # 且必须真的往下传 src = inspect.getsource(dy.dividend_yield_series) for name in inner: assert f"{name}={name}" in src, f"包装函数未把 {name} 传给 ttm_dps_series" def test_wrapper_accepts_smooth_spikes() -> None: """直接以关键字调用，确保签名真的可用（不只是字符串包含）。""" from hdiv.factor.dividend_yield import dividend_yield_series d = _daily("2021-01-01", "2022-12-31") close = pd.Series(10.0, index=d) ev = _events([("2021-06-01", 1.0), ("2022-05-27", 1.2)]) for flag in (True, False): out = dividend_yield_series(close, ev, ttm_days=365, grace_days=45, smooth_spikes=flag) assert not out.empty assert "dividend_yield" in out.columns def test_all_ttm_callers_pass_the_unified_params() -> None: """五处调用点都必须显式传 smooth_spikes，不能靠默认值（否则与配置脱钩）。""" from pathlib import Path root = Path(__file__).resolve().parents[1] / "src" / "hdiv" for rel in ("profile/builder.py", "backtest/engine.py", "backtest/walk_forward.py", "web/analysis.py"): src = (root / rel).read_text(encoding="utf-8") assert "smooth_spikes" in src, f"{rel} 未传 smooth_spikes（会与配置脱钩）"