diff --git a/docs/factors/d5_saturation_feasibility.md b/docs/factors/d5_saturation_feasibility.md index 12ee7a1..42419b4 100644 --- a/docs/factors/d5_saturation_feasibility.md +++ b/docs/factors/d5_saturation_feasibility.md @@ -136,7 +136,7 @@ per-symbol 饱和当作已验证的等价改写。** 3. `valley_price_quantile` 的 D5 绑定（C4 交付物）**继承同一路径**——lead 已明令 不许用固定深度 trim 绕过，per-symbol 饱和是它唯一合法的加载形态。 -## 五、未做 / 留给后续 +## 五、未做 / 留给后续（原文保留；第 1 条已由 D4b 完成，见第六节） - **per-symbol 饱和的实现与四腿验收**：本探测只证明了「必须这么做」与「这么做在两个 pooled 因子上逐位等价」，**没有实现它**，也没有对 `intraday_amp_cut` 做端到端对账。 @@ -145,3 +145,145 @@ per-symbol 饱和当作已验证的等价改写。** 确实从 2015 开始，但 327 票晚得多），但**改 floor 派生方式本身**同样是定义相邻变更， 与 per-symbol 饱和是两件事，不要合并处理。 - 本探测未测 `concat` 峰值与 GC 行为，线性外推对峰值是**乐观**的。 + +--- + +# 六、D4b 结果：修完之后的实测（本节为「后」，一至四节是「前」） + +D4b 把 materializer 改成 **per-symbol 流式**：读一票 → 立刻聚合到该票日频中间量 → +丢弃 bars；**截面 combine 在装配完整 universe 面板之后跑一次**。因子数学、参数、阈值 +（含 `AMP_CUT_MIN_CROSS_SECTION=10`）一个字未动。复测命令： + +``` +python -m qt.saturation_probe --mode stream-scale --factors \ + --cache-root <...> --universe-panel <...> --out <...>.jsonl +``` + +一次一个因子、一个进程（`ru_maxrss` 才是该因子自己的峰值），JSONL 可续跑。 + +## 6.1 逐格对账：流式 vs 整 universe 单帧（真实缓存，40 票 × 6 个月） + +**可复跑**（评审必须能从仓库里重跑这张表，不是自写脚本）： + +``` +python -m qt.saturation_probe --mode reconcile --cache-root <...> --universe-panel <...> +``` + +`2023-01-03..2023-06-30`，参照系 = **被替换掉的那个几何**（全部票进一个 frame、一次 +cutoff、一次整因子调用），单帧 3,711,400 行 / 0.69 GB；`live_calls=0`。 + +⚠️ **参照系的加载深度是承重参数，必须写明**：参照单帧按 emit 窗口前 **400 个日历日** +加载。对 8 个 pooled 因子，流式一侧的深度由**饱和扩展**决定，所以参照太浅会变成"比 +两段不同的历史"、太深则改变浮点累加的前缀。400 天让两侧每个因子的 trailing 窗口都 +满深度——**这也解释了本表 5 个因子有 ~1e-14 残差、而「分支 vs main 真引擎」的对照能 +得到精确 0.000e+00**：后者两侧加载几何完全相同，前者刻意不同。 + +下表是该命令输出的**逐字转录**（列与顺序一致，便于评审直接 diff）： + +| 因子 | idx== | 可比 cell | NaN 集合差异 | max\|abs\| | +|---|---|---|---|---| +| `amp_marginal_anomaly_vol_20` | True | 4,720 | **0** | **0.000e+00** | +| **`intraday_amp_cut_10`** | True | 4,720 | **0** | 2.531e-14 | +| `jump_amount_corr_20` | True | 4,720 | **0** | 3.275e-14 | +| `minute_ideal_amp_10` | True | 4,720 | **0** | **0.000e+00** | +| `peak_interval_kurtosis_20` | True | 4,720 | **0** | **0.000e+00** | +| `peak_ridge_amount_ratio_20` | True | 2,247 | **0** | 3.331e-16 | +| `ridge_minute_return_20` | True | 2,283 | **0** | **0.000e+00** | +| `valley_relative_vwap_20` | True | 4,719 | **0** | 2.220e-16 | +| `valley_ridge_vwap_ratio_20` | True | 2,295 | **0** | 2.220e-16 | +| `volume_peak_count_20` | True | 4,720 | **0** | **0.000e+00** | + +`PROBLEMS: none`（任一 index 不同或 NaN 集合有差异即列入 PROBLEMS 并以 rc=1 退出）。 + +**NaN 集合差异全为 0** 是这张表里最承重的一列：finite↔NaN 翻转是覆盖率变化，不是浮点 +现象。5/10 逐位相同；其余 5 个残差 ≤3.3e-14，是 pandas 滚动累加在不同长度前缀上的重排 +（D2/D4 已编目的可归因项），比任何真实缺行效应低若干个数量级。 + +⚠️ **准确地说：这是本样本上的结果，不是「本步绝不允许翻转」的政策**（此前措辞如此， +是错的）。**per-symbol trim（§6.5）是本步唯一已披露的、能够产生 finite↔NaN 翻转的语义 +变更**，而且它**实测确实会**（三个 bounded 因子里两个）——这 40 票样本只是没有足够稀疏 +的票去触发它。表里的 0 是**该样本未触发**，不是**不可能触发**。 + +`intraday_amp_cut_10` 在 4,720 个真实 cell 上对上，**这是第三节留下的那笔账**：它是 +唯一 pooled × 截面双耦合的因子，切口放错（按整因子逐票切）会让它全 NaN——该负对照已 +committed 为测试，不是一次性观察。 + +## 6.2 规模证明：真实评估面（CSI500 995 票 × 5 年） + +**名义与峰值分开报**——一至四节实测 frame:峰值RSS = 2.2–2.45×，只报名义会把需求少说 +一半。以最危险的因子 `intraday_amp_cut_10`（唯一 pooled × 截面双耦合）为例： + +| | 名义最大单帧 | 峰值 RSS | 可用内存 | 判定 | +|---|---|---|---|---| +| 旧：整 universe 单帧（floor 深度） | **122.2 GB** | **~115 GB**（评估窗口档）| 56 GB | **装不下** | +| 新：per-symbol 流式 | **0.0656 GB** | **0.964 GB** | 56 GB | **可跑（1.7% 内存）** | + +峰值 RSS 降到 **约 1/119**，且**不随 universe 规模增长**——引擎每次只向 provider 要 +**一票**（`max_symbols_per_call=1`，测试锁定，mutation 实证）。 + +## 6.3 每因子实测（`--mode stream-scale`，cache-only，一因子一进程） + +CSI500 995 票 / `2021-07-01..2026-06-30`： + +**三种加载几何各取一个**（这是完整的几何覆盖，不是抽样）： + +| 因子 | 几何 | 输出行 | 有值 | 出现的票 | 每次请求票数 | 名义最大帧 | **峰值 RSS** | wall | live_calls | +|---|---|---|---|---|---|---|---|---|---| +| `intraday_amp_cut_10` | pooled × **截面** | 1,158,842 | 1,158,842 | **995/995** | **1** | 0.0656 GB | **0.964 GB** | 1990.5 s | **0** | +| `ridge_minute_return_20` | pooled | 593,245 | 592,431 | **995/995** | **1** | 0.0713 GB | **0.854 GB** | 1133.7 s | **0** | +| `jump_amount_corr_20` | bounded | 1,159,263 | 1,158,552 | **995/995** | **1** | 0.0658 GB | **0.730 GB** | 417.7 s | **0** | + +**995/995**：没有任何一只被请求的票在流式化后消失（I5a 红线）。峰值 RSS 落在 +**0.73–0.96 GB**，与因子和输出规模基本无关——峰值由「一票的 bars + 日频面板」决定， +不由 universe 决定。`live_calls=0` 全程 cache-only。 + +wall 按几何排序符合预期：bounded 无扩张（418 s）＜ pooled 20 chunk（1134 s）＜ pooled +60 天 chunk（1990 s，见 6.4）。**剩余 7 个因子未跑**——那已经是 D5 的全量 run 本身， +不是 D4b 的验收；同一命令可续跑，JSONL 里已记录的因子会被跳过（实测 SKIP 生效）。 + +⚠️ **wall 是上界，不是净耗时**：这台机器同时在跑其它验收作业（逐格对账、手算锚、 +pytest），20 核，争用主要在磁盘。**内存数字不受影响**（`ru_maxrss` 是各进程自己的 +高水位），wall 应读作「在有争用的条件下不超过这个数」。 + +## 6.4 代价结构：钱花在哪 84 只票上（实测，非推断） + +`intraday_amp_cut_10`，各取 10 票，同 5 年 emit 窗口： + +| 票的类型 | provider 调用/票 | wall/票 | 单票最大 frame | +|---|---|---|---| +| dense（2015 上市） | **1.0** | **0.59 s** | 0.066 GB | +| late（首根 bar ≥ emit_start） | **40.0** | **14.95 s** | 0.063 GB | + +dense 票在**第一个 chunk 就饱和**；late 票攒不出 emit_start 之前的 10 个有效日，于是 +一路扩到声明 floor，**每个 chunk 重读一遍它自己的全部历史**。40 次正是 +`_saturation_chunk_calendar_days = warmup*2+40`（amp_cut 为 60 天）除 2015→2021 的 +2368 天所得（预测 39.5，实测 40.0）。 + +**这正是 D4b 买到的东西**：扩到 floor 的代价现在**只落在这 84 只票上**，而不是拖着 +另外 911 只一起走——旧规则下 995 票全都要付 floor 深度的加载，而那个加载**根本装不下** +（一至四节）。 + +## 6.5 per-symbol trim：不是「更保守的细化」，是在修 D4 的一个真缺陷（方向订正） + +trailing trim 的日历从「装载 universe 的日期并集」改成「该票自己的交易日」。**本文档 +先前把它记成一条保守的语义细化，并预警「若 D5 在稀疏票上对不上 D1 冻结基线，先看这里」 +——方向是反的**（反在安全侧）。跨引擎实测（同一稀疏 fixture，三个 bounded 因子，各动 +1 个 cell）： + +| 因子 | (a) per-symbol（现） | (b) union（D4） | (c) 几何无关真值 | 判定 | +|---|---|---|---|---| +| `jump_amount_corr_20` | **0.090115** | 0.157115 | **0.090115** | union **错** | +| `amp_marginal_anomaly_vol_20` | **0.199165** | **NaN** | **0.199165** | union **错**（少算一个 finite） | +| `minute_ideal_amp_10` | **−0.002685** | **NaN** | **−0.002685** | union **错**（同上） | + +三个因子**全部** (a) == (c) 且 (b) != (c)：union 规则不是更宽松，是**给出与加载几何有关的 +错值**（红线 #6）。且四个旧 runner 都是**逐票读全窗口、根本没有 union trim** ⇒ per-symbol +trim 是**朝 D1 冻结基线靠拢**，不是背离。 + +**因此正确的 D5 预警是**：稀疏票上若出现与冻结基线的差异，per-symbol trim 是**把它拉近** +的那一侧；真要排查该先看饱和深度与 anchor 截断（D4 已具名的归因项），不是这里。 + +⚠️ **步长是可以改的，但本步刻意没改**：D4 的 docstring 明写 chunk "only a STEP SIZE, +never a correctness input"，所以把它调大或改成几何增长是**纯性能**改动，能把这 40 次 +大幅压低。但它仍然是取值路径的加载几何改动，需要自己的逐格对账，**不能夹带进 D4b**。 +留给 D5 作为显式的性能项。 diff --git a/factors/compute/minute/binding.py b/factors/compute/minute/binding.py index da5fca7..169b397 100644 --- a/factors/compute/minute/binding.py +++ b/factors/compute/minute/binding.py @@ -10,6 +10,35 @@ pass (they pass them explicitly but equal to these defaults), so a materialized minute factor matches the frozen baseline's math. +TWO GRANULARITIES (D4b). ``minute_raw_from_bars`` is the WHOLE-factor call: bars +in, daily values out. It requires every symbol's bars in ONE frame, which does +not fit the evaluation plane (995 symbols x 5 years: measured 52.7 GB nominal, +~115 GB peak RSS vs 56 GB available — ``docs/factors/d5_saturation_feasibility.md``). +So the streaming materializer uses the SPLIT pair instead: + +* :func:`minute_stats_from_bars` — the PER-SYMBOL stage (memory-bounded: one + symbol's minute history in, its small daily intermediate out); +* :func:`combine_minute_stats` — the CROSS-SECTIONAL combine, run ONCE on the + assembled full-universe intermediate. + +The cut sits BEFORE the cross-section on purpose. Ten of the eleven minute +factors are per-symbol pure, so their intermediate IS their value and the combine +is the identity; ``intraday_amp_cut`` is not — its step 4 z-scores each date +across the covered universe and needs >= ``AMP_CUT_MIN_CROSS_SECTION`` (=10) +finite pairs, so a ONE-SYMBOL cross-section is all-NaN BY DEFINITION (measured: +1692/1692 rows NaN). Splitting around the whole factor would destroy it; splitting +before the combine reproduces it exactly. Its module already organises the math +that way (``compute_amp_cut_stats`` then ``combine_amp_cut_cross_section``), as do +the eval runners — this binding surfaces that existing seam rather than inventing +one. + +CONTRACT the streaming caller relies on: ``combine`` is a PER-DATE reduction (a +date's output depends only on that date's intermediate rows), so restricting the +intermediate to the emit window before combining cannot change a kept date's +value. Both combines satisfy it (identity; ``groupby(date)`` z-score). +``combine(per_symbol(bars)) == minute_raw_from_bars(bars)`` is pinned by test for +all eleven factors, so the two granularities cannot drift apart. + This lives in ``factors/compute/minute`` (the factor layer) rather than duplicated 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. @@ -25,6 +54,8 @@ from __future__ import annotations from collections.abc import Callable +from dataclasses import dataclass +from typing import NoReturn import pandas as pd @@ -35,6 +66,8 @@ ) from factors.compute.minute.intraday_amp_cut import ( IntradayAmpCutFactor, + combine_amp_cut_cross_section, + compute_amp_cut_stats, compute_intraday_amp_cut, ) from factors.compute.minute.jump_amount_corr import ( @@ -107,6 +140,95 @@ def _call(factor: Factor, bars: pd.DataFrame) -> pd.Series: PeakRidgeAmountRatioFactor: _bind(compute_peak_ridge_amount_ratio), } +# --------------------------------------------------------------------------- # +# Two-stage (streaming) binding: per-symbol stage + cross-sectional combine (D4b) +# --------------------------------------------------------------------------- # +#: Column name of the per-symbol intermediate for the factors whose per-symbol +#: stage ALREADY yields the final value (the combine is then the identity). +STATS_VALUE_COL = "value" + + +@dataclass(frozen=True) +class MinuteStreamBinding: + """The split of a minute factor into per-symbol work + one cross-section pass. + + ``per_symbol(factor, bars)`` -> ``MultiIndex(date, symbol)`` intermediate frame + for the ONE symbol carried by ``bars`` (it also accepts many symbols — it is + the same math, just not memory-bounded then). + + ``combine(factor, stats)`` -> the factor's daily raw Series from the assembled + full-universe intermediate. It MUST be a per-date reduction (see the module + docstring's CONTRACT), which is what lets the streaming materializer restrict + the intermediate to the emit window before combining. + """ + + per_symbol: Callable[[Factor, pd.DataFrame], pd.DataFrame] + combine: Callable[[Factor, pd.DataFrame], pd.Series] + + +def _pure_stream(compute_fn) -> MinuteStreamBinding: + """Stream binding for a PER-SYMBOL PURE factor: intermediate == value. + + Measured, not assumed: whole-universe vs per-symbol-concatenated agree to + 0.000e+00 on real bars for these (probe leg 4 / the D4b reconciliation). + """ + + def _per_symbol(factor: Factor, bars: pd.DataFrame) -> pd.DataFrame: + series = compute_fn( + bars, lookback_days=factor.lookback_days, name=factor.name # type: ignore[attr-defined] + ) + return series.to_frame(STATS_VALUE_COL) + + def _combine(factor: Factor, stats: pd.DataFrame) -> pd.Series: + if STATS_VALUE_COL not in stats.columns: + raise ValueError( + f"{factor.name}: the per-symbol intermediate must carry the " + f"'{STATS_VALUE_COL}' column; got {list(stats.columns)}." + ) + return stats[STATS_VALUE_COL].rename(factor.name).sort_index(kind="mergesort") + + return MinuteStreamBinding(per_symbol=_per_symbol, combine=_combine) + + +def _amp_cut_per_symbol(factor: Factor, bars: pd.DataFrame) -> pd.DataFrame: + """Steps 1-3 of ``intraday_amp_cut``: the ``(v_mean, v_std)`` intermediate.""" + return compute_amp_cut_stats( + bars, lookback_days=factor.lookback_days # type: ignore[attr-defined] + ) + + +def _amp_cut_combine(factor: Factor, stats: pd.DataFrame) -> pd.Series: + """Step 4 of ``intraday_amp_cut``: the per-date cross-sectional z-score. + + Run ONCE on the assembled universe — the ``min_cross_section`` gate (=10) is a + factor DEFINITION constant and is never relaxed to accommodate streaming. + """ + 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. +_MINUTE_STREAM_BINDINGS: dict[type[Factor], MinuteStreamBinding] = { + JumpAmountCorrFactor: _pure_stream(compute_jump_amount_corr), + MinuteIdealAmplitudeFactor: _pure_stream(compute_minute_ideal_amplitude), + AmpMarginalAnomalyVolFactor: _pure_stream(compute_amp_marginal_anomaly_vol), + VolumePeakCountFactor: _pure_stream(compute_volume_peak_count), + IntradayAmpCutFactor: MinuteStreamBinding( + per_symbol=_amp_cut_per_symbol, combine=_amp_cut_combine + ), + PeakIntervalKurtosisFactor: _pure_stream(compute_peak_interval_kurtosis), + ValleyRelativeVwapFactor: _pure_stream(compute_valley_relative_vwap), + ValleyRidgeVwapRatioFactor: _pure_stream(compute_valley_ridge_vwap_ratio), + RidgeMinuteReturnFactor: _pure_stream(compute_ridge_minute_return), + PeakRidgeAmountRatioFactor: _pure_stream(compute_peak_ridge_amount_ratio), +} + +#: 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: ( @@ -223,12 +345,28 @@ def is_minute_bound(factor: Factor) -> bool: def minute_raw_from_bars(factor: Factor, bars: pd.DataFrame) -> pd.Series: """Compute ``factor``'s raw daily Series from (cutoff-filtered) 1min ``bars``. + The WHOLE-factor granularity: every symbol's bars must be in ``bars``. Prefer + :func:`minute_stats_from_bars` + :func:`combine_minute_stats` when the universe + does not fit in memory (see the module docstring). + Readable error for a minute factor that is deferred (needs extra inputs) or a non-minute-bound factor — never a silent wrong result. """ binding = _MINUTE_BINDINGS.get(type(factor)) if binding is not None: return binding(factor, bars) + _raise_unbound(factor) + + +def _stream_binding(factor: Factor) -> MinuteStreamBinding: + binding = _MINUTE_STREAM_BINDINGS.get(type(factor)) + if binding is None: + _raise_unbound(factor) + return binding + + +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}") @@ -238,12 +376,45 @@ def minute_raw_from_bars(factor: Factor, bars: pd.DataFrame) -> pd.Series: ) +def minute_stats_from_bars(factor: Factor, bars: pd.DataFrame) -> pd.DataFrame: + """``factor``'s PER-SYMBOL intermediate from (cutoff-filtered) 1min ``bars``. + + Memory-bounded stage: ``bars`` normally carries ONE symbol, and the returned + ``MultiIndex(date, symbol)`` frame is daily-sized, so the caller can discard + the minute bars before reading the next symbol. For a per-symbol-pure factor + the frame is the value itself (column ``STATS_VALUE_COL``); for + ``intraday_amp_cut`` it is the pre-cross-section ``(v_mean, v_std)`` panel. + """ + return _stream_binding(factor).per_symbol(factor, bars) + + +def combine_minute_stats(factor: Factor, stats: pd.DataFrame) -> 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. + """ + return _stream_binding(factor).combine(factor, stats) + + +def is_cross_sectional_minute(factor: Factor) -> bool: + """True iff ``factor``'s combine is NOT the identity (a real cross-section).""" + return type(factor) in CROSS_SECTIONAL_MINUTE_FACTORS + + __all__ = [ + "CROSS_SECTIONAL_MINUTE_FACTORS", + "STATS_VALUE_COL", "VALID_DAY_POOLED_FACTORS", "BindingFn", + "MinuteStreamBinding", + "combine_minute_stats", + "is_cross_sectional_minute", "is_minute_bound", "is_valid_day_pooled", "minute_raw_from_bars", + "minute_stats_from_bars", "pooled_baseline_days", "pooled_lookback_days", ] diff --git a/factors/materialize.py b/factors/materialize.py index e7d8a12..8114466 100644 --- a/factors/materialize.py +++ b/factors/materialize.py @@ -22,6 +22,30 @@ path calls the ``factors.compute.minute.binding`` for the factor and the cutoff-filtered bars. The forward-return boundary is elsewhere: a factor value never sees a future return (invariant #1) — the materializer only reads history. + +MINUTE LOADING IS PER-SYMBOL STREAMING (D4b). One symbol's bars are read, +reduced to that symbol's daily intermediate, and discarded before the next; the +intermediates are assembled and the factor's cross-sectional combine runs ONCE. +The all-symbol minute panel is never materialized — measured on the evaluation +plane it is 52.7 GB nominal and ~115 GB peak RSS against 56 GB available, so the +previous single-frame load could not run at all, for POOLED and BOUNDED factors +alike (``docs/factors/d5_saturation_feasibility.md``). Two consequences are +deliberate and disclosed: + +* PER-SYMBOL TRIM: the trailing-trading-day trim and the pooled saturation + criterion now read THIS SYMBOL's trading days instead of the loaded universe's + union. Both are strictly more conservative (a symbol's own day list is a subset + of the union, so the same "N trailing days" reaches at least as far back), and + both remove a cross-symbol coupling — under the union rule another symbol's + calendar could decide how much history this symbol was warmed with, which is + itself a load-geometry dependence (red line #6). The factor math, its + parameters and its thresholds are untouched. +* The pooled cost D4 recorded ("any requested symbol that cannot accumulate + ``lookback_days`` valid days drags the WHOLE universe's load back to the + declared floor") is gone: each symbol expands to its own terminal. Measured on + the evaluation universe, 84 of 995 symbols list after the emit start and can + never satisfy the criterion, so under the old rule the floor load was + CONSTRUCTIVE for every pooled factor. """ from __future__ import annotations @@ -37,9 +61,11 @@ from data.clean.schema import DATE_LEVEL, SYMBOL_LEVEL from factors.base import Factor from factors.compute.minute.binding import ( + combine_minute_stats, is_minute_bound, is_valid_day_pooled, minute_raw_from_bars, + minute_stats_from_bars, pooled_baseline_days, pooled_lookback_days, ) @@ -163,6 +189,39 @@ def is_minute_factor(factor: Factor) -> bool: # --------------------------------------------------------------------------- # # Trailing-trading-day trim (the P8 correctness floor) # --------------------------------------------------------------------------- # +def _requested_universe(symbols) -> list[str]: + """The requested symbols as strings, DE-DUPLICATED, first occurrence order. + + The engine owns this the same way ``_symbol_bars`` owns "the provider honoured + its ``symbols`` argument" — a caller-side and a provider-side spelling of one + invariant: a (date, symbol) cell must be produced exactly once. + + Duplicates used to be absorbed by the providers, whose ``isin`` filter is + idempotent, so the single-frame engine returned one row per name whatever the + caller passed. Streaming iterates the list instead, so a repeat would be + materialized TWICE: duplicate index rows, and — for a factor with a real + cross-sectional combine — the repeated name entering its date's cross-section + twice, moving that date's mean and std for EVERY member. Measured on the + streaming test fixture with two names repeated: ``volume_peak_count_20`` gained + 8 duplicate rows with unchanged values, ``intraday_amp_cut_10`` gained 8 + duplicate rows AND changed all 48 shared cells. + + De-duplicating (rather than raising) is deliberate: it reproduces exactly what + the single-frame engine returned for the same call, which keeps D4b a change of + memory profile and nothing else. A raise would be a NEW failure mode for input + the previous engine accepted — a behaviour change in the opposite direction, + and one that belongs to whoever validates a universe, not to the value engine. + """ + seen: set[str] = set() + out: list[str] = [] + for s in symbols: + name = str(s) + if name not in seen: + seen.add(name) + out.append(name) + return out + + def _warmup_start(dates: pd.DatetimeIndex, emit_start: pd.Timestamp, warmup: int): """The date ``warmup`` trading days before ``emit_start`` (or the earliest). @@ -200,6 +259,7 @@ def materialize_range( rows. Single-date and batch calls give bit-identical values (design §3.5 P8). """ resolved_view = View(view) + symbols = _requested_universe(symbols) # one engine-level normalization emit_start = pd.Timestamp(emit_start).normalize() emit_end = pd.Timestamp(emit_end).normalize() w = int(factor.spec.lookback_depth) if warmup is None else int(warmup) @@ -242,6 +302,28 @@ def _materialize_daily( def _materialize_minute( factor, view, symbols, load_start, emit_start, emit_end, warmup, sources, decision_cutoff, ) -> pd.Series: + """STREAMING minute materialization (D4b): one symbol's bars at a time. + + Loads ONE symbol's minute history, reduces it to that symbol's small daily + intermediate, and DISCARDS the bars before the next symbol — the multi-year + all-symbol minute panel is never materialized. That is a structural + requirement, not an optimization: the evaluation plane's single frame is + 52.7 GB nominal / ~115 GB peak RSS against 56 GB of memory, so the + whole-universe load cannot run at all (``docs/factors/d5_saturation_feasibility.md``). + + THE CUT IS BEFORE THE CROSS-SECTION. The per-symbol intermediates are + assembled into the full-universe panel and the factor's combine runs ONCE on + it, so ``intraday_amp_cut``'s date-wise z-score still sees the whole covered + universe (a per-WHOLE-FACTOR split would hand it a one-symbol cross-section, + which its ``AMP_CUT_MIN_CROSS_SECTION`` gate turns entirely into NaN — + measured, and never worked around by relaxing that definition constant). + + 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 sources.minute is None: raise ValueError( f"{factor.name} is a minute factor but no MinuteBarProvider was injected." @@ -249,30 +331,86 @@ def _materialize_minute( if not is_minute_bound(factor): # Readable error (e.g. valley_price_quantile needs the daily panel too). return minute_raw_from_bars(factor, sources.minute.minute_bars([], load_start, emit_end)) - # Load one extra calendar day past emit_end so bar cutoffs on emit_end resolve. - # POOLED (valid-day trailing window): the pool counts VALID days, so its - # CALENDAR depth is data-dependent and UNBOUNDED — a fixed trim would make the - # value depend on load geometry (review HIGH / red line #6). Load by - # SATURATION-EXPANSION instead (below); bounded factors keep the fixed trim. - if is_valid_day_pooled(factor): - return _materialize_pooled( - factor, view, list(symbols), emit_start, emit_end, warmup, sources, decision_cutoff, - ) - bars = sources.minute.minute_bars( - list(symbols), load_start, emit_end + pd.Timedelta(days=1) + + pooled = is_valid_day_pooled(factor) + pooled_kw = {} + if pooled: + pooled_kw = { + "floor": _provider_earliest(sources.minute, list(symbols), factor), + "baseline_days": pooled_baseline_days(factor), + "lookback_days": pooled_lookback_days(factor), + } + + parts: list[pd.DataFrame] = [] + for symbol in symbols: + sym = str(symbol) + if pooled: + stats = _pooled_symbol_stats( + factor, view, sym, emit_start, emit_end, warmup, sources, + decision_cutoff, **pooled_kw, + ) + else: + stats = _bounded_symbol_stats( + factor, view, sym, load_start, emit_start, emit_end, warmup, sources, + decision_cutoff, + ) + if stats is None or stats.empty: + continue + # Restrict to the emit window BEFORE assembling: the combine is a per-date + # reduction (binding CONTRACT), so a kept date's value is unaffected, and + # the assembled panel stays daily-sized. + parts.append(_restrict_emit_frame(stats, emit_start, emit_end)) + + parts = [p for p in parts if not p.empty] + if not parts: + return _empty_series(factor.name) + return combine_minute_stats(factor, pd.concat(parts).sort_index(kind="mergesort")) + + +def _symbol_bars(provider, symbol: str, start, end) -> pd.DataFrame: + """One symbol's bars from ``provider``, RE-FILTERED to that symbol. + + The re-filter is not redundant belt-and-braces: streaming makes the engine's + correctness depend on the provider honouring its ``symbols`` argument, and a + provider that ignores it (returning the whole store) would silently emit every + other symbol's rows once PER STREAMED SYMBOL — duplicated index entries, an + inflated cross-section, and for a pooled factor a saturation decision taken on + the wrong symbol's valid days. The engine owns that invariant rather than + trusting an injected object with it. + """ + bars = provider.minute_bars([symbol], start, end) + if bars.empty: + return bars + mine = pd.Index(bars.index.get_level_values(SYMBOL_LEVEL)) == symbol + return bars if bool(mine.all()) else bars[mine] + + +def _bounded_symbol_stats( + factor, view, symbol, load_start, emit_start, emit_end, warmup, sources, decision_cutoff, +) -> pd.DataFrame | None: + """One BOUNDED minute factor's per-symbol intermediate over the fixed window. + + Load one extra calendar day past ``emit_end`` so bar cutoffs on ``emit_end`` + resolve, then apply the same trailing-trading-day trim as the single-frame + engine. The trim's day list is now THIS SYMBOL's trading days rather than the + loaded universe's union — see the module docstring's PER-SYMBOL TRIM note. + """ + bars = _symbol_bars( + sources.minute, symbol, load_start, emit_end + pd.Timedelta(days=1) ) if bars.empty: - return _empty_series(factor.name) + return None if view is View.DECISION: bars = minute_decision_cutoff(bars, decision_time=decision_cutoff) bars = _trim_minute(bars, emit_start, warmup) - return minute_raw_from_bars(factor, bars) + return minute_stats_from_bars(factor, bars) -def _materialize_pooled( - factor, view, symbols, emit_start, emit_end, warmup, sources, decision_cutoff, -) -> pd.Series: - """Saturation-expanding load for a valid-day-pooled factor (design §3.3, review HIGH). +def _pooled_symbol_stats( + factor, view, symbol, emit_start, emit_end, warmup, sources, decision_cutoff, + *, floor, baseline_days, lookback_days, +) -> pd.DataFrame | None: + """Saturation-expanding load for ONE symbol of a valid-day-pooled factor. STRUCTURAL saturation criterion (NOT a value-stability fixed point — that was disproved: one expansion chunk landing entirely inside a long no-bar gap @@ -303,36 +441,48 @@ def _materialize_pooled( is read off the engine's own result rather than a re-implementation of each factor's validity rule, and a gap simply contributes no output dates (the count stalls and the loop keeps expanding, which is the desired behaviour). + + PER SYMBOL, and why that is the SAME terminal (D4b, not a weakening): + + * The criterion's whole content is "further history cannot change any emit + value", and it is evaluated on the symbol's own valid days and its own + trailing pool. So a symbol that satisfies it at depth D has, BY THE + CRITERION, the same values it would have at any deeper load — including the + universe-wide floor the single-frame engine used to drag everyone to. + * The locked head is now dropped from THIS symbol's loaded days rather than + from the loaded UNION. The union is a superset, so its ``baseline_days``-th + day is at or before the symbol's — i.e. the per-symbol locked window is at + or inside the union one, and the criterion is at least as hard to satisfy. + Per-symbol therefore loads at least as deep; it never terminates earlier. + * The requested-symbol VETO (below) still applies with this symbol as the + requested symbol: a symbol with no output at the current depth counts zero + valid days and cannot terminate its own loop. What it can no longer do is + drag 994 other symbols to the floor with it, which is the entire cost D4's + docstring recorded as "a per-symbol saturation start is the D5 optimization". """ chunk = pd.Timedelta(days=_saturation_chunk_calendar_days(warmup)) end = emit_end + pd.Timedelta(days=1) - floor = _provider_earliest(sources.minute, symbols, factor) - baseline_days = pooled_baseline_days(factor) - lookback_days = pooled_lookback_days(factor) - load_start = emit_start - chunk - last: pd.Series | None = None for _ in range(_MAX_SATURATION_CHUNKS): at_floor = load_start <= floor if at_floor: load_start = floor - bars = sources.minute.minute_bars(symbols, load_start, end) + bars = _symbol_bars(sources.minute, symbol, load_start, end) if bars.empty: if at_floor: - return _empty_series(factor.name) + return None load_start = load_start - chunk continue work = bars if view is View.DECISION: work = minute_decision_cutoff(work, decision_time=decision_cutoff) - full = minute_raw_from_bars(factor, work) - last = _restrict_emit(full, emit_start, emit_end, factor.name) + stats = minute_stats_from_bars(factor, work) if at_floor or _pooled_pool_saturated( - full, work, emit_start, + combine_minute_stats(factor, stats), work, emit_start, baseline_days=baseline_days, lookback_days=lookback_days, - symbols=symbols, + symbols=[symbol], ): - return last + return stats load_start = load_start - chunk # Never silently return an UNSATURATED result (red line #9: no silent # degradation). Unreachable in practice — 500 chunks is >160 years of history @@ -340,7 +490,7 @@ def _materialize_pooled( raise RuntimeError( f"{factor.name}: pooled saturation did not converge after " f"{_MAX_SATURATION_CHUNKS} expansion chunks (expanded back to " - f"{load_start.date()} for {len(symbols)} symbol(s), declared floor " + f"{load_start.date()} for symbol {symbol}, declared floor " f"{floor.date()}). Refusing to return an unsaturated value, which would " f"depend on load geometry." ) @@ -389,15 +539,19 @@ def _pooled_pool_saturated( days vetoes termination, so the loop expands until it too is saturated or the declared floor is reached. - COST, stated plainly: any requested symbol that cannot accumulate - ``lookback_days`` valid days before ``emit_start`` (a recent listing, a long - suspension, a coverage hole) drags the WHOLE universe's load back to the - declared floor. That cost already existed for symbols with SOME output but too - few valid days; this fix only extends the same conservative behaviour to - zero-output symbols. A per-symbol saturation start is the D5 optimization — - and NOT a pure one: ``intraday_amp_cut`` z-scores across the loaded - cross-section, so changing per-symbol load depth changes its cross-section - composition. + D4b: ``symbols`` is now the ONE symbol whose load depth is being decided (the + caller streams). The veto is unchanged in meaning — a symbol with no output at + the current depth still counts zero valid days and still cannot terminate ITS + OWN expansion — and its blast radius shrinks: it no longer forces the other 994 + symbols to the floor. The failure mode the veto was added for (a name silently + missing from one fill and present in the other) is additionally foreclosed by + construction now, because a symbol's own load window always covers the whole + emit window, so a symbol that emits on an emit date always has bars there. + The COST D4 recorded (any thin symbol dragging the whole universe to the + declared floor) is therefore retired; see :func:`_pooled_symbol_stats` for why + the per-symbol terminal is the same terminal, and note that the + ``intraday_amp_cut`` coupling D4 flagged is handled by cutting BEFORE its + cross-sectional combine rather than around the whole factor. """ loaded_days = pd.DatetimeIndex( pd.unique(pd.DatetimeIndex(bars.index.get_level_values("time")).normalize()) @@ -459,6 +613,12 @@ def _restrict_emit(raw, emit_start, emit_end, name) -> pd.Series: return raw[within].sort_index(kind="mergesort").rename(name) +def _restrict_emit_frame(stats: pd.DataFrame, emit_start, emit_end) -> pd.DataFrame: + """Emit-window slice of a per-symbol intermediate (safe: per-date combine).""" + dates = stats.index.get_level_values(DATE_LEVEL) + return stats[(dates >= emit_start) & (dates <= emit_end)] + + def _empty_series(name: str) -> pd.Series: index = pd.MultiIndex.from_arrays( [pd.DatetimeIndex([]), pd.Index([], dtype=object)], diff --git a/qt/factor_hotpath_smoke.py b/qt/factor_hotpath_smoke.py index 4c7056f..f41eb6d 100644 --- a/qt/factor_hotpath_smoke.py +++ b/qt/factor_hotpath_smoke.py @@ -4,8 +4,8 @@ magnitude gap between: * NAIVE per-factor loading — each factor materialized separately, so the 1min - read + normalize is paid ONCE PER FACTOR (what the D4 service does today: each - ``materialize_range`` call loads its own bars); and + read + normalize is paid ONCE PER FACTOR (each ``materialize_range`` call still + loads its own bars); and * SHARED loading — the 1min read + normalize + cutoff paid ONCE, then every factor computed from the shared bars (the ``factors.compute.minute.primitives`` amortization the design §3.2 calls the load-bearing hot-path mechanism). @@ -14,6 +14,13 @@ factor count) and the two wall times are printed as-is, with an explicit caveat that the numbers do NOT extrapolate to full-A (~6x the CSI500 read). +CALIBER NOTE (D4b): BOTH arms here load the whole 40-symbol sample into one +frame, which is no longer the engine's geometry — the materializer streams per +symbol since D4b. What this smoke measures is unchanged and still the point (how +much a repeated 1min read + normalize costs relative to paying it once), but do +not read the NAIVE arm as a picture of the engine's memory profile; for that use +``qt.saturation_probe --mode stream-scale``, which measures the real path. + Run: ``python -m qt.factor_hotpath_smoke`` (deliberately NOT in qt/cli). It reads ``artifacts/cache/tushare/v1`` (symlink the main checkout's artifacts into the worktree first; remove the symlink after so ``git status`` stays clean). diff --git a/qt/saturation_probe.py b/qt/saturation_probe.py index 9034102..8ec2e93 100644 --- a/qt/saturation_probe.py +++ b/qt/saturation_probe.py @@ -1,5 +1,12 @@ """D5 feasibility probe: can the D4 materializer run the real evaluation universe? +MODE 1 (``--mode feasibility``, the default) MEASURES THE PRE-D4b ENGINE. Its +answer was no, and D4b acted on it: the materializer now streams per symbol, so +item 3 below describes a load geometry the engine NO LONGER USES. The mode is +kept because its four measurements are the "before" half of the comparison and +items 1, 2 and 4 are facts about the DATA, not about the engine — but read item 3 +as history, and use ``--mode stream-scale`` (below) to measure what runs today. + D4's materializer was accepted against a 40-symbol hot-path smoke. D5 needs it on the eleven-factor evaluation plane (CSI500, 995 symbols, 2021-07-01..2026-06-30), so before committing to a multi-hour run this probe MEASURES, on the real cache, @@ -12,8 +19,10 @@ to the declared floor; 2. **on-disk volume** at the evaluation window vs at the floor; 3. **in-memory cost** of a whole-universe single-frame load, measured on a sample - and extrapolated linearly (the materializer loads every symbol into ONE frame: - ``sources.minute.minute_bars(list(symbols), ...)``); + and extrapolated linearly. HISTORICAL as of D4b: the materializer used to load + every symbol into ONE frame (``sources.minute.minute_bars(list(symbols), ...)``) + and now loads one symbol at a time. This measurement is what that geometry + would have cost, i.e. the reason it was replaced — not what runs today; 4. **per-symbol equivalence** — whether materializing symbol-by-symbol and concatenating reproduces the whole-universe values exactly. This is the decisive question for the fix, and it is answered per factor rather than @@ -21,21 +30,39 @@ z-scores across the loaded cross-section, ``AMP_CUT_MIN_CROSS_SECTION=10``). Run: ``python -m qt.saturation_probe`` (cache-only; zero live calls). + +SECOND MODE — ``--mode stream-scale`` (D4b): the same instrument, now pointed at +the FIXED engine. It materializes ONE factor over the whole evaluation plane +through ``factors.materialize`` (which streams per symbol since D4b) and reports +the two numbers that must not be conflated: + +* NOMINAL — the largest single bar frame the engine ever held (``memory_usage``), + i.e. what the old single-frame load would have had to hold for ALL symbols; and +* PEAK RSS — the process high-water mark, which is what actually has to fit. The + feasibility mode measured frame:peakRSS at 2.2-2.45x, so reporting only the + nominal number would understate the requirement by about half. + +One factor per process (so ``ru_maxrss`` is that factor's own peak) and results +appended to ``--out`` as JSON lines, so a long sweep is resumable: re-running skips +factors already present unless ``--force``. """ from __future__ import annotations import argparse +import json import resource import time from pathlib import Path import pandas as pd +from data.availability_policy import View from data.cache.intraday_cache import ENDPOINT as INTRADAY_ENDPOINT from data.clean.intraday_schema import RAW_INTRADAY_FREQ from factors import registry as factor_registry from factors.compute.minute.binding import minute_raw_from_bars +from factors.materialize import MaterializeSources, materialize_range from qt.factor_hotpath_smoke import CACHE_MINUTE_DATA_START, CacheMinuteProvider DEFAULT_CACHE_ROOT = "artifacts/cache/tushare/v1" @@ -152,12 +179,227 @@ def per_symbol_equivalence( return rows +# --------------------------------------------------------------------------- # +# Mode 2 (D4b): does the FIXED engine run the whole evaluation plane? +# --------------------------------------------------------------------------- # +class MeasuringMinuteProvider(CacheMinuteProvider): + """Cache-only provider that records what the engine actually asked it for. + + The engine's memory profile is decided by the biggest frame it ever holds, so + that is measured here rather than inferred: ``max_symbols_per_call`` shows the + load is per-symbol at all, and ``max_frame_gb`` is the largest frame handed + back. Neither is the number that must fit — see ``peak_rss_gb``. + """ + + def __init__(self, root: str) -> None: + super().__init__(root) + self.max_symbols_per_call = 0 + self.max_frame_rows = 0 + self.max_frame_gb = 0.0 + + def minute_bars(self, symbols, start, end): + self.max_symbols_per_call = max(self.max_symbols_per_call, len(list(symbols))) + bars = super().minute_bars(symbols, start, end) + if len(bars): + self.max_frame_rows = max(self.max_frame_rows, len(bars)) + self.max_frame_gb = max( + self.max_frame_gb, float(bars.memory_usage(deep=True).sum()) / GB + ) + return bars + + +def stream_scale( + factor_id: str, symbols: list[str], cache_root: str, *, view: View = View.DECISION +) -> dict: + """Materialize ``factor_id`` over the whole evaluation plane; measure the cost.""" + factor = factor_registry.build(factor_id) + provider = MeasuringMinuteProvider(cache_root) + rss_before = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / (1024**2) + t0 = time.monotonic() + values = materialize_range( + factor, + view=view, + symbols=symbols, + emit_start=EVAL_START, + emit_end=EVAL_END, + sources=MaterializeSources(minute=provider), + ) + wall = time.monotonic() - t0 + finite = int(values.notna().sum()) + return { + "factor": factor_id, + "view": str(view.value), + "n_symbols": len(symbols), + "emit": f"{EVAL_START.date()}..{EVAL_END.date()}", + "rows": int(len(values)), + "finite": finite, + "symbols_with_rows": int( + pd.Index(values.index.get_level_values("symbol")).nunique() + ) if len(values) else 0, + "provider_calls": provider.calls, + "live_calls": provider.live_calls, + "max_symbols_per_call": provider.max_symbols_per_call, + "nominal_max_frame_gb": round(provider.max_frame_gb, 4), + "nominal_max_frame_rows": provider.max_frame_rows, + "peak_rss_gb": round(resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / (1024**2), 3), + "rss_before_gb": round(rss_before, 3), + "wall_seconds": round(wall, 1), + } + + +# --------------------------------------------------------------------------- # +# Mode 3 (D4b): cell-by-cell reconciliation against the geometry that was replaced +# --------------------------------------------------------------------------- # +#: Reconciliation caliber: small enough that the WHOLE-UNIVERSE SINGLE FRAME — +#: the geometry D4b replaced — still fits, which is the whole point. The +#: evaluation plane does not fit and never did; that is what mode 1 measured. +RECONCILE_SYMBOLS = 40 +RECONCILE_START = pd.Timestamp("2023-01-03") +RECONCILE_END = pd.Timestamp("2023-06-30") +#: Reference load depth in CALENDAR DAYS before the emit window. It matters: the +#: pooled factors' saturation expansion decides how deep the STREAMED side loads, +#: so a reference shallower than saturation would compare two different histories, +#: and one deeper changes the float-accumulation prefix. 400 days puts every +#: factor's trailing window at full depth on both sides. +RECONCILE_REF_DEPTH_DAYS = 400 + + +def reconcile_streaming_vs_single_frame( + cache_root: str, universe_panel: str, factor_ids: list[str] +) -> list[dict]: + """Streamed values vs the whole-universe single frame, cell for cell. + + The reference is deliberately the REPLACED geometry (every symbol in ONE + frame, one cutoff, one whole-factor call), so agreement is a statement about + two load geometries rather than two spellings of one loop. + """ + from factors.materialize import MaterializeSources, materialize_range + from factors.view_lag import minute_decision_cutoff + + universe = universe_from_frozen_panel(universe_panel)[:RECONCILE_SYMBOLS] + provider = CacheMinuteProvider(cache_root) + frame = provider.minute_bars( + universe, + RECONCILE_START - pd.Timedelta(days=RECONCILE_REF_DEPTH_DAYS), + RECONCILE_END + pd.Timedelta(days=1), + ) + print( + f"universe {len(universe)} symbols | emit {RECONCILE_START.date()}..{RECONCILE_END.date()}\n" + f"single-frame reference: {len(frame):,} rows, " + f"{float(frame.memory_usage(deep=True).sum()) / GB:.2f} GB, " + f"loaded {RECONCILE_REF_DEPTH_DAYS} calendar days deep\n" + ) + cut = minute_decision_cutoff(frame, decision_time="14:50:00") + + rows = [] + for factor_id in factor_ids: + factor = factor_registry.build(factor_id) + streamed = materialize_range( + factor, view=View.DECISION, symbols=universe, + emit_start=RECONCILE_START, emit_end=RECONCILE_END, + sources=MaterializeSources(minute=provider), + ).sort_index() + full = minute_raw_from_bars(factor, cut) + d = full.index.get_level_values("date") + ref = full[(d >= RECONCILE_START) & (d <= RECONCILE_END)].sort_index() + + entry: dict = {"factor": factor_id, "index_same": bool(streamed.index.equals(ref.index))} + if entry["index_same"]: + a, b = streamed.to_numpy(), ref.to_numpy() + both = ~pd.isna(a) & ~pd.isna(b) + entry["nan_set_diff"] = int((pd.isna(a) != pd.isna(b)).sum()) + entry["compared_cells"] = int(both.sum()) + entry["max_abs_diff"] = float(abs(a[both] - b[both]).max()) if both.any() else None + else: + entry.update(nan_set_diff=-1, compared_cells=0, max_abs_diff=None) + rows.append(entry) + rows.append({"factor": "__live_calls__", "live_calls": provider.live_calls}) + return rows + + +def run_reconcile(args) -> int: + factor_ids = args.factors or sorted(_MINUTE_STREAM_BINDINGS_IDS()) + rows = reconcile_streaming_vs_single_frame(args.cache_root, args.universe_panel, factor_ids) + hdr = f"{'factor':30s} {'idx==':>6s} {'cells':>8s} {'NaN diff':>9s} {'max|abs|':>12s}" + print(hdr) + print("-" * len(hdr)) + problems = [] + for r in rows: + if r["factor"] == "__live_calls__": + continue + if not r["index_same"] or r["nan_set_diff"]: + problems.append(r["factor"]) + shown = "n/a" if r["max_abs_diff"] is None else f"{r['max_abs_diff']:.3e}" + print( + f"{r['factor']:30s} {str(r['index_same']):>6s} {r['compared_cells']:>8,} " + f"{r['nan_set_diff']:>9d} {shown:>12s}" + ) + print(f"\nlive minute calls: {rows[-1]['live_calls']} (cache-only)") + print("PROBLEMS:", problems or "none") + return 0 if not problems else 1 + + +def _MINUTE_STREAM_BINDINGS_IDS() -> list[str]: + """Every bars-only minute factor id, derived from the binding table.""" + from factors.compute.minute.binding import _MINUTE_STREAM_BINDINGS + + return [cls().name for cls in _MINUTE_STREAM_BINDINGS] + + +def _done_factors(out_path: Path) -> set[str]: + if not out_path.exists(): + return set() + done = set() + for line in out_path.read_text().splitlines(): + line = line.strip() + if line: + done.add(json.loads(line)["factor"]) + return done + + +def run_stream_scale(args) -> int: + symbols = universe_from_frozen_panel(args.universe_panel) + out_path = Path(args.out) + out_path.parent.mkdir(parents=True, exist_ok=True) + done = set() if args.force else _done_factors(out_path) + print(f"universe: {len(symbols)} symbols | emit {EVAL_START.date()}..{EVAL_END.date()}") + print(f"already recorded: {sorted(done) or 'none'}") + for factor_id in args.factors: + if factor_id in done: + print(f"{factor_id}: SKIP (already in {out_path})") + continue + print(f"{factor_id}: running...", flush=True) + record = stream_scale(factor_id, symbols, args.cache_root) + with out_path.open("a") as fh: + fh.write(json.dumps(record) + "\n") + print( + f"{factor_id}: rows={record['rows']:,} finite={record['finite']:,} " + f"symbols={record['symbols_with_rows']} " + f"max_symbols_per_call={record['max_symbols_per_call']} " + f"NOMINAL max frame={record['nominal_max_frame_gb']:.4f} GB " + f"PEAK RSS={record['peak_rss_gb']:.3f} GB " + f"wall={record['wall_seconds']:.1f}s live_calls={record['live_calls']}", + flush=True, + ) + return 0 + + def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) parser.add_argument("--cache-root", default=DEFAULT_CACHE_ROOT) parser.add_argument("--universe-panel", default=DEFAULT_UNIVERSE_PANEL) parser.add_argument("--sample-symbols", type=int, default=20) + parser.add_argument( + "--mode", choices=("feasibility", "stream-scale", "reconcile"), default="feasibility" + ) + parser.add_argument("--factors", nargs="*", default=[]) + parser.add_argument("--out", default="stream_scale.jsonl") + parser.add_argument("--force", action="store_true") args = parser.parse_args(argv) + if args.mode == "stream-scale": + return run_stream_scale(args) + if args.mode == "reconcile": + return run_reconcile(args) symbols = universe_from_frozen_panel(args.universe_panel) symset = set(symbols) diff --git a/tests/test_factor_materialize_streaming.py b/tests/test_factor_materialize_streaming.py new file mode 100644 index 0000000..e29a2f1 --- /dev/null +++ b/tests/test_factor_materialize_streaming.py @@ -0,0 +1,799 @@ +"""D4b: per-symbol STREAMING minute materialization (design §九 D4b / A1). + +Network-free. D4's materializer loaded every symbol's minute bars into ONE frame, +which the evaluation plane cannot hold (measured: 52.7 GB nominal / ~115 GB peak +RSS vs 56 GB available — ``docs/factors/d5_saturation_feasibility.md``). D4b reads +one symbol at a time, reduces it to that symbol's daily intermediate, discards the +bars, and runs the factor's cross-sectional combine ONCE on the assembled panel. + +This file pins the properties that make that a value-preserving change: + +* the two granularities agree — ``combine(per_symbol(bars)) == minute_raw_from_bars`` + for every bound minute factor, so the split cannot drift from the whole-factor call; +* streaming reproduces the load-geometry-free truth (compute on the whole history in + ONE frame) cell-for-cell, INCLUDING ``intraday_amp_cut``, the one factor whose + combine is a real cross-section; +* the cut is BEFORE that cross-section: splitting around the WHOLE factor instead is + kept as a committed NEGATIVE CONTROL (it turns every value into NaN), and + ``AMP_CUT_MIN_CROSS_SECTION`` is asserted un-relaxed; +* memory-boundedness is structural (the engine never asks a provider for more than + one symbol) rather than a claim; +* per-symbol saturation reaches the same terminal per symbol, and a thin symbol no + longer drags a dense one's load depth; +* the engine re-filters what a provider returns instead of trusting it. + +Every invariance claim here has recorded mutation evidence in its docstring, and each +mutation was first asserted to actually change its target. +""" + +from __future__ import annotations + +import importlib +import inspect +import pathlib + +import numpy as np +import pandas as pd +import pytest + +from data.availability_policy import View +from data.clean.intraday_schema import normalize_intraday_bars +from factors import registry as factor_registry +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, + combine_minute_stats, + is_cross_sectional_minute, + is_valid_day_pooled, + minute_raw_from_bars, + minute_stats_from_bars, +) +from factors.compute.minute.intraday_amp_cut import ( + AMP_CUT_MIN_CROSS_SECTION, + IntradayAmpCutFactor, +) +from factors.materialize import MaterializeSources, materialize_range +from factors.view_lag import minute_decision_cutoff + +def _streamed_factor_ids() -> tuple[str, ...]: + """Every bars-only minute factor, DERIVED from the binding table. + + Hand-maintaining this list would mean a new minute factor could enter both + binding tables — satisfying the coverage test below — and still slip past all + the parametrized reconciliations without anything saying so. Nothing here may + depend on a person remembering to add a line. + + A default-constructed instance's ``name`` IS its registered factor id (the + window is part of the name by construction), so the ids come from the bound + classes themselves rather than from a copy of the registry's internals; the + round-trip through ``factor_registry.build`` below is what proves it. + """ + ids = [] + for cls in _MINUTE_STREAM_BINDINGS: + factor_id = cls().name + rebuilt = factor_registry.build(factor_id) + assert type(rebuilt) is cls, ( + f"{cls.__name__}() is named {factor_id!r} but that id builds a " + f"{type(rebuilt).__name__} — the binding table and the registry disagree" + ) + ids.append(factor_id) + 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). +STREAMED_FACTOR_IDS = _streamed_factor_ids() + +#: The BOUNDED minute factors (fixed trailing trim, no saturation expansion) — +#: also derived, so the trim reconciliation cannot silently miss one. +BOUNDED_FACTOR_IDS = tuple( + fid for fid in STREAMED_FACTOR_IDS + if not is_valid_day_pooled(factor_registry.build(fid)) +) + +#: >= AMP_CUT_MIN_CROSS_SECTION symbols, so intraday_amp_cut's date-wise gate can +#: pass and the reconciliation below is non-vacuous for it. +N_SYMBOLS = 12 +SYMBOLS = [f"6000{i:02d}.SH" for i in range(N_SYMBOLS)] +DATES = pd.bdate_range("2021-01-04", periods=48) +EMIT_START, EMIT_END = DATES[44], DATES[47] +BARS_PER_DAY = 238 + + +def _session(rows, rng, symbol, day, n_bars): + base = pd.Timestamp(day) + pd.Timedelta("09:31:00") + price = 100.0 + rng.normal(0, 2) + for i in range(n_bars): + 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, symbol, price, hi, lo, cl, vol, cl * vol)) + + +def _frame(rows) -> pd.DataFrame: + cols = ["time", "symbol", "open", "high", "low", "close", "volume", "amount"] + return normalize_intraday_bars(pd.DataFrame(rows, columns=cols), freq="1min") + + +def _dense_minute() -> pd.DataFrame: + rng = np.random.RandomState(31) + rows: list[tuple] = [] + for s in SYMBOLS: + for d in DATES: + _session(rows, rng, s, d, BARS_PER_DAY) + return _frame(rows) + + +DENSE = _dense_minute() + + +class StreamProv: + """Well-behaved provider: honours ``symbols``, records what it was asked for.""" + + def __init__(self, bars: pd.DataFrame = DENSE): + self._b = bars + self.symbol_counts: list[int] = [] + self.max_rows_served = 0 + + def minute_bars(self, symbols, start, end): + self.symbol_counts.append(len(list(symbols))) + if not symbols: + return self._b.iloc[0:0] + t = self._b.index.get_level_values("time") + sym = pd.Index(self._b.index.get_level_values("symbol")) + out = self._b[ + (t >= pd.Timestamp(start)) & (t <= pd.Timestamp(end)) & sym.isin([str(s) for s in symbols]) + ] + self.max_rows_served = max(self.max_rows_served, len(out)) + return out + + def earliest_available(self, symbols): + return DATES[0] + + +class SloppyProv(StreamProv): + """Contract-violating provider: IGNORES ``symbols`` and returns everything.""" + + def minute_bars(self, symbols, start, end): + self.symbol_counts.append(len(list(symbols))) + if not symbols: + return self._b.iloc[0:0] + t = self._b.index.get_level_values("time") + return self._b[(t >= pd.Timestamp(start)) & (t <= pd.Timestamp(end))] + + +def _truth(factor, bars: pd.DataFrame = DENSE, view=View.CLOSE) -> pd.Series: + """Load-geometry-free reference: the WHOLE history in ONE frame, one call. + + Independent of the streaming engine (it calls the factor's own whole-factor + binding), so agreeing with it is evidence, not a tautology. This is the same + ground truth ``test_locking_offset_is_load_bearing`` uses. + """ + work = bars if view is View.CLOSE else minute_decision_cutoff(bars, decision_time="14:50:00") + full = minute_raw_from_bars(factor, work) + d = full.index.get_level_values("date") + return full[(d >= EMIT_START) & (d <= EMIT_END)].sort_index(kind="mergesort") + + +def _streamed(factor, provider=None, view=View.CLOSE, symbols=None) -> pd.Series: + return materialize_range( + factor, + view=view, + symbols=list(symbols or SYMBOLS), + emit_start=EMIT_START, + emit_end=EMIT_END, + sources=MaterializeSources(minute=provider or StreamProv()), + decision_cutoff="14:50:00", + ) + + +#: Per-factor budget for the ONE attributable difference between the streamed +#: load and the whole-history load: pandas' rolling accumulation runs over a +#: differently-sized prefix, so a factor that accumulates can differ in the last +#: bits. MEASURED on this fixture: nine of the ten factors are EXACTLY equal +#: (max|abs| = 0.000e+00) and only ``jump_amount_corr_20`` moves, by max|abs| = +#: 5.551e-16 / max|rel| = 4.893e-14 — four orders of magnitude under the budget +#: below, so a real missing-row or wrong-window effect could not hide inside it +#: (that effect is O(1), cf. the 416.0-vs-438.0 locking-offset witness). The +#: exactness of the other nine is PINNED here on purpose: if a change turned one +#: of them from exact into merely-close, this table would have to be edited, and +#: that edit is the review signal. +_RECONCILE_ATOL: dict[str, float] = {"jump_amount_corr_20": 1e-12} + + +def _assert_bit_identical(got: pd.Series, want: pd.Series, label: str, atol: float = 0.0) -> int: + """Index and NaN mask must match EXACTLY; values within ``atol`` (default 0). + + The NaN mask is never given a tolerance: a finite<->NaN flip is the coverage + change this whole step must not make, and it is not a float artefact. + """ + assert got.index.equals(want.index), f"{label}: index differs" + g, w = got.to_numpy(), want.to_numpy() + assert np.array_equal(np.isnan(g), np.isnan(w)), f"{label}: NaN mask differs" + finite = ~np.isnan(g) + if atol == 0.0: + assert np.array_equal(g[finite], w[finite]), f"{label}: finite values differ" + else: + assert np.allclose(g[finite], w[finite], rtol=0.0, atol=atol), ( + f"{label}: finite values differ by more than the float-reorder budget" + ) + return int(finite.sum()) + + +# --------------------------------------------------------------------------- # +# 1. The two granularities cannot drift apart +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize("factor_id", STREAMED_FACTOR_IDS) +def test_split_stages_reproduce_the_whole_factor_call(factor_id): + """``combine(per_symbol(bars)) == minute_raw_from_bars(bars)`` on the SAME bars. + + This is the structural link between the streaming engine and the whole-factor + call the eval runners / panel freeze use: if a future edit changed one stage + without the other, the two granularities would disagree here. + """ + factor = factor_registry.build(factor_id) + want = minute_raw_from_bars(factor, DENSE) + got = combine_minute_stats(factor, minute_stats_from_bars(factor, DENSE)) + n = _assert_bit_identical(got.sort_index(), want.sort_index(), 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) + 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")) + + +def test_every_minute_factor_class_is_classified_by_a_binding_table(): + """A minute factor missing from BOTH tables must fail here, not go untested. + + Deriving the parametrize lists from the binding tables (above) fixed one + direction only: a factor ADDED to a table is automatically reconciled. The + other direction stayed open — a factor that is in NEITHER table simply + disappears from every parametrized reconciliation, and the suite goes green + with fewer tests. Review measured exactly that: dropping one factor from + both tables left this file at 53 passed and the full suite at 2169 passed, + with nothing red. + + So the source of truth here is NOT a table: it is the set of ``Factor`` + subclasses DEFINED under ``factors/compute/minute`` (one factor per file, + §3.2). Adding a factor file without classifying it fails this assertion. + Values are never silently wrong either way — an unbound factor raises via + ``_raise_unbound`` — but the coverage loss is silent, which is what this + closes (§六.8: nothing may depend on a person remembering). + """ + package = pathlib.Path(binding_module.__file__).parent + defined: dict[type, str] = {} + for path in sorted(package.glob("*.py")): + if path.name == "__init__.py": + continue + module = importlib.import_module(f"factors.compute.minute.{path.stem}") + for obj in vars(module).values(): + if ( + inspect.isclass(obj) + and issubclass(obj, Factor) + and obj is not Factor + and obj.__module__ == module.__name__ + ): + defined[obj] = path.name + + classified = set(_MINUTE_STREAM_BINDINGS) | set(_DEFERRED) + 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"{unclassified} — they would vanish from every parametrized " + f"reconciliation without a single test going red" + ) + stale = {cls.__name__ for cls in classified if cls not in defined} + assert not stale, ( + f"binding table(s) classify {stale}, which no longer live under " + f"factors/compute/minute — the tables outlived the code" + ) + + +def test_deferred_factor_still_raises_readably_through_the_split_stages(): + """valley_price_quantile is deferred: both stages must say so, not mis-compute.""" + 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()) + + +# --------------------------------------------------------------------------- # +# 2. Cell-by-cell reconciliation: streaming == whole-universe single frame +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize("factor_id", STREAMED_FACTOR_IDS) +def test_streaming_reproduces_the_whole_universe_value(factor_id): + """Per-symbol streaming == the one-frame, whole-history truth, cell for cell. + + Index and NaN mask exact; values exact for nine of the ten factors and within + the measured float-reorder budget for the tenth (see ``_RECONCILE_ATOL``). + + MUTATION (run, rc=1): dropping the per-symbol re-filter in + ``factors.materialize._symbol_bars`` while a sloppy provider is in play makes + every symbol see the whole universe -> duplicated index entries; asserted + separately in ``test_sloppy_provider_is_re_filtered``. + """ + factor = factor_registry.build(factor_id) + n = _assert_bit_identical( + _streamed(factor), _truth(factor), factor_id, atol=_RECONCILE_ATOL.get(factor_id, 0.0) + ) + assert n > 0, f"{factor_id}: vacuous (no finite values in the emit window)" + + +@pytest.mark.parametrize("factor_id", ["volume_peak_count_20", "intraday_amp_cut_10"]) +def test_streaming_reproduces_the_whole_universe_value_decision_view(factor_id): + """Same, through the DECISION view — the 14:50 cutoff is applied per symbol.""" + factor = factor_registry.build(factor_id) + n = _assert_bit_identical( + _streamed(factor, view=View.DECISION), + _truth(factor, view=View.DECISION), + factor_id, + ) + assert n > 0, f"{factor_id}: vacuous" + + +# --------------------------------------------------------------------------- # +# 3. The cut must sit BEFORE the cross-section (the amp_cut coupling) +# --------------------------------------------------------------------------- # +def test_splitting_around_the_whole_factor_destroys_the_cross_sectional_factor(): + """NEGATIVE CONTROL, committed: the WRONG cut turns intraday_amp_cut all-NaN. + + Splitting per symbol around the whole factor hands its step-4 z-score a + one-symbol cross-section, which is below ``AMP_CUT_MIN_CROSS_SECTION`` (=10) + by definition -> every value NaN while the index still looks right. The probe + measured exactly this on real bars (1692/1692 rows NaN, 0 comparable cells). + Keeping it as a test is what stops a future "simplification" from moving the + cut around the factor and shipping a silently NaN column. + """ + factor = factor_registry.build("intraday_amp_cut_10") + per_factor_parts = [] + for s in SYMBOLS: + one = DENSE[pd.Index(DENSE.index.get_level_values("symbol")) == s] + per_factor_parts.append(minute_raw_from_bars(factor, one)) # the WRONG cut + wrong = pd.concat(per_factor_parts).sort_index() + right = _truth(factor) + + d = wrong.index.get_level_values("date") + wrong_win = wrong[(d >= EMIT_START) & (d <= EMIT_END)].sort_index() + assert wrong_win.index.equals(right.index), "the wrong cut keeps the index — that is the trap" + assert int(wrong_win.notna().sum()) == 0, "the wrong cut must be entirely NaN" + assert int(right.notna().sum()) > 0, "vacuous: the right cut produced nothing to lose" + # ...and the streaming engine takes the RIGHT cut. + _assert_bit_identical(_streamed(factor), right, "intraday_amp_cut streamed") + + +@pytest.mark.parametrize("factor_id", STREAMED_FACTOR_IDS) +def test_cross_sectional_classification_matches_the_factors_actual_behaviour(factor_id): + """``CROSS_SECTIONAL_MINUTE_FACTORS`` is a CHECKED fact, not a declaration. + + A factor is per-symbol pure iff splitting around the WHOLE factor reproduces + the whole-universe values. This asserts BOTH directions on every bound factor: + the nine declared pure ones must survive that split bit-identically, and the + one declared cross-sectional must not. A factor that acquired a cross-section + without being added to the set would be caught here — the alternative is a + comment that is true on the day it is written. + """ + factor = factor_registry.build(factor_id) + parts = [] + for s in SYMBOLS: + one = DENSE[pd.Index(DENSE.index.get_level_values("symbol")) == s] + parts.append(minute_raw_from_bars(factor, one)) + split = pd.concat(parts).sort_index() + whole = minute_raw_from_bars(factor, DENSE).sort_index() + + assert split.index.equals(whole.index), f"{factor_id}: per-symbol split changed the index" + sv, wv = split.to_numpy(), whole.to_numpy() + same = np.array_equal(np.isnan(sv), np.isnan(wv)) and np.array_equal( + sv[~np.isnan(sv)], wv[~np.isnan(wv)] + ) + assert int((~np.isnan(wv)).sum()) > 0, f"{factor_id}: vacuous (whole-universe all NaN)" + if is_cross_sectional_minute(factor): + assert not same, ( + f"{factor_id} is declared cross-sectional but survives a per-whole-factor " + f"split — either the declaration or the factor is wrong" + ) + else: + assert same, ( + f"{factor_id} is declared per-symbol pure but a per-whole-factor split " + f"changes its values — it needs a real combine and a place in " + f"CROSS_SECTIONAL_MINUTE_FACTORS" + ) + + +def test_amp_cut_cross_section_gate_is_not_relaxed_for_streaming(): + """The definition constant stays at 10 and still bites BELOW it (honest NaN). + + Streaming would be trivially "fixed" by widening or dropping this gate; that + would be a definition change. With 9 symbols requested the date's finite pair + count is below the gate and every value is NaN — through the streaming path. + """ + assert AMP_CUT_MIN_CROSS_SECTION == 10 + factor = factor_registry.build("intraday_amp_cut_10") + thin = _streamed(factor, symbols=SYMBOLS[:9]) + assert len(thin) > 0, "vacuous: no rows emitted at all" + assert int(thin.notna().sum()) == 0, "a sub-threshold cross-section must be all NaN" + # the same universe one symbol larger clears the gate (so the gate, not the + # plumbing, is what produced the NaNs above) + wide = _streamed(factor, symbols=SYMBOLS[:10]) + assert int(wide.notna().sum()) > 0 + + +# --------------------------------------------------------------------------- # +# 4. Memory-boundedness is structural, not a claim +# --------------------------------------------------------------------------- # +@pytest.mark.parametrize("factor_id", STREAMED_FACTOR_IDS) +def test_provider_is_never_asked_for_more_than_one_symbol(factor_id): + """The engine asks for ONE symbol per call — the whole-universe frame is never + requested, so it can never be built. + + MUTATION (run, rc=1): restoring the whole-universe request + (``minute_bars(list(symbols), ...)``) makes the recorded call carry 12 symbols + and this fails; restored -> rc=0. + """ + prov = StreamProv() + _streamed(factor_registry.build(factor_id), provider=prov) + assert prov.symbol_counts, "vacuous: the provider was never called" + assert max(prov.symbol_counts) == 1, ( + f"{factor_id}: engine requested {max(prov.symbol_counts)} symbols in one call" + ) + # and the biggest frame it ever handled is one symbol's worth of bars + assert prov.max_rows_served <= len(DATES) * BARS_PER_DAY + + +@pytest.mark.parametrize("factor_id", ["volume_peak_count_20", "intraday_amp_cut_10"]) +def test_a_repeated_symbol_in_the_request_changes_nothing(factor_id): + """The CALLER-SIDE half of the "produced exactly once" invariant. + + ``_symbol_bars`` owns "the provider honoured its ``symbols`` argument"; this + owns "the caller did not repeat a name". The single-frame engine was immune + because the providers' ``isin`` filter is idempotent, so a repeat cost nothing; + streaming iterates the list and would materialize the name twice. + + Measured BEFORE the fix, on this fixture with two names repeated: + ``volume_peak_count_20`` 48 -> 56 rows with 8 duplicated (date, symbol) entries + and values unchanged; ``intraday_amp_cut_10`` the same 8 duplicates AND ALL 48 + shared cells changed, because the repeated name entered its date's + cross-section twice and moved that date's mean and std for every member. + + MUTATION (run, rc=1): removing the ``_requested_universe`` call in + ``materialize_range`` restores both symptoms; restored -> rc=0. The probe that + proves the mutation bit is the duplicate row count going 0 -> 8. + """ + factor = factor_registry.build(factor_id) + clean = _streamed(factor, symbols=SYMBOLS) + repeated = _streamed(factor, symbols=[*SYMBOLS, SYMBOLS[0], SYMBOLS[3]]) + assert not repeated.index.duplicated().any(), "a repeated request produced duplicate rows" + n = _assert_bit_identical(repeated, clean, f"{factor_id} with a repeated symbol") + assert n > 0, "vacuous: no finite values" + + +def test_sloppy_provider_is_re_filtered(): + """A provider that ignores ``symbols`` must not corrupt the streamed result. + + Streaming makes the engine depend on the provider honouring its argument; a + provider returning the whole store would emit every other symbol's rows once + per streamed symbol. The engine re-filters, so the sloppy provider produces + the SAME values as the well-behaved one. + + MUTATION (run, rc=1): removing the re-filter in ``_symbol_bars`` makes this + raise on duplicate index entries / mismatched length; restored -> rc=0. + """ + factor = factor_registry.build("volume_peak_count_20") + clean = _streamed(factor, provider=StreamProv()) + sloppy = _streamed(factor, provider=SloppyProv()) + n = _assert_bit_identical(sloppy, clean, "sloppy provider") + assert n > 0 + assert not sloppy.index.duplicated().any(), "duplicate (date, symbol) rows" + + +# --------------------------------------------------------------------------- # +# 5. Per-symbol saturation: same terminal, no cross-symbol drag +# --------------------------------------------------------------------------- # +#: A thin symbol that can NEVER accumulate ``lookback_days`` valid days before the +#: emit window (it lists inside it) — the shape that, under the universe-wide +#: criterion, dragged every other symbol's load back to the declared floor (84 of +#: the evaluation universe's 995 symbols are of this kind). +_LATE_SYMBOL = "600099.SH" + + +def _dense_plus_late() -> pd.DataFrame: + rng = np.random.RandomState(77) + rows: list[tuple] = [] + for d in DATES[-6:]: # lists only just before the emit window + _session(rows, rng, _LATE_SYMBOL, d, BARS_PER_DAY) + return pd.concat([DENSE, _frame(rows)]).sort_index(kind="mergesort") + + +DENSE_PLUS_LATE = _dense_plus_late() + + +def test_a_late_listing_symbol_does_not_change_the_others_values(): + """Per-symbol saturation: each symbol expands to ITS OWN terminal. + + Adding a symbol that can never satisfy the criterion leaves every other + symbol's value bit-identical (they are decided on their own history), which is + the whole point of moving the decision per symbol. Under the universe-wide + criterion this symbol forced the entire load to the declared floor. + + Note the values compared are the per-symbol-pure ones; ``intraday_amp_cut`` + has its own test below because a new symbol legitimately joins its + cross-section. + """ + factor = factor_registry.build("volume_peak_count_20") + without = _streamed(factor, provider=StreamProv()) + with_late = _streamed( + factor, provider=StreamProv(DENSE_PLUS_LATE), symbols=[*SYMBOLS, _LATE_SYMBOL] + ) + shared = with_late[ + pd.Index(with_late.index.get_level_values("symbol")).isin(SYMBOLS) + ] + n = _assert_bit_identical(shared, without, "late-listing neighbour") + assert n > 0 + + +#: A LONG, NARROW fixture: long enough that a dense symbol's first expansion +#: chunk (``lookback_depth*2 + 40`` = 120 calendar days at depth 40) stops well +#: short of the declared floor, so "saturated early" and "loaded to the floor" are +#: DISTINGUISHABLE. On the short fixture above both coincide, which would let this +#: test pass with the old universe-wide drag still in place. +LONG_DATES = pd.bdate_range("2020-01-01", periods=200) +LONG_DENSE = ["600200.SH", "600201.SH"] +LONG_EMIT_START, LONG_EMIT_END = LONG_DATES[196], LONG_DATES[199] + + +def _long_minute() -> pd.DataFrame: + rng = np.random.RandomState(4242) + rows: list[tuple] = [] + for s in LONG_DENSE: + for d in LONG_DATES: + _session(rows, rng, s, d, BARS_PER_DAY) + for d in LONG_DATES[-6:]: # the late listing: inside the emit neighbourhood + _session(rows, rng, _LATE_SYMBOL, d, BARS_PER_DAY) + return _frame(rows) + + +LONG_MINUTE = _long_minute() + + +class LongProv(StreamProv): + def __init__(self): + super().__init__(LONG_MINUTE) + + def earliest_available(self, symbols): + return LONG_DATES[0] + + +def test_a_late_listing_symbol_still_reaches_its_own_floor(): + """...and the thin symbol is NOT dropped: it is loaded to its declared floor, + while its dense neighbours stop at their own (much shallower) saturation. + + The zero-output veto lives on per symbol — what changed is only that it no + longer vetoes on behalf of the other 995. Under the universe-wide criterion + every symbol here would share the thin symbol's floor-deep load start. + + MUTATION (run, rc=1): forcing the criterion to False (never saturate) drags + the dense symbols to the floor too and ``late_deepest < dense_deepest`` fails + — i.e. this assertion is what distinguishes per-symbol saturation from the + universe-wide drag; see ``test_universe_wide_drag_mutation``. + """ + factor = factor_registry.build("volume_peak_count_20") + starts: dict[str, list[pd.Timestamp]] = {} + + class RecordingProv(LongProv): + def minute_bars(self, symbols, start, end): + if symbols: + starts.setdefault(str(list(symbols)[0]), []).append(pd.Timestamp(start)) + return super().minute_bars(symbols, start, end) + + materialize_range( + factor, view=View.CLOSE, symbols=[*LONG_DENSE, _LATE_SYMBOL], + emit_start=LONG_EMIT_START, emit_end=LONG_EMIT_END, + sources=MaterializeSources(minute=RecordingProv()), + ) + assert _LATE_SYMBOL in starts, "the thin symbol was never loaded (silently dropped)" + late_deepest = min(starts[_LATE_SYMBOL]) + dense_deepest = min(min(v) for k, v in starts.items() if k != _LATE_SYMBOL) + assert late_deepest <= LONG_DATES[0], "the thin symbol must expand to its declared floor" + assert late_deepest < dense_deepest, ( + "the thin symbol must go DEEPER than its dense neighbours — otherwise this " + "test would pass with the old universe-wide drag still in place" + ) + + +def test_universe_wide_drag_mutation(monkeypatch): + """MUTATION for the test above: with saturation disabled every symbol is + dragged to the declared floor, so the depth contrast vanishes. + + The mutation is asserted to actually bite first (the dense symbols' deepest + load start MOVES), which is what makes the contrast above evidence rather + than a coincidence of the fixture. + """ + import factors.materialize as mat + + factor = factor_registry.build("volume_peak_count_20") + + def record() -> dict[str, list[pd.Timestamp]]: + starts: dict[str, list[pd.Timestamp]] = {} + + class RecordingProv(LongProv): + def minute_bars(self, symbols, start, end): + if symbols: + starts.setdefault(str(list(symbols)[0]), []).append(pd.Timestamp(start)) + return super().minute_bars(symbols, start, end) + + materialize_range( + factor, view=View.CLOSE, symbols=[*LONG_DENSE, _LATE_SYMBOL], + emit_start=LONG_EMIT_START, emit_end=LONG_EMIT_END, + sources=MaterializeSources(minute=RecordingProv()), + ) + return starts + + clean = record() + monkeypatch.setattr(mat, "_pooled_pool_saturated", lambda *a, **k: False) + mutated = record() + + clean_dense = min(min(clean[s]) for s in LONG_DENSE) + mutated_dense = min(min(mutated[s]) for s in LONG_DENSE) + assert mutated_dense < clean_dense, "the mutation did not change its target" + assert mutated_dense <= LONG_DATES[0], "the mutation should drag the dense names to the floor" + + +def test_an_early_saturating_symbol_matches_the_whole_history_value(): + """The soundness claim of per-symbol saturation, on a fixture where it BITES. + + A dense symbol here stops at its first expansion chunk — far short of the + declared floor — so this compares a genuinely EARLY-terminated load against + computing on the symbol's whole history. That is the whole content of the + criterion: "further history cannot change any emit value". + + The early termination is asserted HERE, not borrowed from the test above: if + the fixture ever grew short enough that the first chunk already reached the + floor, both sides would load the same bars and this would agree for the wrong + reason — a test that cannot fail. + """ + factor = factor_registry.build("volume_peak_count_20") + starts: list[pd.Timestamp] = [] + + class RecordingProv(LongProv): + def minute_bars(self, symbols, start, end): + starts.append(pd.Timestamp(start)) + return super().minute_bars(symbols, start, end) + + got = materialize_range( + factor, view=View.CLOSE, symbols=list(LONG_DENSE), + emit_start=LONG_EMIT_START, emit_end=LONG_EMIT_END, + sources=MaterializeSources(minute=RecordingProv()), + ) + assert min(starts) > LONG_DATES[0], ( + "the dense symbols reached the declared floor, so 'early saturation' is not " + "being exercised and this comparison would be vacuous" + ) + full = minute_raw_from_bars(factor, LONG_MINUTE) + d = full.index.get_level_values("date") + want = full[ + (d >= LONG_EMIT_START) & (d <= LONG_EMIT_END) + & pd.Index(full.index.get_level_values("symbol")).isin(LONG_DENSE) + ].sort_index(kind="mergesort") + n = _assert_bit_identical(got, want, "early-saturating symbol") + assert n > 0, "vacuous: no finite values" + + +def test_cross_sectional_factor_sees_the_whole_universe_not_the_streamed_symbol(): + """intraday_amp_cut's combine runs on the ASSEMBLED panel: a new symbol joins + the cross-section and moves the z-scores, exactly as in the one-frame path. + + The move is asserted, not assumed. If the extra member left every existing + value untouched, agreeing with the one-frame reference would prove nothing + about WHICH cross-section the combine saw — the streamed values would be + consistent with a per-symbol cross-section too. + """ + factor = factor_registry.build("intraday_amp_cut_10") + got = _streamed( + factor, provider=StreamProv(DENSE_PLUS_LATE), symbols=[*SYMBOLS, _LATE_SYMBOL] + ) + want = _truth(factor, bars=DENSE_PLUS_LATE) + n = _assert_bit_identical(got, want, "amp_cut with a late-listing member") + assert n > 0 + + without = _streamed(factor) # the same 12 symbols, no new member + shared = got[pd.Index(got.index.get_level_values("symbol")).isin(SYMBOLS)] + common = without.index.intersection(shared.index) + assert len(common) > 0, "vacuous: the two runs share no cell" + moved = int((shared.loc[common].to_numpy() != without.loc[common].to_numpy()).sum()) + assert moved > 0, ( + "the extra cross-section member changed nothing, so this test cannot tell " + "a whole-universe combine from a per-symbol one" + ) + + +# --------------------------------------------------------------------------- # +# 6. The per-symbol trailing trim (disclosed semantic refinement) +# --------------------------------------------------------------------------- # +_SPARSE_SYMBOL = "600098.SH" + + +def _dense_plus_sparse() -> pd.DataFrame: + """One symbol trading only every third day: its own trailing N trading days + reach further back than the loaded universe's union of N days.""" + rng = np.random.RandomState(99) + rows: list[tuple] = [] + for i, d in enumerate(DATES): + if i % 3 == 0: + _session(rows, rng, _SPARSE_SYMBOL, d, BARS_PER_DAY) + return pd.concat([DENSE, _frame(rows)]).sort_index(kind="mergesort") + + +DENSE_PLUS_SPARSE = _dense_plus_sparse() + + +@pytest.mark.parametrize("factor_id", BOUNDED_FACTOR_IDS) +def test_per_symbol_trim_uses_the_symbols_own_trading_days(factor_id): + """The trailing trim counts THIS symbol's trading days — and that is a FIX. + + Under the union rule a densely-traded neighbour's calendar decided how much + history a sparse symbol was warmed with, which is itself a load-geometry + dependence (red line #6). Measured across engines on this fixture, all three + bounded factors move exactly one cell — the sparse symbol's — and in all three + the per-symbol result is the load-geometry-free truth while the union result is + NOT: ``jump_amount_corr_20`` 0.1571 (union) vs 0.0901 (truth = per-symbol), and + the other two flip NaN (union) -> finite (per-symbol). So this is not a + conservative refinement of a correct rule, it corrects a wrong one. + + ALL THREE bounded factors are covered, derived from the binding table: two of + the three express the disagreement as a coverage FLIP, which is precisely what + ``_assert_bit_identical`` singles out as the change this step must not make by + accident — testing only the value-moving one would have missed the two that + matter most. + + Both halves are pinned so neither can go vacuous: + + 1. the two rules really disagree on this fixture (the union's ``warmup``-th day + back is LATER than the sparse symbol's own, so the union rule keeps LESS of + its history); + 2. the streamed value equals the load-geometry-free truth. + """ + from factors.materialize import _warmup_start + + factor = factor_registry.build(factor_id) + warmup = int(factor.spec.lookback_depth) + + # (1) the rules disagree — computed from the fixture, not asserted by fiat. + all_days = pd.DatetimeIndex( + pd.unique(pd.DatetimeIndex(DENSE_PLUS_SPARSE.index.get_level_values("time")).normalize()) + ) + sym_level = pd.Index(DENSE_PLUS_SPARSE.index.get_level_values("symbol")) + sparse_bars = DENSE_PLUS_SPARSE[sym_level == _SPARSE_SYMBOL] + own_days = pd.DatetimeIndex( + pd.unique(pd.DatetimeIndex(sparse_bars.index.get_level_values("time")).normalize()) + ) + union_keep = _warmup_start(all_days, EMIT_START, warmup) + own_keep = _warmup_start(own_days, EMIT_START, warmup) + assert own_keep < union_keep, ( + "vacuous fixture: the union rule and the per-symbol rule pick the same " + "cutoff here, so this test would not exercise the refinement" + ) + + # (2) the streamed value is the load-geometry-free one. + got = _streamed( + factor, provider=StreamProv(DENSE_PLUS_SPARSE), symbols=[*SYMBOLS, _SPARSE_SYMBOL] + ) + want = _truth(factor, bars=DENSE_PLUS_SPARSE) + mine = pd.Index(got.index.get_level_values("symbol")) == _SPARSE_SYMBOL + assert int(mine.sum()) > 0, "vacuous: the sparse symbol emitted nothing" + _assert_bit_identical( + got, want, "per-symbol trim", atol=_RECONCILE_ATOL["jump_amount_corr_20"] + ) diff --git a/tests/test_factor_service.py b/tests/test_factor_service.py index 5c76102..32b393c 100644 --- a/tests/test_factor_service.py +++ b/tests/test_factor_service.py @@ -461,19 +461,44 @@ def test_zero_output_symbol_is_not_silently_dropped(): def test_zero_output_symbol_veto_mutation(monkeypatch): - """MUTATION for the HIGH: restricting the criterion to OUTPUT symbols (the + """MUTATION for the HIGH: letting a ZERO-OUTPUT symbol count as saturated (the pre-fix behaviour) reintroduces the silent drop -> this test's assertion of - the drop proves the requested-symbol iteration is load-bearing.""" + the drop proves the zero-output veto is load-bearing. + + THE MUTATION'S EXPRESSION CHANGED WITH D4b; THE DEFECT IT ENCODES DID NOT. + It used to be written as "restrict the criterion's symbol list to the symbols + present in the output", which was the faithful shape while ONE criterion call + decided the load depth for the WHOLE universe: the thin name was absent from + that shared output, so filtering the list removed its veto. D4b decides depth + PER SYMBOL, so the list holds exactly one name and the filtered-list form + became INERT — measured on this fixture: 72 criterion calls (70 from the + per-date fill, 2 from the batch fill), the real and filtered forms answer + identically in 72 of 72, because all 20 calls with zero output also have + ``loaded_days <= baseline_days`` and are already vetoed by the guard above the + symbol loop. + + STATED PRECISELY, because the distinction matters: an inert mutation makes the + assertion below — which asserts that the drop HAPPENS — impossible to satisfy. + So the old expression did not become a test that cannot fail; it became a test + that CANNOT PASS, and it went red on the D4b engine. It was not deleted or + weakened to get green (§六.16): the property it protects still holds and is + still asserted, with the same fixture, the same factor and the same + assertions; only the mutation's wording moved to a form that still bites under + per-symbol saturation ("no output -> no veto"). It reproduces the original + divergence exactly: single-fill carries only 600000.SH (40 rows) while the + batch fill carries both (50 rows). + """ import factors.materialize as mat real = mat._pooled_pool_saturated - def output_only(full, bars, emit_start, *, baseline_days, lookback_days, symbols): - present = [s for s in symbols if str(s) in set(map(str, full.index.get_level_values("symbol")))] + def zero_output_is_saturated(full, bars, emit_start, *, baseline_days, lookback_days, symbols): + if full.empty: # the defect: a symbol that produced nothing gets no veto + return True return real(full, bars, emit_start, baseline_days=baseline_days, - lookback_days=lookback_days, symbols=present) + lookback_days=lookback_days, symbols=symbols) - monkeypatch.setattr(mat, "_pooled_pool_saturated", output_only) + monkeypatch.setattr(mat, "_pooled_pool_saturated", zero_output_is_saturated) dates = list(_ZO_DATES[90:130]) # the review's emit window (indices 90..129) fid = "volume_peak_count_20" with tempfile.TemporaryDirectory() as ta, tempfile.TemporaryDirectory() as tb: @@ -495,7 +520,7 @@ def output_only(full, bars, emit_start, *, baseline_days, lookback_days, symbols "single-vs-batch contrast below proves nothing" ) assert syms_a != syms_b or _ZO_B not in syms_a, ( - "the output-symbols-only criterion should drop the thin symbol from " + "the zero-output-is-saturated criterion should drop the thin symbol from " "the single-date fill" )