feat: factor refactor D4b — stream minute materialization per symbol

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 <id> \
+    --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 作为显式的性能项。

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",
 ]