From fbdf0c69d8bbbcf524d73909327c9784d050a3b1 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Sat, 25 Jul 2026 11:57:48 -0700 Subject: [PATCH 01/10] feat(factors): stream minute materialization per symbol (D4b) D4's materializer loaded every symbol's 1min bars into ONE frame. On the real evaluation plane that frame is 52.7 GB nominal and ~115 GB peak RSS against 56 GB of memory, so the D5 four-leg reconciliation could not be run at all -- not slowly, at all. It also affected BOUNDED factors, not just pooled ones, so per-symbol streaming is a structural prerequisite for all eleven minute factors rather than an optimization for eight of them. The materializer now reads one symbol's bars, reduces them to that symbol's daily intermediate, discards the bars, and runs the factor's cross-sectional combine ONCE on the assembled panel. THE CUT SITS BEFORE THE CROSS-SECTION. Ten 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-score needs >= AMP_CUT_MIN_CROSS_SECTION (=10) finite pairs on a date, so a one-symbol cross-section is all-NaN by definition. Splitting around the WHOLE factor would silently zero it out; the split before its combine reproduces it exactly. That constant is untouched -- no factor math, parameter or threshold changes here. The seam already existed in the factor module and in the eval runners; the binding surfaces it. Saturation is now decided PER SYMBOL, which is the same terminal, not a weakening: the criterion's whole content is "further history cannot change any emit value", so a symbol satisfying it at depth D has the values it would have at any deeper load. The locked head is dropped from the symbol's own day list rather than the loaded union, and the union is a superset, so the per-symbol window is at or inside the union one -- it never terminates earlier. The requested-symbol veto, the locking offset, the provider's declared floor and the loud exhaustion raise are all unchanged; what a thin symbol can no longer do is drag the other 994 to the floor (measured: 84 of the evaluation universe's 995 symbols list after the emit start, so that drag was constructive for every pooled factor). Two disclosed consequences: * PER-SYMBOL TRIM: the trailing-trading-day trim reads the symbol's own trading days instead of the loaded universe's union. Strictly more conservative (a subset reaches at least as far back) and it removes a cross-symbol coupling that was itself a load-geometry dependence -- another symbol's calendar could decide how much history this one was warmed with. Pinned by a test that first asserts the two rules really disagree on its fixture. * The engine RE-FILTERS what a provider returns. Streaming makes correctness depend on the provider honouring its symbols argument; a provider ignoring it would emit every other symbol's rows once per streamed symbol. The engine owns that invariant rather than trusting an injected object with it. test_zero_output_symbol_veto_mutation keeps its fixture, factor and assertions but re-expresses its mutation. The defect it encodes is unchanged ("a symbol that produced nothing gets no veto"); its old wording ("restrict the criterion's symbol list to symbols present in the output") was faithful only while one criterion call decided the depth for the whole universe. Measured under streaming: 70 criterion calls, real and filtered forms answer identically in 70 of 70, because all 20 zero-output calls are already vetoed by the guard above the symbol loop. Left alone it would be a mutation test that cannot fail. Re-expressed directly it reproduces the original divergence on the same fixture: single-fill 40 rows / one symbol vs batch 50 rows / two symbols. qt.saturation_probe gains --mode stream-scale: the same instrument pointed at the fixed engine, one factor per process (so ru_maxrss is that factor's own peak) and resumable via a JSONL out file. It reports the nominal largest frame AND the peak RSS separately, because the feasibility mode measured frame:peakRSS at 2.2-2.45x and the nominal number alone understates the requirement by about half. --- factors/compute/minute/binding.py | 171 ++++++ factors/materialize.py | 204 +++++-- qt/saturation_probe.py | 130 +++++ tests/test_factor_materialize_streaming.py | 613 +++++++++++++++++++++ tests/test_factor_service.py | 32 +- 5 files changed, 1104 insertions(+), 46 deletions(-) create mode 100644 tests/test_factor_materialize_streaming.py 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..b38471c 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, ) @@ -242,6 +268,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 +297,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 +407,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 +456,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 +505,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 +579,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/saturation_probe.py b/qt/saturation_probe.py index 9034102..bb003d4 100644 --- a/qt/saturation_probe.py +++ b/qt/saturation_probe.py @@ -21,21 +21,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 +170,124 @@ 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), + } + + +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"), 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) 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..c5c576d --- /dev/null +++ b/tests/test_factor_materialize_streaming.py @@ -0,0 +1,613 @@ +"""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 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.compute.minute.binding import ( + _MINUTE_BINDINGS, + _MINUTE_STREAM_BINDINGS, + CROSS_SECTIONAL_MINUTE_FACTORS, + combine_minute_stats, + is_cross_sectional_minute, + 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 + +#: The ten minute factors with a bars-only binding (valley_price_quantile is the +#: deliberately deferred eleventh — it also needs the daily panel). +STREAMED_FACTOR_IDS = ( + "jump_amount_corr_20", + "minute_ideal_amp_10", + "amp_marginal_anomaly_vol_20", + "volume_peak_count_20", + "intraday_amp_cut_10", + "peak_interval_kurtosis_20", + "valley_relative_vwap_20", + "valley_ridge_vwap_ratio_20", + "ridge_minute_return_20", + "peak_ridge_amount_ratio_20", +) + +#: >= 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_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") + + +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 + + +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 (asserted above) — 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". + """ + factor = factor_registry.build("volume_peak_count_20") + got = materialize_range( + factor, view=View.CLOSE, symbols=list(LONG_DENSE), + emit_start=LONG_EMIT_START, emit_end=LONG_EMIT_END, + sources=MaterializeSources(minute=LongProv()), + ) + 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.""" + 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 + + +# --------------------------------------------------------------------------- # +# 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() + + +def test_per_symbol_trim_uses_the_symbols_own_trading_days(): + """DISCLOSED REFINEMENT: the trailing trim counts THIS symbol's trading days. + + Under the union rule a densely-traded neighbour's calendar decided how much + history a sparse symbol was warmed with — itself a load-geometry dependence + (red line #6). This pins BOTH halves so the refinement is a fact, not a + side effect: + + 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) — if they agreed, half of this test would be vacuous; + 2. the streamed value equals the load-geometry-free truth (the symbol's whole + history in one frame), which is the property the trim exists to deliver. + """ + from factors.materialize import _warmup_start + + factor = factor_registry.build("jump_amount_corr_20") # a BOUNDED minute factor + 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..a3d123c 100644 --- a/tests/test_factor_service.py +++ b/tests/test_factor_service.py @@ -461,19 +461,37 @@ 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 is + provably inert here — measured on this fixture: 70 criterion calls, the real + and filtered forms answer identically in 70 of 70, because all 20 calls with + zero output also have ``loaded_days <= baseline_days`` and are already vetoed + by the guard above the symbol loop. Left as-is it would be a mutation test + that cannot fail, which is the exact failure mode this project has twice paid + for; so it is re-expressed to say the same thing directly ("no output -> no + veto") and it reproduces the original divergence on the SAME fixture, the + SAME factor and the SAME assertions: 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 +513,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" ) From 21f02bb824fd653859d65150de0db607f4a08716 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Sat, 25 Jul 2026 12:10:39 -0700 Subject: [PATCH 02/10] docs(qt): stop two probes describing a load geometry the engine no longer uses D4b moved the materializer from a whole-universe single frame to per-symbol streaming. Two tools still described the old shape, which is the exact failure mode this repo has paid for three times (#76, #78, #82): the behaviour changed and the sentence about it did not. * qt/saturation_probe feasibility mode measures the PRE-D4b engine. Its item 3 said "the materializer loads every symbol into ONE frame", present tense. It is now marked HISTORICAL -- it is what that geometry would have cost, i.e. the reason it was replaced, not what runs today -- and the module header points at --mode stream-scale for the current path. Items 1, 2 and 4 are facts about the DATA, not the engine, so they stand as written. * qt/factor_hotpath_smoke loads its 40-symbol sample into one frame in BOTH arms. What it measures is unchanged and still the point (repeated 1min read + normalize vs paying it once), but the NAIVE arm is no longer a picture of the engine's memory profile, and a reader could reasonably take it for one. Said so, with a pointer to the tool that does measure the real path. Deliberately NOT swept: the eleven eval runners' "the multi-year all-symbol minute panel is NEVER materialized". Those describe the runners, which still stream per symbol -- a careless sed here would turn true statements false, which is the trap #82 called out by name. --- qt/factor_hotpath_smoke.py | 11 +++++++++-- qt/saturation_probe.py | 13 +++++++++++-- 2 files changed, 20 insertions(+), 4 deletions(-) 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 bb003d4..ac48ba7 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 From e8f24a426cbeccebc1d0822415fc9d4db8c6d2b1 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Sat, 25 Jul 2026 12:13:05 -0700 Subject: [PATCH 03/10] test(factors): close two vacuity holes in the D4b streaming tests Both tests could have agreed for the wrong reason if their fixture drifted, and neither would have said so: * the early-saturation test borrowed "the dense symbols stop short of the floor" from the test above it, in prose. On a shorter fixture the first expansion chunk would already reach the floor, both sides would load the same bars, and the comparison would pass while exercising nothing. It now records the load starts itself and asserts the early stop. * the cross-section test compared the streamed values against the one-frame reference with an extra universe member present. If that member changed nothing, agreement would be equally consistent with a per-symbol cross-section -- exactly the failure it is meant to exclude. It now asserts that the extra member actually moved existing cells (measured: it does). --- tests/test_factor_materialize_streaming.py | 42 +++++++++++++++++++--- 1 file changed, 37 insertions(+), 5 deletions(-) diff --git a/tests/test_factor_materialize_streaming.py b/tests/test_factor_materialize_streaming.py index c5c576d..a623085 100644 --- a/tests/test_factor_materialize_streaming.py +++ b/tests/test_factor_materialize_streaming.py @@ -514,15 +514,31 @@ 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 (asserted above) — 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". + 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=LongProv()), + 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") @@ -536,7 +552,13 @@ def test_an_early_saturating_symbol_matches_the_whole_history_value(): 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 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] @@ -545,6 +567,16 @@ def test_cross_sectional_factor_sees_the_whole_universe_not_the_streamed_symbol( 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) From ba50723fd9846c0cf30c284989c839182f2f648a Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Sat, 25 Jul 2026 12:41:07 -0700 Subject: [PATCH 04/10] docs(factors): record the D4b measurements against the probe's own baseline The feasibility probe's section 5 left "per-symbol saturation: implemented and accepted" open, and said explicitly that it had NOT reconciled intraday_amp_cut end to end. Section 6 closes that in the same file, so the before and the after sit next to each other instead of in two documents that can drift apart. Recorded, all measured: * cell-by-cell reconciliation on REAL bars (40 symbols x 6 months) against the geometry that was replaced -- every symbol in one frame. Index identical and NaN-set difference ZERO on all ten factors; five bit-identical, the other five within 3.3e-14 (pandas rolling accumulation over a differently-sized prefix, the residual D2/D4 already catalogued). The NaN column is the load-bearing one: a finite<->NaN flip is a coverage change, not a float artefact. * the scale proof on the real evaluation plane for the riskiest factor (intraday_amp_cut, the only pooled x cross-sectional one): nominal largest frame 0.0656 GB, PEAK RSS 0.964 GB against 56 GB available, 995 of 995 symbols present, live_calls 0. The two numbers are reported separately because the probe measured frame:peakRSS at 2.2-2.45x and the nominal figure alone understates the requirement by about half. * where the cost actually falls: a dense symbol saturates in ONE provider call (0.59 s), a late-listing symbol takes 40 (14.95 s) because it re-reads its own history on every expansion chunk. 40 is exactly the chunk arithmetic. That cost now lands on those 84 symbols instead of dragging all 995 -- which is the point, since the all-995 version did not fit in memory at all. Also disclosed: wall times are upper bounds (other acceptance jobs were running on the same box); the memory figures are per-process high-water marks and are not affected. And the expansion STEP SIZE is a documented non-correctness knob that could cut those 40 reads, but changing it is still a load-geometry change that owes its own reconciliation, so it is left to D5 rather than bundled here. --- docs/factors/d5_saturation_feasibility.md | 94 ++++++++++++++++++++++- 1 file changed, 93 insertions(+), 1 deletion(-) diff --git a/docs/factors/d5_saturation_feasibility.md b/docs/factors/d5_saturation_feasibility.md index 12ee7a1..ca8b6af 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,95 @@ 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 个月） + +`2023-01-03..2023-06-30`，参照系 = **被替换掉的那个几何**（全部票进一个 frame、一次 +cutoff、一次整因子调用），单帧 3,711,400 行 / 0.69 GB；`live_calls=0`。 + +| 因子 | index 相同 | 可比 cell | NaN 集合差异 | max\|abs\| | max\|rel\| | +|---|---|---|---|---|---| +| `jump_amount_corr_20` | True | 4,720 | **0** | 3.275e-14 | 2.387e-13 | +| `minute_ideal_amp_10` | True | 4,720 | **0** | **0.000e+00** | 0.000e+00 | +| `amp_marginal_anomaly_vol_20` | True | 4,720 | **0** | **0.000e+00** | 0.000e+00 | +| `volume_peak_count_20` | True | 4,720 | **0** | **0.000e+00** | 0.000e+00 | +| **`intraday_amp_cut_10`** | True | 4,720 | **0** | 2.531e-14 | 2.155e-12 | +| `peak_interval_kurtosis_20` | True | 4,720 | **0** | **0.000e+00** | 0.000e+00 | +| `valley_relative_vwap_20` | True | 4,719 | **0** | 2.220e-16 | 2.232e-16 | +| `valley_ridge_vwap_ratio_20` | True | 2,295 | **0** | 2.220e-16 | 2.246e-16 | +| `ridge_minute_return_20` | True | 2,283 | **0** | **0.000e+00** | 0.000e+00 | +| `peak_ridge_amount_ratio_20` | True | 2,247 | **0** | 3.331e-16 | 4.943e-16 | + +**NaN 集合差异全为 0** 是这张表里最承重的一列：finite↔NaN 翻转是覆盖率变化，不是浮点 +现象，本步绝不允许。5/10 逐位相同；其余 5 个残差 ≤3.3e-14，是 pandas 滚动累加在不同 +长度前缀上的重排（D2/D4 已编目的可归因项），比任何真实缺行效应低若干个数量级。 + +`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` | 1,158,842 | 1,158,842 | **995/995** | **1** | 0.0656 GB | **0.964 GB** | 1990.5 s | **0** | + +**995/995**：没有任何一只被请求的票在流式化后消失（I5a 红线）。`live_calls=0` 全程 +cache-only。其余因子的 sweep 可用同一命令续跑（JSONL 已记录的因子会被跳过）。 + +⚠️ **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 深度的加载，而那个加载**根本装不下** +（一至四节）。 + +⚠️ **步长是可以改的，但本步刻意没改**：D4 的 docstring 明写 chunk "only a STEP SIZE, +never a correctness input"，所以把它调大或改成几何增长是**纯性能**改动，能把这 40 次 +大幅压低。但它仍然是取值路径的加载几何改动，需要自己的逐格对账，**不能夹带进 D4b**。 +留给 D5 作为显式的性能项。 From 4d49cc2de208c331e1f4d99c02e0f1d2f5a4e7e0 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Sat, 25 Jul 2026 12:42:36 -0700 Subject: [PATCH 05/10] test(factors): make the cross-sectional classification a checked fact CROSS_SECTIONAL_MINUTE_FACTORS was a declaration with a comment explaining it. The comment is true today; nothing made it stay true. A minute factor that later grew a cross-sectional step without being added to the set would be streamed with an identity combine and would silently return per-symbol values -- the same class of defect as splitting around the whole factor, which this step exists to prevent. The classification is now derived and asserted on every bound factor, in both directions: a factor is per-symbol pure iff splitting around the WHOLE factor reproduces the whole-universe values. Nine must survive that split bit-identically and intraday_amp_cut must not. Both halves are non-vacuous (the whole-universe side is asserted to have finite values first). --- tests/test_factor_materialize_streaming.py | 38 ++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/tests/test_factor_materialize_streaming.py b/tests/test_factor_materialize_streaming.py index a623085..1bda5b1 100644 --- a/tests/test_factor_materialize_streaming.py +++ b/tests/test_factor_materialize_streaming.py @@ -299,6 +299,44 @@ def test_splitting_around_the_whole_factor_destroys_the_cross_sectional_factor() _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). From d06eb4b3c6ff5b2c2ad1db8fb084e69c785c6eba Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Sat, 25 Jul 2026 12:51:04 -0700 Subject: [PATCH 06/10] docs(factors): complete the D4b scale table with all three load geometries One factor per load geometry, on the real evaluation plane, one process each: * intraday_amp_cut_10 (pooled x CROSS-SECTIONAL) peak RSS 0.964 GB, 1990.5 s * ridge_minute_return_20 (pooled) peak RSS 0.854 GB, 1133.7 s * jump_amount_corr_20 (bounded) peak RSS 0.730 GB, 417.7 s That is complete coverage of the geometries, not a sample of the factors: every minute factor takes one of these three paths. All three carry 995 of 995 symbols and live_calls 0. The peak sits in a 0.73-0.96 GB band regardless of factor or output size, because it is set by one symbol's bars plus the daily panel and not by the universe -- which is the property that makes the run possible at all, against the 52.7 GB nominal / ~115 GB peak the replaced geometry needed. Wall time orders exactly by geometry: no expansion < 20-chunk expansion < 60-day-chunk expansion. The remaining seven factors are deliberately not run here: that is the D5 full run, not D4b's acceptance. The sweep skips factors already in its JSONL (verified by running it again against a recorded factor), so D5 continues rather than restarts. --- docs/factors/d5_saturation_feasibility.md | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/docs/factors/d5_saturation_feasibility.md b/docs/factors/d5_saturation_feasibility.md index ca8b6af..3bd9912 100644 --- a/docs/factors/d5_saturation_feasibility.md +++ b/docs/factors/d5_saturation_feasibility.md @@ -204,12 +204,21 @@ committed 为测试，不是一次性观察。 CSI500 995 票 / `2021-07-01..2026-06-30`： -| 因子 | 输出行 | 有值 | 出现的票 | 每次请求票数 | 名义最大帧 | **峰值 RSS** | wall | live_calls | -|---|---|---|---|---|---|---|---|---| -| `intraday_amp_cut_10` | 1,158,842 | 1,158,842 | **995/995** | **1** | 0.0656 GB | **0.964 GB** | 1990.5 s | **0** | +**三种加载几何各取一个**（这是完整的几何覆盖，不是抽样）： -**995/995**：没有任何一只被请求的票在流式化后消失（I5a 红线）。`live_calls=0` 全程 -cache-only。其余因子的 sweep 可用同一命令续跑（JSONL 已记录的因子会被跳过）。 +| 因子 | 几何 | 输出行 | 有值 | 出现的票 | 每次请求票数 | 名义最大帧 | **峰值 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` 是各进程自己的 From c8a2bd5c1e18c00556d02c0b2dd7aa1e49647787 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Sat, 25 Jul 2026 13:40:29 -0700 Subject: [PATCH 07/10] fix(factors): de-duplicate the requested universe before streaming (MED-1) Review found the one real engine defect D4b introduced. The single-frame engine was immune to a repeated symbol because the providers' isin filter is idempotent -- whatever the caller passed, one row per name came back. Streaming iterates the list, so a repeat is materialized twice. Measured on the streaming fixture with two names repeated, before the fix: volume_peak_count_20 48 -> 56 rows, 8 duplicated (date, symbol) entries, values unchanged (the factor is per-symbol pure) intraday_amp_cut_10 the same 8 duplicates AND ALL 48 shared cells changed The second one is the serious half: a factor with a real cross-sectional combine sees the repeated name twice in its date's cross-section, which moves that date's mean and std for EVERY member. FactorValueStore.upsert's duplicated(keep="last") would de-duplicate the ROWS while keeping the contaminated VALUES, so it would land in a persistent artifact. No live path can reach this today (the service is not wired to a runner until D6), which is why it is not a HIGH -- but the mirror of it was already taken seriously: the engine re-filters what a PROVIDER returns and explains that it owns that invariant rather than trusting an injected object. The caller-side half of the same invariant was not owned. Now it is, in one place at the engine's entry point, covering the daily path too. DE-DUPLICATING RATHER THAN RAISING, and why: 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. MUTATION (run): removing the call restores 56 rows / 8 duplicates / 48 changed cells (probe asserted BEFORE trusting the test), test rc=1; restored rc=0. --- factors/materialize.py | 34 ++++++++++++++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/factors/materialize.py b/factors/materialize.py index b38471c..8114466 100644 --- a/factors/materialize.py +++ b/factors/materialize.py @@ -189,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). @@ -226,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) From 401fad39c3f8b3082f40753289a0f5737b06a55c Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Sat, 25 Jul 2026 13:40:52 -0700 Subject: [PATCH 08/10] test(factors): correct the JC5 wording, derive the factor list, widen the trim test Three review items, all in tests. MED-2 -- the JC5 docstring stated its own reasoning backwards. It said leaving the old mutation expression in place "would be a mutation test that cannot fail". It would not: the assertion asserts that the drop HAPPENS, so an inert mutation makes it impossible to SATISFY. The old expression went RED on the D4b engine; it was a test that cannot pass. That distinction is not pedantic here -- the docstring was invoking this project's heaviest methodology rule while describing its exact opposite, next to a test whose real situation is the other rule (do not quietly edit a test the new engine fails). Restated honestly, and the call count corrected from 70 to 72: the earlier figure counted only the per-date fill and missed the batch fill's 2 calls. Re-measured: 72 calls, 0 divergences, 20 zero-output calls, all of them already vetoed by the guard above the symbol loop. LOW-4 -- the per-symbol trim test covered only jump_amount_corr_20, whose disagreement with the union rule shows up as a CHANGED VALUE. The other two bounded factors express the same disagreement as a COVERAGE FLIP (NaN under the union rule, finite under per-symbol) -- and finite<->NaN is precisely what _assert_bit_identical singles out as the change this step must not make by accident. Testing only the value-moving one missed the two that matter most. Now parametrized over all three, derived. Also recorded there, because it reverses this file's earlier framing: the trim is not a conservative refinement of a correct rule, it CORRECTS a wrong one. Measured across engines, all three bounded factors move exactly one cell and in all three the per-symbol result equals the load-geometry-free truth while the union result does not (jump_amount_corr 0.157115 union vs 0.090115 truth; the other two NaN vs finite). LOW-5 -- STREAMED_FACTOR_IDS was a hand-maintained tuple with nothing tying it to the binding tables. A new minute factor could enter both tables, satisfy the coverage test, and still slip past every parametrized reconciliation silently. Derived from the binding table now, with the id round-tripped through the registry so the derivation itself is checked; BOUNDED_FACTOR_IDS likewise. --- tests/test_factor_materialize_streaming.py | 111 ++++++++++++++++----- tests/test_factor_service.py | 27 +++-- 2 files changed, 103 insertions(+), 35 deletions(-) diff --git a/tests/test_factor_materialize_streaming.py b/tests/test_factor_materialize_streaming.py index 1bda5b1..bd0619e 100644 --- a/tests/test_factor_materialize_streaming.py +++ b/tests/test_factor_materialize_streaming.py @@ -41,6 +41,7 @@ CROSS_SECTIONAL_MINUTE_FACTORS, combine_minute_stats, is_cross_sectional_minute, + is_valid_day_pooled, minute_raw_from_bars, minute_stats_from_bars, ) @@ -51,19 +52,40 @@ from factors.materialize import MaterializeSources, materialize_range from factors.view_lag import minute_decision_cutoff -#: The ten minute factors with a bars-only binding (valley_price_quantile is the -#: deliberately deferred eleventh — it also needs the daily panel). -STREAMED_FACTOR_IDS = ( - "jump_amount_corr_20", - "minute_ideal_amp_10", - "amp_marginal_anomaly_vol_20", - "volume_peak_count_20", - "intraday_amp_cut_10", - "peak_interval_kurtosis_20", - "valley_relative_vwap_20", - "valley_ridge_vwap_ratio_20", - "ridge_minute_return_20", - "peak_ridge_amount_ratio_20", +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 @@ -377,6 +399,33 @@ def test_provider_is_never_asked_for_more_than_one_symbol(factor_id): 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. @@ -636,23 +685,35 @@ def _dense_plus_sparse() -> pd.DataFrame: DENSE_PLUS_SPARSE = _dense_plus_sparse() -def test_per_symbol_trim_uses_the_symbols_own_trading_days(): - """DISCLOSED REFINEMENT: the trailing trim counts THIS symbol's trading days. +@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 — itself a load-geometry dependence - (red line #6). This pins BOTH halves so the refinement is a fact, not a - side effect: - - 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) — if they agreed, half of this test would be vacuous; - 2. the streamed value equals the load-geometry-free truth (the symbol's whole - history in one frame), which is the property the trim exists to deliver. + 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("jump_amount_corr_20") # a BOUNDED minute factor + factor = factor_registry.build(factor_id) warmup = int(factor.spec.lookback_depth) # (1) the rules disagree — computed from the fixture, not asserted by fiat. diff --git a/tests/test_factor_service.py b/tests/test_factor_service.py index a3d123c..32b393c 100644 --- a/tests/test_factor_service.py +++ b/tests/test_factor_service.py @@ -470,16 +470,23 @@ def test_zero_output_symbol_veto_mutation(monkeypatch): 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 is - provably inert here — measured on this fixture: 70 criterion calls, the real - and filtered forms answer identically in 70 of 70, because all 20 calls with - zero output also have ``loaded_days <= baseline_days`` and are already vetoed - by the guard above the symbol loop. Left as-is it would be a mutation test - that cannot fail, which is the exact failure mode this project has twice paid - for; so it is re-expressed to say the same thing directly ("no output -> no - veto") and it reproduces the original divergence on the SAME fixture, the - SAME factor and the SAME assertions: single-fill carries only 600000.SH (40 - rows) while the batch fill carries both (50 rows). + 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 From 1619846833051831d8688c9de466025c8a1c2f48 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Sat, 25 Jul 2026 13:41:18 -0700 Subject: [PATCH 09/10] feat(qt): make the D4b reconciliation table reproducible, and fix three doc claims NIT-6 -- --mode stream-scale reproduced the scale table, but NOTHING in the repo reproduced section 6.1, the most load-bearing table in the document: a reviewer could only re-derive it by writing their own script. The probe is already the project's "committed, re-runnable instrument"; this holds it to that standard. --mode reconcile runs the 40-symbol x 6-month comparison against the whole-universe single frame and exits rc=1 if any factor's index or NaN set differs. Its output is now transcribed into 6.1 literally, same columns and same order, so the two can be diffed directly. Verified: the committed mode reproduces every number. NIT-7 -- 6.1 never stated the reference's load depth, which is load-bearing: the streamed side's depth is decided by saturation expansion for the 8 pooled factors, so a shallower reference compares two different histories and a deeper one changes the float-accumulation prefix. Stated (400 calendar days), with the reason, and it explains why this table shows ~1e-14 residuals while a branch-vs-main comparison of the real engine gets exactly 0.000e+00 -- there both sides load identically. MED-3 -- 6.1 declared a policy this step does not actually keep: that finite<->NaN flips are "never allowed here". per-symbol trim CAN produce them and measurably does (two of three bounded factors); the 40-symbol sample simply has no symbol sparse enough to trigger it. Corrected to say what is true: zero flips ON THIS SAMPLE, with the one disclosed semantic change that can produce them named. Direction reversal (new 6.5) -- this document warned that if D5 mismatches the D1 frozen baseline on sparse symbols, the trim is the first place to look. That is backwards, on the safe side. Cross-engine measurement, three bounded factors, one cell each: per-symbol == the load-geometry-free truth in all three, union != truth in all three. The union rule was not looser, it was WRONG (red line #6: a value that depends on load geometry). And the eleven eval runners read each symbol's whole window with no union trim at all, so per-symbol moves TOWARD the frozen baseline, not away. The warning is restated pointing at saturation depth and anchor truncation, which are the named attribution items. --- docs/factors/d5_saturation_feasibility.md | 69 +++++++++++--- qt/saturation_probe.py | 105 +++++++++++++++++++++- 2 files changed, 159 insertions(+), 15 deletions(-) diff --git a/docs/factors/d5_saturation_feasibility.md b/docs/factors/d5_saturation_feasibility.md index 3bd9912..42419b4 100644 --- a/docs/factors/d5_saturation_feasibility.md +++ b/docs/factors/d5_saturation_feasibility.md @@ -163,25 +163,46 @@ python -m qt.saturation_probe --mode stream-scale --factors \ ## 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`。 -| 因子 | index 相同 | 可比 cell | NaN 集合差异 | max\|abs\| | max\|rel\| | -|---|---|---|---|---|---| -| `jump_amount_corr_20` | True | 4,720 | **0** | 3.275e-14 | 2.387e-13 | -| `minute_ideal_amp_10` | True | 4,720 | **0** | **0.000e+00** | 0.000e+00 | -| `amp_marginal_anomaly_vol_20` | True | 4,720 | **0** | **0.000e+00** | 0.000e+00 | -| `volume_peak_count_20` | True | 4,720 | **0** | **0.000e+00** | 0.000e+00 | -| **`intraday_amp_cut_10`** | True | 4,720 | **0** | 2.531e-14 | 2.155e-12 | -| `peak_interval_kurtosis_20` | True | 4,720 | **0** | **0.000e+00** | 0.000e+00 | -| `valley_relative_vwap_20` | True | 4,719 | **0** | 2.220e-16 | 2.232e-16 | -| `valley_ridge_vwap_ratio_20` | True | 2,295 | **0** | 2.220e-16 | 2.246e-16 | -| `ridge_minute_return_20` | True | 2,283 | **0** | **0.000e+00** | 0.000e+00 | -| `peak_ridge_amount_ratio_20` | True | 2,247 | **0** | 3.331e-16 | 4.943e-16 | +⚠️ **参照系的加载深度是承重参数，必须写明**：参照单帧按 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 已编目的可归因项），比任何真实缺行效应低若干个数量级。 +现象。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——该负对照已 @@ -242,6 +263,26 @@ dense 票在**第一个 chunk 就饱和**；late 票攒不出 emit_start 之前 另外 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**。 diff --git a/qt/saturation_probe.py b/qt/saturation_probe.py index ac48ba7..8ec2e93 100644 --- a/qt/saturation_probe.py +++ b/qt/saturation_probe.py @@ -247,6 +247,105 @@ def stream_scale( } +# --------------------------------------------------------------------------- # +# 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() @@ -290,13 +389,17 @@ def main(argv: list[str] | None = None) -> int: 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"), default="feasibility") + 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) From 074522599d33ed6c4fa88ff1e567cac43afbe860 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Sat, 25 Jul 2026 13:59:18 -0700 Subject: [PATCH 10/10] =?UTF-8?q?test(factors):=20close=20the=20other=20di?= =?UTF-8?q?rection=20=E2=80=94=20an=20unclassified=20minute=20factor=20mus?= =?UTF-8?q?t=20fail?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deriving the parametrize lists from the binding tables only guarded factors ADDED to a table. Review measured the open direction: dropping one factor from both tables left this file at 53 passed and the suite at 2169 passed, all green, with the factor silently absent from every reconciliation. The source of truth is now the set of Factor subclasses defined under factors/compute/minute, so a new factor file that nobody classifies fails here. Mutation witness: removing PeakIntervalKurtosisFactor from both tables (occurrences 4 -> 0, asserted before running) turns this test red; restoring makes binding.py byte-identical and the file green at 58. --- tests/test_factor_materialize_streaming.py | 55 ++++++++++++++++++++++ 1 file changed, 55 insertions(+) diff --git a/tests/test_factor_materialize_streaming.py b/tests/test_factor_materialize_streaming.py index bd0619e..e29a2f1 100644 --- a/tests/test_factor_materialize_streaming.py +++ b/tests/test_factor_materialize_streaming.py @@ -28,6 +28,10 @@ from __future__ import annotations +import importlib +import inspect +import pathlib + import numpy as np import pandas as pd import pytest @@ -35,7 +39,10 @@ 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, @@ -248,6 +255,54 @@ def test_stream_bindings_cover_exactly_the_bound_minute_factors(): 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")