diff --git a/docs/factors/d5_runner_difference_catalogue.md b/docs/factors/d5_runner_difference_catalogue.md index fb0d0b6..783e6d3 100644 --- a/docs/factors/d5_runner_difference_catalogue.md +++ b/docs/factors/d5_runner_difference_catalogue.md @@ -378,3 +378,84 @@ anchors 腿同判据：hand 侧按旧几何算、service 侧按新几何算， 6. jump 的 `spec.description` 与 `sections[7].payload.factor_version`（连同原已特判的 `spec.version`）纳入 `registered_correction_effect`——仅当新 JSON 带 `corrections` 结构化更正承载（契约 v1.1）时成立。 + +### 七之四、C4b（vpq 绑定 + runner 启用）引入的 artifact 漂移 —— 预登记（2026-07-28） + +1. **`valley_price_quantile_20` 的 reports 腿：新增 add-Section `neutralization_coverage`**。 + 旧 runner 对 NeutralizationCoverage 只 **log + 进 result dataclass**（§三 机制 B），从未 + add-Section；统一 runner 把它经 §3.6 扩展点装配进报告。冻结 exec artifact 没有这一节， + 故新 artifact 的 JSON `sections[8].*` 全部叶子与 MD 的整节渲染行（`## + neutralization_coverage` + 标题、note 行、9 个 payload `- key:` 行及分节空行）是**新增**。登记方式：按 section **名字** + 匹配（`qt/factor_eval_reconcile.py::REGISTERED_EXTRA_SECTIONS`，JSON 侧读新 JSON 该下标 + section 的 `name` 字段核对，**绝不按裸下标放行**；MD 侧前缀从 dataclass 字段派生，改名即红）。 + 任何其他新增 section 不登记、仍失败。 +2. **vpq 的 panels 腿预期差异类**：与 pooled 因子同口径——`warmup_left_extension`（旧 runner + 左端锚 `data.start` 的 anchor 截断 vs materializer 饱和左延到 2015-01-05；早区 + [2021-07-01, 2021-10-31] 三方向 + 按月非递增）。vpq 的反转中性化在旧几何是 + `reversal_20`（未滞后面板）、新几何是 `reversal_20_shifted`（决策滞后面板），两者 + **逐位等价**（C4b 双侧钉死），故平移面板本身**不产生**类外差异；左端 22 个交易日窗口的 + 零 NaN 差异已在 fixture 层钉死。若真实缓存对账出现类外差异，如实全报。 + +### 七之五、C4b vpq 真实对账（2026-07-28 实测）——panels 腿差异分解与 anchors 修复登记 + +真实对账（cache-only，`stk_mins_live_calls=0`；store 由本次 run 冷填，pooled 饱和加载 +1132s）：reports 4/4 OK；anchors 修复后 5/5（ok=4 + warmup=1）；**panels FAIL，差异已全量 +分解如下（三块，类外 0 未解释）——具名类边界**已由 lead 裁定（2026-07-28）并落地**， +三条裁定与边界参数见本节末**： + +1. **warmup 块（24,516 格，全 finite→finite，2021-07-30→2021-09-30）**：即登记的 + `warmup_left_extension`（anchor 截断 vs 饱和左延）。**但 vpq 的逐月计数结构性地 + 从部分月开始**（2021-07: 902 → 2021-08: 19,975 → 2021-09: 3,639）——冻结面板的残差 + 要到 ~07-29 才存在（qbar 需 ≥10 有效日 + rev20 需 21 收盘），7 月天然非满月，pooled + 规则「按月非递增」在首月必假。monotonicity 起点需按因子推迟（如从首个满月起算）。 +2. **600623.SH 分类翻转块（19,143 格，2023-06-01→2023-07-14）**：直接受影响行只有 + **600623.SH 一只、30 个交易日**（max |Δ|=1.60e-04）；其余 19,113 格（959 票 ≤ 5.5e-07） + 是该股输入变化经**逐日 OLS 系数**污染全截面的二阶效应。与 C4 首轮登记的 + `threshold_flip_tail`（volume_peak_count，**同一 symbol 同一窗口** 600623.SH + 2023-06-15..07-14）同族：D2 迁移前后 PRV 分类在临界 bar 上的翻转（rolling-sigma 浮点 + 噪声 × 整数成交量），在 vpq 身上表现为 valley VWAP→q_day→qbar 的连续值变化 + + 截面污染，而非 ±1 计数。qbar 本身新旧几何**逐位一致**（max 1.1e-16，旧锚 vs 饱和 + 三票抽样 141 格）；日频/分钟缓存 ledger 自冻结（2026-07-24）前零写入（日频 max + fetched_at 2026-07-18、分钟 2026-07-16），**输入侧无变化**——差异源是冻结面板由 + pre-D2 旧代码路径产生、当前引擎是迁移后 primitives。 +3. **浮点尘（~232 格）**：rel>5e-12 中 abs ≤ 1e-8 的部分（残差量级 ~1e-5，rel 判据在 + 近零值上失真，abs 实为 ~1e-16 机器精度）；加上 rel ≤ 5e-12 的 707 格**超过** + `FLOAT_TAIL_MAX_CELLS`=101 的既有上限。`float_reordering_tail` 的 cap 是按 jump 实测 + 标定的，对截面 OLS 残差（近零值多、格子多）系统性偏紧——需要 abs 下限或按因子标定。 + +**anchors 模式修复（已落地）**：截面因子（`stores_intermediate`）的 served 值是请求 +universe 的函数，只请求锚行 symbol（5 只 < `min_cross_section`=10）会全 NaN——改为 +**按 config 全 universe 服务再查锚格**（per-symbol 因子保持锚-only 请求）。实测：修复前 +4/5 行 service=NaN（假 FAIL），修复后 4 行精确对上（2 行 rel=0.0、2 行 ~1e-15）+ 1 行 +warmup_end（2021-07-30，rel 2.61e-01，属登记 warmup 类）。**intraday_amp_cut 有同样性质， +C5 全量对账会踩到同一处**——修复是通用的。 + +**第三条路（shifted vs legacy rev）在真实全 universe 上逐位一致**：served（引擎 shifted +路径）vs `residualize_on_reversal(qbar, reversal_20(未滞后面板))`，1,135,926 格 +max|diff|=0.0、NaN 集合互差为 0——平移面板左端效应在真实缓存上**不存在**，与 fixture +层钉死一致。 + +**lead 三裁定（2026-07-28，已落地 `qt/factor_eval_reconcile.py` + 正反向测试）**： + +1. **pooled 月单调性起点**：`warmup_left_extension` 的 pooled 按月非递增检查豁免**一个 + 结构性锚定的部分月**——豁免月 = 冻结面板上该因子**首个 finite 值所在的月**（残差/值 + 存在性起始的那个部分月，不参与单调性判定；该月差异仍必须在早区窗内且方向合规）， + 豁免月之后仍强制非递增，违反即失败。*理由*：vpq 冻结面板的残差 ~07-29 才存在 + （qbar 需 ≥10 有效日 + rev20 需 21 收盘），首月结构性必为部分月，对它施加非递增是 + 判一个构造上必假的命题。**锚定必须是结构性的（评审 LOW-1 修正）**：初版实现是位置性 + 的——豁免「第一个**有** warmup diff 的月」，评审实测「豁免月零 diff + 随后 Aug→Sep + 增长形」被放行，裁定理由被绕过；改为从**冻结网格**读首个 finite 值月后，该 probe + 反转（FAIL）。冻结面板全 NaN（无 finite 值）→ 不豁免（保守方向）。 +2. **新具名类 `threshold_flip_contamination`（仅截面因子）**：窗口 + [2023-06-01, 2023-07-14]；直接 symbol（600623.SH）|diff| ≤ **2e-04**；其余 symbol + 污染格 |diff| ≤ **1e-06**；总 cell 数 ≤ **20,000**。这是本类全部参数，逐格校验， + 任何越界即 unclassified → FAIL。*理由*：与 `threshold_flip_tail` 同 symbol 同窗口 + 同族（PRV 临界 bar 分类翻转 × rolling-sigma 浮点噪声 × 整数成交量），vpq 表现为 + 连续值（600623.SH 30 日 max 1.60e-04）+ 逐日截面 OLS 系数污染全截面（19,113 格 + ≤5.5e-07）；非输入侧变化——两缓存 ledger 自冻结前零写入、qbar 新旧几何逐位一致 + （max 1.1e-16）、同 symbol+窗口翻转已对 volume_peak 登记过。 +3. **float 尾判据修两处**：① 增加 abs 下限——|diff| ≤ **1e-12** 一律计 float dust + （近零残差上 rel 判据失真，实测 abs ~1e-16）；② cap 分档——bars-only 因子 ≤101 + （不变），**截面因子 ≤1,000**（实测 707 + 余量）。全局容差不动。*理由*：101 的 cap + 按 jump（bars-only）标定，对近零值多、格子多的截面 OLS 残差系统性偏紧；abs 下限 + 把「rel 在近零值上失真」从 cap 压力里剥离，两处都不放宽对真回归的牙。 diff --git a/factors/compute/minute/binding.py b/factors/compute/minute/binding.py index a3afeef..2b704c3 100644 --- a/factors/compute/minute/binding.py +++ b/factors/compute/minute/binding.py @@ -43,10 +43,16 @@ in ``qt`` because it is FACTOR knowledge; ``qt.panel_freeze`` keeps its own recipe copy (a frozen D1 tool) and D6 may later fold it onto this binding. -Deferred (readable error, D5): ``valley_price_quantile`` also needs the DAILY -close panel (its reversal neutralization), so it does not fit the pure -``(factor, bars) -> series`` shape and is NOT bound here — the materializer -raises for it rather than silently mis-computing. +``valley_price_quantile`` (D5 C4b) is the one factor whose combine ALSO needs the +DAILY close panel (its reversal neutralization), declared via +:class:`DailyCombineInput` on its stream binding — a DECLARATION the engine +consults, never an isinstance dispatch (red line #5). Its per-symbol stage (the +raw trailing-mean quantile ``qbar``) is universe-independent and is what the +store persists; the residualization runs at read-assembly over the requested +universe, fed a DECISION-LAGGED daily close panel (the shifted-panel third path: +the panel's row d already carries close_{d-1}, so the reversal's internal T-1 +shift is OFF — ``reversal_20_shifted`` — which is algebraically identical to the +legacy ``reversal_20`` on the un-lagged panel, pinned bit-for-bit by test). Layering: factor layer only (never qt / feeds). """ @@ -92,7 +98,13 @@ RidgeMinuteReturnFactor, compute_ridge_minute_return, ) -from factors.compute.minute.valley_price_quantile import ValleyPriceQuantileFactor +from factors.compute.minute.valley_price_quantile import ( + VALLEY_QUANTILE_REVERSAL_DAYS, + ValleyPriceQuantileFactor, + compute_valley_price_quantile_stats, + residualize_on_reversal, + reversal_20_shifted, +) from factors.compute.minute.valley_relative_vwap import ( ValleyRelativeVwapFactor, compute_valley_relative_vwap, @@ -127,8 +139,9 @@ def _call(factor: Factor, bars: pd.DataFrame) -> pd.Series: #: factor class -> its bars-based raw compute. Bound: the 10 minute factors whose -#: raw compute needs ONLY the 1min bars. NOT bound: valley_price_quantile (needs -#: the daily panel too — deferred, readable error below). +#: raw compute needs ONLY the 1min bars. NOT bound here: valley_price_quantile +#: (its value also needs the daily panel, so it has no bars-only whole-factor +#: form — it IS bound in the two-stage table below, with a declared daily input). _MINUTE_BINDINGS: dict[type[Factor], BindingFn] = { JumpAmountCorrFactor: _bind(compute_jump_amount_corr), MinuteIdealAmplitudeFactor: _bind(compute_minute_ideal_amplitude), @@ -150,6 +163,43 @@ def _call(factor: Factor, bars: pd.DataFrame) -> pd.Series: STATS_VALUE_COL = "value" +@dataclass(frozen=True) +class DailyCombineInput: + """DECLARED daily-panel input of a minute factor's combine (D5 C4b). + + A factor whose cross-sectional combine also reads the DAILY panel declares + that need here; the engine (``factors.materialize.combine_daily_panel``) + consults the declaration to load, view-lag and trailing-trim the panel, and + hands it to ``combine``. No isinstance dispatch anywhere (red line #5): the + engine asks the binding, the binding answers. + + ``columns``: the daily panel columns the combine reads (``("close",)`` for + valley_price_quantile's reversal neutralization). + + ``warmup_days``: the trailing trading days of the LAGGED panel the combine + needs before the emit start, DECLARED (never inferred) so the trim is exact + and single-fill / batch-fill combines read bit-identical inputs (§3.5 P8). + For a T-1-shifted ``days``-span reversal this is ``days + 2``: the reversal + consumes ``days + 1`` lagged rows (d .. d-days) and the decision lag's own + leading NaN row must land one day before that window (§六.18 nested-lookback + trap: an under-deep panel would fabricate edge NaNs the legacy path never + had). + + ``combine(factor, stats, daily)`` -> the factor's daily raw Series from the + assembled full-universe intermediate plus the (lagged, trimmed) daily panel. + Same CONTRACT as ``MinuteStreamBinding.combine`` — a PER-DATE reduction in + ``stats`` given a fixed ``daily`` panel — PLUS one more, load-bearing for the + pooled saturation loop: it returns EXACTLY the intermediate's rows (values + may be NaN), so the value's output dates can be read off the intermediate + without loading the daily panel (``residualize_on_reversal`` already returns + over exactly ``qbar``'s rows; pinned by test). + """ + + columns: tuple[str, ...] + warmup_days: int + combine: Callable[[Factor, pd.DataFrame, pd.DataFrame], pd.Series] + + @dataclass(frozen=True) class MinuteStreamBinding: """The split of a minute factor into per-symbol work + one cross-section pass. @@ -182,6 +232,11 @@ class MinuteStreamBinding: #: for diagnostics re-runs the per-symbol math on bars ALREADY in memory (no #: extra I/O, no effect on any value); that cost is disclosed at the call site. per_symbol_diagnostics: Callable[[Factor, pd.DataFrame], pd.DataFrame] | None = None + #: OPTIONAL declared daily-panel input of the combine (D5 C4b). ``None`` + #: means the combine needs only the assembled intermediate (the ten + #: bars-only factors); valley_price_quantile declares its reversal + #: neutralization's daily close need here. + combine_daily: DailyCombineInput | None = None def _pure_stream(compute_fn) -> MinuteStreamBinding: @@ -259,9 +314,79 @@ def _amp_cut_combine(factor: Factor, stats: pd.DataFrame) -> pd.Series: return combine_amp_cut_cross_section(stats, name=factor.name) -#: factor class -> its two-stage streaming binding. Same coverage as -#: ``_MINUTE_BINDINGS`` (the ten bars-only factors); a factor bound in one table -#: and not the other is a readable error, pinned by test. +#: Column name of valley_price_quantile's per-symbol intermediate: the RAW +#: trailing-mean price quantile (steps 1-5), BEFORE the reversal neutralization. +#: Named ``raw_qbar`` rather than ``STATS_VALUE_COL`` on purpose: for the pure +#: factors the intermediate IS the value; here it is the pre-neutralization +#: input to the cross-sectional combine, and the store artifact should say so. +RAW_QBAR_COL = "raw_qbar" + + +def _vpq_per_symbol(factor: Factor, bars: pd.DataFrame) -> pd.DataFrame: + """Steps 1-5 of ``valley_price_quantile``: the raw ``qbar`` intermediate. + + Universe-independent (the classification, the daily quantile and the trailing + valid-day mean are all strictly per symbol), so THIS is what the store + persists (D4c); the reversal neutralization is the cross-sectional combine. + """ + series = compute_valley_price_quantile_stats( + bars, lookback_days=factor.lookback_days, name=factor.name # type: ignore[attr-defined] + ) + return series.to_frame(RAW_QBAR_COL) + + +def _vpq_combine_with_daily( + factor: Factor, stats: pd.DataFrame, daily: pd.DataFrame +) -> pd.Series: + """Step 6 of ``valley_price_quantile``: the per-date reversal neutralization. + + Run ONCE on the assembled universe at read-assembly, exactly like the legacy + runner — but ``daily`` is the DECISION-LAGGED close panel, so the reversal is + taken by :func:`reversal_20_shifted` (internal T-1 OFF), which is + algebraically identical to the legacy ``reversal_20`` on the un-lagged panel + (pinned bit-for-bit by ``tests/test_minute_binding_vpq.py``). + """ + if RAW_QBAR_COL not in stats.columns: + raise ValueError( + f"{factor.name}: the per-symbol intermediate must carry the " + f"'{RAW_QBAR_COL}' column; got {list(stats.columns)}." + ) + rev = reversal_20_shifted(daily, days=VALLEY_QUANTILE_REVERSAL_DAYS) + return residualize_on_reversal(stats[RAW_QBAR_COL], rev, name=factor.name).sort_index( + kind="mergesort" + ) + + +#: valley_price_quantile's declared daily need: the front-adjusted close column, +#: ``reversal_days + 2`` lagged trailing trading days of warmup (the reversal +#: consumes reversal_days + 1 lagged rows; the decision lag's leading NaN row +#: must land one day earlier — see DailyCombineInput). +_VPQ_DAILY_INPUT = DailyCombineInput( + columns=("close",), + warmup_days=VALLEY_QUANTILE_REVERSAL_DAYS + 2, + combine=_vpq_combine_with_daily, +) + + +def _requires_daily_combine(factor: Factor, stats: pd.DataFrame) -> pd.Series: + """The plain ``combine`` slot of a factor whose combine needs the daily panel. + + Such a combine cannot run on the intermediate alone, so the two-argument + entry point is a readable error directing the caller to the declared + ``combine_daily`` path — never a silent mis-compute with a missing input. + """ + raise ValueError( + f"{factor.name}: its combine declares a daily-panel input " + f"(combine_daily); call combine_minute_stats(factor, stats, daily=...) " + f"with the view-lagged daily panel." + ) + + +#: factor class -> its two-stage streaming binding. Superset of +#: ``_MINUTE_BINDINGS``: the ten bars-only factors live in both tables; +#: valley_price_quantile lives ONLY here (its combine declares a daily input, so +#: it has no bars-only whole-factor form). The exact relationship is pinned by +#: ``tests/test_factor_materialize_streaming.py``. _MINUTE_STREAM_BINDINGS: dict[type[Factor], MinuteStreamBinding] = { JumpAmountCorrFactor: _pure_stream(compute_jump_amount_corr), MinuteIdealAmplitudeFactor: _pure_stream(compute_minute_ideal_amplitude), @@ -274,6 +399,15 @@ def _amp_cut_combine(factor: Factor, stats: pd.DataFrame) -> pd.Series: ), PeakIntervalKurtosisFactor: _pure_stream(compute_peak_interval_kurtosis), ValleyRelativeVwapFactor: _pure_stream(compute_valley_relative_vwap), + ValleyPriceQuantileFactor: MinuteStreamBinding( + per_symbol=_vpq_per_symbol, + # The combine needs the daily panel; the declared form lives in + # ``combine_daily`` and the engine routes the panel to it. The plain + # ``combine`` slot is a readable error, never a silent mis-compute. + combine=_requires_daily_combine, + intermediate_columns=(RAW_QBAR_COL,), + combine_daily=_VPQ_DAILY_INPUT, + ), # The three factors that publish a per-day gate-attrition disclosure. ValleyRidgeVwapRatioFactor: _pure_stream_with_diagnostics( compute_valley_ridge_vwap_ratio @@ -286,17 +420,12 @@ def _amp_cut_combine(factor: Factor, stats: pd.DataFrame) -> pd.Series: #: The factors whose cross-sectional combine is NOT the identity — i.e. the ones #: for which "split per symbol around the WHOLE factor" would be wrong. Declared -#: so the property is a checkable fact rather than a comment. -CROSS_SECTIONAL_MINUTE_FACTORS: frozenset[type[Factor]] = frozenset({IntradayAmpCutFactor}) - -#: Minute factors deliberately NOT bound (need extra inputs), with the reason. -_DEFERRED: dict[type[Factor], str] = { - ValleyPriceQuantileFactor: ( - "valley_price_quantile also consumes the DAILY close panel (its reversal " - "neutralization), so it does not fit the pure (factor, bars) binding; the " - "materializer defers it to D5 (a readable error, never a silent mis-compute)." - ), -} +#: so the property is a checkable fact rather than a comment. Both members store +#: their per-symbol intermediate (D4c); valley_price_quantile's combine +#: additionally declares a daily-panel input. +CROSS_SECTIONAL_MINUTE_FACTORS: frozenset[type[Factor]] = frozenset( + {IntradayAmpCutFactor, ValleyPriceQuantileFactor} +) #: VALID-DAY POOLED factors (design §3.3 review HIGH): their trailing window @@ -316,8 +445,8 @@ def _amp_cut_combine(factor: Factor, stats: pd.DataFrame) -> pd.Series: #: volume_peak_count / peak_interval_kurtosis / intraday_amp_cut / #: valley_relative_vwap roll over valid days too and are the "happens-to-be-clean" #: representatives (#82 lesson: never fix only what you observed). -#: valley_price_quantile is DEFERRED (needs the daily panel) but declared HERE so -#: D5 cannot bind it carrying the fixed-depth defect. +#: valley_price_quantile rolls over valid days as well (its raw qbar stage); it is +#: classified HERE so its D5 C4b binding cannot carry the fixed-depth defect. #: pooled factor class -> its same-slot BASELINE depth in trading days (the #: "locking offset", review point 1): a loaded day is CLASSIFICATION-FINAL only #: once it has this many strictly-prior trading days (the baseline reaches full @@ -433,6 +562,26 @@ def is_minute_bound(factor: Factor) -> bool: return type(factor) in _MINUTE_BINDINGS +def is_minute_stream_bound(factor: Factor) -> bool: + """True iff ``factor`` has a two-stage streaming binding here. + + Superset of :func:`is_minute_bound`: valley_price_quantile is stream-bound + (with a declared daily combine input) but has no bars-only whole-factor form. + """ + return type(factor) in _MINUTE_STREAM_BINDINGS + + +def minute_combine_daily_spec(factor: Factor) -> DailyCombineInput | None: + """The DECLARED daily-panel input of ``factor``'s combine, or ``None``. + + ``None`` means the combine needs only the assembled intermediate. The engine + consults this — a declaration lookup, not an isinstance dispatch (red line + #5) — to decide whether it must load and hand over the view-lagged daily + panel. + """ + return _stream_binding(factor).combine_daily + + def minute_raw_from_bars(factor: Factor, bars: pd.DataFrame) -> pd.Series: """Compute ``factor``'s raw daily Series from (cutoff-filtered) 1min ``bars``. @@ -457,10 +606,7 @@ def _stream_binding(factor: Factor) -> MinuteStreamBinding: def _raise_unbound(factor: Factor) -> NoReturn: - """The shared readable error for a deferred / unbound minute factor.""" - deferred = _DEFERRED.get(type(factor)) - if deferred is not None: - raise NotImplementedError(f"{factor.name}: {deferred}") + """The shared readable error for an unbound minute factor.""" raise KeyError( f"{factor.name} ({type(factor).__name__}) has no minute-bars binding; it " f"is not a minute-derived factor bound in factors.compute.minute.binding." @@ -507,14 +653,36 @@ def minute_intermediate_columns(factor: Factor) -> tuple[str, ...]: return tuple(_stream_binding(factor).intermediate_columns) -def combine_minute_stats(factor: Factor, stats: pd.DataFrame) -> pd.Series: +def combine_minute_stats( + factor: Factor, stats: pd.DataFrame, *, daily: pd.DataFrame | None = None +) -> pd.Series: """``factor``'s daily raw Series from the ASSEMBLED per-symbol intermediates. Run ONCE on the full-universe intermediate: this is where a cross-sectional factor's date-wise standardization happens, so its universe is the whole covered set exactly as in the single-frame path. + + ``daily``: the view-lagged, trailing-trimmed daily panel, REQUIRED iff the + binding declares a ``combine_daily`` input (valley_price_quantile) and + refused otherwise — a missing declared input or an undeclared extra one are + both readable errors, never a silent mis-compute. The engine builds it via + ``factors.materialize.combine_daily_panel``. """ - return _stream_binding(factor).combine(factor, stats) + binding = _stream_binding(factor) + spec = binding.combine_daily + if spec is None: + if daily is not None: + raise ValueError( + f"{factor.name}: a daily panel was passed but its combine declares " + f"no daily input — the panel would be silently ignored." + ) + return binding.combine(factor, stats) + if daily is None: + raise ValueError( + f"{factor.name}: its combine declares a daily-panel input " + f"(combine_daily) but no daily panel was passed." + ) + return spec.combine(factor, stats, daily) def is_cross_sectional_minute(factor: Factor) -> bool: @@ -525,16 +693,20 @@ def is_cross_sectional_minute(factor: Factor) -> bool: __all__ = [ "CROSS_SECTIONAL_MINUTE_FACTORS", "NOT_DECISION_CUTOFF_SAFE", + "RAW_QBAR_COL", "STATS_VALUE_COL", "VALID_DAY_POOLED_FACTORS", "BindingFn", + "DailyCombineInput", "MinuteStreamBinding", "combine_minute_stats", "has_minute_diagnostics", "is_cross_sectional_minute", "is_decision_cutoff_safe", "is_minute_bound", + "is_minute_stream_bound", "is_valid_day_pooled", + "minute_combine_daily_spec", "minute_diagnostics_from_bars", "minute_intermediate_columns", "minute_raw_from_bars", diff --git a/factors/compute/minute/valley_price_quantile.py b/factors/compute/minute/valley_price_quantile.py index 0505a9e..7af788d 100644 --- a/factors/compute/minute/valley_price_quantile.py +++ b/factors/compute/minute/valley_price_quantile.py @@ -432,6 +432,73 @@ def reversal_20( return rev.rename(name) +def reversal_20_shifted( + closes: pd.DataFrame | pd.Series, + *, + days: int = VALLEY_QUANTILE_REVERSAL_DAYS, + close_col: str = "close", + name: str = "rev20", +) -> pd.Series: + """``days``-day reversal on an ALREADY decision-lagged close panel (T-1 shift OFF). + + The decision view makes the previous day's close the legal same-day input + (availability policy: every daily field except ``open`` is visible at + ``<= d-1``), so the materializer hands the factor a panel whose row ``d`` + already carries ``close_{d-1}`` (``factors.view_lag.daily_decision_lag``). + Applying :func:`reversal_20`'s internal T-1 shift on top of that would lag + TWICE (``close_{d-2}``) — the defect the shifted-panel construction exists to + avoid. This variant therefore takes the ratio WITHOUT the internal shift: + + rev(d) = -(p_d / p_{d-days} - 1) where p_d = close_{d-1} + = -(close_{d-1} / close_{d-(days+1)} - 1) == reversal_20(closes) + + The two constructions are ALGEBRAICALLY IDENTICAL (measured bit-for-bit: + ``max|diff| = 0.0``, ``Series.equals`` True — the D5b review's third path), + and that equivalence is pinned on BOTH sides by + ``tests/test_minute_binding_vpq.py`` (including a sabotage control showing the + un-lagged panel passed here differs materially, so the equivalence cannot + pass vacuously). The shifted panel must carry one extra leading day of + history so the shift's first-row NaN lands before the needed window; the + engine's declared warmup (``binding.DailyCombineInput.warmup_days``) covers it. + + The cleaning order matters and is preserved: non-positive / non-finite closes + are blanked on the panel AS HANDED IN (post-lag), so a bad print poisons + exactly the same output rows as the un-lagged construction (blanked pre-shift + there, post-shift here — the same (date, symbol) cells either way). + + Args: + closes: ``MultiIndex(date, symbol)`` DECISION-LAGGED daily panel + (DataFrame with ``close_col``, or a Series of lagged closes). + days: reversal span in trading days (definition, not a tuned knob). + close_col: the close column name when ``closes`` is a DataFrame. + name: the returned Series name. + + Returns: + ``MultiIndex(date, symbol)`` Series over the SAME rows as the input. + Pure: never mutates ``closes``. + """ + if days < 1: + raise ValueError(f"days must be >= 1; got {days!r}.") + if isinstance(closes, pd.DataFrame): + if close_col not in closes.columns: + raise ValueError( + f"reversal_20_shifted needs a '{close_col}' column; got " + f"{list(closes.columns)}." + ) + series = closes[close_col] + else: + series = closes + if series.empty: + return empty_factor_series(name) + + s = series.astype(float).sort_index() + clean = s.where(np.isfinite(s.to_numpy(dtype=float)) & (s.to_numpy(dtype=float) > 0.0)) + by_symbol = clean.groupby(level=SYMBOL_LEVEL, sort=False) + base = by_symbol.shift(days) # p_{d-days} = close_{d-(days+1)} + rev = -(clean / base - 1.0) + return rev.rename(name) + + def residualize_on_reversal( qbar: pd.Series, rev: pd.Series, @@ -733,5 +800,6 @@ def compute(self, panel: pd.DataFrame) -> pd.Series: "compute_valley_price_quantile_stats", "residualize_on_reversal", "reversal_20", + "reversal_20_shifted", "valley_price_quantile_by_day", ] diff --git a/factors/materialize.py b/factors/materialize.py index ee30271..38d8619 100644 --- a/factors/materialize.py +++ b/factors/materialize.py @@ -56,6 +56,7 @@ from __future__ import annotations +import logging from collections.abc import Callable from dataclasses import dataclass from typing import Protocol @@ -69,8 +70,9 @@ from factors.compute.minute.binding import ( combine_minute_stats, is_cross_sectional_minute, - is_minute_bound, + is_minute_stream_bound, is_valid_day_pooled, + minute_combine_daily_spec, minute_diagnostics_from_bars, minute_intermediate_columns, minute_raw_from_bars, @@ -85,6 +87,8 @@ minute_decision_cutoff, ) +logger = logging.getLogger(__name__) + #: Extra calendar days loaded before the emit window so the exact trailing-trading #: -day trim always has enough history (covers weekends + holiday clusters + the #: daily shift's one extra day). Generous by design: the CORRECTNESS floor is the @@ -296,10 +300,20 @@ def _warmup_start(dates: pd.DatetimeIndex, emit_start: pd.Timestamp, warmup: int trading days precede ``emit_start`` the earliest available date is used (near the data start the emit rows are honestly under-warmed -> NaN, and single/batch stay consistent because both trim to the same earliest date). + The clamp is DISCLOSED with one INFO log line (the clamp fact, the actual + start date, and the declared warmup depth) — the saturation disclosures + set the convention that a silently narrowed load window is never silent. """ order = dates.sort_values() pos = int(order.searchsorted(emit_start, side="left")) # index of emit_start (or ins.) keep_idx = max(0, pos - int(warmup)) + if pos - int(warmup) < 0 and len(order): + logger.info( + "warmup trim clamped to the data start %s: only %d trading day(s) " + "precede emit_start %s, fewer than the declared warmup depth %d " + "(emit rows near the data start are honestly under-warmed).", + order[0].date(), pos, pd.Timestamp(emit_start).date(), int(warmup), + ) return order[keep_idx] @@ -453,14 +467,19 @@ def _materialize_minute( which its ``AMP_CUT_MIN_CROSS_SECTION`` gate turns entirely into NaN — measured, and never worked around by relaxing that definition constant). + A factor whose combine declares a DAILY input (valley_price_quantile's + reversal neutralization) is handed the view-lagged, trailing-trimmed daily + panel from :func:`combine_daily_panel` — the shifted-panel third path, + bit-identical to the legacy un-lagged construction by pinned test. + Two load geometries, unchanged in meaning from the single-frame engine: bounded factors keep the fixed trailing-trading-day trim; valid-day-POOLED factors expand to saturation. Both are now decided PER SYMBOL — see :func:`_pooled_symbol_stats` for why that is the same terminal, not a weakening. """ - if not is_minute_bound(factor): - # Readable error (e.g. valley_price_quantile needs the daily panel too). + if not is_minute_stream_bound(factor): + # Readable error for a factor with no minute binding at all. if sources.minute is None: raise ValueError( f"{factor.name} is a minute factor but no MinuteBarProvider was injected." @@ -472,7 +491,74 @@ def _materialize_minute( ) if stats.empty: return _empty_series(factor.name) - return combine_minute_stats(factor, stats) + daily = None + if minute_combine_daily_spec(factor) is not None: + daily = combine_daily_panel( + factor, view=view, symbols=symbols, emit_start=emit_start, + emit_end=emit_end, sources=sources, + ) + return combine_minute_stats(factor, stats, daily=daily) + + +def combine_daily_panel( + factor: Factor, + *, + view: object, + symbols: list[str], + emit_start: pd.Timestamp, + emit_end: pd.Timestamp, + sources: MaterializeSources, +) -> pd.DataFrame: + """The daily panel a factor's DECLARED ``combine_daily`` combine consumes. + + Loads the declared columns from the injected daily provider with the same + generous calendar buffer the factor-input path uses, applies the (source, + view) availability lag (decision view: the prev-day shift — so the panel's + row d carries ``close_{d-1}``, the shifted panel of the third path), then + trims to EXACTLY the declared ``warmup_days`` trailing trading days before + ``emit_start`` — the same trailing-trim discipline as the factor-input path + (design §3.5 P8), so a single-date fill and a batch fill hand the combine + bit-identical daily inputs. The warmup depth is the binding's DECLARATION + (never inferred): for valley_price_quantile it is ``reversal_days + 2`` — + the reversal consumes ``reversal_days + 1`` lagged rows and the decision + lag's own leading NaN row must land one day before that window (§六.18: + an under-deep panel would fabricate edge NaNs the legacy path never had). + + A missing daily provider or a factor with no declared daily combine input + is a readable error. + """ + spec = minute_combine_daily_spec(factor) + if spec is None: + raise ValueError( + f"{factor.name}: its combine declares no daily-panel input, so there " + f"is no combine daily panel to build." + ) + if sources.daily is None: + raise ValueError( + f"{factor.name}: its combine declares a daily-panel input " + f"(columns {list(spec.columns)}) but no DailyPanelProvider was injected." + ) + resolved_view = View(view) + symbols = requested_universe(symbols) # idempotent: callers normalize too + emit_start = pd.Timestamp(emit_start).normalize() + emit_end = pd.Timestamp(emit_end).normalize() + w = int(spec.warmup_days) + if w < 1: + raise ValueError(f"{factor.name}: combine daily warmup must be >= 1; got {w}.") + load_start = emit_start - pd.Timedelta(days=_load_buffer_calendar_days(w)) + panel = sources.daily.daily_panel(list(symbols), load_start, emit_end) + if panel.empty: + return panel + missing = [c for c in spec.columns if c not in panel.columns] + if missing: + raise ValueError( + f"{factor.name}: its combine declares daily columns {list(spec.columns)} " + f"but the injected daily panel lacks {missing}; got {list(panel.columns)}." + ) + panel = panel[list(spec.columns)] + if resolved_view is View.DECISION: + panel = daily_decision_lag(panel) # prev-day shift; 'close' is not same-day + return _trim_daily(panel, emit_start, w) def _minute_intermediate( @@ -659,8 +745,20 @@ def _pooled_symbol_stats( if view is View.DECISION: work = minute_decision_cutoff(work, decision_time=decision_cutoff) stats = minute_stats_from_bars(factor, work) + # The saturation criterion reads the value's OUTPUT DATES. A factor whose + # combine declares a DAILY input cannot run its combine here (the daily + # panel is deliberately NOT loaded inside the per-symbol saturation loop); + # its declared combine contract returns EXACTLY the intermediate's rows, + # so the intermediate's dates ARE the value's output dates. Every other + # factor reads the combined single-symbol value exactly as before (zero + # behaviour change for the ten bars-only factors and intraday_amp_cut). + saturation_output = ( + stats + if minute_combine_daily_spec(factor) is not None + else combine_minute_stats(factor, stats) + ) if at_floor or _pooled_pool_saturated( - combine_minute_stats(factor, stats), work, emit_start, + saturation_output, work, emit_start, baseline_days=baseline_days, lookback_days=lookback_days, symbols=[symbol], ): @@ -698,7 +796,7 @@ def _provider_earliest(provider, symbols, factor) -> pd.Timestamp: def _pooled_pool_saturated( - full: pd.Series, + full: pd.Series | pd.DataFrame, bars: pd.DataFrame, emit_start: pd.Timestamp, *, @@ -711,7 +809,11 @@ def _pooled_pool_saturated( True iff, for EVERY REQUESTED symbol, the LOCKED sub-window (the loaded trading days after dropping the first ``baseline_days``) holds at least ``lookback_days`` valid days at or before ``emit_start`` — valid days being - the factor's own output dates. + the factor's own output dates. Only ``full``'s INDEX is read (dates and + symbols), so the caller may pass either the combined value Series or — for a + factor whose combine needs the daily panel, which this loop does not load — + the per-symbol intermediate frame, whose rows its declared combine preserves + exactly. ITERATING THE REQUESTED SYMBOLS, NOT THE OUTPUT SYMBOLS, IS LOAD-BEARING (review HIGH): a symbol that produces ZERO rows in the current load window (a @@ -862,6 +964,7 @@ def _recompute(emit_start, end, warmup) -> pd.Series: "MaterializeSources", "MinuteBarProvider", "build_horizon_config", + "combine_daily_panel", "is_minute_factor", "make_recompute_fn", "materialize_intermediate_range", diff --git a/factors/service.py b/factors/service.py index 3e61e15..bfca08d 100644 --- a/factors/service.py +++ b/factors/service.py @@ -62,9 +62,10 @@ from data.clean.schema import DATE_LEVEL, SYMBOL_LEVEL from factors import registry as factor_registry from factors.base import Factor -from factors.compute.minute.binding import combine_minute_stats +from factors.compute.minute.binding import combine_minute_stats, minute_combine_daily_spec from factors.materialize import ( MaterializeSources, + combine_daily_panel, materialize_intermediate_range, materialize_range, payload_columns, @@ -305,6 +306,9 @@ def _assemble( payload_by_id: dict[str, pd.DataFrame], dates: pd.DatetimeIndex, symbols: list[str], + *, + sources: MaterializeSources, + view: View, ) -> pd.DataFrame: """Slice each stored payload to (dates x universe) and stack the values. @@ -314,6 +318,11 @@ def _assemble( universe standardizes them" is a property of the READ, not of the artifact — so the same store serves a 12-name and a 24-name request correctly, and neither can pollute the other (D4c / design revision A2). + + A factor whose combine DECLARES a daily input (valley_price_quantile's + reversal neutralization) is handed the view-lagged, trailing-trimmed daily + panel built by ``materialize.combine_daily_panel`` from the injected sources + — the declaration is consulted, never an isinstance dispatch (red line #5). """ symbol_set = set(map(str, symbols)) columns: dict[str, pd.Series] = {} @@ -326,7 +335,16 @@ def _assemble( sym = payload.index.get_level_values(SYMBOL_LEVEL) sliced = payload[d.isin(dates) & pd.Index(sym).isin(symbol_set)] if stores_intermediate(factor): - columns[fid] = combine_minute_stats(factor, sliced) + if sliced.empty: + columns[fid] = pd.Series([], index=_empty_index(), dtype=float, name=fid) + elif minute_combine_daily_spec(factor) is not None: + daily = combine_daily_panel( + factor, view=view, symbols=symbols, emit_start=dates.min(), + emit_end=dates.max(), sources=sources, + ) + columns[fid] = combine_minute_stats(factor, sliced, daily=daily) + else: + columns[fid] = combine_minute_stats(factor, sliced) elif sliced.empty: columns[fid] = pd.Series([], index=_empty_index(), dtype=float, name=fid) else: @@ -389,7 +407,7 @@ def panel( sources=sources, view=resolved_view, cutoff=cutoff, diagnostics=diagnostics, ) - return _assemble(factors, payload_by_id, dates, symbols) + return _assemble(factors, payload_by_id, dates, symbols, sources=sources, view=resolved_view) def cross_section( @@ -414,4 +432,51 @@ def cross_section( ) -__all__ = ["DecisionPoint", "cross_section", "panel"] +def stored_payload( + factor_id: str, + universe: Iterable[str], + decisions: list[DecisionPoint], + *, + store: FactorValueStore, + sources: MaterializeSources, + view: object = View.DECISION, + basis: object = ReturnBasis.EXEC_TO_EXEC, + params_by_id: Mapping[str, Mapping[str, object]] | None = None, +) -> pd.DataFrame: + """The factor's STORED payload over ``decisions`` x ``universe`` (read-through). + + For most factors the payload IS the value; for a cross-sectional factor it is + the per-symbol INTERMEDIATE whose combine runs at read-assembly (D4c). A + disclosure that is reduced from the raw intermediate alongside the served + value (valley_price_quantile's NeutralizationCoverage — catalogue §三 + mechanism B) reads the intermediate HERE: the same key + fingerprint + + coverage engine as :func:`panel`, sliced to exactly this request, so the + disclosure can never be computed from a payload the value read did not also + see, and never from a superset a wider earlier request happened to leave in + the store. + """ + resolved_view, _ = require_legal_pairing(view, basis) + symbols = requested_universe(universe) + if not decisions: + raise ValueError("stored_payload() needs at least one DecisionPoint.") + cutoff = _uniform_cutoff(decisions) + dates = pd.DatetimeIndex( + sorted({pd.Timestamp(d.date).normalize() for d in decisions}) + ) + factor = _build_factor(factor_id, params_by_id) + fp = _fingerprint(factor) + key = store_key( + factor, view=resolved_view.value, params=(params_by_id or {}).get(factor_id) + ) + payload = _ensure_coverage( + factor, key, fp, dates=dates, symbols=symbols, store=store, + sources=sources, view=resolved_view, cutoff=cutoff, diagnostics=None, + ) + if payload.empty: + return payload + d = payload.index.get_level_values(DATE_LEVEL) + sym = payload.index.get_level_values(SYMBOL_LEVEL) + return payload[d.isin(dates) & pd.Index(sym).isin(set(map(str, symbols)))] + + +__all__ = ["DecisionPoint", "cross_section", "panel", "stored_payload"] diff --git a/factors/store/code_hash.py b/factors/store/code_hash.py index be5dad3..6354652 100644 --- a/factors/store/code_hash.py +++ b/factors/store/code_hash.py @@ -5,7 +5,10 @@ 1. the factor's own module file; 2. the ENUMERATED shared set every factor leans on: - ``{factors.compute.minute.primitives, factors.ops.*, factors.base, factors.spec}``; + ``{factors.compute.minute.primitives, factors.compute.minute.binding, + factors.ops.*, factors.base, factors.spec}`` (the binding joined in D5 C4b — + it carries load-bearing value semantics the column-shape validation cannot + see); 3. the factor module's DIRECT project-internal imports that are NOT already in the shared set — folded ONE HOP, module-granular, derived from the AST (never a manual list, red line #6). @@ -55,8 +58,19 @@ # --------------------------------------------------------------------------- # #: Single-module members (design §3.4). ``factors.ops`` is a PACKAGE, expanded to #: every ``*.py`` under it (so a new operator file joins the set automatically). +#: ``factors.compute.minute.binding`` joined in D5 C4b: it carries LOAD-BEARING +#: value semantics (which compute function a factor is bound to, the per-symbol / +#: combine split, the declared daily combine input), while the D4c shape +#: validation covers only the intermediate's COLUMN NAMES — so a value-carrying +#: edit to the binding moved no store key and neither guard could see it. +#: Folding it here invalidates every factor's key on a binding edit +#: (over-invalidate-safe; the direction the store prefers). ONE-TIME EFFECT, +#: disclosed: the first fold changes all 11 minute factors' keys (and the daily +#: factors' — the shared set is global), so a factor store filled before the +#: fold is wholesale invalid and must be recomputed once. _SHARED_SET_SINGLE_MODULES: tuple[str, ...] = ( "factors.compute.minute.primitives", + "factors.compute.minute.binding", "factors.base", "factors.spec", ) diff --git a/qt/cli.py b/qt/cli.py index 95b3cbf..6a1e44d 100644 --- a/qt/cli.py +++ b/qt/cli.py @@ -634,6 +634,7 @@ def _cmd_run_factor_eval_reconcile(args: argparse.Namespace) -> int: f"warmup={len(result.by_class('warmup_left_extension'))} " f"float_tail={len(result.by_class('float_reordering_tail'))} " f"threshold_flip={len(result.by_class('threshold_flip_tail'))} " + f"flip_contamination={len(result.by_class('threshold_flip_contamination'))} " f"nan_footprint={result.nan_footprint_rows} " f"unclassified={len(unclassified)} " f"max_rel_diff={result.max_rel_diff:.3e}" diff --git a/qt/factor_eval_disclosures.py b/qt/factor_eval_disclosures.py index 346d0a0..561685c 100644 --- a/qt/factor_eval_disclosures.py +++ b/qt/factor_eval_disclosures.py @@ -66,6 +66,7 @@ from data.clean.schema import DATE_LEVEL from factors.compute.minute.peak_ridge_amount_ratio import PeakRidgeAmountRatioFactor from factors.compute.minute.ridge_minute_return import RidgeMinuteReturnFactor +from factors.compute.minute.valley_price_quantile import ValleyPriceQuantileFactor from factors.compute.minute.valley_ridge_vwap_ratio import ValleyRidgeVwapRatioFactor # Percentiles reported for the realized ridge/peak-bar distributions. @@ -632,13 +633,33 @@ def disclosure_binding_for(factor) -> DisclosureBinding | None: return _DISCLOSURE_BY_CLASS.get(type(factor)) +#: The add-Section name of the ONE mechanism-B disclosure (catalogue §三). +NEUTRALIZATION_SECTION_NAME = "neutralization_coverage" + +#: factor class -> publishes the NeutralizationCoverage disclosure. Mechanism B +#: (catalogue §三): NO diagnostics sink — the disclosure is reduced from the +#: raw + reversal + residual panels AFTER the loop, so it cannot ride +#: :func:`disclosure_binding_for`'s sink table. Class-keyed like that table, +#: so the two mechanisms agree by construction. +_NEUTRALIZATION_DISCLOSURE_CLASSES: frozenset[type] = frozenset( + {ValleyPriceQuantileFactor} +) + + +def publishes_neutralization_disclosure(factor) -> bool: + """True iff ``factor`` publishes the mechanism-B NeutralizationCoverage.""" + return type(factor) in _NEUTRALIZATION_DISCLOSURE_CLASSES + + __all__ = [ "DisclosureBinding", + "NEUTRALIZATION_SECTION_NAME", "NeutralizationCoverage", "PeakCoverage", "RidgeCoverage", "RidgeReturnCoverage", "disclosure_binding_for", + "publishes_neutralization_disclosure", "summarize_neutralization", "summarize_peak_coverage", "summarize_ridge_coverage", diff --git a/qt/factor_eval_reconcile.py b/qt/factor_eval_reconcile.py index d802849..4ca0352 100644 --- a/qt/factor_eval_reconcile.py +++ b/qt/factor_eval_reconcile.py @@ -26,17 +26,48 @@ and a finite value on a row the frozen panel does not have); * valid-day-POOLED factor: the early region [2021-07-01, 2021-10-31], same three directions, and the per-month counts must be - non-increasing ("按月递减至零"; a violation fails the mode). + non-increasing ("按月递减至零"; a violation fails the mode) with ONE + STRUCTURALLY ANCHORED EXEMPTION — lead ruling 1 (catalogue §七之五): + the exempt month is the month of the frozen panel's FIRST FINITE + VALUE, the structural partial month in which residual/value + existence starts (vpq: the frozen panel's first values exist only + from ~07-29), so it is exempt from the monotonicity check; its cells + must still sit inside the early region with a registered direction. + The anchor is read from the FROZEN GRID, never from where warmup + diffs happen to land: a zero-diff exempt month followed by a rising + shape still FAILS. If the frozen panel has no finite value at all + there is no exemption (the conservative direction). Months after the + exempt month stay strictly gated. 2. ``float_reordering_tail`` — scattered finite-vs-finite cells with rel diff <= 5e-12 (rolling-correlation summation order; the JC1 1e-12 gate - is the attributable floor, this is the measured tail above it). The - class is CAPPED at 101 cells (the measured jump count); more fails. + is the attributable floor, this is the measured tail above it), OR with + abs diff <= 1e-12 regardless of rel (lead ruling 3: on near-zero + cross-sectional OLS residuals the rel criterion is meaningless — the + measured dust is ~1e-16 machine precision on ~1e-5 residuals). The + class is CAPPED — 101 cells for bars-only factors (the measured jump + count), 1000 for cross-sectional ones (measured vpq 707 + headroom); + more fails. 3. ``threshold_flip_tail`` — count factors (volume_peak): rolling-sigma float noise (~4e-10) times an integer volume sitting on the peak threshold flips the count by EXACTLY +/-1 on a sparse (symbol, day) cluster (measured: 20 cells, 600623.SH 2023-06-15..07-14). Bounds: |delta| == 1 exactly, rel <= 1e-2, at most 25 cells; more fails. - 4. the jump cutoff reference — handled by the reference-path selection above. + 4. ``threshold_flip_contamination`` — CROSS-SECTIONAL factors only (lead + ruling 2, catalogue §七之五): the same PRV critical-bar flip family as + (3), observed on vpq as a CONTINUOUS value change (valley VWAP -> + q_day -> qbar) plus second-order contamination of the whole cross + section through the per-date OLS coefficients. Bounds (checked cell by + cell; any violation is UNCLASSIFIED and fails): window + [2023-06-01, 2023-07-14]; the directly affected symbol (600623.SH) + |diff| <= 2e-04; every other symbol |diff| <= 1e-06; at most 20,000 + cells in total. Not an input-side change: (a) both cache ledgers had + ZERO writes since the panel freeze (daily max fetched_at 2026-07-18, + minute 2026-07-16, freeze 2026-07-24); (b) qbar is bitwise-identical + across the old/new loading geometries (max 1.1e-16); (c) the flip sits + on the same symbol and window already registered for volume_peak as + (3) — the frozen panel was produced by the pre-D2 code path, the + served one by the migrated primitives. + 5. the jump cutoff reference — handled by the reference-path selection above. Anything else (finite->NaN, finite-vs-finite beyond every named tail, a finite value on a row the frozen panel does not have outside the warmup @@ -75,6 +106,12 @@ class is CAPPED at 101 cells (the measured jump count); more fails. new JSON carries the ``corrections`` block (contract v1.1); without it they fail. Jump's value-level verification lives in the panels leg (cutoff reference) and in the post-fix restated numbers, not here. + * ``valley_price_quantile_20`` (§七之四): the unified runner publishes the + NeutralizationCoverage disclosure as an add-Section; the legacy runner + only LOGGED it (catalogue §三 mechanism B), so the frozen artifact has no + such section. The whole ``neutralization_coverage`` subtree (JSON leaves + + the section's MD lines) is a registered addition, matched by section NAME + (``REGISTERED_EXTRA_SECTIONS``), never by index. Markdown is diffed as a line set, then same-key lines are PAIRED into row-level changes before classification (a bare set diff reports every @@ -87,13 +124,18 @@ class is CAPPED at 101 cells (the measured jump count); more fails. * ``--mode anchors`` — the service path produces engine values for the ``hand_anchors_d2.json`` rows (same real-cache bars) and reconciles against - the HAND side. The hand side was computed with the OLD loading geometry - (left edge anchored at 2021-07-01), so a mismatch inside the warmup - boundary is the registered ``warmup_left_extension`` class — for bounded - factors the frozen grid's first ``lookback_depth - 1`` trading dates (the - harness never had this class for bounded factors before §七之三; the - first-run failures were exactly this asymmetry). EXPECTED SIGNAL (do not - misread as a regression): jump's NON-warmup rows must reconcile — the + the HAND side. A cross-sectional factor (``stores_intermediate`` — its + served value is a function of the requested universe, D4c) is served over + the FULL config universe, because the hand side was computed over it and a + 5-name request both empties the panel (below the combine's cross-section + floor) and asks a different question; per-symbol factors keep the cheap + anchor-symbols-only request. The hand side was computed with the OLD + loading geometry (left edge anchored at 2021-07-01), so a mismatch inside + the warmup boundary is the registered ``warmup_left_extension`` class — + for bounded factors the frozen grid's first ``lookback_depth - 1`` trading + dates (the harness never had this class for bounded factors before §七之三; + the first-run failures were exactly this asymmetry). EXPECTED SIGNAL (do + not misread as a regression): jump's NON-warmup rows must reconcile — the service carries the corrected, truncated definition, so the frozen-engine mismatches on random dates go GREEN here (rel ~1e-15). A jump mismatch outside the warmup boundary FAILS this mode. @@ -145,6 +187,16 @@ class is CAPPED at 101 cells (the measured jump count); more fails. #: failing; a flood of float-noise cells fails the cap. FLOAT_TAIL_REL_TOL = 5e-12 FLOAT_TAIL_MAX_CELLS = 101 +#: Lead ruling 3 (catalogue §七之五): an ABS lower bound — on near-zero +#: cross-sectional OLS residuals the rel criterion is meaningless (measured +#: dust: abs ~1e-16 machine precision on ~1e-5 residuals, rel >> 5e-12), so +#: |diff| <= 1e-12 is float dust regardless of rel. +FLOAT_TAIL_ABS_TOL = 1e-12 +#: Lead ruling 3: the cell cap is tiered by factor kind. The 101 cap was +#: calibrated on jump (bars-only); cross-sectional OLS residuals produce +#: systematically more dust cells (measured vpq: 707), so the cross-sectional +#: cap is 1000 (measured + headroom). Bars-only stays 101. +FLOAT_TAIL_MAX_CELLS_CROSS_SECTIONAL = 1000 #: ``threshold_flip_tail`` (catalogue §七之三): count factors (volume_peak) — #: rolling-sigma float noise (~4e-10) times an integer volume sitting on the @@ -153,6 +205,32 @@ class is CAPPED at 101 cells (the measured jump count); more fails. THRESHOLD_FLIP_REL_TOL = 1e-2 THRESHOLD_FLIP_MAX_CELLS = 25 +#: ``threshold_flip_contamination`` (catalogue §七之五, lead ruling 2) — +#: CROSS-SECTIONAL factors ONLY. Mechanism: the same PRV critical-bar +#: classification flip as ``threshold_flip_tail`` (rolling-sigma float noise +#: x an integer volume on the threshold), but on a continuous-value factor +#: (vpq: valley VWAP -> q_day -> qbar) the flip shows as a small continuous +#: change on the directly affected symbol, and the per-date cross-sectional +#: OLS coefficients propagate it as a second-order contamination of EVERY +#: symbol on those dates. Why this is NOT an input-side change: (a) both +#: cache ledgers had ZERO writes since the panel freeze (daily max +#: fetched_at 2026-07-18, minute 2026-07-16; freeze 2026-07-24); (b) qbar is +#: bitwise-identical across the old and new loading geometries (max +#: 1.1e-16); (c) the same symbol+window flip is already registered as +#: ``threshold_flip_tail`` for volume_peak — the frozen panel came from the +#: pre-D2 code path, the served one from the migrated primitives. +#: THE FULL PARAMETER SET of the class (lead ruling: cell-by-cell; any +#: violation lands in ``unclassified_finite_vs_finite`` and fails): +THRESHOLD_FLIP_CONTAMINATION_LO = pd.Timestamp("2023-06-01") +THRESHOLD_FLIP_CONTAMINATION_HI = pd.Timestamp("2023-07-14") +THRESHOLD_FLIP_CONTAMINATION_SYMBOL = "600623.SH" +#: measured direct-symbol max |diff| 1.60e-04 -> bound 2e-04 +THRESHOLD_FLIP_CONTAMINATION_DIRECT_ABS_TOL = 2e-04 +#: measured contaminated-symbol max |diff| 5.5e-07 -> bound 1e-06 +THRESHOLD_FLIP_CONTAMINATION_CROSS_ABS_TOL = 1e-06 +#: measured 19,143 cells -> cap 20,000 +THRESHOLD_FLIP_CONTAMINATION_MAX_CELLS = 20_000 + EARLY_REGION_LO = pd.Timestamp("2021-07-01") EARLY_REGION_HI = pd.Timestamp("2021-10-31") @@ -208,6 +286,73 @@ class is CAPPED at 101 cells (the measured jump count); more fails. "- ⚠️ CORRECTION (", ) +#: factor_id -> the add-Sections the unified runner publishes that the frozen +#: artifact does NOT carry (§七之四, D5 C4b). valley_price_quantile's +#: NeutralizationCoverage was only LOGGED by the legacy runner (catalogue §三 +#: mechanism B — never an add-Section), so the unified runner's new section is +#: an ADDITION against the frozen artifact, registered here per factor; any +#: other added section stays unregistered and fails. +REGISTERED_EXTRA_SECTIONS: dict[str, tuple[str, ...]] = { + "valley_price_quantile_20": ("neutralization_coverage",), +} + +#: The MD note-line prefix of ``NeutralizationCoverage.render()`` — the +#: disclosure's one-line summary, rendered as the section's note. Kept as a +#: constant with a pinning test (the alternative — re-rendering a zero +#: instance — would couple the reconcile gate to the renderer's format by +#: construction instead of by assertion). +_NEUTRALIZATION_NOTE_PREFIX = "neutralization (T-1 rev20):" + + +def _registered_section_md_prefixes(section_names: tuple[str, ...]) -> tuple[str, ...]: + """The MD line prefixes belonging to a registered add-Section's rendering. + + An extra section renders (``analytics/eval/render.py._render_section``) as + an unnumbered ``## + `` heading, the note line, and one + ``- :`` payload line per dataclass field (derived from the + dataclass, so a field rename breaks the pinning test instead of silently + unregistering). An unknown section name is a readable error — registering + a section whose MD rendering is not catalogued would be a guess. + """ + from dataclasses import fields as dc_fields + + from qt.factor_eval_disclosures import ( + NEUTRALIZATION_SECTION_NAME, + NeutralizationCoverage, + ) + + prefixes: list[str] = [] + for name in section_names: + if name == NEUTRALIZATION_SECTION_NAME: + prefixes.append(f"## + {NEUTRALIZATION_SECTION_NAME}") + prefixes.append(_NEUTRALIZATION_NOTE_PREFIX) + prefixes.extend(f"- {f.name}:" for f in dc_fields(NeutralizationCoverage)) + else: + raise ValueError( + f"no MD rendering is registered for added section {name!r}." + ) + return tuple(prefixes) + + +def _is_registered_section_addition( + path: str, new: dict, section_names: tuple[str, ...] +) -> bool: + """True iff ``path`` is a leaf of one of ``section_names`` in ``new``. + + The check reads the NEW JSON's section at the path's index and matches its + NAME — never the index alone — so a different extra section landing at the + same index stays unregistered. + """ + match = re.match(r"sections\[(\d+)\]", path) + if match is None or not section_names: + return False + sections = new.get("sections") or [] + idx = int(match.group(1)) + if idx >= len(sections) or not isinstance(sections[idx], dict): + return False + return sections[idx].get("name") in section_names + + MAX_EXAMPLES = 10 @@ -354,7 +499,13 @@ def _classify_value_change( def diff_report_json( - old: dict, new: dict, *, name: str, strict: bool, correction_expected: bool + old: dict, + new: dict, + *, + name: str, + strict: bool, + correction_expected: bool, + registered_sections: tuple[str, ...] = (), ) -> ReportDiff: """Diff one (frozen, new) JSON pair leaf by leaf against the registered list. @@ -364,6 +515,8 @@ def diff_report_json( registered; hiding it would be, gating it would false-positive. ``correction_expected`` (jump): value differences are the declared correction — accepted ONLY if the new JSON carries a ``corrections`` block. + ``registered_sections``: add-Section names whose whole subtree is a + registered addition (§七之四 — the frozen artifact predates the section). """ result = ReportDiff(name=name, strict=strict) old_flat, new_flat = _flatten(old), _flatten(new) @@ -387,11 +540,12 @@ def diff_report_json( ) result.diffs.append(LeafDiff(path, old_v, new_v, cls)) elif in_new: - cls = ( - "registered_addition" - if _is_registered_addition(path) - else "unregistered_addition" - ) + if _is_registered_addition(path): + cls = "registered_addition" + elif _is_registered_section_addition(path, new, registered_sections): + cls = "registered_section_addition" + else: + cls = "unregistered_addition" result.diffs.append(LeafDiff(path, None, new_flat[path], cls)) else: result.diffs.append(LeafDiff(path, old_flat[path], None, "unregistered_removal")) @@ -482,7 +636,13 @@ def _classify_md_change( def diff_report_md( - old_text: str, new_text: str, *, name: str, strict: bool = True, correction_expected: bool + old_text: str, + new_text: str, + *, + name: str, + strict: bool = True, + correction_expected: bool, + registered_section_lines: tuple[str, ...] = (), ) -> ReportDiff: """Markdown: line-set diff, then pair same-key lines into row-level CHANGES. @@ -492,6 +652,8 @@ def diff_report_md( the numeric-attribution ladder (the same one as the JSON leg), keeps the teeth where they belong: UNPAIRED additions must be registered additions, UNPAIRED removals are never registered (except jump's correction prose). + ``registered_section_lines``: the line prefixes of a registered + add-Section's rendering (§七之四). """ result = ReportDiff(name=name, strict=strict) old_counts = Counter(old_text.splitlines()) @@ -510,11 +672,19 @@ def diff_report_md( cls = "registered_correction_effect" if correction_expected else "unregistered_removal" result.diffs.append(LeafDiff("", line, None, cls)) for line in unpaired_additions: - cls = ( - "registered_addition" - if line.startswith(ALLOWED_ADDED_MD_PREFIXES) - else ("registered_correction_effect" if correction_expected else "unregistered_addition") - ) + if line.startswith(ALLOWED_ADDED_MD_PREFIXES): + cls = "registered_addition" + elif registered_section_lines and ( + line == "" or line.startswith(registered_section_lines) + ): + # the section's rendering includes its blank separators; blank-line + # additions register ONLY alongside a registered section (they are + # unregistered everywhere else). + cls = "registered_section_addition" + elif correction_expected: + cls = "registered_correction_effect" + else: + cls = "unregistered_addition" result.diffs.append(LeafDiff("", None, line, cls)) result.ok = not any(d.classification.startswith("unregistered") for d in result.diffs) return result @@ -543,7 +713,8 @@ class PanelCellDiff: frozen: object new: object classification: str # warmup_left_extension | float_reordering_tail | - # threshold_flip_tail | unclassified_* | unregistered_* + # threshold_flip_tail | threshold_flip_contamination | + # unclassified_* | unregistered_* @dataclass @@ -559,6 +730,7 @@ class PanelDiff: warmup_by_month: dict[str, int] = field(default_factory=dict) warmup_by_direction: dict[str, int] = field(default_factory=dict) warmup_monotonic: bool = True + warmup_exempt_month: str | None = None ok: bool = True def by_class(self, classification: str) -> list[PanelCellDiff]: @@ -572,11 +744,12 @@ def classify_panel_differences( factor_id: str, is_pooled: bool, lookback_depth: int, + is_cross_sectional: bool = False, tol: float = PANEL_REL_TOL, early_lo: pd.Timestamp = EARLY_REGION_LO, early_hi: pd.Timestamp = EARLY_REGION_HI, float_tail_tol: float = FLOAT_TAIL_REL_TOL, - float_tail_max: int = FLOAT_TAIL_MAX_CELLS, + float_tail_max: int | None = None, flip_rel_tol: float = THRESHOLD_FLIP_REL_TOL, flip_max: int = THRESHOLD_FLIP_MAX_CELLS, ) -> PanelDiff: @@ -591,8 +764,25 @@ def classify_panel_differences( dates are under-warmed), the materializer saturates to the cache's real start. Bounded: the grid's first ``lookback_depth - 1`` trading dates. Pooled: the early region [early_lo, early_hi] with non-increasing - per-month counts. + per-month counts, with ONE structurally anchored exemption (lead + ruling 1): the exempt month is the month of the frozen panel's FIRST + FINITE VALUE — the structural partial month in which residual/value + existence starts. The anchor comes from the frozen grid, not from + where diffs land; a frozen panel with no finite value gets no + exemption (conservative). Later months stay gated. + + ``is_cross_sectional`` (D4c ``stores_intermediate`` factors — the served + value is a per-date cross-sectional combine): enables the + ``threshold_flip_contamination`` class and the wider float-tail cap + (lead rulings 2 and 3); a bars-only factor never gets either. + ``float_tail_max=None`` resolves the cap from the factor kind. """ + if float_tail_max is None: + float_tail_max = ( + FLOAT_TAIL_MAX_CELLS_CROSS_SECTIONAL + if is_cross_sectional + else FLOAT_TAIL_MAX_CELLS + ) result = PanelDiff(factor_id=factor_id) frozen = frozen.copy() frozen["date"] = pd.to_datetime(frozen["date"]) @@ -660,19 +850,49 @@ def _warmup_cell(date, symbol, frozen_v, new_v, direction: str) -> None: result.equal += 1 continue if pd.notna(frozen_v) and pd.notna(new_v): + abs_diff = abs(frozen_v - new_v) denom = max(abs(frozen_v), abs(new_v)) - rel = abs(frozen_v - new_v) / denom if denom > 0 else 0.0 + rel = abs_diff / denom if denom > 0 else 0.0 result.max_rel_diff = max(result.max_rel_diff, rel) if rel <= tol: result.within_tolerance += 1 elif _in_warmup(date, symbol): _warmup_cell(date, symbol, frozen_v, new_v, "finite_to_finite") - elif rel <= float_tail_tol: + elif ( + is_cross_sectional + and THRESHOLD_FLIP_CONTAMINATION_LO + <= date + <= THRESHOLD_FLIP_CONTAMINATION_HI + ): + # Lead ruling 2: INSIDE the registered window a cross-sectional + # cell must meet the contamination bounds — the direct symbol + # (600623.SH) at <= 2e-04, every other symbol at <= 1e-06. + # Any overshoot is UNCLASSIFIED here, never a fall-through to + # the generic tails (the window's mechanism is adjudicated; + # a bigger move inside it is a new fact, not float noise). + bound = ( + THRESHOLD_FLIP_CONTAMINATION_DIRECT_ABS_TOL + if symbol == THRESHOLD_FLIP_CONTAMINATION_SYMBOL + else THRESHOLD_FLIP_CONTAMINATION_CROSS_ABS_TOL + ) + cls = ( + "threshold_flip_contamination" + if abs_diff <= bound + else "unclassified_finite_vs_finite" + ) + result.diffs.append( + PanelCellDiff(str(date.date()), str(symbol), float(frozen_v), + float(new_v), cls) + ) + elif abs_diff <= FLOAT_TAIL_ABS_TOL or rel <= float_tail_tol: + # Lead ruling 3: |diff| <= 1e-12 is float dust REGARDLESS of + # rel (the rel criterion is meaningless on near-zero + # cross-sectional OLS residuals). result.diffs.append( PanelCellDiff(str(date.date()), str(symbol), float(frozen_v), float(new_v), "float_reordering_tail") ) - elif abs(frozen_v - new_v) == 1.0 and rel <= flip_rel_tol: + elif abs_diff == 1.0 and rel <= flip_rel_tol: result.diffs.append( PanelCellDiff(str(date.date()), str(symbol), float(frozen_v), float(new_v), "threshold_flip_tail") @@ -698,14 +918,36 @@ def _warmup_cell(date, symbol, frozen_v, new_v, direction: str) -> None: "unclassified_nan_to_finite") ) - counts = [result.warmup_by_month[m] for m in sorted(result.warmup_by_month)] - result.warmup_monotonic = all(b <= a for a, b in zip(counts, counts[1:])) + # Lead ruling 1, STRUCTURALLY anchored: the exempt month is the month of + # the frozen panel's FIRST FINITE VALUE — the partial month in which + # residual/value existence starts (vpq: the frozen panel's first values + # exist only from ~07-29, so July is structurally partial). The anchor is + # read from the frozen GRID, never from where warmup diffs happen to land: + # an exempt month with ZERO diffs followed by a rising shape must still + # fail (the positional "first month WITH diffs" reading let exactly that + # through). A frozen panel with no finite value at all gets NO exemption + # (the conservative direction). Every other month's counts must be + # non-increasing ("按月递减至零"); a violation fails the mode. + if is_pooled: + finite_dates = frozen_s.index[frozen_s.notna().to_numpy()].get_level_values( + "date" + ) + if len(finite_dates): + result.warmup_exempt_month = str(finite_dates.min().to_period("M")) + gated = [ + result.warmup_by_month[m] + for m in sorted(result.warmup_by_month) + if m != result.warmup_exempt_month + ] + result.warmup_monotonic = all(b <= a for a, b in zip(gated, gated[1:])) result.ok = ( not any(d.classification.startswith("unclassified") or d.classification.startswith("unregistered") for d in result.diffs) and result.warmup_monotonic and len(result.by_class("float_reordering_tail")) <= float_tail_max and len(result.by_class("threshold_flip_tail")) <= flip_max + and len(result.by_class("threshold_flip_contamination")) + <= THRESHOLD_FLIP_CONTAMINATION_MAX_CELLS ) return result @@ -865,22 +1107,29 @@ def run_panels_mode(config_path: str, factor_id: str, repo_root: Path) -> PanelD ) factor = factor_registry.build(factor_id) frozen = pd.read_parquet(frozen_panel_path(factor_id, repo_root)) + from factors.materialize import stores_intermediate + + is_cross = stores_intermediate(factor) result = classify_panel_differences( served[factor_id], frozen, factor_id=factor_id, is_pooled=is_valid_day_pooled(factor), lookback_depth=int(factor.spec.lookback_depth), + is_cross_sectional=is_cross, ) logger.info( "panels %s: rows frozen=%d new=%d equal=%d tol=%d warmup=%d(%s) " - "float_tail=%d threshold_flip=%d footprint=%d unclassified=%d " - "max_rel=%.3e live_calls=%d ok=%s", + "exempt_month=%s " + "float_tail=%d threshold_flip=%d flip_contamination=%d footprint=%d " + "unclassified=%d max_rel=%.3e live_calls=%d ok=%s", factor_id, result.rows_frozen, result.rows_new, result.equal, result.within_tolerance, len(result.by_class("warmup_left_extension")), result.warmup_by_direction, + result.warmup_exempt_month, len(result.by_class("float_reordering_tail")), len(result.by_class("threshold_flip_tail")), + len(result.by_class("threshold_flip_contamination")), result.nan_footprint_rows, len([d for d in result.diffs if d.classification.startswith("un")]), result.max_rel_diff, live_calls, result.ok, @@ -898,6 +1147,8 @@ def run_reports_mode( report_dir = report_dir or Path(cfg.output.report_dir) report_name = _report_name(factor_id) correction_expected = factor_id == "jump_amount_corr_20" + registered_sections = REGISTERED_EXTRA_SECTIONS.get(factor_id, ()) + section_md_prefixes = _registered_section_md_prefixes(registered_sections) stem = f"factor_eval_{factor_id}" results: list[ReportDiff] = [] @@ -921,6 +1172,7 @@ def run_reports_mode( diff_report_json( frozen_json, new_json, name=f"{stem}_exec_{book}.json[{label}]", strict=strict, correction_expected=correction_expected, + registered_sections=registered_sections, ) ) new_md = (report_dir / f"{stem}_exec_{book}{'_bookclose' if 'bookclose' in label else ''}.md").read_text() @@ -929,6 +1181,7 @@ def run_reports_mode( diff_report_md( frozen_md, new_md, name=f"{stem}_exec_{book}.md[{label}]", strict=strict, correction_expected=correction_expected, + registered_section_lines=section_md_prefixes, ) ) problems = check_new_pair_consistency(new_no_book, new_with_book) @@ -975,14 +1228,26 @@ def run_anchors_mode(config_path: str, factor_id: str, repo_root: Path) -> Ancho Path(cfg.output.log_dir) / f"factor_eval_reconcile_anchors_{factor_id}.log", name="qt.factor_eval_reconcile", ) - symbols = sorted({row["symbol"] for row in rows}) + anchor_symbols = sorted({row["symbol"] for row in rows}) value_factors = () if factor_id in ("value_ep", "value_bp"): from qt.factor_eval_runner import _build_book_factors value_factors = tuple(_build_book_factors()) + # A CROSS-SECTIONAL factor's served value is a function of the REQUESTED + # universe (D4c: the combine runs at read-assembly over exactly what was + # asked for) — and the hand anchors were computed over the FULL evaluation + # universe. Requesting just the anchor symbols both empties the panel + # (5 names < the combine's cross-section floor -> all-NaN, measured on + # valley_price_quantile) and asks a different question than the hand side + # answered. So such a factor is served over the full config universe and + # the anchor cells are looked up inside it; per-symbol factors keep the + # cheap anchor-only request (their values are universe-independent). + from factors.materialize import stores_intermediate + + request_symbols = None if stores_intermediate(factor) else anchor_symbols store, sources, _panel, symbols, _cache = _build_bundle( - cfg, logger, symbols=symbols, value_factors=value_factors + cfg, logger, symbols=request_symbols, value_factors=value_factors ) decisions = [factor_service.DecisionPoint(pd.Timestamp(row["date"])) for row in rows] served = factor_service.panel( @@ -1023,14 +1288,23 @@ def run_anchors_mode(config_path: str, factor_id: str, repo_root: Path) -> Ancho "AnchorsDiff", "EARLY_REGION_HI", "EARLY_REGION_LO", + "FLOAT_TAIL_ABS_TOL", "FLOAT_TAIL_MAX_CELLS", + "FLOAT_TAIL_MAX_CELLS_CROSS_SECTIONAL", "FLOAT_TAIL_REL_TOL", "LeafDiff", "METRIC_REL_TOL", "PANEL_REL_TOL", "PanelDiff", + "REGISTERED_EXTRA_SECTIONS", "ReconciliationError", "ReportDiff", + "THRESHOLD_FLIP_CONTAMINATION_CROSS_ABS_TOL", + "THRESHOLD_FLIP_CONTAMINATION_DIRECT_ABS_TOL", + "THRESHOLD_FLIP_CONTAMINATION_HI", + "THRESHOLD_FLIP_CONTAMINATION_LO", + "THRESHOLD_FLIP_CONTAMINATION_MAX_CELLS", + "THRESHOLD_FLIP_CONTAMINATION_SYMBOL", "THRESHOLD_FLIP_MAX_CELLS", "THRESHOLD_FLIP_REL_TOL", "check_new_pair_consistency", diff --git a/qt/factor_eval_runner.py b/qt/factor_eval_runner.py index 00c7292..edb1002 100644 --- a/qt/factor_eval_runner.py +++ b/qt/factor_eval_runner.py @@ -14,7 +14,10 @@ 3. the add-Section coverage disclosures — ``qt.factor_eval_disclosures`` (mechanism A diagnostics sink through ``factors.service.panel``'s ``diagnostics=``; the sink re-materializes the request so a warm store can - still serve the disclosure — see ``factors.service._ensure_coverage``). + still serve the disclosure — see ``factors.service._ensure_coverage``; + mechanism B — valley_price_quantile's NeutralizationCoverage — is reduced + from the stored intermediate + the reversal + the served residual AFTER the + panel read, see ``_summarize_subject_neutralization``). 4. metric keys — ``qt.exec_basis_eval``'s extraction. The subject factor's VALUES come from the factor SERVICE @@ -54,9 +57,14 @@ RunRegistry appends are a DELIBERATE DEFERRAL to D7's governance surface: this runner writes its artifacts but does not register runs. -``valley_price_quantile`` is NOT served: its raw compute also needs the DAILY -close panel (reversal neutralization), so it has no minute binding yet — that -lands in PR-C4b. Asking for it is a readable error, never a silent mis-compute. +``valley_price_quantile`` IS served (PR-C4b): its per-symbol ``raw_qbar`` +intermediate rides the same store read-through as every other factor, and its +cross-sectional reversal neutralization runs at the service's read-assembly +over exactly the requested universe (the binding's DECLARED daily combine +input). Its NeutralizationCoverage disclosure is catalogue §三 mechanism B — +no diagnostics sink — so the runner reduces it from the STORED intermediate + +the reversal + the served residual AFTER the panel read +(:func:`_summarize_subject_neutralization`), never from a second engine. """ from __future__ import annotations @@ -73,12 +81,26 @@ from data.clean.schema import DATE_LEVEL, SYMBOL_LEVEL from factors import registry as factor_registry from factors import service as factor_service -from factors.compute.minute.binding import has_minute_diagnostics, is_minute_bound -from factors.compute.minute.valley_price_quantile import ValleyPriceQuantileFactor +from factors.compute.minute.binding import ( + RAW_QBAR_COL, + has_minute_diagnostics, + is_minute_bound, +) +from factors.compute.minute.valley_price_quantile import ( + VALLEY_QUANTILE_MIN_CROSS_SECTION, + VALLEY_QUANTILE_REVERSAL_DAYS, + reversal_20, +) from factors.spec import FactorSpec from qt.config import RootConfig, load_config from qt.exec_basis_eval import ExecBasisEvaluation, run_exec_basis_evaluation -from qt.factor_eval_disclosures import disclosure_binding_for, to_section +from qt.factor_eval_disclosures import ( + NEUTRALIZATION_SECTION_NAME, + disclosure_binding_for, + publishes_neutralization_disclosure, + summarize_neutralization, + to_section, +) from qt.factor_eval_providers import EvalServiceBundle, build_eval_service from qt.pipeline import _make_logger, _process_factors @@ -208,18 +230,6 @@ def _build_eval_config(cfg: RootConfig) -> EvalConfig: ) -def _check_subject_supported(factor) -> None: - """The one subject this runner cannot serve yet (PR-C4b), as a readable error.""" - if isinstance(factor, ValleyPriceQuantileFactor): - raise ValueError( - f"run-factor-eval cannot serve {factor.name!r} yet: its raw compute " - "also needs the DAILY close panel (its reversal neutralization), so " - "it has no minute binding in factors.compute.minute.binding — the " - "binding lands in PR-C4b. Asking for it here must fail loudly, never " - "silently mis-compute." - ) - - # --------------------------------------------------------------------------- # # The evaluation flow # --------------------------------------------------------------------------- # @@ -257,6 +267,48 @@ def _load_subject_raw( ) +def _summarize_subject_neutralization( + bundle: EvalServiceBundle, + factor, + subject_raw: pd.DataFrame, + decisions: list[factor_service.DecisionPoint], +): + """valley_price_quantile's NeutralizationCoverage (catalogue §三 mechanism B). + + No diagnostics sink exists for this disclosure: it is reduced AFTER the + panel read from the three panels the legacy runner also used — + + * raw — the STORED per-symbol ``raw_qbar`` intermediate, read back + through :func:`factors.service.stored_payload` (the same store the value + read just came from, sliced to exactly this request — never a second + engine, never a recomputation); + * rev — ``reversal_20`` on the bundle's close-view daily panel (NO new + data source). This is the legacy runner's exact computation, and the C4b + binding pins it bit-for-bit equal to what the service's read-assembly + combine actually applied (``reversal_20_shifted`` on the decision-lagged + panel); + * residual — the served subject values (pre-processing), i.e. the combine's + output over exactly this universe. + """ + payload = factor_service.stored_payload( + factor.name, + bundle.symbols, + decisions, + store=bundle.store, + sources=bundle.sources, + view=View.DECISION, + basis=ReturnBasis.EXEC_TO_EXEC, + ) + raw = payload[RAW_QBAR_COL] + rev = reversal_20(bundle.panel[["close"]], days=VALLEY_QUANTILE_REVERSAL_DAYS) + return summarize_neutralization( + raw, + rev, + subject_raw[factor.name], + min_cross_section=VALLEY_QUANTILE_MIN_CROSS_SECTION, + ) + + def _load_book_raw( bundle: EvalServiceBundle, decisions: list[factor_service.DecisionPoint], @@ -350,7 +402,6 @@ def run_factor_eval( _check_preconditions(cfg) _check_config_book(cfg) factor = factor_registry.build(factor_id) - _check_subject_supported(factor) spec = factor.spec eval_cfg = _build_eval_config(cfg) @@ -389,6 +440,21 @@ def run_factor_eval( coverage = binding.summarize(sink) logger.info("%s", coverage.render()) extra_sections.append(to_section(binding.section_name, coverage)) + elif publishes_neutralization_disclosure(factor): + # Catalogue §三 mechanism B: no sink — reduce the disclosure from the + # stored intermediate + the reversal + the served residual, AFTER the + # panel read. The cost is the store re-read of a payload already + # materialized by the subject read (a warm-store read, zero minute + # bars); the COLD fill behind that payload is vpq's real cost — it is + # pooled + cross-sectional + daily-bound, so a cold store means the + # saturation load (the declared 2015-01-05 floor), disclosed here and + # visible in the run log's elapsed time, same as every other factor's + # cold fill. + coverage = _summarize_subject_neutralization( + bundle, factor, subject_raw, decisions + ) + logger.info("%s", coverage.render()) + extra_sections.append(to_section(NEUTRALIZATION_SECTION_NAME, coverage)) subject_processed = _process_factors( cfg, subject_raw[[spec.factor_id]], bundle.panel diff --git a/tests/test_decision_cutoff_visibility.py b/tests/test_decision_cutoff_visibility.py index 8c9f761..345f9a8 100644 --- a/tests/test_decision_cutoff_visibility.py +++ b/tests/test_decision_cutoff_visibility.py @@ -15,11 +15,15 @@ the project already uses, aimed at the WITHIN-DAY boundary rather than the across-day one. -RESULT, ENCODED RATHER THAN DESCRIBED. All ten bars-bound minute factors are now -clean. ``jump_amount_corr_20`` was NOT when this file was written: its compute was -the ONLY one of the eleven that applied no decision-time truncation of its own -(grep: zero ``decision_time`` / ``prepare_visible_minute_bars`` references in its -module), so under the old runners it saw 09:30-15:00 of day d. +RESULT, ENCODED RATHER THAN DESCRIBED. All eleven stream-bound minute factors +are now clean — the ten bars-bound ones measured through their whole-factor +call, valley_price_quantile (stream-bound with a declared daily combine input, +D5 C4b) measured through its per-symbol qbar stage, which carries its entire +minute-bar dependence. ``jump_amount_corr_20`` was NOT when this file was +written: its compute was the ONLY one of the eleven that applied no +decision-time truncation of its own (grep: zero ``decision_time`` / +``prepare_visible_minute_bars`` references in its module), so under the old +runners it saw 09:30-15:00 of day d. It was recorded here as a KNOWN, NAMED exception rather than fixed, because truncating it changes a published factor's values and could move its verdict — a @@ -63,7 +67,10 @@ from factors.compute.minute.binding import ( _MINUTE_STREAM_BINDINGS, NOT_DECISION_CUTOFF_SAFE, + RAW_QBAR_COL, + is_minute_bound, minute_raw_from_bars, + minute_stats_from_bars, ) CUTOFF = "14:50:00" @@ -137,8 +144,17 @@ def _poison(bars: pd.DataFrame) -> pd.DataFrame: def _moved_cells(factor_id: str) -> tuple[int, int, float]: factor = factor_registry.build(factor_id) - before = minute_raw_from_bars(factor, BARS) - after = minute_raw_from_bars(factor, _poison(BARS)) + if is_minute_bound(factor): + before = minute_raw_from_bars(factor, BARS) + after = minute_raw_from_bars(factor, _poison(BARS)) + else: + # valley_price_quantile has no bars-only whole-factor form (its combine + # declares a daily-panel input, D5 C4b). Its ENTIRE minute-bar dependence + # lives in the per-symbol qbar stage — the combine reads only the daily + # panel — so that stage is exactly what gets measured here. Routed by the + # binding tables, not by isinstance. + before = minute_stats_from_bars(factor, BARS)[RAW_QBAR_COL] + after = minute_stats_from_bars(factor, _poison(BARS))[RAW_QBAR_COL] joined = pd.DataFrame({"a": before, "b": after}).dropna() if joined.empty: return 0, 0, 0.0 @@ -184,7 +200,7 @@ def test_the_deny_list_is_empty_and_that_emptiness_is_a_measurement(): """ assert KNOWN_POST_CUTOFF_DEPENDENT_IDS == frozenset() assert set(CLEAN_FACTOR_IDS) == set(BOUND_FACTOR_IDS) - assert len(BOUND_FACTOR_IDS) == 10, ( + assert len(BOUND_FACTOR_IDS) == 11, ( "the bound minute-factor set changed; the emptiness above only covers " "what this file measures, so re-check the new one before trusting it" ) diff --git a/tests/test_factor_eval_disclosures.py b/tests/test_factor_eval_disclosures.py index ec77efe..0143737 100644 --- a/tests/test_factor_eval_disclosures.py +++ b/tests/test_factor_eval_disclosures.py @@ -374,3 +374,34 @@ def test_an_extra_section_may_never_shadow_a_mandatory_name(): FactorEvalReport.assemble( base.spec, base.cfg, [*base.sections, bad], thresholds=base.thresholds ) + + +def test_neutralization_disclosure_is_class_keyed_mechanism_b(): + """Mechanism B (catalogue §三): vpq publishes the NeutralizationCoverage — + with NO sink binding — and the two mechanisms never both claim one factor.""" + from factors.compute.minute.peak_ridge_amount_ratio import ( + PeakRidgeAmountRatioFactor, + ) + from factors.compute.minute.ridge_minute_return import RidgeMinuteReturnFactor + from factors.compute.minute.valley_price_quantile import ( + ValleyPriceQuantileFactor, + ) + from factors.compute.minute.valley_ridge_vwap_ratio import ( + ValleyRidgeVwapRatioFactor, + ) + from qt.factor_eval_disclosures import ( + NEUTRALIZATION_SECTION_NAME, + disclosure_binding_for, + publishes_neutralization_disclosure, + ) + + vpq = ValleyPriceQuantileFactor() + assert publishes_neutralization_disclosure(vpq) is True + assert disclosure_binding_for(vpq) is None # mechanism B has no sink + assert NEUTRALIZATION_SECTION_NAME == "neutralization_coverage" + for cls in ( + ValleyRidgeVwapRatioFactor, + RidgeMinuteReturnFactor, + PeakRidgeAmountRatioFactor, + ): + assert publishes_neutralization_disclosure(cls()) is False diff --git a/tests/test_factor_eval_reconcile.py b/tests/test_factor_eval_reconcile.py index 66b69d7..c70c475 100644 --- a/tests/test_factor_eval_reconcile.py +++ b/tests/test_factor_eval_reconcile.py @@ -396,6 +396,134 @@ def test_md_prose_WORD_change_does_not_pair_and_fails(): assert result.by_class("unregistered_removal") +# --------------------------------------------------------------------------- # +# reports mode — registered add-Section additions (§七之四, D5 C4b vpq) +# --------------------------------------------------------------------------- # +def _with_neutralization_section(new: dict) -> dict: + """The unified runner's vpq artifact: one extra add-Section the frozen lacks.""" + new = dict(new) + new["sections"] = [ + *new["sections"], + { + "name": "neutralization_coverage", + "note": "neutralization (T-1 rev20): raw_rows=100 rev_paired=90 " + "residual_rows=80 dates=10/10 cross_section min/med/max=11/12.0/13 " + "mean_spearman(raw,rev20)=-0.1234", + "payload": { + "raw_rows": 100, + "rev_rows": 90, + "residual_rows": 80, + "dates_total": 10, + "dates_residualized": 10, + "cross_section_min": 11, + "cross_section_median": 12.0, + "cross_section_max": 13, + "raw_rev_spearman_mean": -0.1234, + }, + }, + ] + return new + + +def test_json_registered_section_addition_passes_for_vpq(): + new = _with_neutralization_section(_frozen_like()) + result = diff_report_json( + _frozen_like(), new, name="t", strict=True, correction_expected=False, + registered_sections=("neutralization_coverage",), + ) + assert result.ok, result.diffs + classes = {d.classification for d in result.diffs} + assert classes == {"registered_section_addition"} + assert all(d.path.startswith("sections[1]") for d in result.diffs) + + +def test_json_extra_section_without_registration_fails(): + new = _with_neutralization_section(_frozen_like()) + result = diff_report_json( + _frozen_like(), new, name="t", strict=True, correction_expected=False + ) + assert not result.ok + assert result.by_class("unregistered_addition") + + +def test_json_section_addition_is_matched_by_NAME_not_index(): + """A DIFFERENT section at the registered index stays unregistered.""" + new = _with_neutralization_section(_frozen_like()) + new["sections"][1]["name"] = "some_other_coverage" + result = diff_report_json( + _frozen_like(), new, name="t", strict=True, correction_expected=False, + registered_sections=("neutralization_coverage",), + ) + assert not result.ok + # every leaf of the foreign section is unregistered, incl. the payload + assert len(result.by_class("unregistered_addition")) >= 10 + + +_NEUTRALIZATION_MD = ( + "## + neutralization_coverage\n" + "\n" + "neutralization (T-1 rev20): raw_rows=100 rev_paired=90 residual_rows=80 " + "dates=10/10 cross_section min/med/max=11/12.0/13 " + "mean_spearman(raw,rev20)=-0.1234\n" + "\n" + "- cross_section_max: 13\n" + "- cross_section_median: 12.0\n" + "- cross_section_min: 11\n" + "- dates_residualized: 10\n" + "- dates_total: 10\n" + "- raw_rev_spearman_mean: -0.1234\n" + "- raw_rows: 100\n" + "- residual_rows: 80\n" + "- rev_rows: 90\n" +) + + +def test_md_registered_section_lines_pass_for_vpq(): + from qt.factor_eval_reconcile import _registered_section_md_prefixes + + prefixes = _registered_section_md_prefixes(("neutralization_coverage",)) + result = diff_report_md( + _MD_OLD, _MD_OLD + _NEUTRALIZATION_MD, name="t", correction_expected=False, + registered_section_lines=prefixes, + ) + assert result.ok, result.diffs + assert {d.classification for d in result.diffs} == {"registered_section_addition"} + + +def test_md_section_lines_without_registration_fail(): + result = diff_report_md( + _MD_OLD, _MD_OLD + _NEUTRALIZATION_MD, name="t", correction_expected=False + ) + assert not result.ok + assert result.by_class("unregistered_addition") + + +def test_md_section_prefixes_are_derived_from_the_dataclass_fields(): + """A NeutralizationCoverage field rename breaks the registration loudly.""" + from dataclasses import fields as dc_fields + + from qt.factor_eval_disclosures import NeutralizationCoverage + from qt.factor_eval_reconcile import _registered_section_md_prefixes + + prefixes = _registered_section_md_prefixes(("neutralization_coverage",)) + for f in dc_fields(NeutralizationCoverage): + assert f"- {f.name}:" in prefixes + # the note prefix really is the render() format's head (not a stale copy) + cov = NeutralizationCoverage( + raw_rows=1, rev_rows=1, residual_rows=1, dates_total=1, + dates_residualized=1, cross_section_min=1, cross_section_median=1.0, + cross_section_max=1, raw_rev_spearman_mean=0.0, + ) + assert cov.render().startswith("neutralization (T-1 rev20):") + + +def test_md_section_prefixes_reject_an_unknown_section(): + from qt.factor_eval_reconcile import _registered_section_md_prefixes + + with pytest.raises(ValueError, match="no MD rendering is registered"): + _registered_section_md_prefixes(("bogus_coverage",)) + + # --------------------------------------------------------------------------- # # panels mode — cell classification # --------------------------------------------------------------------------- # @@ -574,6 +702,66 @@ def test_panels_pooled_warmup_after_early_region_is_unclassified(): def test_panels_pooled_warmup_non_monotonic_monthly_counts_fail(): + # REVERSE (lead ruling 1): the structurally anchored exempt month (July, + # where the frozen panel's FIRST finite value lives) is exempt, but + # months AFTER it must still be non-increasing — {07:1, 08:1, 09:2} + # rises after the exempt month and must FAIL. + frozen = _frozen_panel( + [ + ("2021-07-01", "A", NAN), + ("2021-07-15", "B", 9.9), # first finite frozen value -> exempt 2021-07 + ("2021-08-02", "A", NAN), + ("2021-09-01", "A", NAN), ("2021-09-02", "A", NAN), + ], + "f", + ) + new = _new_series( + [ + ("2021-07-01", "A", 1.0), + ("2021-07-15", "B", 9.9), + ("2021-08-02", "A", 2.0), + ("2021-09-01", "A", 3.0), ("2021-09-02", "A", 4.0), + ] + ) + result = classify_panel_differences( + new, frozen, factor_id="f", is_pooled=True, lookback_depth=20 + ) + assert result.warmup_exempt_month == "2021-07" + assert not result.ok # 2021-09 (2) > 2021-08 (1) violates 按月递减至零 + assert not result.warmup_monotonic + + +def test_panels_pooled_warmup_zero_diff_exempt_month_does_not_reset_the_gate(): + # REVIEW PROBE, reversed (LOW-1): the exemption is anchored to the + # STRUCTURAL partial month (the frozen panel's first-finite-value month), + # NOT to the first month that HAPPENS to have warmup diffs. Here July + # (the structural partial month) has ZERO diffs and the counts then rise + # {08:2, 09:5} — the positional "first month WITH diffs is exempt" + # reading let exactly this through; the structural anchor must FAIL it. + frozen = _frozen_panel( + [("2021-07-29", "B", 9.9)] # first finite frozen value -> exempt 2021-07 + + [(f"2021-08-{d + 1:02d}", f"S{i}", NAN) for i, d in enumerate(range(2))] + + [(f"2021-09-{d + 1:02d}", f"S{i}", NAN) for i, d in enumerate(range(5))], + "f", + ) + new = _new_series( + [("2021-07-29", "B", 9.9)] + + [(f"2021-08-{d + 1:02d}", f"S{i}", 1.0) for i, d in enumerate(range(2))] + + [(f"2021-09-{d + 1:02d}", f"S{i}", 1.0) for i, d in enumerate(range(5))] + ) + result = classify_panel_differences( + new, frozen, factor_id="f", is_pooled=True, lookback_depth=20 + ) + assert result.warmup_exempt_month == "2021-07" + assert result.warmup_by_month == {"2021-08": 2, "2021-09": 5} + assert not result.warmup_monotonic + assert not result.ok + + +def test_panels_pooled_warmup_all_nan_frozen_panel_gets_no_exemption(): + # DEGENERATE: a frozen panel with NO finite value at all has no + # structural partial month to anchor to -> no exemption (conservative): + # {07:1, 08:2} rises from the very first month and must FAIL. frozen = _frozen_panel( [("2021-07-01", "A", NAN), ("2021-08-02", "A", NAN), ("2021-08-03", "A", NAN)], "f", @@ -584,8 +772,33 @@ def test_panels_pooled_warmup_non_monotonic_monthly_counts_fail(): result = classify_panel_differences( new, frozen, factor_id="f", is_pooled=True, lookback_depth=20 ) - assert not result.ok # 2021-08 (2) > 2021-07 (1) violates 按月递减至零 + assert result.warmup_exempt_month is None + assert result.warmup_by_month == {"2021-07": 1, "2021-08": 2} assert not result.warmup_monotonic + assert not result.ok + + +def test_panels_pooled_warmup_partial_first_month_is_exempt_from_monotonicity(): + # FORWARD (lead ruling 1): the exempt month is the month of the frozen + # panel's FIRST FINITE VALUE — the structural partial month in which + # residual/value existence starts (vpq: the frozen panel's first values + # exist only from ~07-29) — it is exempt from the monotonicity check, so + # {07:5, 08:10, 09:3} passes even though 2021-08 > 2021-07. + rows = ( + [("2021-07-29", "B", 9.9)] # first finite frozen value -> exempt 2021-07 + + [(f"2021-07-{d + 1:02d}", f"S{i}", NAN) for i, d in enumerate(range(5))] + + [(f"2021-08-{d + 1:02d}", f"S{i}", NAN) for i, d in enumerate(range(10))] + + [(f"2021-09-{d + 1:02d}", f"S{i}", NAN) for i, d in enumerate(range(3))] + ) + frozen = _frozen_panel(rows, "f") + new = _new_series([(d, s, v if pd.notna(v) else 1.0) for d, s, v in rows]) + result = classify_panel_differences( + new, frozen, factor_id="f", is_pooled=True, lookback_depth=20 + ) + assert result.ok, result.diffs + assert result.warmup_exempt_month == "2021-07" + assert result.warmup_by_month == {"2021-07": 5, "2021-08": 10, "2021-09": 3} + assert result.warmup_monotonic def test_panels_float_reordering_tail_within_bounds_passes(): @@ -662,6 +875,157 @@ def test_panels_threshold_flip_beyond_cell_cap_fails(): assert len(result.by_class("threshold_flip_tail")) == n +# --------------------------------------------------------------------------- # +# panels mode — threshold_flip_contamination (catalogue §七之五, lead ruling 2) +# --------------------------------------------------------------------------- # +def test_panels_threshold_flip_contamination_within_bounds_passes(): + # FORWARD: the measured vpq shape — the direct symbol (600623.SH) at + # 1.6e-04 and cross-sectionally contaminated symbols at <= 5.5e-07, all + # inside [2023-06-01, 2023-07-14], on a CROSS-SECTIONAL factor. + rows = [ + ("2023-06-01", "600623.SH", 1.0), ("2023-07-14", "600623.SH", 1.0), + ("2023-06-15", "S1", 2.0), ("2023-07-14", "S2", 3.0), + ] + deltas = [1.6e-04, -1.6e-04, 5.5e-07, -5.5e-07] + frozen = _frozen_panel(rows, "f") + new = _new_series([(d, s, v + dv) for (d, s, v), dv in zip(rows, deltas)]) + result = classify_panel_differences( + new, frozen, factor_id="f", is_pooled=False, lookback_depth=1, + is_cross_sectional=True, + ) + assert result.ok, result.diffs + assert len(result.by_class("threshold_flip_contamination")) == 4 + + +def test_panels_contamination_direct_symbol_above_2e_04_fails(): + # REVERSE (bound 1 of 4): the direct symbol overshoots 2e-04 inside the + # window -> UNCLASSIFIED, never a fall-through to the generic tails. + frozen = _frozen_panel([("2023-06-15", "600623.SH", 1.0)], "f") + new = _new_series([("2023-06-15", "600623.SH", 1.0 + 3e-04)]) + result = classify_panel_differences( + new, frozen, factor_id="f", is_pooled=False, lookback_depth=1, + is_cross_sectional=True, + ) + assert not result.ok + assert len(result.by_class("unclassified_finite_vs_finite")) == 1 + assert not result.by_class("threshold_flip_contamination") + + +def test_panels_contamination_other_symbol_above_1e_06_fails(): + # REVERSE (bound 2 of 4): a contaminated symbol overshoots 1e-06 inside + # the window. + frozen = _frozen_panel([("2023-07-01", "S1", 1.0)], "f") + new = _new_series([("2023-07-01", "S1", 1.0 + 2e-06)]) + result = classify_panel_differences( + new, frozen, factor_id="f", is_pooled=False, lookback_depth=1, + is_cross_sectional=True, + ) + assert not result.ok + assert len(result.by_class("unclassified_finite_vs_finite")) == 1 + + +def test_panels_contamination_outside_window_fails(): + # REVERSE (bound 3 of 4): the same magnitude one day AFTER the window + # (2023-07-15) is a new fact, not the adjudicated mechanism. + frozen = _frozen_panel([("2023-07-15", "600623.SH", 1.0)], "f") + new = _new_series([("2023-07-15", "600623.SH", 1.0 + 1.6e-04)]) + result = classify_panel_differences( + new, frozen, factor_id="f", is_pooled=False, lookback_depth=1, + is_cross_sectional=True, + ) + assert not result.ok + assert len(result.by_class("unclassified_finite_vs_finite")) == 1 + + +def test_panels_contamination_beyond_cell_cap_fails(): + # REVERSE (bound 4 of 4): 20,020 in-window cells within the abs bounds + # (1001 symbols x 20 dates) exceed the 20,000 cap. + symbols = [f"S{i}" for i in range(1001)] + dates = [f"2023-06-{d + 1:02d}" for d in range(20)] + rows = [(d, s, 1.0) for d in dates for s in symbols] + frozen = _frozen_panel(rows, "f") + new = _new_series([(d, s, 1.0 + 5e-07) for d, s, _v in rows]) + result = classify_panel_differences( + new, frozen, factor_id="f", is_pooled=False, lookback_depth=1, + is_cross_sectional=True, + ) + assert not result.ok + assert len(result.by_class("threshold_flip_contamination")) == len(rows) + + +def test_panels_contamination_not_available_for_bars_only_factor(): + # REVERSE: the class is CROSS-SECTIONAL-ONLY — the same in-window cells + # on a bars-only factor (is_cross_sectional=False) are unclassified. + frozen = _frozen_panel([("2023-06-15", "600623.SH", 1.0)], "f") + new = _new_series([("2023-06-15", "600623.SH", 1.0 + 1.6e-04)]) + result = classify_panel_differences( + new, frozen, factor_id="f", is_pooled=False, lookback_depth=1, + ) + assert not result.ok + assert len(result.by_class("unclassified_finite_vs_finite")) == 1 + + +# --------------------------------------------------------------------------- # +# panels mode — float-tail abs floor + tiered cap (lead ruling 3) +# --------------------------------------------------------------------------- # +def test_panels_float_dust_abs_floor_passes_regardless_of_rel(): + # FORWARD: a near-zero cross-sectional OLS residual — abs 5e-13 (machine + # precision on a ~1e-6 residual) but rel ~5e-07 >> 5e-12. The abs floor + # (|diff| <= 1e-12) classes it as float dust on ANY factor kind. + frozen = _frozen_panel([("2022-11-03", "A", 1e-06)], "f") + new = _new_series([("2022-11-03", "A", 1e-06 + 5e-13)]) + result = classify_panel_differences( + new, frozen, factor_id="f", is_pooled=False, lookback_depth=1 + ) + assert result.ok, result.diffs + assert len(result.by_class("float_reordering_tail")) == 1 + + +def test_panels_float_dust_above_abs_floor_with_rel_overage_fails(): + # REVERSE: abs 2e-12 > the 1e-12 floor AND rel ~2e-06 > 5e-12 — neither + # criterion catches it, so it is unclassified (the floor is not a + # tolerance widening). + frozen = _frozen_panel([("2022-11-03", "A", 1e-06)], "f") + new = _new_series([("2022-11-03", "A", 1e-06 + 2e-12)]) + result = classify_panel_differences( + new, frozen, factor_id="f", is_pooled=False, lookback_depth=1 + ) + assert not result.ok + assert len(result.by_class("unclassified_finite_vs_finite")) == 1 + + +def test_panels_float_tail_cross_sectional_cap_allows_up_to_1000(): + # FORWARD: 500 float-tail cells pass on a cross-sectional factor (the + # bars-only 101 cap would fail them — the cap is tiered, measured vpq + # 707 + headroom). + n = 500 + frozen = _frozen_panel([(f"2022-11-{(i % 28) + 1:02d}", f"S{i}", 1.0) for i in range(n)], "f") + new = _new_series( + [(f"2022-11-{(i % 28) + 1:02d}", f"S{i}", 1.0 * (1 + 3e-12)) for i in range(n)] + ) + result = classify_panel_differences( + new, frozen, factor_id="f", is_pooled=False, lookback_depth=1, + is_cross_sectional=True, + ) + assert result.ok, result.diffs + assert len(result.by_class("float_reordering_tail")) == n + + +def test_panels_float_tail_cross_sectional_cap_1001_fails(): + # REVERSE: 1001 float-tail cells — one above the cross-sectional cap. + n = 1001 + frozen = _frozen_panel([(f"2022-11-{(i % 28) + 1:02d}", f"S{i}", 1.0) for i in range(n)], "f") + new = _new_series( + [(f"2022-11-{(i % 28) + 1:02d}", f"S{i}", 1.0 * (1 + 3e-12)) for i in range(n)] + ) + result = classify_panel_differences( + new, frozen, factor_id="f", is_pooled=False, lookback_depth=1, + is_cross_sectional=True, + ) + assert not result.ok + assert len(result.by_class("float_reordering_tail")) == n + + def test_panels_frozen_finite_new_nan_never_allowed(): frozen = _frozen_panel([("2021-07-01", "A", 1.0)], "f") new = _new_series([("2021-07-01", "A", NAN)]) diff --git a/tests/test_factor_eval_runner.py b/tests/test_factor_eval_runner.py index c14a9e6..506de3e 100644 --- a/tests/test_factor_eval_runner.py +++ b/tests/test_factor_eval_runner.py @@ -6,12 +6,14 @@ code). What THESE tests pin is the runner's own logic: the config gates (catalogue C1/C2 collapse), the BUG 5 config-book closure, the exec identity on the EvalConfig, the two book modes' ``book_view`` derivation, the -add-Section passthrough, the artifact stem isolation (incl. the ``_bookclose`` -suffix), and the readable refusal of the deferred ``valley_price_quantile``. +add-Section passthrough (mechanism A sink AND mechanism B neutralization), +and the artifact stem isolation (incl. the ``_bookclose`` suffix). """ from __future__ import annotations +import math + import numpy as np import pandas as pd import pytest @@ -203,16 +205,107 @@ def test_invalid_book_mode_is_rejected_before_any_work(tmp_path): run_factor_eval("ignored.yaml", "jump_amount_corr_20", book_mode="bogus") -def test_valley_price_quantile_is_a_readable_deferral(tmp_path, monkeypatch): - monkeypatch.setattr( - "qt.factor_eval_runner.load_config", lambda path: _min_config(tmp_path) +def test_valley_price_quantile_runs_end_to_end_with_neutralization_section( + monkeypatch, tmp_path +): + """vpq is SERVED (PR-C4b): mechanism-B disclosure -> exactly one add-Section. + + The service seams are faked (``panel`` returns a finite grid; + ``stored_payload`` returns a synthetic raw_qbar intermediate); the expected + coverage is recomputed INDEPENDENTLY from the same fakes via the real + ``summarize_neutralization`` + ``reversal_20``, so a wrong wiring (wrong + raw, wrong residual, wrong floor) fails the equality, not a tautology. + """ + from dataclasses import asdict + + import factors.service as service_mod + from factors.compute.minute.binding import RAW_QBAR_COL + from factors.compute.minute.valley_price_quantile import ( + VALLEY_QUANTILE_MIN_CROSS_SECTION, + VALLEY_QUANTILE_REVERSAL_DAYS, + reversal_20, ) - monkeypatch.setattr( - "qt.factor_eval_runner.build_eval_service", - lambda *a, **k: pytest.fail("service must not be built for a deferred factor"), + from qt.factor_eval_disclosures import ( + NEUTRALIZATION_SECTION_NAME, + NeutralizationCoverage, + summarize_neutralization, + ) + + captured: dict = {} + bundle = _wire(monkeypatch, tmp_path, captured) + + idx = pd.MultiIndex.from_product( + [pd.DatetimeIndex(DATES), SYMS], names=["date", "symbol"] ) - with pytest.raises(ValueError, match="PR-C4b"): - run_factor_eval("ignored.yaml", "valley_price_quantile_20") + + def fake_panel(factor_ids, universe, decisions, **kwargs): + return pd.DataFrame( + {fid: np.arange(len(idx), dtype=float) + 1.0 for fid in factor_ids}, + index=idx, + ) + + payload_frame = pd.DataFrame( + {RAW_QBAR_COL: np.linspace(-0.5, 0.5, len(idx))}, index=idx + ) + payload_calls: dict = {} + + def fake_stored_payload(factor_id, universe, decisions, **kwargs): + payload_calls.update( + factor_id=factor_id, universe=list(universe), decisions=decisions, + kwargs=kwargs, + ) + return payload_frame + + monkeypatch.setattr(service_mod, "panel", fake_panel) + monkeypatch.setattr(service_mod, "stored_payload", fake_stored_payload) + + result = run_factor_eval("ignored.yaml", "valley_price_quantile_20") + + # the mechanism-B inputs were requested for THIS factor / universe / decisions + assert payload_calls["factor_id"] == "valley_price_quantile_20" + assert payload_calls["universe"] == SYMS + assert payload_calls["kwargs"]["store"] is bundle.store + assert payload_calls["kwargs"]["sources"] is bundle.sources + + # the coverage equals the INDEPENDENTLY recomputed one + residual = fake_panel(["valley_price_quantile_20"], SYMS, None)[ + "valley_price_quantile_20" + ] + rev = reversal_20( + bundle.panel[["close"]], days=VALLEY_QUANTILE_REVERSAL_DAYS + ) + expected = summarize_neutralization( + payload_frame[RAW_QBAR_COL], + rev, + residual, + min_cross_section=VALLEY_QUANTILE_MIN_CROSS_SECTION, + ) + assert isinstance(result.coverage, NeutralizationCoverage) + # field-wise with NaN tolerance: the 2-symbol fixture is below + # min_cross_section, so raw_rev_spearman_mean is legitimately NaN on BOTH + # sides (and NaN != NaN); every count field is an exact check. + got, want = asdict(result.coverage), asdict(expected) + assert got.keys() == want.keys() + for key in got: + g, w = got[key], want[key] + if isinstance(g, float) and isinstance(w, float) and math.isnan(g) and math.isnan(w): + continue + assert g == w, key + + # ...and it reached the exec tail as exactly one add-Section + extras = captured["extra_sections"] + assert extras is not None and len(extras) == 1 + section = extras[0] + assert isinstance(section, Section) + assert section.name == NEUTRALIZATION_SECTION_NAME + assert section.payload.keys() == got.keys() # no derived props on this coverage + for key in section.payload: + g, w = section.payload[key], got[key] + if isinstance(g, float) and isinstance(w, float) and math.isnan(g) and math.isnan(w): + continue + assert g == w, key + assert section.note == expected.render() + assert captured["stem"] == "factor_eval_valley_price_quantile_20" # --------------------------------------------------------------------------- # diff --git a/tests/test_factor_materialize.py b/tests/test_factor_materialize.py index eec609b..9f58419 100644 --- a/tests/test_factor_materialize.py +++ b/tests/test_factor_materialize.py @@ -375,3 +375,34 @@ def test_build_horizon_config_from_live_config(): def test_build_horizon_config_refuses_missing_attr(): with pytest.raises(ValueError, match="fina_tail_days"): build_horizon_config(_PartialCacheCfg()) + + +# --------------------------------------------------------------------------- # +# warmup-trim clamp disclosure (D5 C4 review NIT-1) +# --------------------------------------------------------------------------- # +def test_warmup_start_clamp_to_data_start_logs_one_disclosure_line(caplog): + import logging + + from factors.materialize import _warmup_start + + dates = pd.DatetimeIndex(["2021-07-01", "2021-07-02", "2021-07-05"]) + with caplog.at_level(logging.INFO, logger="factors.materialize"): + keep = _warmup_start(dates, pd.Timestamp("2021-07-05"), warmup=20) + assert keep == pd.Timestamp("2021-07-01") # clamped to the earliest date + lines = [r for r in caplog.records if "warmup trim clamped" in r.message] + assert len(lines) == 1 + msg = lines[0].message + assert "2021-07-01" in msg # the actual start date + assert "20" in msg # the declared warmup depth + + +def test_warmup_start_without_clamp_is_silent(caplog): + import logging + + from factors.materialize import _warmup_start + + dates = pd.DatetimeIndex([f"2021-07-{d:02d}" for d in range(1, 10)]) + with caplog.at_level(logging.INFO, logger="factors.materialize"): + keep = _warmup_start(dates, pd.Timestamp("2021-07-09"), warmup=3) + assert keep == pd.Timestamp("2021-07-06") + assert not [r for r in caplog.records if "warmup trim clamped" in r.message] diff --git a/tests/test_factor_materialize_streaming.py b/tests/test_factor_materialize_streaming.py index 6ac41bf..8f2a76c 100644 --- a/tests/test_factor_materialize_streaming.py +++ b/tests/test_factor_materialize_streaming.py @@ -42,7 +42,6 @@ from factors.base import Factor from factors.compute.minute import binding as binding_module from factors.compute.minute.binding import ( - _DEFERRED, _MINUTE_BINDINGS, _MINUTE_STREAM_BINDINGS, CROSS_SECTIONAL_MINUTE_FACTORS, @@ -73,7 +72,7 @@ def _streamed_factor_ids() -> tuple[str, ...]: round-trip through ``factor_registry.build`` below is what proves it. """ ids = [] - for cls in _MINUTE_STREAM_BINDINGS: + for cls in _MINUTE_BINDINGS: factor_id = cls().name rebuilt = factor_registry.build(factor_id) assert type(rebuilt) is cls, ( @@ -84,8 +83,9 @@ def _streamed_factor_ids() -> tuple[str, ...]: return tuple(sorted(ids)) -#: The bars-only minute factors (valley_price_quantile is the deliberately -#: deferred eleventh — it also needs the daily panel, so it has no bars binding). +#: The bars-only minute factors. valley_price_quantile is NOT here: its combine +#: declares a daily-panel input (D5 C4b), so it has no bars-only whole-factor +#: binding; it IS stream-bound and reconciled in tests/test_minute_binding_vpq.py. STREAMED_FACTOR_IDS = _streamed_factor_ids() #: The BOUNDED minute factors (fixed trailing trim, no saturation expansion) — @@ -247,12 +247,31 @@ def test_split_stages_reproduce_the_whole_factor_call(factor_id): assert n > 0, f"{factor_id}: vacuous (no finite values on the fixture)" -def test_stream_bindings_cover_exactly_the_bound_minute_factors(): - """A factor bound at one granularity but not the other is a readable error.""" - assert set(_MINUTE_STREAM_BINDINGS) == set(_MINUTE_BINDINGS) +def test_stream_bindings_cover_the_bars_only_factors_plus_the_daily_bound_one(): + """The stream table is the bars-only table PLUS exactly valley_price_quantile. + + The ten bars-only factors must live in BOTH tables (a factor bound at one + granularity but not the other is a readable error); valley_price_quantile is + stream-only BY DESIGN — its combine declares a daily-panel input, so it has + no bars-only whole-factor form — and that daily declaration is asserted here + rather than assumed (D5 C4b). + """ + from factors.compute.minute.binding import minute_combine_daily_spec + from factors.compute.minute.valley_price_quantile import ValleyPriceQuantileFactor + + assert set(_MINUTE_STREAM_BINDINGS) - set(_MINUTE_BINDINGS) == { + ValleyPriceQuantileFactor + } + assert set(_MINUTE_BINDINGS) < set(_MINUTE_STREAM_BINDINGS) assert CROSS_SECTIONAL_MINUTE_FACTORS <= set(_MINUTE_STREAM_BINDINGS) assert is_cross_sectional_minute(IntradayAmpCutFactor()) assert not is_cross_sectional_minute(factor_registry.build("volume_peak_count_20")) + # The daily declaration is on EXACTLY the one stream-only factor. + for cls in _MINUTE_STREAM_BINDINGS: + factor = cls() + assert (minute_combine_daily_spec(factor) is not None) is ( + cls is ValleyPriceQuantileFactor + ), cls.__name__ def test_every_minute_factor_class_is_classified_by_a_binding_table(): @@ -288,11 +307,11 @@ def test_every_minute_factor_class_is_classified_by_a_binding_table(): ): defined[obj] = path.name - classified = set(_MINUTE_STREAM_BINDINGS) | set(_DEFERRED) + classified = set(_MINUTE_STREAM_BINDINGS) assert defined, "found no minute factor classes — the walk itself is broken" unclassified = {cls.__name__: defined[cls] for cls in defined if cls not in classified} assert not unclassified, ( - f"minute factor(s) in neither _MINUTE_STREAM_BINDINGS nor _DEFERRED: " + f"minute factor(s) missing from _MINUTE_STREAM_BINDINGS: " f"{unclassified} — they would vanish from every parametrized " f"reconciliation without a single test going red" ) @@ -303,13 +322,18 @@ def test_every_minute_factor_class_is_classified_by_a_binding_table(): ) -def test_deferred_factor_still_raises_readably_through_the_split_stages(): - """valley_price_quantile is deferred: both stages must say so, not mis-compute.""" +def test_whole_factor_entry_point_still_refuses_the_daily_bound_factor(): + """valley_price_quantile has NO bars-only whole-factor form (its combine + needs the daily panel): the whole-factor entry point must refuse it + readably, while its stream stage works (the binding proper is reconciled in + tests/test_minute_binding_vpq.py).""" factor = factor_registry.build("valley_price_quantile_20") - with pytest.raises(NotImplementedError, match="valley_price_quantile"): - minute_stats_from_bars(factor, DENSE) - with pytest.raises(NotImplementedError, match="valley_price_quantile"): - combine_minute_stats(factor, pd.DataFrame()) + with pytest.raises(KeyError, match="valley_price_quantile"): + minute_raw_from_bars(factor, DENSE) + assert not minute_stats_from_bars(factor, DENSE).empty + # ...and the daily-declared combine refuses a missing daily panel readably. + with pytest.raises(ValueError, match="daily"): + combine_minute_stats(factor, minute_stats_from_bars(factor, DENSE)) # --------------------------------------------------------------------------- # diff --git a/tests/test_factor_service.py b/tests/test_factor_service.py index c5493a6..64f8380 100644 --- a/tests/test_factor_service.py +++ b/tests/test_factor_service.py @@ -810,3 +810,64 @@ def test_panel_requires_uniform_cutoff(): [DecisionPoint(date=DATES[40], cutoff="14:50:00"), DecisionPoint(date=DATES[41], cutoff="14:45:00")], store=store, sources=src) + + +# --------------------------------------------------------------------------- # +# stored_payload: the disclosure read-back of the STORED payload (D5 C4b) +# --------------------------------------------------------------------------- # +def test_stored_payload_value_factor_matches_the_panel_read_and_is_request_scoped(): + """For a value-payload factor the stored payload IS the value, read back + through the same engine; a narrower request is served the SLICE, never the + superset a wider earlier request left in the store.""" + dates = list(DATES[40:56]) + with tempfile.TemporaryDirectory() as td: + store = FactorValueStore(td) + src = _sources() + decisions = [DecisionPoint(date=d) for d in dates] + served = panel(["momentum_20"], SYMS, decisions, store=store, sources=src) + + payload = service_mod.stored_payload( + "momentum_20", SYMS, decisions, store=store, sources=src + ) + got = payload["momentum_20"].sort_index() + want = served["momentum_20"].sort_index() + assert got.index.equals(want.index) + gv, wv = got.to_numpy(), want.to_numpy() + assert np.array_equal(np.isnan(gv), np.isnan(wv)) + assert np.array_equal(gv[~np.isnan(gv)], wv[~np.isnan(wv)]) # BIT identical + + sub = service_mod.stored_payload( + "momentum_20", SYMS[:1], decisions[:4], store=store, sources=src + ) + assert set(map(str, sub.index.get_level_values("symbol"))) == {SYMS[0]} + assert set(sub.index.get_level_values("date")) == set(dates[:4]) + + +def test_stored_payload_cross_sectional_factor_is_the_intermediate_no_recompute(): + """For the cross-sectional factor the payload is the per-symbol INTERMEDIATE + (v_mean/v_std — never the combined value), and a warm store is NOT + re-materialized for the disclosure read (zero minute-bars reads).""" + dates = list(DATES[42:54]) + with tempfile.TemporaryDirectory() as td: + store = FactorValueStore(td) + src = _sources() + decisions = [DecisionPoint(date=d) for d in dates] + served = panel( + ["intraday_amp_cut_10"], SYMS, decisions, store=store, sources=src + ) + calls_after_panel = src.minute.calls + assert calls_after_panel > 0 # the cold fill really read bars + + payload = service_mod.stored_payload( + "intraday_amp_cut_10", SYMS, decisions, store=store, sources=src + ) + assert src.minute.calls == calls_after_panel # warm: NO re-materialization + assert set(payload.columns) == {"v_mean", "v_std"} + assert not payload.empty + # the per-symbol intermediate really carried values on this fixture... + assert payload["v_mean"].notna().any() + # ...and the served value is the COMBINE of exactly this payload, not a + # column of it (all-NaN here: 2 symbols < the combine's cross-section + # floor — the served shape is the combine's business, D4c) + assert "intraday_amp_cut_10" not in payload.columns + assert served["intraday_amp_cut_10"].isna().all() diff --git a/tests/test_factor_store_keys.py b/tests/test_factor_store_keys.py index a3fb299..bb14c31 100644 --- a/tests/test_factor_store_keys.py +++ b/tests/test_factor_store_keys.py @@ -69,13 +69,20 @@ def test_params_hash_none_equals_empty(): # --------------------------------------------------------------------------- # # code_hash: the shared set + module identity # --------------------------------------------------------------------------- # -def test_shared_set_is_the_enumerated_four_plus_ops(): +def test_shared_set_is_the_enumerated_five_plus_ops(): labels = {label for label, _ in shared_set_labeled_files()} assert "factors.compute.minute.primitives" in labels + # D5 C4b: the minute binding carries load-bearing value semantics the + # column-shape validation cannot see, so it is folded into every code hash. + assert "factors.compute.minute.binding" in labels assert "factors.base" in labels assert "factors.spec" in labels # factors.ops expands to every module under the package assert any(label.startswith("factors.ops") for label in labels) + # ...and files that merely CONSUME the binding (engine/service) are NOT + # folded: a code-hash invalidation must stay a factor-semantics event. + assert "factors.materialize" not in labels + assert "factors.service" not in labels def test_code_hash_is_stable(): @@ -126,6 +133,72 @@ def test_code_hash_changes_when_a_shared_set_member_content_changes(tmp_path): assert code_hash(factor) == base +def test_code_hash_changes_when_the_binding_content_changes(tmp_path): + """MUTATION (D5 C4b): a content change to the minute BINDING module must move + the code hash — the binding decides which compute function a factor is bound + to and how the per-symbol/combine split runs, and the D4c shape validation + covers only the intermediate's column names, so a value-carrying binding edit + was previously invisible to the store key. An engine file that merely + CONSUMES the binding (factors/materialize.py) must stay OUT of the fold. + """ + factor = _vol() + folded = _folded_items(factor) + base = content_hash_of_labeled_files(folded) + + target_label = "factors.compute.minute.binding" + mutated_items = [] + swapped = False + for label, path in folded: + if label == target_label: + copy = tmp_path / "binding_mutated.py" + copy.write_bytes(Path(path).read_bytes() + b"\n# D5 C4b mutation\n") + mutated_items.append((label, copy)) + swapped = True + else: + mutated_items.append((label, path)) + assert swapped, "shared set unexpectedly lacks the binding module" + assert content_hash_of_labeled_files(mutated_items) != base + + # And the real code_hash equals the un-mutated fold (the binding IS in it). + assert code_hash(factor) == base + # Control: the engine consumer is not folded, so an edit to IT moves nothing. + assert all(label != "factors.materialize" for label, _ in folded) + + +def test_binding_mutation_moves_every_registered_factor_code_hash(tmp_path): + """The binding fold is GLOBAL: mutating its content moves EVERY registered + factor's code hash (over-invalidate-safe — the one-time wholesale store + invalidation this causes is disclosed in the code_hash module docstring). + """ + from factors.registry.registry import DEFAULT_REGISTRY + + classes = { + entry.factor_cls + for entry in list(DEFAULT_REGISTRY._exact.values()) + + list(DEFAULT_REGISTRY._prefixes) + } + assert len(classes) >= 14, "the closing factor set alone has 14 members" + binding_label, binding_path = "factors.compute.minute.binding", None + for label, path in shared_set_labeled_files(): + if label == binding_label: + binding_path = path + assert binding_path is not None + mutated_copy = tmp_path / "binding_mutated.py" + mutated_copy.write_bytes(Path(binding_path).read_bytes() + b"\n# D5 C4b mutation\n") + + for cls in classes: + folded = _folded_items(cls) + base = content_hash_of_labeled_files(folded) + mutated = [ + (label, mutated_copy if label == binding_label else path) + for label, path in folded + ] + assert content_hash_of_labeled_files(mutated) != base, ( + f"{cls.__name__}: a binding content change did not move its code hash" + ) + assert code_hash(cls) == base + + def test_one_hop_folds_a_composed_module_but_not_unrelated_modules(): # Review MEDIUM: candidates.py composes MomentumFactor, so momentum.py is a # ONE-HOP dep of every candidates factor (reversal / value / volatility) — a diff --git a/tests/test_factor_store_universe.py b/tests/test_factor_store_universe.py index c742bd3..08e3db8 100644 --- a/tests/test_factor_store_universe.py +++ b/tests/test_factor_store_universe.py @@ -41,6 +41,8 @@ from factors import service as service_mod from factors.compute.minute.binding import ( _MINUTE_STREAM_BINDINGS, + CROSS_SECTIONAL_MINUTE_FACTORS, + is_cross_sectional_minute, minute_intermediate_columns, minute_stats_from_bars, ) @@ -50,8 +52,10 @@ V_STD_COL, IntradayAmpCutFactor, ) +from factors.compute.minute.valley_price_quantile import ValleyPriceQuantileFactor from factors.materialize import ( MaterializeSources, + combine_daily_panel, make_recompute_fn, materialize_intermediate_range, materialize_range, @@ -414,16 +418,34 @@ def _probe_requested_universe(universe): return pd.Series(0.0, index=pd.Index(requested_universe(universe), name="symbol")) +def _probe_combine_daily_panel(universe): + return combine_daily_panel( + factor_registry.build("valley_price_quantile_20"), view=View.DECISION, + symbols=universe, emit_start=EMIT[0], emit_end=EMIT[-1], + sources=MaterializeSources(daily=DailyProv()), + ) + + +def _probe_stored_payload(universe): + with tempfile.TemporaryDirectory() as td: + return service_mod.stored_payload( + PURE_MINUTE_ID, universe, [DecisionPoint(date=d) for d in EMIT], + store=FactorValueStore(td), sources=_sources(PURE_MINUTE_ID), + ) + + #: entry point -> how to call it with a caller list. A new entry point without a #: recipe fails the surface test below; the recipes cannot be derived (each #: signature differs), but WHICH ones must exist is. _ENTRY_PROBES = { "factors.service.panel": _probe_panel, "factors.service.cross_section": _probe_cross_section, + "factors.service.stored_payload": _probe_stored_payload, "factors.materialize.materialize_range": _probe_materialize_range, "factors.materialize.materialize_intermediate_range": _probe_materialize_intermediate_range, "factors.materialize.make_recompute_fn": _probe_make_recompute_fn, "factors.materialize.requested_universe": _probe_requested_universe, + "factors.materialize.combine_daily_panel": _probe_combine_daily_panel, } @@ -735,15 +757,22 @@ def counted(factor, stats): # Payload shape: what each kind of factor stores # --------------------------------------------------------------------------- # def test_only_the_cross_sectional_factor_stores_an_intermediate(): - """The other ten minute factors and the daily factors are NOT dragged into the - two-stage form — their values are universe-free, so they store their value.""" + """The other nine minute factors and the daily factors are NOT dragged into + the two-stage form — their values are universe-free, so they store their + value. Derived from the DECLARED set (``is_cross_sectional_minute``), so a + third cross-sectional factor cannot sneak in untested; today the set is + intraday_amp_cut (v_mean/v_std) + valley_price_quantile (raw_qbar, D5 C4b).""" for cls in _MINUTE_STREAM_BINDINGS: factor = factor_registry.build(cls().name) - expect = cls is IntradayAmpCutFactor + expect = is_cross_sectional_minute(factor) assert stores_intermediate(factor) is expect, factor.name assert payload_columns(factor) == ( - (V_MEAN_COL, V_STD_COL) if expect else (factor.name,) + minute_intermediate_columns(factor) if expect else (factor.name,) ), factor.name + assert set(CROSS_SECTIONAL_MINUTE_FACTORS) == { + IntradayAmpCutFactor, + ValleyPriceQuantileFactor, + } for fid in (DAILY_ID, "volatility_20", "value_ep"): factor = factor_registry.build(fid) assert stores_intermediate(factor) is False, fid diff --git a/tests/test_minute_binding_vpq.py b/tests/test_minute_binding_vpq.py new file mode 100644 index 0000000..c929b7a --- /dev/null +++ b/tests/test_minute_binding_vpq.py @@ -0,0 +1,331 @@ +"""D5 C4b: valley_price_quantile binding + the shifted-panel third path. + +Network-free. Pins BOTH sides of the lead-authorized third path for binding +valley_price_quantile (vpq) into the streaming engine: + +* BINDING SIDE — the shifted-panel reversal is algebraically identical to the + legacy construction: ``reversal_20_shifted(daily_decision_lag(closes))`` == + ``reversal_20(closes)`` BIT-FOR-BIT (the factor's internal T-1 shift composed + with the decision lag would double-lag to close_{d-2}). A sabotage control + (the shifted variant fed the UN-lagged panel) must differ materially, so the + equivalence can never pass vacuously; a perturbation control shows both + constructions track a changed close identically. +* RUNNER SIDE — the full service read-through (store + materializer + combine) + reproduces the LEGACY runner semantics cell for cell: per-symbol + ``compute_valley_price_quantile_stats`` over the full history + ONE + cross-sectional ``residualize_on_reversal`` against ``reversal_20`` on the + un-lagged close panel. Including the hard assertion that the left-edge + 22-trading-day window shows ZERO NaN-set divergence (the shifted panel's + leading NaN row must land BEFORE the needed window — §六.18). + +The real-cache vpq reconciliation against the frozen exec baseline is the NEXT +commit (the unified runner enablement); this file pins the semantics on fixtures. +""" + +from __future__ import annotations + +import tempfile + +import numpy as np +import pandas as pd +import pytest + +from data.clean.intraday_schema import normalize_intraday_bars +from factors import registry as factor_registry +from factors.compute.minute.binding import ( + RAW_QBAR_COL, + combine_minute_stats, + minute_combine_daily_spec, + minute_stats_from_bars, +) +from factors.compute.minute.valley_price_quantile import ( + VALLEY_QUANTILE_REVERSAL_DAYS, + compute_valley_price_quantile_stats, + residualize_on_reversal, + reversal_20, + reversal_20_shifted, +) +from factors.materialize import MaterializeSources +from factors.service import DecisionPoint, cross_section, panel +from factors.store.values import FactorValueStore +from factors.view_lag import daily_decision_lag + +FID = "valley_price_quantile_20" +#: >= VALLEY_QUANTILE_MIN_CROSS_SECTION (=10) so the per-date OLS can run. +N_SYMBOLS = 12 +SYMBOLS = [f"6000{i:02d}.SH" for i in range(N_SYMBOLS)] +DATES = pd.bdate_range("2021-01-04", periods=75) +EMIT_START, EMIT_END = DATES[40], DATES[59] +BARS_PER_DAY = 238 +#: The declared daily-combine warmup (rev20's 21 closes + the lag's leading row). +DAILY_WARMUP = VALLEY_QUANTILE_REVERSAL_DAYS + 2 + + +def _minute() -> pd.DataFrame: + rng = np.random.RandomState(5) + rows: list[tuple] = [] + for si, s in enumerate(SYMBOLS): + for d in DATES: + base = pd.Timestamp(d) + pd.Timedelta("09:31:00") + price = 100.0 + si * 3 + rng.normal(0, 2) + for i in range(BARS_PER_DAY): + t = base + pd.Timedelta(minutes=i) + price += rng.normal(0, 0.05) + slot = 1e4 * (1.0 + 0.3 * np.sin(i / 12.0)) + erupt = 6.0 if (rng.rand() < 0.06) else 1.0 + vol = slot * erupt * (1.0 + 0.1 * rng.rand()) + w = 0.15 * price * (1.0 + (2.0 if erupt > 1 else 0.0)) * (0.5 + rng.rand()) + hi, lo = price + abs(w) * rng.rand(), price - abs(w) * rng.rand() + cl = lo + (hi - lo) * rng.rand() + rows.append((t, s, price, hi, lo, cl, vol, cl * vol)) + cols = ["time", "symbol", "open", "high", "low", "close", "volume", "amount"] + return normalize_intraday_bars(pd.DataFrame(rows, columns=cols), freq="1min") + + +def _daily() -> pd.DataFrame: + rng = np.random.RandomState(3) + rows = [] + for si, s in enumerate(SYMBOLS): + px = 100.0 + si * 2 + np.cumsum(rng.normal(0, 1.0, len(DATES))) + for d, p in zip(DATES, px): + rows.append((d, s, p - 0.3, p + 0.5, p - 0.5, p, 1e5, p * 1e5)) + cols = ["date", "symbol", "open", "high", "low", "close", "volume", "amount"] + return pd.DataFrame(rows, columns=cols).set_index(["date", "symbol"]).sort_index() + + +MINUTE = _minute() +DAILY = _daily() +CLOSES = DAILY[["close"]] + + +class MinuteProv: + """Honours ``symbols`` (like the cache reader); declares the fixture start.""" + + def minute_bars(self, symbols, start, end): + if not symbols: + return MINUTE.iloc[0:0] + t = MINUTE.index.get_level_values("time") + sym = MINUTE.index.get_level_values("symbol") + keep = (t >= pd.Timestamp(start)) & (t <= pd.Timestamp(end)) & sym.isin(list(symbols)) + return MINUTE[keep] + + def earliest_available(self, symbols): + return DATES[0] + + +class DailyProv: + def daily_panel(self, symbols, start, end): + d = DAILY.index.get_level_values("date") + sym = DAILY.index.get_level_values("symbol") + return DAILY[(d >= pd.Timestamp(start)) & (d <= pd.Timestamp(end)) & sym.isin(list(symbols))] + + +def _sources() -> MaterializeSources: + return MaterializeSources(daily=DailyProv(), minute=MinuteProv()) + + +def _factor(): + return factor_registry.build(FID) + + +def _assert_bit_identical(got: pd.Series, want: pd.Series, label: str) -> int: + assert got.index.equals(want.index), f"{label}: index differs" + a, b = got.to_numpy(), want.to_numpy() + assert np.array_equal(np.isnan(a), np.isnan(b)), f"{label}: NaN mask differs" + finite = ~np.isnan(a) + assert np.array_equal(a[finite], b[finite]), f"{label}: finite values differ" + return int(finite.sum()) + + +# --------------------------------------------------------------------------- # +# BINDING SIDE: shifted panel + T-1 OFF == un-lagged panel + internal T-1 +# --------------------------------------------------------------------------- # +def test_shifted_panel_reversal_is_bit_identical_to_internal_t1(): + """The third path's core identity, exact (the D5b review measured 0.0).""" + want = reversal_20(CLOSES).sort_index() + got = reversal_20_shifted(daily_decision_lag(CLOSES)).sort_index() + assert want.equals(got), "shifted+no-T1 must equal un-lagged+T1 bit-for-bit" + assert int(want.notna().sum()) > 0, "vacuous: no finite reversal values" + + +def test_shifted_reversal_on_the_UNLAGGED_panel_differs_materially(): + """Sabotage control: the equivalence above cannot pass vacuously. + + Feeding the shifted variant the UN-lagged panel computes a DIFFERENT + quantity (one day too early); the measured max|diff| on this fixture is + ~4.6e-2 (the D5b review's sabotage scale, 7.3e-2, on its own data). If a + future edit made the two constructions quietly identical, THIS assertion — + not the equivalence — turns red first. + """ + want = reversal_20(CLOSES).sort_index() + wrong = reversal_20_shifted(CLOSES).sort_index() + diff = (want - wrong).abs() + assert diff.max() > 1e-3, ( + f"the shifted variant on the un-lagged panel should differ materially " + f"(measured max|diff|={diff.max():.3e}); if it does not, the equivalence " + f"test above is vacuous" + ) + + +def test_shifted_reversal_tracks_a_perturbed_close_bit_identically(): + """Perturbation control: both constructions move TOGETHER under a bad print. + + Picks a close deep inside the reversal window of the emit range, bumps it, + and asserts the two constructions stay bit-identical AND actually moved — + the two failure modes this guards are 'the shifted path ignores part of the + panel' and 'the test compares two constants'. + """ + target = CLOSES.index[int(len(CLOSES) * 0.6)] + perturbed = CLOSES.copy() + perturbed.loc[target, "close"] = float(perturbed.loc[target, "close"]) * 1.5 + want = reversal_20(perturbed).sort_index() + got = reversal_20_shifted(daily_decision_lag(perturbed)).sort_index() + assert want.equals(got), "a perturbed close must move both constructions identically" + assert not want.equals(reversal_20(CLOSES).sort_index()), ( + "vacuous: the perturbation moved nothing" + ) + + +def test_combine_preserves_the_intermediate_rows_and_is_a_per_date_reduction(): + """The two combine CONTRACTS the engine relies on (DailyCombineInput). + + (1) The residualized value has EXACTLY the intermediate's rows — the pooled + saturation loop reads the value's output dates off the intermediate without + loading the daily panel. (2) Restricting the intermediate to the emit + window before combining cannot change a kept date's value — the streaming + materializer and the store slice rely on it. + """ + factor = _factor() + stats = minute_stats_from_bars(factor, MINUTE) + daily = daily_decision_lag(CLOSES) + full = combine_minute_stats(factor, stats, daily=daily) + assert full.index.equals(stats.index), ( + "the combine must return exactly the intermediate's rows" + ) + d = stats.index.get_level_values("date") + windowed = stats[(d >= EMIT_START) & (d <= EMIT_END)] + got = combine_minute_stats(factor, windowed, daily=daily) + want = full[(full.index.get_level_values("date") >= EMIT_START)] + want = want[want.index.get_level_values("date") <= EMIT_END] + _assert_bit_identical(got.sort_index(), want.sort_index(), "per-date reduction") + + +# --------------------------------------------------------------------------- # +# RUNNER SIDE: the service read-through == the legacy runner semantics +# --------------------------------------------------------------------------- # +def _legacy_reference() -> pd.Series: + """The LEGACY runner semantics on the same fixtures (the truth to match). + + Per-symbol ``compute_valley_price_quantile_stats`` over the FULL history + (the legacy runner reads each symbol's whole cached minute window; the + compute applies its own 14:50 cutoff), then ONE cross-sectional + residualization against ``reversal_20`` on the UN-lagged close panel. + """ + series = [] + for s in SYMBOLS: + bars = MINUTE[MINUTE.index.get_level_values("symbol") == s] + series.append(compute_valley_price_quantile_stats(bars, lookback_days=20)) + raw = pd.concat(series).sort_index() + rev = reversal_20(CLOSES) + factor = residualize_on_reversal(raw, rev, name=FID) + d = factor.index.get_level_values("date") + return factor[(d >= EMIT_START) & (d <= EMIT_END)].sort_index() + + +def _service_values(dates) -> pd.Series: + with tempfile.TemporaryDirectory() as td: + got = panel( + [FID], SYMBOLS, [DecisionPoint(date=d) for d in dates], + store=FactorValueStore(td), sources=_sources(), + ) + return got[FID].sort_index() + + +EMIT_DATES = DATES[(DATES >= EMIT_START) & (DATES <= EMIT_END)] + + +def test_service_read_through_reproduces_the_legacy_runner_cell_for_cell(): + """service.panel(vpq) vs the legacy construction: values + NaN masks exact. + + The served panel may carry EXTRA all-NaN rows: the D4c fill footprint writes + an explicit NaN row for every covered-but-valueless (date, symbol) cell, and + the combine passes them through as NaN. Those rows are a recorded-shape + disclosure, not a divergence — asserted all-NaN, then stripped for the + cell-for-cell comparison. + """ + got = _service_values(EMIT_DATES) + ref = _legacy_reference() + extra = got[~got.index.isin(ref.index)] + assert int(extra.notna().sum()) == 0, "footprint rows must be NaN, never values" + aligned = got[got.index.isin(ref.index)] + n = _assert_bit_identical(aligned, ref, "service vs legacy") + assert n > 0, "vacuous: no finite values to compare" + + +def test_left_edge_window_has_zero_nan_set_divergence(): + """HARD assertion on the §六.18 trap: within the first ``DAILY_WARMUP`` + trading days of the emit window the served NaN set equals the legacy one + EXACTLY — the shifted panel's leading NaN row must not eat into it.""" + got = _service_values(EMIT_DATES) + ref = _legacy_reference() + edge_end = DATES[DATES.get_loc(EMIT_START) + DAILY_WARMUP - 1] + gd = got.index.get_level_values("date") + rd = ref.index.get_level_values("date") + got_edge = got[gd <= edge_end] + ref_edge = ref[rd <= edge_end] + aligned = got_edge[got_edge.index.isin(ref_edge.index)] + assert aligned.index.equals(ref_edge.index), "left-edge rows differ" + assert np.array_equal( + np.isnan(aligned.to_numpy()), np.isnan(ref_edge.to_numpy()) + ), "left-edge NaN sets diverge — the shifted panel is under-warmed" + assert int((~np.isnan(ref_edge.to_numpy())).sum()) > 0, ( + "vacuous: the legacy reference has no finite value in the left-edge window" + ) + + +def test_single_date_fill_equals_batch_fill_for_vpq(): + """§3.5 P8 through the store for the daily-bound cross-sectional factor.""" + dates = pd.DatetimeIndex([EMIT_START, DATES[45], EMIT_END]) + with tempfile.TemporaryDirectory() as ta, tempfile.TemporaryDirectory() as tb: + a = FactorValueStore(ta) + for d in dates: + cross_section([FID], SYMBOLS, DecisionPoint(date=d), store=a, sources=_sources()) + b = FactorValueStore(tb) + panel([FID], SYMBOLS, [DecisionPoint(date=d) for d in dates], store=b, sources=_sources()) + fa = panel([FID], SYMBOLS, [DecisionPoint(date=d) for d in dates], store=a, sources=_sources()) + fb = panel([FID], SYMBOLS, [DecisionPoint(date=d) for d in dates], store=b, sources=_sources()) + x, y = fa[FID].sort_index(), fb[FID].sort_index() + _assert_bit_identical(x, y, "single-fill vs batch-fill (served)") + assert int(x.notna().sum()) > 0, "vacuous: no finite served values" + + +# --------------------------------------------------------------------------- # +# Engine wiring: declared daily input, loud on misuse +# --------------------------------------------------------------------------- # +def test_service_refuses_a_missing_daily_provider_for_vpq(): + """No DailyPanelProvider -> a readable error at assembly, never a silent NaN.""" + with tempfile.TemporaryDirectory() as td: + with pytest.raises(ValueError, match="DailyPanelProvider"): + panel( + [FID], SYMBOLS, [DecisionPoint(date=EMIT_START)], + store=FactorValueStore(td), + sources=MaterializeSources(minute=MinuteProv()), + ) + + +def test_combine_refuses_a_daily_panel_for_a_factor_that_declares_none(): + """Passing a daily panel to a bars-only factor's combine is refused, not + silently ignored.""" + other = factor_registry.build("volume_peak_count_20") + stats = minute_stats_from_bars(other, MINUTE) + with pytest.raises(ValueError, match="declares no daily input"): + combine_minute_stats(other, stats, daily=daily_decision_lag(CLOSES)) + + +def test_declared_daily_input_shape_is_the_pinned_third_path(): + """The declaration itself: close column only, rev20+2 lagged warmup days.""" + spec = minute_combine_daily_spec(_factor()) + assert spec is not None + assert spec.columns == ("close",) + assert spec.warmup_days == DAILY_WARMUP + assert RAW_QBAR_COL == "raw_qbar"