From 6e5b9074196c2645337cc6321ce4beb7c6d1a4cf Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Mon, 15 Jun 2026 19:09:45 +0800 Subject: [PATCH 1/8] feat(data): MMP minute factor + EW daily aggregation (I5c) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the exploratory Minute Microstructure Pressure (MMP) factor to the I3 minute -> daily aggregation. Per 1min bar t (single symbol/day session, sorted by bar_end): mid=(high+low)/2; S=(close-mid)/mid; V=sqrt(volume/median(vol[t-20:t])); B=|close-open|/(high-low+eps); R=(high-low)/(mean(hl[t-20:t])+eps); MMP_t=S*V*B*R, eps=1e-6. - compute_minute_mmp: rolling baselines use ONLY the prior 20 bars t-20..t-1 (rolling(20).shift(1)) — never bar t, never a later bar, never the prior day's tail; the first 20 bars of each session are NaN. Invalid denominators (mid<=0, median vol<=0, NaN baseline) yield NaN, never inf. - new feature key 'mmp_ew' -> column intraday_mmp20_ew_0930_1450 = the EQUAL -WEIGHT mean of valid MMP_t visible at the cutoff (the volume term is already inside MMP_t; the daily aggregation is NOT additionally volume-weighted). No valid minute -> NaN. - mmp_ew is SELECTABLE-ONLY: DEFAULT_FEATURE_KEYS keeps the original four, so a no-args asof_daily_features call has the same columns/cost as before. - mmp_valid_minute_counts: report-only per-(date,symbol) valid-minute count under the same PIT cutoff, reusing compute_minute_mmp (one MMP source of truth). PIT unchanged: available_time <= trade_date + decision_time filters per-bar timestamps BEFORE daily grouping. Data-layer only: no returns / Tushare / cache / portfolio / execution. --- data/clean/intraday_aggregate.py | 160 +++++++++++++++++++++++++++++-- 1 file changed, 154 insertions(+), 6 deletions(-) diff --git a/data/clean/intraday_aggregate.py b/data/clean/intraday_aggregate.py index 3d62308..57f2a38 100644 --- a/data/clean/intraday_aggregate.py +++ b/data/clean/intraday_aggregate.py @@ -47,18 +47,88 @@ DATE_LEVEL = "date" DAILY_INDEX_NAMES: list[str] = [DATE_LEVEL, SYMBOL_LEVEL] -# Feature keys (selectable via the ``features`` argument); column NAMES below -# encode the cutoff and are derived from the time arguments. +# All selectable feature keys (validated against the ``features`` argument and +# mirrored by qt.config.IntradayCfg.score_feature). Column NAMES below encode the +# cutoff and are derived from the time arguments. INTRADAY_FEATURE_KEYS: tuple[str, ...] = ( "ret", "realized_vol", "vwap", "last30m_ret", + "mmp_ew", +) +# Default feature set when ``features`` is None — the original cheap four. ``mmp_ew`` +# (the I5c rolling MMP factor) is selectable-only, so the default output columns and +# the cost of a no-args call stay EXACTLY as before (existing callers unchanged). +DEFAULT_FEATURE_KEYS: tuple[str, ...] = ( + "ret", + "realized_vol", + "vwap", + "last30m_ret", ) DEFAULT_DECISION_TIME = "14:50:00" DEFAULT_SESSION_OPEN = "09:30:00" DEFAULT_LAST_WINDOW_MINUTES = 30 +# Minute Microstructure Pressure (MMP, I5c): rolling baseline window (prior bars +# t-MMP_LOOKBACK..t-1) and the default denominator epsilon. EXPLORATORY factor; +# the window is part of the factor definition, not a tuned parameter. +MMP_LOOKBACK = 20 +DEFAULT_EPSILON = 1e-6 + + +def compute_minute_mmp( + open_: np.ndarray, + high: np.ndarray, + low: np.ndarray, + close: np.ndarray, + volume: np.ndarray, + *, + lookback: int = MMP_LOOKBACK, + epsilon: float = DEFAULT_EPSILON, +) -> np.ndarray: + """Per-bar Minute Microstructure Pressure ``MMP_t`` for ONE symbol/day session. + + Inputs are equal-length 1D arrays ORDERED by ``bar_end`` ascending and + belonging to a SINGLE ``(symbol, trade_date)`` session. The rolling baselines + use ONLY the prior ``lookback`` bars (``t-lookback..t-1``) — never bar ``t`` + itself, never a later bar, never the prior day's tail — so the first + ``lookback`` bars have NaN ``MMP``. + + mid_t = (high_t + low_t) / 2 + S_t = (close_t - mid_t) / mid_t (NaN if mid_t <= 0) + V_t = sqrt(volume_t / median(volume[t-lookback:t])) + (NaN if baseline <= 0 / NaN) + B_t = |close_t - open_t| / (high_t - low_t + epsilon) + R_t = (high_t - low_t) / (mean(hl[t-lookback:t]) + epsilon) + (NaN if baseline is NaN) + MMP_t = S_t * V_t * B_t * R_t + + Invalid denominators yield NaN, never ``inf``. Pure: reads no returns / no + future bars / no token. + """ + open_ = np.asarray(open_, dtype=float) + high = np.asarray(high, dtype=float) + low = np.asarray(low, dtype=float) + close = np.asarray(close, dtype=float) + volume = np.asarray(volume, dtype=float) + + hl = high - low + mid = (high + low) / 2.0 + + # Prior-`lookback` baselines: rolling over t-lookback+1..t THEN shift(1) so + # position t holds the statistic of bars t-lookback..t-1 (excludes bar t). + med_vol = pd.Series(volume).rolling(lookback).median().shift(1).to_numpy() + ma_hl = pd.Series(hl).rolling(lookback).mean().shift(1).to_numpy() + + with np.errstate(divide="ignore", invalid="ignore"): + s_t = np.where(mid > 0.0, (close - mid) / mid, np.nan) + ratio = np.where(med_vol > 0.0, volume / med_vol, np.nan) + v_t = np.sqrt(ratio) + b_t = np.abs(close - open_) / (hl + epsilon) + r_t = hl / (ma_hl + epsilon) + mmp = s_t * v_t * b_t * r_t + return mmp def _hhmm(time_str: str) -> str: @@ -75,7 +145,7 @@ def _hhmm_minus(time_str: str, minutes: int) -> str: def _resolve_feature_keys(features: list[str] | None) -> list[str]: if features is None: - return list(INTRADAY_FEATURE_KEYS) + return list(DEFAULT_FEATURE_KEYS) unknown = [f for f in features if f not in INTRADAY_FEATURE_KEYS] if unknown: raise ValueError( @@ -98,6 +168,8 @@ def _column_name( if key == "last30m_ret": start = _hhmm_minus(decision_time, last_window_minutes) return f"intraday_last{int(last_window_minutes)}m_ret_{start}_{c}" + if key == "mmp_ew": + return f"intraday_mmp{MMP_LOOKBACK}_ew_{o}_{c}" raise ValueError(f"Unhandled intraday feature key: {key!r}.") @@ -112,7 +184,11 @@ def _empty_daily(colnames: list[str]) -> pd.DataFrame: def _compute_group( - g: pd.DataFrame, decision_time: str, last_window_minutes: int + g: pd.DataFrame, + decision_time: str, + last_window_minutes: int, + keys: list[str], + epsilon: float, ) -> dict[str, float]: """Per-(date, symbol) features from already PIT-filtered, bar_end-sorted bars.""" closes = g["close"].to_numpy(dtype=float) @@ -143,12 +219,35 @@ def _compute_group( else: last30 = float("nan") - return { + out = { "ret": ret, "realized_vol": realized_vol, "vwap": vwap, "last30m_ret": last30, } + # MMP is the only rolling feature; compute it ONLY when requested (I5c). + if "mmp_ew" in keys: + out["mmp_ew"] = _mmp_ew_daily(g, epsilon) + return out + + +def _mmp_ew_daily(g: pd.DataFrame, epsilon: float) -> float: + """Equal-weight mean of valid per-minute ``MMP_t`` over one PIT-filtered group. + + ``g`` is one ``(date, symbol)`` session's visible bars, sorted by ``bar_end``. + Every valid minute ``MMP_t`` gets EQUAL weight (no extra volume weighting — the + volume term already lives inside ``MMP_t``). No valid minute -> NaN. + """ + mmp = compute_minute_mmp( + g["open"].to_numpy(dtype=float), + g["high"].to_numpy(dtype=float), + g["low"].to_numpy(dtype=float), + g["close"].to_numpy(dtype=float), + g["volume"].to_numpy(dtype=float), + epsilon=epsilon, + ) + valid = mmp[~np.isnan(mmp)] + return float(np.mean(valid)) if valid.size else float("nan") def asof_daily_features( @@ -158,6 +257,7 @@ def asof_daily_features( session_open: str = DEFAULT_SESSION_OPEN, last_window_minutes: int = DEFAULT_LAST_WINDOW_MINUTES, features: list[str] | None = None, + epsilon: float = DEFAULT_EPSILON, ) -> pd.DataFrame: """PIT-safe daily features from normalized 1min ``bars``. @@ -189,7 +289,7 @@ def asof_daily_features( index_tuples: list[tuple] = [] data: dict[str, list[float]] = {c: [] for c in colnames} for (date, sym), g in visible.groupby(["trade_date", SYMBOL_LEVEL], sort=True): - feats = _compute_group(g, decision_time, last_window_minutes) + feats = _compute_group(g, decision_time, last_window_minutes, keys, epsilon) index_tuples.append((date, str(sym))) for key, col in zip(keys, colnames): data[col].append(feats[key]) @@ -198,6 +298,54 @@ def asof_daily_features( return pd.DataFrame(data, index=index)[colnames].sort_index() +def mmp_valid_minute_counts( + bars: pd.DataFrame, + *, + decision_time: str = DEFAULT_DECISION_TIME, + epsilon: float = DEFAULT_EPSILON, +) -> pd.Series: + """Per-``(date, symbol)`` count of valid (non-NaN) ``MMP_t`` minutes (I5c report). + + Report-only diagnostic: applies the SAME PIT cutoff + (``available_time <= trade_date + decision_time``) and per-session grouping as + :func:`asof_daily_features`, then counts the visible minutes that yielded a + valid ``MMP_t`` (the first ``MMP_LOOKBACK`` bars of a session never do). Reuses + :func:`compute_minute_mmp` so there is a single MMP source of truth. + """ + validate_intraday_bars(bars) + empty = pd.Series( + [], dtype=int, + index=pd.MultiIndex.from_arrays( + [pd.DatetimeIndex([]), pd.Index([], dtype=object)], + names=DAILY_INDEX_NAMES, + ), + ) + if len(bars) == 0: + return empty + work = bars.reset_index() + work["trade_date"] = work["bar_end"].dt.normalize() + cutoff = work["trade_date"] + pd.Timedelta(decision_time) + visible = work.loc[work["available_time"] <= cutoff].copy() + if visible.empty: + return empty + visible = visible.sort_values([SYMBOL_LEVEL, "bar_end"]) + index_tuples: list[tuple] = [] + counts: list[int] = [] + for (date, sym), g in visible.groupby(["trade_date", SYMBOL_LEVEL], sort=True): + mmp = compute_minute_mmp( + g["open"].to_numpy(dtype=float), + g["high"].to_numpy(dtype=float), + g["low"].to_numpy(dtype=float), + g["close"].to_numpy(dtype=float), + g["volume"].to_numpy(dtype=float), + epsilon=epsilon, + ) + index_tuples.append((date, str(sym))) + counts.append(int(np.count_nonzero(~np.isnan(mmp)))) + index = pd.MultiIndex.from_tuples(index_tuples, names=DAILY_INDEX_NAMES) + return pd.Series(counts, index=index, dtype=int).sort_index() + + def resample_intraday_bars(bars: pd.DataFrame, freq: str) -> pd.DataFrame: """Derive coarser intraday bars from normalized 1min ``bars``. From ccabb24b7c4eb0237eb0824f9192adbf53726f60 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Mon, 15 Jun 2026 19:10:09 +0800 Subject: [PATCH 2/8] feat(config): configurable intraday score_feature + report title (I5c) Add IntradayCfg.score_feature (Literal mirroring INTRADAY_FEATURE_KEYS; default 'ret' reproduces the I5a/I5b smoke score, 'mmp_ew' selects the MMP factor) and OutputCfg.intraday_report_title so a reused intraday runner can name the actual study instead of a stale phase label (same precedent as subset_report_title). All existing configs validate unchanged and keep the 'ret' default; an unknown score_feature fails readably at validation. --- qt/config.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/qt/config.py b/qt/config.py index eaf8162..11c85e6 100644 --- a/qt/config.py +++ b/qt/config.py @@ -261,6 +261,14 @@ class IntradayCfg(_Strict): execution_window: tuple[str, str] = ("14:51:00", "14:56:59") require_cache_coverage: bool = True missing_execution: Literal["block"] = "block" + # I5c: which PIT-safe daily intraday feature the tail-rebalance score uses. + # Default "ret" reproduces the I5a/I5b smoke (intraday_ret_0930_1450); "mmp_ew" + # selects the exploratory Minute Microstructure Pressure factor. The allowed + # set mirrors data.clean.intraday_aggregate.INTRADAY_FEATURE_KEYS (a drift test + # locks them equal); an unknown key fails readably at validation. + score_feature: Literal[ + "ret", "realized_vol", "vwap", "last30m_ret", "mmp_ew" + ] = "ret" # I5b execution-time price-limit feasibility. OFF by default, so every I5a / # daily config validates and behaves unchanged. When enabled, the intraday # tail model gates buys at the raw upper limit and sells at the raw lower @@ -345,6 +353,12 @@ class OutputCfg(_Strict): # execution-feasibility config sets its own so it never overwrites the # accepted I5a artifact (same precedent as baseline_report_name). intraday_report_name: str | None = None + # H1 title for the intraday tail report. None keeps the renderer's + # study-aware default (I5c for the MMP factor / I5b when price-limit on / else + # I5a); a config that reuses the runner for a DIFFERENT study sets its own so + # the heading names the actual study, not a stale phase label (I5c precedent, + # same as subset_report_title). + intraday_report_title: str | None = None class OOSCfg(_Strict): From d20a1f4d24fa92c44b2e6a120e5a7c0512350bd2 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Mon, 15 Jun 2026 19:10:09 +0800 Subject: [PATCH 3/8] feat(qt): select MMP score + I5c report/diagnostics + config Make the intraday tail runner consume intraday.score_feature: _score_panel requests ONLY the selected feature from asof_daily_features and uses its returned column EXACTLY (no hardcoded prefix matching), so default 'ret' reproduces I5b and 'mmp_ew' selects intraday_mmp20_ew_0930_1450. _report_heading is study-aware (I5c for mmp_ew / I5b when price-limit on / else I5a) with an intraday_report_title override. Add report-only factor diagnostics: MMP formula + PIT disclosure, daily score coverage, valid-MMP-minute distribution, and per-settled-rebalance cross-sectional score stats + Spearman IC of the decision-date score vs the same period's exec-to-exec return (read in the report layer only, never fed back to factors/alpha). I5b raw stk_limit feasibility stays ON and disclosed. Add config/phase_i5c_mmp_minute_factor.yaml: same SSE50 SH/SZ window as I5b with score_feature=mmp_ew, price_limit_check=true, its own report name + title. --- config/phase_i5c_mmp_minute_factor.yaml | 122 ++++++++++ qt/intraday_tail_framework.py | 293 +++++++++++++++++++++--- 2 files changed, 382 insertions(+), 33 deletions(-) create mode 100644 config/phase_i5c_mmp_minute_factor.yaml diff --git a/config/phase_i5c_mmp_minute_factor.yaml b/config/phase_i5c_mmp_minute_factor.yaml new file mode 100644 index 0000000..e0378d8 --- /dev/null +++ b/config/phase_i5c_mmp_minute_factor.yaml @@ -0,0 +1,122 @@ +# Phase I5c — MMP minute-factor study (EXPLORATORY, NOT a performance claim). +# +# Runs ONE user-proposed minute factor — Minute Microstructure Pressure (MMP) — +# as a PIT-safe daily score through the I5a intraday tail event model with I5b raw +# stk_limit execution feasibility ON. Per 1min bar t: +# mid=(high+low)/2; S=(close-mid)/mid; V=sqrt(volume/median(vol[t-20:t])); +# B=|close-open|/(high-low+eps); R=(high-low)/(mean(hl[t-20:t])+eps); +# MMP_t = S*V*B*R (eps=1e-6) +# Daily score intraday_mmp20_ew_0930_1450 = EQUAL-WEIGHT mean of valid MMP_t over +# bars visible at the 14:50 cutoff (rolling baselines use only prior 20 bars within +# the same symbol/day session; first 20 bars of a day are NaN). No tuning, no +# learned/IC weights, no robustness matrix. +# +# Minute bars are read from the EXISTING intraday cache only (read-only; a miss is +# a hard blocker -> zero stk_mins live calls). Raw stk_limit flows through the same +# P4 read-through cache as the daily endpoints. Same short SSE50 SH/SZ covered +# window as I5b. The report writes to its own name/title so it never overwrites the +# accepted I5a/I5b artifacts. + +project: + name: quantitative_trading_phase_i5c_mmp_minute_factor + timezone: Asia/Shanghai +data: + source: tushare + freq: D + start: '2026-03-03' + end: '2026-06-12' + external_secret_file: /home/shaofl/Projects/financial_projects/.config.json + tushare_token_key: tushare.token + output_name: i5c_daily + cache: + enabled: true + root_dir: artifacts/cache/tushare/v1 + refresh_recent_days: 14 + refresh_dimension_days: 30 + force_refresh: [] +universe: + type: index + index_code: 000016.SH + symbols: [] + min_listing_days: 60 + filters: + # Selection-level daily tradability flags stay DISABLED (as in I5b): the daily + # close limit flag is an EOD value not known at 14:50. The honest direction-aware + # blocking is execution-time raw stk_limit vs the raw execution-minute close. + missing_close: true + suspended: false + st: false + limit_up_down: false +# factors is required by the schema but the runner ignores it: the score is the I3 +# mmp_ew feature (selected via intraday.score_feature), not a config factor. +factors: +- name: momentum_20 + enabled: true + params: + window: 20 + price_col: close +processing: + drop_missing: true + standardize: + enabled: true + method: zscore + winsorize: + enabled: false + method: mad + n: 3.0 + neutralize: + enabled: false + industry_col: industry + size_col: market_cap + industry_level: L1 +alpha: + model: equal_weight + params: {} +portfolio: + constructor: topn_equal_weight + top_n: 10 + long_only: true + max_weight: null + turnover_cap: null +backtest: + initial_nav: 1.0 + rebalance: monthly + event_order: intraday_tail_rebalance + cash_return: 0.0 +intraday: + enabled: true + decision_time: '14:50:00' + data_lag: 1min + session_open: '09:30:00' + execution_model: next_minute_close + execution_window: + - '14:51:00' + - '14:56:59' + require_cache_coverage: true + missing_execution: block + # I5b execution-time raw price-limit feasibility stays ON. + price_limit_check: true + require_price_limit_coverage: true + limit_tolerance: 1.0e-06 + # I5c: select the exploratory MMP minute factor as the PIT-safe daily score. + score_feature: mmp_ew +cost: + fee_rate: 0.001 + slippage_rate: 0.0 + turnover_formula: l1 +analytics: + forward_return_periods: + - 1 + - 5 + - 20 + quantiles: 5 + benchmark: null +output: + root_dir: artifacts + data_dir: artifacts/data + factor_dir: artifacts/factors + report_dir: artifacts/reports + log_dir: artifacts/logs + overwrite: true + intraday_report_name: phase_i5c_mmp_minute_factor + intraday_report_title: Phase I5c — MMP Minute Factor Study diff --git a/qt/intraday_tail_framework.py b/qt/intraday_tail_framework.py index b048ffd..bdf3206 100644 --- a/qt/intraday_tail_framework.py +++ b/qt/intraday_tail_framework.py @@ -26,6 +26,7 @@ from dataclasses import dataclass from pathlib import Path +import numpy as np import pandas as pd from data.cache.intervals import subtract_intervals @@ -33,7 +34,7 @@ from data.cache.intraday_cache import TushareIntradayCache from data.cache.intraday_coverage import IntradayCoverageLedger from data.cache.intraday_parquet_store import IntradayParquetStore -from data.clean.intraday_aggregate import asof_daily_features +from data.clean.intraday_aggregate import asof_daily_features, mmp_valid_minute_counts from data.clean.intraday_schema import RAW_INTRADAY_FREQ, normalize_intraday_bars from data.feed.tushare_flags import TushareFlagsFeed from portfolio.construct import TopNEqualWeight @@ -50,7 +51,6 @@ from runtime.backtest.sim_execution import SimExecution from runtime.intraday_execution import IntradayExecutionConfig -_SCORE_FEATURE = "intraday_ret" # the I3 feature family used as the smoke score _LOGGER_NAME = "qt.intraday_tail_framework" _DEFAULT_REPORT_NAME = "phase_i5a_intraday_tail_framework" @@ -60,29 +60,51 @@ def _report_basename(cfg: RootConfig) -> str: return cfg.output.intraday_report_name or _DEFAULT_REPORT_NAME -def _report_heading(price_limit_check: bool) -> tuple[str, str]: - """(H1 title, intro paragraph) — names the actual study (I5a vs I5b). +def _report_heading( + score_feature: str, price_limit_check: bool, title_override: str | None = None +) -> tuple[str, str]: + """(H1 title, intro paragraph) — names the actual study (I5a / I5b / I5c). - When execution-time price-limit feasibility is on this is the I5b - execution-hardening run, so the heading must not stay frozen at the I5a label - (the same stale-wording trap fixed for the P3-7/P3-8 reports). + The intro is keyed on the actual run (MMP factor study / execution-hardening / + architecture smoke); ``title_override`` (config ``intraday_report_title``) wins + for the H1 so a reused runner never keeps a stale phase label — the same + stale-wording trap fixed for the P3-7/P3-8 reports. """ - if price_limit_check: - return ( + if score_feature == "mmp_ew": + title = "# Phase I5c — MMP Minute Factor Study" + intro = ( + "**This is an EXPLORATORY minute-factor study, NOT a performance claim " + "and NOT a broad factor search.** It runs ONE user-proposed minute " + "factor — Minute Microstructure Pressure (MMP) — as a PIT-safe daily " + "score through the I5a intraday tail event model with I5b raw " + "`stk_limit` execution feasibility ON. No parameters were tuned from " + "performance and no learned/IC weights are used." + ) + elif price_limit_check: + title = ( "# Phase I5b — Intraday Execution-Time Price-Limit Feasibility " - "(execution hardening)", + "(execution hardening)" + ) + intro = ( "**This is an execution-feasibility hardening run, NOT a research " "result.** It extends the I5a intraday tail event model with " "direction-aware raw `stk_limit` blocking at the execution minute; the " "score is still a single PIT-safe I3 feature used only to exercise the " - "shared event-driven backtest engine.", + "shared event-driven backtest engine." ) - return ( - "# Phase I5a — Intraday Tail-Rebalance Event Framework (architecture smoke)", - "**This is an architecture/framework run, NOT a research result.** The " - "score is a single PIT-safe I3 feature used solely to exercise the shared " - "event-driven backtest engine with an intraday tail event model.", - ) + else: + title = ( + "# Phase I5a — Intraday Tail-Rebalance Event Framework (architecture smoke)" + ) + intro = ( + "**This is an architecture/framework run, NOT a research result.** The " + "score is a single PIT-safe I3 feature used solely to exercise the " + "shared event-driven backtest engine with an intraday tail event model." + ) + if title_override: + t = title_override.strip() + title = t if t.startswith("#") else f"# {t}" + return title, intro @dataclass(frozen=True) @@ -92,7 +114,8 @@ class I5aResult: config: RootConfig event_order: str exec_cfg: IntradayExecutionConfig - score_feature: str + score_feature: str # the resolved feature COLUMN (e.g. intraday_mmp20_ew_0930_1450) + score_feature_key: str # the config key (e.g. "ret", "mmp_ew") that selected it requested_symbols: int covered_symbols: int uncovered_symbols: tuple[str, ...] @@ -113,6 +136,10 @@ class I5aResult: up_limit_blocked_buys: int down_limit_blocked_sells: int missing_limit_rows: int + # I5c MMP factor diagnostics (report-only). Empty/None unless score is mmp_ew. + score_coverage: dict[str, int] # {rows, valid, nan} over the daily score panel + minute_count_summary: dict[str, float] | None # valid-MMP-minutes-per-(date,symbol) distribution + factor_diagnostics: tuple[dict, ...] # per settled rebalance: n / stats / Spearman IC report_path: Path log_path: Path @@ -243,28 +270,123 @@ def _load_price_limits( def _score_panel(cfg: RootConfig, bars: pd.DataFrame, logger) -> tuple[pd.Series, str]: - """Deterministic PIT-safe score = the I3 intraday return feature. + """PIT-safe daily score = the configured I3 intraday feature (I5c). ``asof_daily_features`` keeps only bars with ``available_time <= decision_time`` before aggregating to (date, symbol), so the score for date T is known at T's - 14:50 cutoff — exactly the information a tail decision may use. Returns the - score Series (named ``score``) and the source feature column name. + 14:50 cutoff — exactly the information a tail decision may use. The feature is + selected by ``intraday.score_feature`` (default ``ret`` reproduces I5a/I5b; + ``mmp_ew`` is the exploratory MMP factor); only that one feature is requested + and its returned column is used EXACTLY (no prefix matching). Returns the score + Series (named ``score``) and the source feature column name. """ ic = cfg.intraday assert ic is not None feats = asof_daily_features( - bars, decision_time=ic.decision_time, session_open=ic.session_open + bars, + decision_time=ic.decision_time, + session_open=ic.session_open, + features=[ic.score_feature], ) - col = next((c for c in feats.columns if c.startswith(_SCORE_FEATURE)), None) - if col is None: + if feats.shape[1] != 1: raise ValueError( - f"no {_SCORE_FEATURE!r} feature column produced by asof_daily_features " - f"(got {list(feats.columns)})." + f"expected exactly one feature column for score_feature=" + f"{ic.score_feature!r}, got {list(feats.columns)}." ) - logger.info("intraday score: feature=%s, %d (date,symbol) rows", col, len(feats)) + col = feats.columns[0] + logger.info( + "intraday score: feature_key=%s, column=%s, %d (date,symbol) rows", + ic.score_feature, col, len(feats), + ) return feats[col].rename("score"), col +# Minimum cross-sectional sample for a meaningful Spearman IC; below it the IC is +# reported as NaN with the sample size, never silently computed on too few names. +_MIN_IC_SAMPLE = 5 + + +def _score_coverage(score_series: pd.Series) -> dict[str, int]: + """{rows, valid, nan} over the daily score panel (report-only).""" + rows = int(score_series.shape[0]) + nan = int(score_series.isna().sum()) + return {"rows": rows, "valid": rows - nan, "nan": nan} + + +def _minute_count_summary(cfg: RootConfig, bars: pd.DataFrame) -> dict[str, float] | None: + """Distribution of valid-MMP minutes per (date, symbol) (report-only). + + Reuses :func:`mmp_valid_minute_counts` (single MMP source of truth) under the + SAME PIT cutoff. Returns None when there is nothing to summarize. + """ + ic = cfg.intraday + assert ic is not None + counts = mmp_valid_minute_counts(bars, decision_time=ic.decision_time) + if counts.empty: + return None + arr = counts.to_numpy(dtype=float) + return { + "groups": float(counts.shape[0]), + "min": float(np.min(arr)), + "p10": float(np.percentile(arr, 10)), + "p50": float(np.percentile(arr, 50)), + "p90": float(np.percentile(arr, 90)), + "max": float(np.max(arr)), + "mean": float(np.mean(arr)), + } + + +def _factor_diagnostics( + score_series: pd.Series, + model: IntradayTailEventModel, + nav_table: pd.DataFrame, + covered: list[str], +) -> list[dict]: + """Per settled-rebalance score stats + Spearman IC vs exec-to-exec return. + + REPORT-ONLY: the IC correlates the decision-date score with the SAME period's + execution-to-execution return the event model realizes (``holding_returns``), + over the covered, non-NaN-scored cross-section. Returns are read here purely + for analytics — never fed back into the factor/alpha layer (invariant #1). + """ + covered_set = {str(s) for s in covered} + settled = {pd.Timestamp(d).normalize() for d in nav_table.index} + periods = [ + p for p in model.holding_periods() + if pd.Timestamp(p.date).normalize() in settled + ] + rows: list[dict] = [] + for p in periods: + d = pd.Timestamp(p.date).normalize() + try: + cross = score_series.xs(d, level="date") + except KeyError: + continue + cross = cross[[s for s in cross.index if str(s) in covered_set]].dropna() + n = int(cross.shape[0]) + if n: + arr = cross.to_numpy(dtype=float) + stats = { + "mean": float(np.mean(arr)), "std": float(np.std(arr)), + "p10": float(np.percentile(arr, 10)), + "p50": float(np.percentile(arr, 50)), + "p90": float(np.percentile(arr, 90)), + } + else: + stats = {k: float("nan") for k in ("mean", "std", "p10", "p50", "p90")} + rets = model.holding_returns(p, list(cross.index)) # exec-to-exec; omits blocked + joined = pd.concat( + [cross.rename("score"), rets.rename("ret")], axis=1 + ).dropna() + ic_n = int(joined.shape[0]) + ic_val = ( + float(joined["score"].corr(joined["ret"], method="spearman")) + if ic_n >= _MIN_IC_SAMPLE else float("nan") + ) + rows.append({"date": d, "n_scored": n, **stats, "ic": ic_val, "ic_n": ic_n}) + return rows + + def run_phase_i5a_intraday(config_path: str) -> I5aResult: """Run the I5a intraday-tail architecture smoke and write its report.""" cfg = load_config(config_path) @@ -332,12 +454,21 @@ def run_phase_i5a_intraday(config_path: str) -> I5aResult: lo, hi = actual_exec.get(d, (t, t)) actual_exec[d] = (min(lo, t), max(hi, t)) + # I5c factor diagnostics (report-only): score coverage, valid-MMP-minute + # distribution (mmp only), and per-rebalance score stats + Spearman IC. + score_coverage = _score_coverage(score_series) + minute_count_summary = ( + _minute_count_summary(cfg, bars) if ic.score_feature == "mmp_ew" else None + ) + factor_diag = _factor_diagnostics(score_series, model, nav_table, covered) + report_path = Path(cfg.output.report_dir) / f"{basename}.md" result = I5aResult( config=cfg, event_order=cfg.backtest.event_order, exec_cfg=exec_cfg, score_feature=score_feature, + score_feature_key=ic.score_feature, requested_symbols=len(symbols), covered_symbols=len(covered), uncovered_symbols=tuple(uncovered), @@ -354,6 +485,9 @@ def run_phase_i5a_intraday(config_path: str) -> I5aResult: up_limit_blocked_buys=model.up_limit_blocked_buys(), down_limit_blocked_sells=model.down_limit_blocked_sells(), missing_limit_rows=model.missing_limit_rows(), + score_coverage=score_coverage, + minute_count_summary=minute_count_summary, + factor_diagnostics=tuple(factor_diag), report_path=report_path, log_path=log_path, ) @@ -385,6 +519,83 @@ def _check_i5a_preconditions(cfg: RootConfig) -> None: ) +def _append_factor_section(lines: list[str], result: I5aResult) -> None: + """Score-feature disclosure + factor diagnostics (report-only, I5c §4/§5).""" + if result.score_feature_key == "mmp_ew": + lines.append("## MMP minute factor (definition & PIT)") + lines.append("") + lines.append( + "Per 1min bar `t`: `mid=(high+low)/2`; `S=(close-mid)/mid`; " + "`V=sqrt(volume/median(vol[t-20:t]))`; `B=|close-open|/(high-low+eps)`; " + "`R=(high-low)/(mean(hl[t-20:t])+eps)`; **`MMP_t = S*V*B*R`** " + "(`eps=1e-6`)." + ) + lines.append( + "- **daily score** `intraday_mmp20_ew_0930_1450` = the EQUAL-WEIGHT mean " + "of valid `MMP_t` over the bars visible at the cutoff " + "(`[session_open, decision_time]`). The volume term lives inside " + "`MMP_t`; the daily aggregation is NOT additionally volume-weighted." + ) + lines.append( + "- **rolling baselines** use ONLY the prior 20 bars `t-20..t-1` within " + "the SAME `(symbol, trade_date)` session — never bar `t`, never a later " + "bar, never the prior day's tail. The first 20 bars of each session " + "have NaN `MMP`." + ) + lines.append( + "- **PIT**: a bar enters only if `available_time <= trade_date + " + "decision_time`; the filter runs on per-bar timestamps BEFORE daily " + "grouping, so post-cutoff / late-available bars cannot move the score." + ) + mc = result.minute_count_summary + if mc is not None: + lines.append( + f"- **valid-MMP minutes per (date,symbol)**: groups " + f"{int(mc['groups'])}; min {int(mc['min'])} / p10 {int(mc['p10'])} / " + f"p50 {int(mc['p50'])} / p90 {int(mc['p90'])} / max {int(mc['max'])} " + f"(mean {mc['mean']:.1f})." + ) + lines.append("") + sc = result.score_coverage + lines.append("## Score coverage & factor diagnostics (report-only)") + lines.append("") + lines.append( + f"- daily score panel: {sc['rows']} (date,symbol) rows; " + f"valid {sc['valid']}; NaN {sc['nan']}." + ) + diag = result.factor_diagnostics + if not diag: + lines.append("- _no settled rebalance to diagnose._") + lines.append("") + return + lines.append( + "- per settled rebalance — cross-sectional score stats over covered, " + "non-NaN-scored names, and **Spearman IC** of the decision-date score vs " + "the SAME period's execution-to-execution return " + f"(report-only; min sample {_MIN_IC_SAMPLE}, else NaN):" + ) + lines.append("") + lines.append("| date | n_scored | mean | std | p10 | p50 | p90 | IC(spearman) | IC_n |") + lines.append("|---|---|---|---|---|---|---|---|---|") + for r in diag: + ic_str = "NaN" if pd.isna(r["ic"]) else f"{r['ic']:.4f}" + lines.append( + f"| {pd.Timestamp(r['date']).date()} | {r['n_scored']} | " + f"{r['mean']:.6f} | {r['std']:.6f} | {r['p10']:.6f} | {r['p50']:.6f} | " + f"{r['p90']:.6f} | {ic_str} | {r['ic_n']} |" + ) + ics = [r["ic"] for r in diag if not pd.isna(r["ic"])] + if ics: + lines.append("") + lines.append( + f"- mean IC across {len(ics)} settled period(s) with sufficient sample: " + f"{float(np.mean(ics)):.4f}. **Report-only, NOT a performance claim** — " + "a 1–3 period intraday smoke is far too small to infer factor quality; " + "returns are read here for analytics only and never feed the factor/alpha." + ) + lines.append("") + + def _write_report(result: I5aResult, *, elapsed: float) -> None: """Write the I5a architecture-smoke markdown report (auditable event basis).""" cfg = result.config @@ -393,7 +604,10 @@ def _write_report(result: I5aResult, *, elapsed: float) -> None: path.parent.mkdir(parents=True, exist_ok=True) nav = result.nav_table - title, intro = _report_heading(result.price_limit_check) + title, intro = _report_heading( + result.score_feature_key, result.price_limit_check, + cfg.output.intraday_report_title, + ) lines: list[str] = [] lines.append(title) lines.append("") @@ -409,12 +623,16 @@ def _write_report(result: I5aResult, *, elapsed: float) -> None: lines.append( f"- execution_window: `[{ec.execution_window[0]}, {ec.execution_window[1]}]`" ) - lines.append(f"- score feature: `{result.score_feature}`") + lines.append( + f"- score feature: `{result.score_feature}` " + f"(key=`{result.score_feature_key}`)" + ) lines.append("") lines.append("**Returns are execution-to-execution, NOT close-to-close.** Holding " "return of a period = exec_price(next) / exec_price(this) - 1, priced " "at the first valid 1min close in the execution window.") lines.append("") + _append_factor_section(lines, result) lines.append("## Minute-cache coverage & data provenance") lines.append("") lines.append(f"- window: `{cfg.data.start}` → `{cfg.data.end}`") @@ -589,10 +807,19 @@ def _write_report(result: I5aResult, *, elapsed: float) -> None: "so they are blocked by the missing-bar rule; explicit ST status is NOT " "consulted at execution time in this smoke." ) - lines.append( - "- The score is a single PIT-safe intraday feature to prove the framework; " - "this is not a performance claim and no parameters were tuned." - ) + if result.score_feature_key == "mmp_ew": + lines.append( + "- **EXPLORATORY single-factor study, NOT a performance claim**: ONE " + "user-proposed minute factor (MMP) on one short SSE50 window. No " + "parameter was tuned from performance, no robustness matrix, no learned " + "/ IC-weighted alpha. The IC above is report-only over 1–3 periods — far " + "too small to infer factor quality; final NAV is reported, not claimed." + ) + else: + lines.append( + "- The score is a single PIT-safe intraday feature to prove the " + "framework; this is not a performance claim and no parameters were tuned." + ) lines.append("") lines.append(f"_elapsed: {elapsed:.1f}s_") path.write_text("\n".join(lines) + "\n", encoding="utf-8") From e3d8fdf6e6606be927bf0a6acd38cdb562cd2a82 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Mon, 15 Jun 2026 19:10:09 +0800 Subject: [PATCH 4/8] test: MMP factor formula/PIT + I5c config/runner coverage 16 I5c tests: exact MMP_t formula on a controlled 22-bar day; rolling baselines exclude bar t (t-20..t-1); first 20 bars NaN; daily score is equal-weight (asserts != volume-weighted); NaN-not-inf denominators; daily NaN when no valid minute; PIT (post-cutoff bars + delayed availability don't move the score); future bars don't change earlier MMP_t; multi-day no prior-tail carryover; multi-symbol isolation; old configs default to 'ret'; I5c config selects mmp_ew + titles the study; invalid score_feature fails; score_feature Literal mirrors feature keys; _score_panel selects MMP without prefix matching; heading names I5c. Update the I5b heading test for the new _report_heading signature. --- tests/test_i5b_execution_feasibility.py | 4 +- tests/test_i5c_mmp_minute_factor.py | 280 ++++++++++++++++++++++++ 2 files changed, 282 insertions(+), 2 deletions(-) create mode 100644 tests/test_i5c_mmp_minute_factor.py diff --git a/tests/test_i5b_execution_feasibility.py b/tests/test_i5b_execution_feasibility.py index 5f3e70f..521929c 100644 --- a/tests/test_i5b_execution_feasibility.py +++ b/tests/test_i5b_execution_feasibility.py @@ -396,8 +396,8 @@ def test_i5b_report_name_default_keeps_i5a_basename(): def test_i5b_report_heading_names_the_actual_study(): # The report H1 must not stay frozen at the I5a label when the limit check is on. from qt.intraday_tail_framework import _report_heading - i5a_title, _ = _report_heading(False) - i5b_title, i5b_intro = _report_heading(True) + i5a_title, _ = _report_heading("ret", False) + i5b_title, i5b_intro = _report_heading("ret", True) assert i5a_title.startswith("# Phase I5a") assert i5b_title.startswith("# Phase I5b") assert "execution-feasibility hardening" in i5b_intro.lower() diff --git a/tests/test_i5c_mmp_minute_factor.py b/tests/test_i5c_mmp_minute_factor.py new file mode 100644 index 0000000..d0d5008 --- /dev/null +++ b/tests/test_i5c_mmp_minute_factor.py @@ -0,0 +1,280 @@ +"""I5c: Minute Microstructure Pressure (MMP) factor — formula, PIT, config/runner. + +Covers the goal's test groups: + +1. MMP formula — exact MMP_t on a controlled 22-bar one-symbol day; rolling + baselines use t-20..t-1 and exclude t; the first 20 bars have NaN MMP; the daily + ``intraday_mmp20_ew_0930_1450`` is the arithmetic (equal-weight) mean of valid + MMP_t (NOT volume-weighted); zero/NaN denominators yield NaN, never inf. +2. PIT/leakage — post-14:50 bars don't change the daily MMP; moving a pre-cutoff + bar's available_time past the cutoff removes its effect; future bars within a day + don't change earlier MMP_t baselines; multi-symbol/multi-day isolation; no + prior-day tail feeds the new day's first-20 baseline. +3. Config/runner — old I5a/I5b configs validate and default to the ``ret`` score; + the I5c config selects ``mmp_ew``; an invalid score_feature fails readably; + ``_score_panel`` selects MMP without prefix matching; the report heading names + I5c; the config Literal mirrors INTRADAY_FEATURE_KEYS (drift guard). +""" + +from __future__ import annotations + +import logging +import typing +from pathlib import Path + +import numpy as np +import pandas as pd +import pytest +import yaml +from pydantic import ValidationError + +from data.clean.intraday_aggregate import ( + DEFAULT_FEATURE_KEYS, + INTRADAY_FEATURE_KEYS, + asof_daily_features, + compute_minute_mmp, + mmp_valid_minute_counts, +) +from data.clean.intraday_schema import normalize_intraday_bars +from qt.config import IntradayCfg, RootConfig, load_config + +_CONFIG_DIR = Path(__file__).resolve().parent.parent / "config" +_I5A_CONFIG = _CONFIG_DIR / "phase_i5a_intraday_tail_framework.yaml" +_I5B_CONFIG = _CONFIG_DIR / "phase_i5b_intraday_execution_feasibility.yaml" +_I5C_CONFIG = _CONFIG_DIR / "phase_i5c_mmp_minute_factor.yaml" + +_MMP_COL = "intraday_mmp20_ew_0930_1450" +_DAY = "2024-01-02" + + +# --------------------------------------------------------------------------- # +# builders +# --------------------------------------------------------------------------- # +def _mbars(specs: list[tuple]) -> pd.DataFrame: + """specs = [(time_str, symbol, open, high, low, close, volume), ...].""" + df = pd.DataFrame( + { + "time": pd.to_datetime([s[0] for s in specs]), + "symbol": [s[1] for s in specs], + "open": [s[2] for s in specs], + "high": [s[3] for s in specs], + "low": [s[4] for s in specs], + "close": [s[5] for s in specs], + "volume": [s[6] for s in specs], + "amount": [s[5] * s[6] for s in specs], + } + ) + return normalize_intraday_bars(df, freq="1min", data_lag="1min") + + +def _controlled_day(symbol="000001.SZ", n=22, start=f"{_DAY} 09:31:00", day=None): + """A controlled n-bar session with varied OHLCV (all pre-14:50).""" + base = pd.Timestamp(start if day is None else f"{day} 09:31:00") + specs = [] + for i in range(n): + t = base + pd.Timedelta(minutes=i) + k = i + 1 + specs.append( + (str(t), symbol, 10.0, 10.0 + 0.1 * k, 10.0 - 0.1 * k, + 10.0 + 0.05 * k, 100.0 + 7.0 * (k % 5)) # varied volume + ) + return specs + + +def _arrays(specs): + o = np.array([s[2] for s in specs], float) + h = np.array([s[3] for s in specs], float) + low = np.array([s[4] for s in specs], float) + c = np.array([s[5] for s in specs], float) + v = np.array([s[6] for s in specs], float) + return o, h, low, c, v + + +# --------------------------------------------------------------------------- # +# 1. MMP formula +# --------------------------------------------------------------------------- # +def test_i5c_mmp_exact_formula_and_first20_nan(): + specs = _controlled_day(n=22) + o, h, low, c, v = _arrays(specs) + mmp = compute_minute_mmp(o, h, low, c, v) + assert len(mmp) == 22 + assert np.all(np.isnan(mmp[:20])) # first 20 have insufficient lookback + assert np.all(~np.isnan(mmp[20:])) # 20, 21 are valid + + t = 20 # manual MMP_20 with baseline = bars 0..19 (excludes bar 20) + mid = (h[t] + low[t]) / 2.0 + S = (c[t] - mid) / mid + V = (v[t] / np.median(v[0:20])) ** 0.5 + B = abs(c[t] - o[t]) / (h[t] - low[t] + 1e-6) + R = (h[t] - low[t]) / (np.mean((h - low)[0:20]) + 1e-6) + assert mmp[t] == pytest.approx(S * V * B * R) + + +def test_i5c_rolling_baseline_excludes_bar_t(): + # Bumping ONLY bar t's volume must not change MMP_t's baseline (which is t-20..t-1). + specs = _controlled_day(n=25) + o, h, low, c, v = _arrays(specs) + base = compute_minute_mmp(o, h, low, c, v) + v2 = v.copy() + v2[22] *= 100.0 # perturb bar 22's own volume + bumped = compute_minute_mmp(o, h, low, c, v2) + # MMP_22 itself changes (its V term), but earlier MMP_t (t<22) baselines are intact. + assert np.allclose(base[:22], bumped[:22], equal_nan=True) + assert base[22] != pytest.approx(bumped[22]) + + +def test_i5c_daily_is_equal_weight_not_volume_weighted(): + specs = _controlled_day(n=25) + o, h, low, c, v = _arrays(specs) + mmp = compute_minute_mmp(o, h, low, c, v) + valid = ~np.isnan(mmp) + ew = float(np.mean(mmp[valid])) + vw = float(np.average(mmp[valid], weights=v[valid])) + assert abs(ew - vw) > 1e-6 # the test actually distinguishes the two + + out = asof_daily_features(_mbars(specs), features=["mmp_ew"]) + assert list(out.columns) == [_MMP_COL] + daily = out[_MMP_COL].iloc[0] + assert daily == pytest.approx(ew) + assert not np.isclose(daily, vw) + + +def test_i5c_invalid_denominators_are_nan_not_inf(): + # zero volume -> median baseline 0 -> V NaN; mid<=0 -> S NaN. Never inf. + flat = np.array([1.0] * 25) + v = np.array([0.0] * 25) + m = compute_minute_mmp(flat, flat, flat, flat, v) + assert np.all(np.isnan(m)) and not np.any(np.isinf(m)) + # mid <= 0 (degenerate negative prices) -> S NaN -> MMP NaN, not inf + neg = np.array([-1.0] * 25) + v2 = np.array([100.0] * 25) + m2 = compute_minute_mmp(neg, neg, neg, neg, v2) + assert not np.any(np.isinf(m2)) + assert np.all(np.isnan(m2[20:])) # where a baseline exists, mid<=0 -> NaN + + +def test_i5c_daily_nan_when_no_valid_minute(): + # only 5 bars -> never reaches the 20-bar lookback -> daily score NaN. + specs = _controlled_day(n=5) + out = asof_daily_features(_mbars(specs), features=["mmp_ew"]) + assert np.isnan(out[_MMP_COL].iloc[0]) + + +# --------------------------------------------------------------------------- # +# 2. PIT / leakage +# --------------------------------------------------------------------------- # +def test_i5c_post_cutoff_bars_do_not_change_daily_mmp(): + specs = _controlled_day(n=22) + base = asof_daily_features(_mbars(specs), features=["mmp_ew"]) + after = specs + [ + (f"{_DAY} 14:51:00", "000001.SZ", 99.0, 99.0, 99.0, 99.0, 9999.0), + (f"{_DAY} 14:55:00", "000001.SZ", 99.0, 99.0, 99.0, 99.0, 9999.0), + ] + perturbed = asof_daily_features(_mbars(after), features=["mmp_ew"]) + pd.testing.assert_frame_equal(base, perturbed) + + +def test_i5c_delayed_availability_excludes_bar(): + specs = _controlled_day(n=22) + bars = _mbars(specs) + base = asof_daily_features(bars, features=["mmp_ew"]) + # push one pre-cutoff bar's availability past the 14:50 cutoff + delayed = bars.copy() + target = pd.Timestamp(f"{_DAY} 09:40:00") + delayed.loc[delayed["bar_end"] == target, "available_time"] = pd.Timestamp( + f"{_DAY} 15:00:00" + ) + got = asof_daily_features(delayed, features=["mmp_ew"]) + dropped = bars[bars["bar_end"] != target] + expected = asof_daily_features(dropped, features=["mmp_ew"]) + pd.testing.assert_frame_equal(got, expected) + # and the exclusion actually changed the score (fewer bars -> different baseline/mean) + assert got[_MMP_COL].iloc[0] != base[_MMP_COL].iloc[0] + + +def test_i5c_future_bars_do_not_change_earlier_mmp(): + specs = _controlled_day(n=30) + o, h, low, c, v = _arrays(specs) + full = compute_minute_mmp(o, h, low, c, v) + prefix = compute_minute_mmp(o[:25], h[:25], low[:25], c[:25], v[:25]) + # earlier per-bar MMP_t use only t-20..t-1, so adding later bars cannot move them + assert np.allclose(full[:25], prefix, equal_nan=True) + + +def test_i5c_multi_day_no_prior_day_tail_carryover(): + # Same symbol, two consecutive days of 22 bars each. Each day's rolling baseline + # resets, so day-2's first 20 bars are NaN -> exactly 2 valid minutes (not 22). + s1 = _controlled_day(n=22, day="2024-01-02") + s2 = _controlled_day(n=22, day="2024-01-03") + counts = mmp_valid_minute_counts(_mbars(s1 + s2)) + assert counts.loc[(pd.Timestamp("2024-01-02"), "000001.SZ")] == 2 + assert counts.loc[(pd.Timestamp("2024-01-03"), "000001.SZ")] == 2 + + +def test_i5c_multi_symbol_isolation(): + a = _controlled_day(symbol="000001.SZ", n=22) + b = _controlled_day(symbol="000002.SZ", n=22) + out = asof_daily_features(_mbars(a + b), features=["mmp_ew"]) + # each symbol independently computed (and equals its solo computation) + solo_a = asof_daily_features(_mbars(a), features=["mmp_ew"]) + assert out.loc[(pd.Timestamp(_DAY), "000001.SZ"), _MMP_COL] == pytest.approx( + solo_a[_MMP_COL].iloc[0] + ) + assert (pd.Timestamp(_DAY), "000002.SZ") in out.index + + +# --------------------------------------------------------------------------- # +# 3. Config / runner +# --------------------------------------------------------------------------- # +def test_i5c_old_configs_default_to_ret_score(): + for path in (_I5A_CONFIG, _I5B_CONFIG): + cfg = load_config(str(path)) + assert cfg.intraday is not None + assert cfg.intraday.score_feature == "ret" # default preserved + + +def test_i5c_config_selects_mmp_and_titles_study(): + cfg = load_config(str(_I5C_CONFIG)) + assert cfg.intraday is not None + assert cfg.intraday.score_feature == "mmp_ew" + assert cfg.intraday.price_limit_check is True # I5b feasibility stays ON + assert cfg.output.intraday_report_name == "phase_i5c_mmp_minute_factor" + assert "I5c" in (cfg.output.intraday_report_title or "") + + +def test_i5c_invalid_score_feature_fails_readably(): + d = yaml.safe_load(_I5C_CONFIG.read_text()) + d["intraday"]["score_feature"] = "bogus" + with pytest.raises(ValidationError, match="score_feature"): + RootConfig(**d) + + +def test_i5c_score_feature_literal_mirrors_feature_keys(): + args = typing.get_args(IntradayCfg.model_fields["score_feature"].annotation) + assert set(args) == set(INTRADAY_FEATURE_KEYS) + assert "mmp_ew" not in DEFAULT_FEATURE_KEYS # selectable-only, not a default column + + +def test_i5c_score_panel_selects_mmp_without_prefix_matching(): + from qt.intraday_tail_framework import _score_panel + + cfg = load_config(str(_I5C_CONFIG)) + bars = _mbars(_controlled_day(n=25)) + series, col = _score_panel(cfg, bars, logging.getLogger("test.i5c")) + assert col == _MMP_COL + assert series.name == "score" + assert (pd.Timestamp(_DAY), "000001.SZ") in series.index + + +def test_i5c_report_heading_names_i5c_study(): + from qt.intraday_tail_framework import _report_heading + + title, intro = _report_heading("mmp_ew", True) + assert title.startswith("# Phase I5c") + assert "exploratory" in intro.lower() + # title override wins for the H1 (no stale phase label) + over, _ = _report_heading("mmp_ew", True, "Phase I5c — MMP Minute Factor Study") + assert over == "# Phase I5c — MMP Minute Factor Study" + # a non-mmp run with the limit check on still reads as I5b + i5b_title, _ = _report_heading("ret", True) + assert i5b_title.startswith("# Phase I5b") From 5afc86141e5677e72a02b65afd62cbb7060e50af Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Mon, 15 Jun 2026 19:27:21 +0800 Subject: [PATCH 5/8] fix(data): MMP daily aggregation honors the [session_open, decision_time] window MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The MMP daily score promised aggregation over bars with bar_end in [session_open, decision_time] but only applied the upper bound (available_time <= cutoff); pre-session bars (before session_open) leaked into the rolling baseline, so the first in-session bar wrongly got a non-NaN MMP and the report's [session_open, decision_time] claim did not hold. Add the lower bound bar_end >= trade_date + session_open on the MMP path only (_mmp_ew_daily + mmp_valid_minute_counts) so pre-session bars enter neither the rolling baseline nor the daily mean; the first 20 IN-SESSION bars are NaN. The existing four features (ret/realized_vol/vwap/last30m_ret) are untouched — only mmp_ew gains the lower bound — so I5a/I5b behavior cannot drift. Regression: 20 pre-session bars (09:00..09:19) + one 09:31 bar with session_open 09:30 -> daily MMP NaN / 0 valid minutes (was a spurious value); prepending pre-session bars leaves a full in-session day's MMP unchanged. Real SSE50 stk_mins has no pre-09:30 continuous bars, so the smoke is unaffected. --- data/clean/intraday_aggregate.py | 67 +++++++++++++++++++++-------- qt/intraday_tail_framework.py | 4 +- tests/test_i5c_mmp_minute_factor.py | 27 ++++++++++++ 3 files changed, 78 insertions(+), 20 deletions(-) diff --git a/data/clean/intraday_aggregate.py b/data/clean/intraday_aggregate.py index 57f2a38..d3058a9 100644 --- a/data/clean/intraday_aggregate.py +++ b/data/clean/intraday_aggregate.py @@ -189,6 +189,7 @@ def _compute_group( last_window_minutes: int, keys: list[str], epsilon: float, + session_open: str, ) -> dict[str, float]: """Per-(date, symbol) features from already PIT-filtered, bar_end-sorted bars.""" closes = g["close"].to_numpy(dtype=float) @@ -227,23 +228,42 @@ def _compute_group( } # MMP is the only rolling feature; compute it ONLY when requested (I5c). if "mmp_ew" in keys: - out["mmp_ew"] = _mmp_ew_daily(g, epsilon) + out["mmp_ew"] = _mmp_ew_daily(g, epsilon, trade_date, session_open) return out -def _mmp_ew_daily(g: pd.DataFrame, epsilon: float) -> float: +def _in_session(g: pd.DataFrame, trade_date: pd.Timestamp, session_open: str) -> pd.DataFrame: + """Bars whose ``bar_end`` is on/after ``session_open`` (the MMP window lower bound). + + The MMP daily score aggregates over ``[session_open, decision_time]`` (the upper + bound is the available_time cutoff already applied upstream). Restricting to the + in-session bars keeps PRE-session bars out of BOTH the rolling baseline and the + daily mean, so the first in-session bar correctly has no prior-20 baseline. + """ + session_start = trade_date + pd.Timedelta(session_open) + return g[g["bar_end"] >= session_start] + + +def _mmp_ew_daily( + g: pd.DataFrame, epsilon: float, trade_date: pd.Timestamp, session_open: str +) -> float: """Equal-weight mean of valid per-minute ``MMP_t`` over one PIT-filtered group. - ``g`` is one ``(date, symbol)`` session's visible bars, sorted by ``bar_end``. + ``g`` is one ``(date, symbol)`` session's visible bars, sorted by ``bar_end``; + only the in-session bars (``bar_end >= session_open``) enter, so the rolling + baseline starts at the session open and the first 20 in-session bars are NaN. Every valid minute ``MMP_t`` gets EQUAL weight (no extra volume weighting — the volume term already lives inside ``MMP_t``). No valid minute -> NaN. """ + gs = _in_session(g, trade_date, session_open) + if gs.empty: + return float("nan") mmp = compute_minute_mmp( - g["open"].to_numpy(dtype=float), - g["high"].to_numpy(dtype=float), - g["low"].to_numpy(dtype=float), - g["close"].to_numpy(dtype=float), - g["volume"].to_numpy(dtype=float), + gs["open"].to_numpy(dtype=float), + gs["high"].to_numpy(dtype=float), + gs["low"].to_numpy(dtype=float), + gs["close"].to_numpy(dtype=float), + gs["volume"].to_numpy(dtype=float), epsilon=epsilon, ) valid = mmp[~np.isnan(mmp)] @@ -289,7 +309,9 @@ def asof_daily_features( index_tuples: list[tuple] = [] data: dict[str, list[float]] = {c: [] for c in colnames} for (date, sym), g in visible.groupby(["trade_date", SYMBOL_LEVEL], sort=True): - feats = _compute_group(g, decision_time, last_window_minutes, keys, epsilon) + feats = _compute_group( + g, decision_time, last_window_minutes, keys, epsilon, session_open + ) index_tuples.append((date, str(sym))) for key, col in zip(keys, colnames): data[col].append(feats[key]) @@ -302,15 +324,17 @@ def mmp_valid_minute_counts( bars: pd.DataFrame, *, decision_time: str = DEFAULT_DECISION_TIME, + session_open: str = DEFAULT_SESSION_OPEN, epsilon: float = DEFAULT_EPSILON, ) -> pd.Series: """Per-``(date, symbol)`` count of valid (non-NaN) ``MMP_t`` minutes (I5c report). - Report-only diagnostic: applies the SAME PIT cutoff - (``available_time <= trade_date + decision_time``) and per-session grouping as - :func:`asof_daily_features`, then counts the visible minutes that yielded a - valid ``MMP_t`` (the first ``MMP_LOOKBACK`` bars of a session never do). Reuses - :func:`compute_minute_mmp` so there is a single MMP source of truth. + Report-only diagnostic: applies the SAME window as the daily MMP score — + ``available_time <= trade_date + decision_time`` (upper bound) AND + ``bar_end >= trade_date + session_open`` (lower bound) — then counts the + in-session minutes that yielded a valid ``MMP_t`` (the first ``MMP_LOOKBACK`` + in-session bars never do). Reuses :func:`compute_minute_mmp` so there is a + single MMP source of truth. """ validate_intraday_bars(bars) empty = pd.Series( @@ -332,12 +356,17 @@ def mmp_valid_minute_counts( index_tuples: list[tuple] = [] counts: list[int] = [] for (date, sym), g in visible.groupby(["trade_date", SYMBOL_LEVEL], sort=True): + gs = _in_session(g, pd.Timestamp(date).normalize(), session_open) + if gs.empty: + index_tuples.append((date, str(sym))) + counts.append(0) + continue mmp = compute_minute_mmp( - g["open"].to_numpy(dtype=float), - g["high"].to_numpy(dtype=float), - g["low"].to_numpy(dtype=float), - g["close"].to_numpy(dtype=float), - g["volume"].to_numpy(dtype=float), + gs["open"].to_numpy(dtype=float), + gs["high"].to_numpy(dtype=float), + gs["low"].to_numpy(dtype=float), + gs["close"].to_numpy(dtype=float), + gs["volume"].to_numpy(dtype=float), epsilon=epsilon, ) index_tuples.append((date, str(sym))) diff --git a/qt/intraday_tail_framework.py b/qt/intraday_tail_framework.py index bdf3206..cef262c 100644 --- a/qt/intraday_tail_framework.py +++ b/qt/intraday_tail_framework.py @@ -321,7 +321,9 @@ def _minute_count_summary(cfg: RootConfig, bars: pd.DataFrame) -> dict[str, floa """ ic = cfg.intraday assert ic is not None - counts = mmp_valid_minute_counts(bars, decision_time=ic.decision_time) + counts = mmp_valid_minute_counts( + bars, decision_time=ic.decision_time, session_open=ic.session_open + ) if counts.empty: return None arr = counts.to_numpy(dtype=float) diff --git a/tests/test_i5c_mmp_minute_factor.py b/tests/test_i5c_mmp_minute_factor.py index d0d5008..23966ce 100644 --- a/tests/test_i5c_mmp_minute_factor.py +++ b/tests/test_i5c_mmp_minute_factor.py @@ -211,6 +211,33 @@ def test_i5c_multi_day_no_prior_day_tail_carryover(): assert counts.loc[(pd.Timestamp("2024-01-03"), "000001.SZ")] == 2 +def test_i5c_pre_session_bars_excluded_from_window(): + # The MMP window is [session_open, decision_time]: a bar before session_open must + # NOT feed the rolling baseline. 20 pre-session bars (09:00..09:19) + ONE + # in-session bar (09:31) -> that bar has no prior-20 IN-SESSION baseline -> NaN. + specs = [ + (f"{_DAY} {9:02d}:{m:02d}:00", "000001.SZ", 10.0, 10.1, 9.9, 10.05, 100.0 + m) + for m in range(20) # 09:00..09:19, all before the 09:30 session open + ] + specs.append((f"{_DAY} 09:31:00", "000001.SZ", 10.0, 10.2, 9.8, 10.1, 200.0)) + out = asof_daily_features(_mbars(specs), features=["mmp_ew"], session_open="09:30:00") + assert np.isnan(out[_MMP_COL].iloc[0]) + counts = mmp_valid_minute_counts(_mbars(specs), session_open="09:30:00") + assert int(counts.iloc[0]) == 0 + + +def test_i5c_pre_session_bars_do_not_change_in_session_score(): + # Prepending pre-session bars must leave a full in-session day's MMP unchanged. + in_session = _controlled_day(n=25) # 09:31.. (all >= 09:30) + base = asof_daily_features(_mbars(in_session), features=["mmp_ew"]) + pre = [ + (f"{_DAY} {9:02d}:{m:02d}:00", "000001.SZ", 9.0, 9.1, 8.9, 9.0, 50.0) + for m in range(10) # 09:00..09:09, before session open + ] + withpre = asof_daily_features(_mbars(pre + in_session), features=["mmp_ew"]) + pd.testing.assert_frame_equal(base, withpre) + + def test_i5c_multi_symbol_isolation(): a = _controlled_day(symbol="000001.SZ", n=22) b = _controlled_day(symbol="000002.SZ", n=22) From 7e321f1a4ed593860cb10c0430615ddc1b942b47 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Mon, 15 Jun 2026 22:04:44 +0800 Subject: [PATCH 6/8] feat(runtime): optional precomputed exec prices for IntradayTailEventModel The execution-price matrix + fills are a pure function of (bars, anchor_dates, symbols, cfg). When several fresh models share the same bars/cfg (one per quantile group in I5d), pass the precomputed (prices, fills) so the heavy matrix is built once and reused while each model keeps its own fresh mutable feasibility diagnostics. Default None reproduces the byte-identical I5a/I5b in-place build (regression tests unchanged). --- runtime/backtest/event_models.py | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/runtime/backtest/event_models.py b/runtime/backtest/event_models.py index d175c2d..e93b211 100644 --- a/runtime/backtest/event_models.py +++ b/runtime/backtest/event_models.py @@ -142,6 +142,7 @@ def __init__( price_limit_check: bool = False, limit_tolerance: float = 1e-6, require_price_limit_coverage: bool = True, + precomputed_prices: tuple[pd.DataFrame, list[ExecutionFill]] | None = None, ) -> None: self._calendar_panel = calendar_panel self._cfg = cfg or IntradayExecutionConfig() @@ -154,7 +155,17 @@ def __init__( ) symbols = sorted({str(s) for s in bars.index.get_level_values("symbol")}) self._symbols = symbols - if anchor_dates and symbols: + # The execution-price matrix + fills are a PURE function of + # (bars, anchor_dates, symbols, cfg). When several models share the SAME + # bars/cfg (e.g. one fresh model per quantile group in I5d), the caller may + # pass ``precomputed_prices`` — the exact ``(prices, fills)`` returned by + # ``build_execution_prices`` over those same inputs — so the heavy matrix is + # built ONCE and reused, while each model keeps its OWN fresh mutable + # feasibility diagnostics. Default None reproduces the original build + # in-place (byte-identical I5a/I5b behaviour, locked by tests). + if precomputed_prices is not None: + prices, fills = precomputed_prices + elif anchor_dates and symbols: prices, fills = build_execution_prices( bars, anchor_dates, symbols, self._cfg ) From a8e245344705025558ed7561c653a6f81add3414 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Mon, 15 Jun 2026 22:05:08 +0800 Subject: [PATCH 7/8] feat(qt): I5d MMP quintile grouped intraday backtest (exploratory) Rank the I5c MMP daily score (intraday_mmp20_ew_0930_1450) on each monthly rebalance date into analytics.quantiles equal-count rank buckets (Q1 = lowest, QN = highest) and run each group as its own long-only equal-weight portfolio through the SAME BacktestEngine + IntradayTailEventModel + SimExecution (fee_rate) with I5b raw stk_limit execution feasibility ON. - intraday_groups.py: assign_quantile_buckets (equal-count rank buckets, deterministic ties, too-few-names leaves high groups empty), GroupScores, EqualWeightAll (whole-bucket equal weight, ignores top_n). - intraday_group_backtest.py: run-phase-i5d-intraday-groups runner. Cache-only, anchor-date-sliced minute load (only rebalance/exit days -> memory-conscious, zero stk_mins live calls), one shared immutable exec-price matrix across N fresh per-group models, per-group metrics, QN-Q1 synthetic spread + group monotonicity (report-only). - intraday_group_figures.py / intraday_group_report.py: Agg PNG figures (NAV curves / Q5-Q1 spread / grouped metric bars) + markdown report. - cli.py: run-phase-i5d-intraday-groups subcommand. - config/phase_i5d_mmp_quintile_5y.yaml: CSI500, 2021-06-01..2026-05-31, mmp_ew, price_limit_check on, quantiles=5, fee_rate=0.001. require_cache_coverage=false drops minute-uncovered constituents (disclosed; window not shortened). Factor math, daily close-to-close, the event model and execution feasibility are unchanged; only the grouping + per-group orchestration is new. EXPLORATORY, not a performance claim, no tuning. --- config/phase_i5d_mmp_quintile_5y.yaml | 129 ++++++ qt/cli.py | 31 ++ qt/intraday_group_backtest.py | 611 ++++++++++++++++++++++++++ qt/intraday_group_figures.py | 139 ++++++ qt/intraday_group_report.py | 345 +++++++++++++++ qt/intraday_groups.py | 131 ++++++ 6 files changed, 1386 insertions(+) create mode 100644 config/phase_i5d_mmp_quintile_5y.yaml create mode 100644 qt/intraday_group_backtest.py create mode 100644 qt/intraday_group_figures.py create mode 100644 qt/intraday_group_report.py create mode 100644 qt/intraday_groups.py diff --git a/config/phase_i5d_mmp_quintile_5y.yaml b/config/phase_i5d_mmp_quintile_5y.yaml new file mode 100644 index 0000000..cf42d12 --- /dev/null +++ b/config/phase_i5d_mmp_quintile_5y.yaml @@ -0,0 +1,129 @@ +# Phase I5d — MMP Quintile 5Y Group Backtest (EXPLORATORY, NOT a performance claim). +# +# Ranks the I5c Minute Microstructure Pressure (MMP) daily score +# (intraday_mmp20_ew_0930_1450) on each monthly rebalance date and splits the +# cross-section into analytics.quantiles EQUAL-COUNT groups (Q1 = lowest score, +# Q5 = highest). Each group runs as its own long-only equal-weight portfolio +# through the SAME I5a/I5b intraday tail event model (14:50 decision / 14:51 +# execution, exec-to-exec returns) with raw stk_limit execution feasibility ON and +# trading cost fee_rate=0.001. Factor math / event model / execution feasibility +# are UNCHANGED — only the grouping + per-group orchestration is new. No tuning. +# +# Window = 5 complete years (2021-06-01 .. 2026-05-31). Primary universe = CSI500 +# (000905.SH): enough cross-section for 5 groups. Minute bars are read from the +# EXISTING intraday cache ONLY on the monthly anchor dates (read-only; a miss is a +# hard blocker -> zero stk_mins live calls). require_cache_coverage=false drops the +# minute-uncovered constituents (DISCLOSED in the report; the WINDOW is not +# shortened). Raw stk_limit flows through the same P4 read-through cache as the +# daily endpoints. + +project: + name: quantitative_trading_phase_i5d_mmp_quintile_5y + timezone: Asia/Shanghai +data: + source: tushare + freq: D + start: '2021-06-01' + end: '2026-05-31' + external_secret_file: /home/shaofl/Projects/financial_projects/.config.json + tushare_token_key: tushare.token + output_name: i5d_daily + cache: + enabled: true + root_dir: artifacts/cache/tushare/v1 + refresh_recent_days: 14 + refresh_dimension_days: 30 + force_refresh: [] +universe: + type: index + index_code: 000905.SH + symbols: [] + min_listing_days: 60 + filters: + # Selection-level daily tradability flags stay DISABLED (as in I5b/I5c): the + # daily close limit flag is an EOD value not known at 14:50. The honest + # direction-aware blocking is execution-time raw stk_limit vs the raw + # execution-minute close. + missing_close: true + suspended: false + st: false + limit_up_down: false +# factors is required by the schema but the runner ignores it: the score is the I3 +# mmp_ew feature (selected via intraday.score_feature), not a config factor. +factors: +- name: momentum_20 + enabled: true + params: + window: 20 + price_col: close +processing: + drop_missing: true + standardize: + enabled: true + method: zscore + winsorize: + enabled: false + method: mad + n: 3.0 + neutralize: + enabled: false + industry_col: industry + size_col: market_cap + industry_level: L1 +alpha: + model: equal_weight + params: {} +portfolio: + # top_n is required by the schema but IGNORED by the grouped backtest: each + # quantile group holds ALL its bucket members (EqualWeightAll), not a top-N cut. + constructor: topn_equal_weight + top_n: 10 + long_only: true + max_weight: null + turnover_cap: null +backtest: + initial_nav: 1.0 + rebalance: monthly + event_order: intraday_tail_rebalance + cash_return: 0.0 +intraday: + enabled: true + decision_time: '14:50:00' + data_lag: 1min + session_open: '09:30:00' + execution_model: next_minute_close + execution_window: + - '14:51:00' + - '14:56:59' + # A 5-year CSI500 minute history is not fully cached for every constituent; drop + # the uncovered names (disclosed in the report) instead of failing the whole run. + # The window is NOT shortened. + require_cache_coverage: false + missing_execution: block + # I5b execution-time raw price-limit feasibility stays ON. + price_limit_check: true + require_price_limit_coverage: true + limit_tolerance: 1.0e-06 + # I5c: select the exploratory MMP minute factor as the PIT-safe daily score. + score_feature: mmp_ew +cost: + fee_rate: 0.001 + slippage_rate: 0.0 + turnover_formula: l1 +analytics: + forward_return_periods: + - 1 + - 5 + - 20 + # number of equal-count quantile groups for the I5d grouped backtest. + quantiles: 5 + benchmark: null +output: + root_dir: artifacts + data_dir: artifacts/data + factor_dir: artifacts/factors + report_dir: artifacts/reports + log_dir: artifacts/logs + overwrite: true + intraday_report_name: phase_i5d_mmp_quintile_5y + intraday_report_title: Phase I5d — MMP Quintile 5Y Group Backtest diff --git a/qt/cli.py b/qt/cli.py index af6b865..c43b85a 100644 --- a/qt/cli.py +++ b/qt/cli.py @@ -179,6 +179,30 @@ def _cmd_run_phase_i5a_intraday(args: argparse.Namespace) -> int: return 0 +def _cmd_run_phase_i5d_intraday_groups(args: argparse.Namespace) -> int: + """Run the I5d MMP quintile grouped intraday-tail backtest + report/figures.""" + from qt.intraday_group_backtest import run_phase_i5d_intraday_groups + + try: + result = run_phase_i5d_intraday_groups(args.config) + except (ConfigError, ValueError, FileNotFoundError, RuntimeError) as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 1 + navs = " ".join( + f"Q{g.group}={g.metrics['final_nav']:.4f}" for g in result.groups + ) + print( + f"OK run-phase-i5d-intraday-groups: groups={result.n_groups}, " + f"rebalances={result.rebalance_count}, " + f"covered={result.covered_symbols}/{result.requested_symbols}, " + f"stk_mins_live_calls={result.minute_live_calls}, " + f"fee_rate={result.config.cost.fee_rate}\n" + f"final_nav: {navs}\n" + f"report: {result.report_path}" + ) + return 0 + + def _cmd_data_update(args: argparse.Namespace) -> int: """Warm/update the tushare caches (P4-3); never runs a backtest.""" from qt.data_updater import format_summary, run_data_update @@ -269,6 +293,13 @@ def build_parser() -> argparse.ArgumentParser: p_i5a.add_argument("--config", required=True, help="Path to the YAML config.") p_i5a.set_defaults(func=_cmd_run_phase_i5a_intraday) + p_i5d = sub.add_parser( + "run-phase-i5d-intraday-groups", + help="Run the I5d MMP quintile grouped intraday-tail backtest (5 groups).", + ) + p_i5d.add_argument("--config", required=True, help="Path to the YAML config.") + p_i5d.set_defaults(func=_cmd_run_phase_i5d_intraday_groups) + for name, func, help_text in ( ("fetch-data", _cmd_fetch_data, "Run the spine, report data fetch."), ("compute-factors", _cmd_compute_factors, "Run the spine, report factor compute."), diff --git a/qt/intraday_group_backtest.py b/qt/intraday_group_backtest.py new file mode 100644 index 0000000..bee0e9a --- /dev/null +++ b/qt/intraday_group_backtest.py @@ -0,0 +1,611 @@ +"""run-phase-i5d-intraday-groups: a 5-quantile grouped intraday-tail backtest of +the I5c MMP daily score (EXPLORATORY factor analysis — NOT a performance claim). + +On each monthly rebalance date the cross-section is ranked by the PIT-safe daily +score ``intraday_mmp20_ew_0930_1450`` (the I5c Minute Microstructure Pressure +factor, visible at the 14:50 cutoff) and split into ``analytics.quantiles`` +EQUAL-COUNT groups (Q1 = lowest score, QN = highest). Each group is run as its own +long-only equal-weight portfolio through the SAME event-driven machinery the daily +and I5a/I5b intraday paths use — a fresh :class:`BacktestEngine` + +:class:`IntradayTailEventModel` + ``SimExecution(fee_rate=...)`` per group — so +14:51 execution pricing, exec-to-exec holding returns, raw ``stk_limit`` execution +feasibility (I5b), turnover, cost and cash are modelled consistently and NEVER +fall back to a daily close-to-close return. + +Two engineering constraints drive the design: + + * **memory** — a 5-year CSI500 minute history is enormous, but the monthly event + model only needs minute bars on the ANCHOR dates (each rebalance/exit date). + :func:`_load_anchor_minute_bars` reads ONLY those days from the persistent + intraday cache (read-only; a miss never warms the cache → zero ``stk_mins`` + live calls), so the run never materialises the full-window minute panel. + * **compute** — the execution-price matrix is a pure function of + ``(bars, anchors, symbols, cfg)``; it is built ONCE and shared (immutable) + across the N fresh per-group models, while each model keeps its OWN mutable + feasibility diagnostics (no cross-group contamination). + +The MMP score, the factor math, the daily backtest, and the execution feasibility +are all UNCHANGED — only the grouping + per-group orchestration is new. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from pathlib import Path + +import numpy as np +import pandas as pd + +from data.cache.intervals import subtract_intervals +from data.cache.intraday_cache import ENDPOINT as INTRADAY_ENDPOINT +from data.cache.intraday_cache import READ_COLUMNS +from data.cache.intraday_coverage import IntradayCoverageLedger +from data.cache.intraday_parquet_store import IntradayParquetStore +from data.clean.intraday_schema import RAW_INTRADAY_FREQ, normalize_intraday_bars +from portfolio.base import PortfolioConstructor +from qt.config import RootConfig, load_config + +# Reuse the stable I5a/I5b/I5c helpers rather than copy-paste them (goal §1: avoid +# drift). These are import-stable private helpers: run preconditions, the exec +# config builder, the configured-feature daily score, and the raw stk_limit loader. +from qt.intraday_groups import EqualWeightAll, GroupScores, assign_quantile_buckets +from qt.intraday_tail_framework import ( + _check_i5a_preconditions, + _exec_cfg_from, + _load_price_limits, + _score_panel, +) +from qt.pipeline import _build_cache, _build_universe, _load_panel, _make_logger +from runtime.backtest.engine import BacktestEngine +from runtime.backtest.event_models import IntradayTailEventModel +from runtime.backtest.events import monthly_anchor_pairs, trading_calendar +from runtime.backtest.sim_execution import SimExecution +from runtime.intraday_execution import IntradayExecutionConfig + +_LOGGER_NAME = "qt.intraday_group_backtest" +_DEFAULT_REPORT_NAME = "phase_i5d_mmp_quintile_5y" +_DEFAULT_REPORT_TITLE = "Phase I5d — MMP Quintile 5Y Group Backtest" + +# Monthly rebalance -> 12 periods/year for annualization. +_PERIODS_PER_YEAR = 12.0 + + +# --------------------------------------------------------------------------- # +# Per-group performance metrics (monthly net-return series -> annualized stats) +# --------------------------------------------------------------------------- # +def _annualized_return(nav: pd.DataFrame) -> float: + """CAGR from the settled NAV path (monthly compounding).""" + if nav.empty: + return float("nan") + final = float(nav["nav"].iloc[-1]) + n = len(nav) + if n <= 0 or final <= 0: + return float("nan") + return final ** (_PERIODS_PER_YEAR / n) - 1.0 + + +def _annualized_vol(nav: pd.DataFrame) -> float: + """Annualized volatility of the monthly net returns.""" + if len(nav) < 2: + return float("nan") + return float(nav["net_return"].std(ddof=1) * np.sqrt(_PERIODS_PER_YEAR)) + + +def _sharpe(nav: pd.DataFrame) -> float: + """Annualized Sharpe (rf=0) = mean(net)*12 / (std(net)*sqrt(12)).""" + sigma = _annualized_vol(nav) + if not np.isfinite(sigma) or sigma == 0.0: + return float("nan") + mu = float(nav["net_return"].mean()) * _PERIODS_PER_YEAR + return mu / sigma + + +def _max_drawdown(nav: pd.DataFrame) -> float: + """Worst peak-to-trough drawdown of the NAV path (incl. the 1.0 start).""" + if nav.empty: + return float("nan") + path = np.concatenate([[1.0], nav["nav"].to_numpy(dtype=float)]) + running_max = np.maximum.accumulate(path) + return float(np.min(path / running_max - 1.0)) + + +def _avg_holdings(holdings_log: pd.DataFrame) -> float: + """Mean number of achieved holdings per settled rebalance date.""" + if holdings_log is None or holdings_log.empty: + return 0.0 + return float(holdings_log.groupby("date").size().mean()) + + +def _group_metrics(nav: pd.DataFrame, holdings_log: pd.DataFrame) -> dict[str, float]: + """Headline per-group metrics from one group's NAV table + holdings log.""" + final_nav = float(nav["nav"].iloc[-1]) if not nav.empty else float("nan") + return { + "final_nav": final_nav, + "annual_return": _annualized_return(nav), + "volatility": _annualized_vol(nav), + "sharpe": _sharpe(nav), + "max_drawdown": _max_drawdown(nav), + "mean_turnover": float(nav["turnover"].mean()) if not nav.empty else float("nan"), + "total_cost": float(nav["cost"].sum()) if not nav.empty else 0.0, + "avg_holdings": _avg_holdings(holdings_log), + "n_periods": int(len(nav)), + } + + +# --------------------------------------------------------------------------- # +# Minute loading (cache-only, anchor-date-sliced -> memory-conscious) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class _MinuteLoad: + """Diagnostics from the cache-only, anchor-date-sliced minute read.""" + + bars: pd.DataFrame + covered: list[str] + uncovered: list[str] + raw_rows: int + normalized_rows: int + anchor_dates: int + live_calls: int + + +def _coverage_partition( + cfg: RootConfig, symbols: list[str], root: str +) -> tuple[list[str], list[str]]: + """Split ``symbols`` into (fully covered, uncovered) over the data window. + + A symbol is covered only if the intraday ledger leaves NO uncovered trading-day + gap across ``[data.start, data.end]`` — exactly the I5a rule, so the realized + universe is auditable and never silently shrunk by a partial gap. + """ + ledger = IntradayCoverageLedger(root) + req_start = pd.Timestamp(cfg.data.start).normalize() + req_end = pd.Timestamp(cfg.data.end).normalize() + covered: list[str] = [] + uncovered: list[str] = [] + for sym in symbols: + gaps = subtract_intervals( + req_start, + req_end, + ledger.covered_day_intervals(INTRADAY_ENDPOINT, sym, RAW_INTRADAY_FREQ), + ) + (uncovered if gaps else covered).append(sym) + return covered, uncovered + + +def _load_anchor_minute_bars( + cfg: RootConfig, + symbols: list[str], + anchor_dates: list[pd.Timestamp], + logger, +) -> _MinuteLoad: + """Read 1min bars for ONLY the ``anchor_dates`` from the cache (read-only). + + Memory-conscious: the monthly event model prices only the rebalance/exit + anchors, so we read one trading day at a time from the month-partitioned store + and keep just those rows — the full multi-year minute history is NEVER + materialised. No fetch closure exists here, so ``stk_mins`` live calls are + provably zero (a miss simply yields no rows for that day/symbol). Uncovered + symbols are handled exactly as I5a: in ``require_cache_coverage`` mode a single + uncovered name is a loud blocker; otherwise they are excluded and disclosed. + """ + root = cfg.data.cache.root_dir + covered, uncovered = _coverage_partition(cfg, symbols, root) + + require_cov = cfg.intraday is not None and cfg.intraday.require_cache_coverage + if require_cov and uncovered: + shown = ", ".join(uncovered[:10]) + more = "" if len(uncovered) <= 10 else f" (+{len(uncovered) - 10} more)" + raise ValueError( + "i5d grouped backtest blocked: require_cache_coverage=true but " + f"{len(uncovered)}/{len(symbols)} requested symbols are NOT fully " + f"covered in the minute cache for [{cfg.data.start}, {cfg.data.end}]: " + f"{shown}{more}. Set intraday.require_cache_coverage=false to drop the " + "uncovered names (disclosed), or shrink the window to fully-covered " + "data. This runner refuses to warm missing minute history." + ) + if uncovered: + logger.info( + "intraday cache: %d/%d symbols fully covered; %d excluded (uncovered, " + "require_cache_coverage=false)", + len(covered), len(symbols), len(uncovered), + ) + if not covered: + raise ValueError( + "i5d grouped backtest blocked: no requested symbol is fully covered in " + f"the minute cache for [{cfg.data.start}, {cfg.data.end}]." + ) + + store = IntradayParquetStore(root) + frames: list[pd.DataFrame] = [] + raw_rows = 0 + for anchor in anchor_dates: + day = pd.Timestamp(anchor).normalize() + day_end = day + pd.Timedelta("23:59:59") + for sym in covered: + part = store.read_range( + INTRADAY_ENDPOINT, sym, RAW_INTRADAY_FREQ, day, day_end + ) + if not part.empty: + hit = part.rename(columns={"bar_end": "time"}) + frames.append(hit[READ_COLUMNS]) + raw_rows += len(part) + if not frames: + raise ValueError( + "i5d grouped backtest blocked: the covered symbols returned no cached " + "1min bars on the anchor dates (unexpected — check the coverage ledger)." + ) + read = pd.concat(frames, ignore_index=True) + bars = normalize_intraday_bars( + read, freq=RAW_INTRADAY_FREQ, data_lag=cfg.intraday.data_lag + ) + logger.info( + "intraday minute load (anchor-sliced): %d anchor dates, %d covered symbols, " + "%d raw rows, %d normalized rows, stk_mins_live_calls=0", + len(anchor_dates), len(covered), raw_rows, len(read), + ) + return _MinuteLoad( + bars=bars, + covered=covered, + uncovered=uncovered, + raw_rows=raw_rows, + normalized_rows=int(len(read)), + anchor_dates=len(anchor_dates), + live_calls=0, + ) + + +# --------------------------------------------------------------------------- # +# Group assignment (single source of truth, shared by all N group engines) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class _GroupAssignment: + """Per-date bucket assignment + per-date diagnostics (immutable).""" + + by_date: dict[pd.Timestamp, dict[str, int]] + per_date_rows: tuple[dict, ...] + + +def _build_group_assignment( + score_series: pd.Series, + universe, + panel: pd.DataFrame, + rebalance_dates: list[pd.Timestamp], + covered: set[str], + n_groups: int, +) -> _GroupAssignment: + """Rank the MMP cross-section into N equal-count buckets on each rebalance date. + + Uses ONLY the daily PIT universe (``universe.tradable``) ∩ covered names ∩ + valid MMP scores — no forward return ever enters. Returns the per-date + ``{symbol: group}`` map and the per-date diagnostics (scored count, group + sizes, score distribution). + """ + by_date: dict[pd.Timestamp, dict[str, int]] = {} + rows: list[dict] = [] + for d in rebalance_dates: + norm = pd.Timestamp(d).normalize() + tradable = [s for s in universe.tradable(d, panel) if str(s) in covered] + try: + cross = score_series.xs(norm, level="date") + except KeyError: + cross = pd.Series(dtype=float) + cross = cross.reindex([str(s) for s in tradable]).dropna() + assignment = assign_quantile_buckets(cross, n_groups) + by_date[norm] = assignment + sizes = [0] * n_groups + for g in assignment.values(): + sizes[g - 1] += 1 + if len(cross) > 0: + arr = cross.to_numpy(dtype=float) + stats = { + "mean": float(np.mean(arr)), + "std": float(np.std(arr)), + "p10": float(np.percentile(arr, 10)), + "p50": float(np.percentile(arr, 50)), + "p90": float(np.percentile(arr, 90)), + } + else: + stats = {k: float("nan") for k in ("mean", "std", "p10", "p50", "p90")} + rows.append( + {"date": norm, "n_scored": int(len(cross)), "sizes": tuple(sizes), **stats} + ) + return _GroupAssignment(by_date=by_date, per_date_rows=tuple(rows)) + + +# --------------------------------------------------------------------------- # +# Result container +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class GroupRunResult: + """One quantile group's settled run + diagnostics (immutable).""" + + group: int + nav_table: pd.DataFrame + holdings_log: pd.DataFrame + metrics: dict[str, float] + up_limit_blocked_buys: int + down_limit_blocked_sells: int + missing_limit_rows: int + blocked_fill_reasons: dict[str, int] + + +@dataclass(frozen=True) +class I5dResult: + """Immutable summary of the I5d grouped backtest run.""" + + config: RootConfig + n_groups: int + score_feature: str + score_feature_key: str + requested_symbols: int + covered_symbols: int + uncovered_symbols: tuple[str, ...] + anchor_dates: int + raw_rows: int + normalized_rows: int + minute_live_calls: int + rebalance_count: int + groups: tuple[GroupRunResult, ...] + spread_per_period: pd.Series + spread_cumulative: pd.Series + spread_summary: dict[str, float] + monotonicity: dict[str, float] + per_date_rows: tuple[dict, ...] + score_coverage: dict[str, int] + price_limit_check: bool + limit_coverage: dict[str, int] + stk_limit_gap_fetches: int + figure_paths: dict[str, Path] + report_path: Path + log_path: Path + elapsed: float + + +# --------------------------------------------------------------------------- # +# Runner +# --------------------------------------------------------------------------- # +def _report_basename(cfg: RootConfig) -> str: + return cfg.output.intraday_report_name or _DEFAULT_REPORT_NAME + + +def _run_one_group( + group: int, + assignment: _GroupAssignment, + *, + cfg: RootConfig, + panel: pd.DataFrame, + bars: pd.DataFrame, + universe, + exec_cfg: IntradayExecutionConfig, + price_limits: pd.DataFrame | None, + shared_prices: tuple[pd.DataFrame, list] | None, + constructor: PortfolioConstructor, +) -> tuple[GroupRunResult, tuple[pd.DataFrame, list]]: + """Run ONE quantile group through a fresh engine/model/execution. + + A FRESH :class:`IntradayTailEventModel` and ``SimExecution`` per group (no + shared mutable state); the immutable execution-price matrix is built by the + first group and reused by the rest (``shared_prices``). Returns the group + result and the shared prices (so the caller can thread them to the next group). + """ + ic = cfg.intraday + assert ic is not None + model = IntradayTailEventModel( + calendar_panel=panel, + bars=bars, + cfg=exec_cfg, + price_limits=price_limits, + price_limit_check=ic.price_limit_check, + limit_tolerance=ic.limit_tolerance, + require_price_limit_coverage=ic.require_price_limit_coverage, + precomputed_prices=shared_prices, + ) + if shared_prices is None: + # First group builds the matrix; reuse it (immutable) for the rest. + shared_prices = (model.execution_prices(), model.fills()) + + execution = SimExecution(fee_rate=cfg.cost.fee_rate) + engine = BacktestEngine( + model=model, + universe=universe, + scores=GroupScores(assignment.by_date, group), + constructor=constructor, + execution=execution, + selection_panel=panel, + initial_nav=cfg.backtest.initial_nav, + cash_return=cfg.backtest.cash_return, + ) + nav = engine.run() + holdings = engine.holdings_log() + blocked: dict[str, int] = {} + for f in model.blocked_fills(): + key = f.reason or "unknown" + blocked[key] = blocked.get(key, 0) + 1 + result = GroupRunResult( + group=group, + nav_table=nav, + holdings_log=holdings, + metrics=_group_metrics(nav, holdings), + up_limit_blocked_buys=model.up_limit_blocked_buys(), + down_limit_blocked_sells=model.down_limit_blocked_sells(), + missing_limit_rows=model.missing_limit_rows(), + blocked_fill_reasons=blocked, + ) + return result, shared_prices + + +def _spread_and_monotonicity( + groups: tuple[GroupRunResult, ...], n_groups: int +) -> tuple[pd.Series, pd.Series, dict[str, float], dict[str, float]]: + """Synthetic QN−Q1 spread series + group monotonicity diagnostics (report-only). + + The spread is the per-period difference of the high-group and low-group NET + returns and its CUMULATIVE SUM — a synthetic long-only leg difference, NOT a + separately executed dollar-neutral portfolio. Monotonicity is the Spearman + rank correlation of the group index (1..N) vs the group annual return / final + NAV. + """ + by_group = {g.group: g for g in groups} + low, high = by_group.get(1), by_group.get(n_groups) + if low is None or high is None or low.nav_table.empty or high.nav_table.empty: + empty = pd.Series(dtype=float) + return empty, empty, {}, {} + aligned = pd.concat( + [high.nav_table["net_return"].rename("high"), + low.nav_table["net_return"].rename("low")], + axis=1, + ).dropna() + per_period = (aligned["high"] - aligned["low"]).rename("spread") + cumulative = per_period.cumsum().rename("cum_spread") + summary = { + "mean_per_period": float(per_period.mean()) if len(per_period) else float("nan"), + "total": float(cumulative.iloc[-1]) if len(cumulative) else float("nan"), + "n_periods": int(len(per_period)), + } + idx = pd.Series( + [g for g in sorted(by_group)], dtype=float, index=sorted(by_group) + ) + annual = pd.Series( + {g: by_group[g].metrics["annual_return"] for g in sorted(by_group)} + ) + final_nav = pd.Series( + {g: by_group[g].metrics["final_nav"] for g in sorted(by_group)} + ) + mono = { + "annual_spearman": float(idx.corr(annual, method="spearman")), + "final_nav_spearman": float(idx.corr(final_nav, method="spearman")), + } + return per_period, cumulative, summary, mono + + +def run_phase_i5d_intraday_groups(config_path: str) -> I5dResult: + """Run the I5d MMP quintile grouped intraday-tail backtest and write its report.""" + cfg = load_config(config_path) + _check_i5a_preconditions(cfg) # real tushare, intraday enabled, event order, cache + ic = cfg.intraday + assert ic is not None + if ic.score_feature != "mmp_ew": + raise ValueError( + "run-phase-i5d-intraday-groups is the MMP quintile study and requires " + f"intraday.score_feature='mmp_ew' (got {ic.score_feature!r})." + ) + n_groups = int(cfg.analytics.quantiles) + if n_groups < 2: + raise ValueError( + f"analytics.quantiles must be >= 2 for a grouped backtest; got {n_groups}." + ) + + basename = _report_basename(cfg) + log_dir = Path(cfg.output.log_dir) + log_dir.mkdir(parents=True, exist_ok=True) + log_path = log_dir / f"{basename}.log" + logger = _make_logger(log_path, name=_LOGGER_NAME) + started = time.monotonic() + + cache = _build_cache(cfg) + universe, symbols = _build_universe(cfg, logger, cache) + panel = _load_panel(cfg, symbols, logger, cache) # daily qfq selection panel + calendar + + # Monthly anchors from the daily calendar; minute bars only on those days. + pairs = monthly_anchor_pairs(trading_calendar(panel)) + rebalance_dates = [pd.Timestamp(p[0]).normalize() for p in pairs] + anchor_dates = sorted({pd.Timestamp(d).normalize() for pair in pairs for d in pair}) + logger.info( + "schedule: %d monthly rebalances, %d anchor dates", len(pairs), len(anchor_dates) + ) + + load = _load_anchor_minute_bars(cfg, symbols, anchor_dates, logger) + score_series, score_feature = _score_panel(cfg, load.bars, logger) + score_coverage = { + "rows": int(score_series.shape[0]), + "valid": int(score_series.notna().sum()), + "nan": int(score_series.isna().sum()), + } + + price_limits, stk_limit_gap_fetches = _load_price_limits( + cfg, load.covered, cache, logger + ) + exec_cfg = _exec_cfg_from(cfg) + + assignment = _build_group_assignment( + score_series, universe, panel, rebalance_dates, set(load.covered), n_groups + ) + + constructor = EqualWeightAll(long_only=cfg.portfolio.long_only) + group_results: list[GroupRunResult] = [] + shared_prices: tuple[pd.DataFrame, list] | None = None + limit_coverage: dict[str, int] = {} + for group in range(1, n_groups + 1): + result, shared_prices = _run_one_group( + group, + assignment, + cfg=cfg, + panel=panel, + bars=load.bars, + universe=universe, + exec_cfg=exec_cfg, + price_limits=price_limits, + shared_prices=shared_prices, + constructor=constructor, + ) + group_results.append(result) + logger.info( + "group Q%d: periods=%d final_nav=%.6f up_blocked=%d down_blocked=%d", + group, result.metrics["n_periods"], result.metrics["final_nav"], + result.up_limit_blocked_buys, result.down_limit_blocked_sells, + ) + # limit coverage is identical across groups (derived from the shared prices); + # read it once from a fresh model so the diagnostic is reported a single time. + if ic.price_limit_check: + cov_model = IntradayTailEventModel( + calendar_panel=panel, bars=load.bars, cfg=exec_cfg, + price_limits=price_limits, price_limit_check=True, + limit_tolerance=ic.limit_tolerance, + require_price_limit_coverage=ic.require_price_limit_coverage, + precomputed_prices=shared_prices, + ) + limit_coverage = cov_model.limit_coverage() + + per_period, cumulative, spread_summary, monotonicity = _spread_and_monotonicity( + tuple(group_results), n_groups + ) + + report_path = Path(cfg.output.report_dir) / f"{basename}.md" + figure_dir = Path(cfg.output.report_dir) / f"{basename}_figures" + result = I5dResult( + config=cfg, + n_groups=n_groups, + score_feature=score_feature, + score_feature_key=ic.score_feature, + requested_symbols=len(symbols), + covered_symbols=len(load.covered), + uncovered_symbols=tuple(load.uncovered), + anchor_dates=load.anchor_dates, + raw_rows=load.raw_rows, + normalized_rows=load.normalized_rows, + minute_live_calls=load.live_calls, + rebalance_count=len(pairs), + groups=tuple(group_results), + spread_per_period=per_period, + spread_cumulative=cumulative, + spread_summary=spread_summary, + monotonicity=monotonicity, + per_date_rows=assignment.per_date_rows, + score_coverage=score_coverage, + price_limit_check=ic.price_limit_check, + limit_coverage=limit_coverage, + stk_limit_gap_fetches=stk_limit_gap_fetches, + figure_paths=_write_figures(figure_dir, tuple(group_results), per_period, + cumulative, n_groups), + report_path=report_path, + log_path=log_path, + elapsed=time.monotonic() - started, + ) + _write_report(result) + logger.info("report: %s", report_path) + return result + + +# Report + figure rendering live in a sibling module for cohesion. +from qt.intraday_group_report import _write_figures, _write_report # noqa: E402 diff --git a/qt/intraday_group_figures.py b/qt/intraday_group_figures.py new file mode 100644 index 0000000..e2443f4 --- /dev/null +++ b/qt/intraday_group_figures.py @@ -0,0 +1,139 @@ +"""Matplotlib (Agg) figures for the I5d MMP quintile grouped backtest. + +Pure plotting helpers: each takes already-computed frames/series and an output +path, renders ONE PNG with a non-interactive ``Agg`` backend (no display needed), +and returns the written path. They are deliberately decoupled from the run logic +so they can be unit-tested with toy NAV frames (goal §Tests 5). + +Direction convention is made explicit on every figure: **Q1 = lowest MMP score, +QN = highest** — the legend/labels say "low"/"high" so the quantile direction can +never be misread. +""" + +from __future__ import annotations + +from pathlib import Path + +import matplotlib + +matplotlib.use("Agg") # headless: render to PNG, never open a display. + +import matplotlib.pyplot as plt # noqa: E402 (must follow the use("Agg") call) +import pandas as pd # noqa: E402 + +# Required metric columns for the grouped-bar figure (one bar group per metric). +METRIC_COLUMNS: tuple[str, ...] = ( + "annual_return", + "max_drawdown", + "mean_turnover", + "total_cost", +) +_METRIC_TITLES = { + "annual_return": "Annualized return", + "max_drawdown": "Max drawdown", + "mean_turnover": "Mean turnover", + "total_cost": "Total cost", +} + + +def _group_label(group: int, n_groups: int) -> str: + """``Q{g}`` with a low/high tag on the extremes (direction is unmissable).""" + if group == 1: + return f"Q{group} (low)" + if group == n_groups: + return f"Q{group} (high)" + return f"Q{group}" + + +def plot_quintile_nav( + nav_by_group: dict[int, pd.DataFrame], out_path: str | Path, n_groups: int +) -> Path: + """NAV curves for Q1..QN on one chart (x = rebalance date, y = NAV).""" + out = Path(out_path) + out.parent.mkdir(parents=True, exist_ok=True) + fig, ax = plt.subplots(figsize=(10, 6)) + for group in sorted(nav_by_group): + nav = nav_by_group[group] + if nav is None or nav.empty or "nav" not in nav.columns: + continue + ax.plot( + [pd.Timestamp(d) for d in nav.index], + nav["nav"].to_numpy(dtype=float), + marker="o", + markersize=3, + label=_group_label(group, n_groups), + ) + ax.set_title("MMP quintile group NAV (Q1 = lowest score, QN = highest)") + ax.set_xlabel("Rebalance date") + ax.set_ylabel("NAV (start = 1.0)") + ax.axhline(1.0, color="grey", linewidth=0.8, linestyle="--") + ax.legend(loc="best", fontsize=9) + ax.grid(True, alpha=0.3) + fig.autofmt_xdate() + fig.tight_layout() + fig.savefig(out, dpi=120) + plt.close(fig) + return out + + +def plot_spread_curve( + spread_curve: pd.Series, + out_path: str | Path, + *, + low_label: str = "Q1", + high_label: str = "Q5", +) -> Path: + """Cumulative synthetic ``high − low`` group spread curve (long-only legs).""" + out = Path(out_path) + out.parent.mkdir(parents=True, exist_ok=True) + fig, ax = plt.subplots(figsize=(10, 5)) + if spread_curve is not None and len(spread_curve) > 0: + ax.plot( + [pd.Timestamp(d) for d in spread_curve.index], + spread_curve.to_numpy(dtype=float), + color="tab:purple", + marker="o", + markersize=3, + ) + ax.axhline(0.0, color="grey", linewidth=0.8, linestyle="--") + ax.set_title( + f"Cumulative synthetic {high_label}−{low_label} spread " + "(long-only leg difference, NOT a dollar-neutral book)" + ) + ax.set_xlabel("Rebalance date") + ax.set_ylabel(f"Cumulative {high_label}−{low_label} net return") + ax.grid(True, alpha=0.3) + fig.autofmt_xdate() + fig.tight_layout() + fig.savefig(out, dpi=120) + plt.close(fig) + return out + + +def plot_group_metrics( + metrics: pd.DataFrame, out_path: str | Path, n_groups: int +) -> Path: + """Compact 2x2 grouped-bar panel: annual return / maxDD / turnover / cost.""" + out = Path(out_path) + out.parent.mkdir(parents=True, exist_ok=True) + groups = sorted(metrics.index) + labels = [_group_label(int(g), n_groups) for g in groups] + fig, axes = plt.subplots(2, 2, figsize=(11, 8)) + for ax, col in zip(axes.flat, METRIC_COLUMNS): + if col in metrics.columns: + values = [float(metrics.loc[g, col]) for g in groups] + else: + values = [float("nan")] * len(groups) + ax.bar(labels, values, color="tab:blue") + ax.set_title(_METRIC_TITLES.get(col, col)) + ax.axhline(0.0, color="grey", linewidth=0.8) + ax.grid(True, axis="y", alpha=0.3) + ax.tick_params(axis="x", labelrotation=0, labelsize=9) + fig.suptitle( + "MMP quintile group metrics (Q1 = lowest score, QN = highest)", + fontsize=12, + ) + fig.tight_layout(rect=(0, 0, 1, 0.97)) + fig.savefig(out, dpi=120) + plt.close(fig) + return out diff --git a/qt/intraday_group_report.py b/qt/intraday_group_report.py new file mode 100644 index 0000000..21d1dd1 --- /dev/null +++ b/qt/intraday_group_report.py @@ -0,0 +1,345 @@ +"""Markdown report + PNG figures for the I5d MMP quintile grouped backtest. + +Kept in a sibling module (cohesion / small files) and decoupled from the run +logic: every function takes already-computed results and only READS them, so there +is no import cycle with :mod:`qt.intraday_group_backtest`. The report makes the +exploratory framing, the Q1/Q5 direction, the ``fee_rate`` and the cache-only +minute provenance explicit, and embeds the three required figures. +""" + +from __future__ import annotations + +from pathlib import Path + +import pandas as pd + +from qt.intraday_group_figures import ( + METRIC_COLUMNS, + plot_group_metrics, + plot_quintile_nav, + plot_spread_curve, +) + + +def _write_figures( + figure_dir: Path, + groups: tuple, + per_period: pd.Series, + cumulative: pd.Series, + n_groups: int, +) -> dict[str, Path]: + """Render the three required PNGs (NAV curves / spread / metric bars).""" + figure_dir.mkdir(parents=True, exist_ok=True) + nav_by_group = {g.group: g.nav_table for g in groups} + metrics = pd.DataFrame( + {g.group: {c: g.metrics.get(c, float("nan")) for c in METRIC_COLUMNS} + for g in groups} + ).T + nav_path = plot_quintile_nav( + nav_by_group, figure_dir / "mmp_quintile_nav.png", n_groups + ) + spread_path = plot_spread_curve( + cumulative, + figure_dir / f"mmp_q{n_groups}_minus_q1_spread.png", + low_label="Q1", + high_label=f"Q{n_groups}", + ) + metrics_path = plot_group_metrics( + metrics, figure_dir / "mmp_quintile_metrics.png", n_groups + ) + return {"nav": nav_path, "spread": spread_path, "metrics": metrics_path} + + +def _fmt(v: float, pct: bool = False, nd: int = 4) -> str: + """Format a float, NaN-safe; ``pct`` renders as a percentage.""" + if v is None or (isinstance(v, float) and pd.isna(v)): + return "NaN" + if pct: + return f"{v * 100:.2f}%" + return f"{v:.{nd}f}" + + +def _intro_lines(result) -> list[str]: + """Title + exploratory framing + study design.""" + cfg = result.config + title = cfg.output.intraday_report_title or "Phase I5d — MMP Quintile 5Y Group Backtest" + t = title.strip() + h1 = t if t.startswith("#") else f"# {t}" + lines = [ + h1, + "", + "**This is an EXPLORATORY grouped factor analysis of the I5c MMP " + "minute-derived daily score — NOT a performance claim and NOT parameter " + "tuning.** On each monthly rebalance date the cross-section is ranked by the " + "PIT-safe daily score and split into equal-count quantile groups; each group " + "is run as its own long-only equal-weight portfolio through the SAME " + "event-driven intraday-tail model (14:50 decision / 14:51 execution, " + "exec-to-exec returns) with I5b raw `stk_limit` execution feasibility ON and " + "trading cost `fee_rate` applied. No parameter, group count, cost, universe, " + "or window was tuned from results.", + "", + "## Study design", + "", + f"- window: `{cfg.data.start}` → `{cfg.data.end}`", + f"- universe: `{cfg.universe.type}` `{cfg.universe.index_code or ''}` " + f"(requested {result.requested_symbols} distinct constituents)", + f"- rebalance: monthly — {result.rebalance_count} settled periods", + f"- score column: `{result.score_feature}` (key=`{result.score_feature_key}`)", + f"- groups: `analytics.quantiles={result.n_groups}` equal-count rank buckets, " + "**Q1 = lowest MMP score, " + f"Q{result.n_groups} = highest**", + f"- decision_time `{cfg.intraday.decision_time}` / execution_window " + f"`[{cfg.intraday.execution_window[0]}, {cfg.intraday.execution_window[1]}]`; " + "returns are execution-to-execution, NEVER close-to-close", + f"- trading cost: `cost.fee_rate={cfg.cost.fee_rate}`, " + f"`slippage_rate={cfg.cost.slippage_rate}` (cost line = turnover × fee_rate " + "inside SimExecution; no extra ad-hoc cost layer)", + "", + ] + return lines + + +def _mmp_and_grouping_lines(result) -> list[str]: + return [ + "## MMP factor & group assignment", + "", + "Per 1min bar `t`: `mid=(high+low)/2`; `S=(close-mid)/mid`; " + "`V=sqrt(volume/median(vol[t-20:t]))`; `B=|close-open|/(high-low+eps)`; " + "`R=(high-low)/(mean(hl[t-20:t])+eps)`; **`MMP_t = S*V*B*R`** (`eps=1e-6`). " + "The daily score `intraday_mmp20_ew_0930_1450` is the EQUAL-WEIGHT mean of " + "valid `MMP_t` over the in-session bars visible at the 14:50 cutoff " + "(`[session_open, decision_time]`); rolling baselines use only the prior 20 " + "in-session bars (first 20 are NaN). The MMP math is UNCHANGED from I5c.", + "- **group rule**: drop NaN/non-finite scores, sort by `(score asc, symbol " + "asc)`, split BY RANK into equal-count buckets. `Q1` = lowest score, " + f"`Q{result.n_groups}` = highest. Equal-count (not value-cut) buckets keep " + "tied/degenerate scores deterministic; a date with too few valid names " + "leaves the high groups empty (no crash).", + "- **no lookahead**: grouping uses only the daily PIT universe ∩ " + "minute-covered names ∩ valid MMP score; forward returns NEVER enter the " + "assignment.", + "", + ] + + +def _provenance_lines(result) -> list[str]: + lines = [ + "## Minute-cache coverage & data provenance", + "", + f"- requested symbols: {result.requested_symbols}; minute-cache fully " + f"covered: {result.covered_symbols}; excluded (uncovered): " + f"{len(result.uncovered_symbols)}", + f"- anchor dates requiring minute bars: {result.anchor_dates} (rebalance ∪ " + "exit dates only — the full multi-year minute history is never loaded)", + f"- raw rows loaded: {result.raw_rows}; normalized rows used: " + f"{result.normalized_rows}", + f"- **stk_mins live API calls during this run: {result.minute_live_calls}** " + "(cache-only, read-only; a miss is never a silent warm/backfill)", + f"- elapsed: {result.elapsed:.1f}s", + ] + if result.uncovered_symbols: + shown = ", ".join(result.uncovered_symbols[:20]) + more = ( + "" if len(result.uncovered_symbols) <= 20 + else f" (+{len(result.uncovered_symbols) - 20} more)" + ) + lines.append( + f"- **excluded (uncovered) symbols** ({len(result.uncovered_symbols)}): " + f"{shown}{more}" + ) + lines.append( + " - ⚠️ dropping uncovered names trades full-window completion for a " + "potential coverage bias (the realized cross-section omits names with " + "no cached minute history). Disclosed, not silent; the 5-year WINDOW is " + "NOT shortened." + ) + sc = result.score_coverage + lines.append( + f"- daily MMP score panel: {sc['rows']} (date,symbol) rows; valid " + f"{sc['valid']}; NaN {sc['nan']}." + ) + lines.append("") + return lines + + +def _feasibility_lines(result) -> list[str]: + cfg = result.config + lines = ["## Execution-time price-limit feasibility (I5b)", ""] + if not result.price_limit_check: + lines.append( + "- **disabled**: feasibility is the base bar-exists rule only " + "(missing/NaN execution bar blocks both directions)." + ) + lines.append("") + return lines + cov = result.limit_coverage + lines.extend([ + "- **enabled** (`intraday.price_limit_check=true`): a buy is blocked at the " + "raw upper limit and a sell at the raw lower limit, comparing the selected " + "execution-minute **raw** 1min close to the raw `stk_limit` band " + "(RAW-vs-RAW; never qfq / daily close / a daily-close-derived flag).", + f"- limit tolerance: `{cfg.intraday.limit_tolerance}`; " + f"require_price_limit_coverage: `{cfg.intraday.require_price_limit_coverage}`", + f"- limit coverage over rebalance anchors (shared across groups): required " + f"{cov.get('required', 0)} (date, symbol) pairs; present " + f"{cov.get('present', 0)}; missing {cov.get('missing', 0)}", + f"- **stk_limit cache gap-fetches this run: {result.stk_limit_gap_fetches}** " + "(read through the existing P4 daily cache — never a minute/stk_mins fetch).", + "", + "Per-group blocked buy/sell counts (a fresh model per group, so counts do " + "NOT double-count across groups):", + "", + "| group | up-limit blocked buys | down-limit blocked sells | unchecked limit rows |", + "|---|---|---|---|", + ]) + for g in result.groups: + lines.append( + f"| Q{g.group} | {g.up_limit_blocked_buys} | " + f"{g.down_limit_blocked_sells} | {g.missing_limit_rows} |" + ) + lines.append("") + return lines + + +def _performance_lines(result) -> list[str]: + lines = [ + "## Per-group NAV / performance", + "", + "| group | final NAV | annual | vol | Sharpe | maxDD | mean turnover | " + "total cost | avg holdings |", + "|---|---|---|---|---|---|---|---|---|", + ] + for g in result.groups: + m = g.metrics + tag = " (low)" if g.group == 1 else (" (high)" if g.group == result.n_groups else "") + lines.append( + f"| Q{g.group}{tag} | {_fmt(m['final_nav'])} | " + f"{_fmt(m['annual_return'], pct=True)} | {_fmt(m['volatility'], pct=True)} " + f"| {_fmt(m['sharpe'])} | {_fmt(m['max_drawdown'], pct=True)} | " + f"{_fmt(m['mean_turnover'])} | {_fmt(m['total_cost'])} | " + f"{_fmt(m['avg_holdings'], nd=1)} |" + ) + lines.append("") + lines.append( + "- turnover/cost count the ACHIEVED book after feasible fills (price-limit " + "blocked + missing-bar blocked names earn nothing, never a daily-close " + "fallback); idle cash earns `cash_return`." + ) + lines.append("") + return lines + + +def _spread_lines(result) -> list[str]: + n = result.n_groups + s = result.spread_summary + mono = result.monotonicity + lines = [ + f"## Q{n}−Q1 synthetic spread & monotonicity (report-only)", + "", + f"The Q{n}−Q1 spread is the per-period difference of the Q{n} (high) and Q1 " + "(low) group NET returns; the cumulative curve is their running sum. It is a " + "**synthetic long-only leg difference, NOT a separately executed " + "dollar-neutral portfolio** (no long-short execution model is run).", + ] + if s: + lines.append( + f"- mean per-period Q{n}−Q1 net return: {_fmt(s.get('mean_per_period'), pct=True)} " + f"over {int(s.get('n_periods', 0))} periods; cumulative (sum): " + f"{_fmt(s.get('total'), pct=True)}." + ) + if mono: + lines.append( + f"- group monotonicity (Spearman of group index 1..{n} vs metric): " + f"annual return {_fmt(mono.get('annual_spearman'))}, final NAV " + f"{_fmt(mono.get('final_nav_spearman'))} " + "(+1 = perfectly increasing Q1→QN, −1 = perfectly decreasing)." + ) + lines.append( + "- **report-only**: returns are read here for analytics only and never feed " + "the factor/alpha; a single overlapping window is far too little to infer " + "factor quality." + ) + lines.append("") + return lines + + +def _per_date_lines(result) -> list[str]: + n = result.n_groups + header = "| date | n_scored | " + " | ".join(f"Q{i}" for i in range(1, n + 1)) + \ + " | score mean | std | p10 | p50 | p90 |" + sep = "|---|---|" + "---|" * n + "---|---|---|---|---|" + lines = [ + "## Per-rebalance group sizes & score distribution", + "", + header, + sep, + ] + for r in result.per_date_rows: + sizes = " | ".join(str(x) for x in r["sizes"]) + lines.append( + f"| {pd.Timestamp(r['date']).date()} | {r['n_scored']} | {sizes} | " + f"{_fmt(r['mean'], nd=6)} | {_fmt(r['std'], nd=6)} | {_fmt(r['p10'], nd=6)} " + f"| {_fmt(r['p50'], nd=6)} | {_fmt(r['p90'], nd=6)} |" + ) + lines.append("") + return lines + + +def _figure_lines(result) -> list[str]: + report_dir = result.report_path.parent + lines = ["## Figures", ""] + captions = { + "nav": "NAV curves for each quantile group (Q1 = lowest MMP score, " + f"Q{result.n_groups} = highest).", + "spread": f"Cumulative synthetic Q{result.n_groups}−Q1 net-return spread " + "(long-only leg difference).", + "metrics": "Grouped bars: annualized return, max drawdown, mean turnover, " + "total cost per group.", + } + for key in ("nav", "spread", "metrics"): + path = result.figure_paths.get(key) + if path is None: + continue + rel = Path(path).relative_to(report_dir).as_posix() + lines.append(f"**{captions[key]}**") + lines.append("") + lines.append(f"![{key}]({rel})") + lines.append("") + return lines + + +def _limitations_lines(result) -> list[str]: + return [ + "## Limitations (explicit)", + "", + "- **EXPLORATORY grouped factor analysis, NOT a performance claim**: one " + "factor, one overlapping 5-year window, one universe. No tuning, no " + "robustness matrix, no learned / IC-weighted alpha.", + "- **coverage bias**: uncovered minute names are dropped (disclosed above); " + "the realized cross-section is the covered subset.", + "- **execution feasibility** models price-limit + bar-existence only; no " + "partial-fill / liquidity / volume cap at the execution minute. Suspended " + "names have no minute bar and are blocked by the missing-bar rule; explicit " + "ST status is not consulted at 14:50.", + f"- **Q{result.n_groups}−Q1 spread is synthetic** (a long-only leg " + "difference), not a separately executed dollar-neutral book.", + "", + ] + + +def _write_report(result) -> None: + """Write the I5d grouped-backtest markdown report (with embedded figures).""" + path = result.report_path + path.parent.mkdir(parents=True, exist_ok=True) + lines: list[str] = [] + lines += _intro_lines(result) + lines += _mmp_and_grouping_lines(result) + lines += _provenance_lines(result) + lines += _performance_lines(result) + lines += _spread_lines(result) + lines += _feasibility_lines(result) + lines += _per_date_lines(result) + lines += _figure_lines(result) + lines += _limitations_lines(result) + lines.append(f"_elapsed: {result.elapsed:.1f}s_") + path.write_text("\n".join(lines) + "\n", encoding="utf-8") diff --git a/qt/intraday_groups.py b/qt/intraday_groups.py new file mode 100644 index 0000000..9b5f6b4 --- /dev/null +++ b/qt/intraday_groups.py @@ -0,0 +1,131 @@ +"""Cross-sectional quantile grouping primitives for the I5d grouped backtest. + +These are the small, pure pieces the grouped intraday-tail backtest +(:mod:`qt.intraday_group_backtest`) layers on TOP of the existing event-driven +machinery — they decide *which names land in which group* and present one group +to the shared :class:`~runtime.backtest.engine.BacktestEngine` as an equal-weight +target, WITHOUT touching the engine, the execution feasibility, or the factor +math: + + * :func:`assign_quantile_buckets` — split one scored cross-section into N + EQUAL-COUNT rank buckets (Q1 = lowest score, QN = highest). Rank/position + buckets (not value cuts) so tied or degenerate scores still produce a + deterministic, auditable assignment; too few names simply leave high groups + empty instead of crashing. + * :class:`GroupScores` — a ``ScoresSource`` that exposes ONE group's members to + the engine (1.0 for a member, NaN otherwise), so the engine's selection picks + exactly that bucket. + * :class:`EqualWeightAll` — a constructor that equal-weights EVERY non-NaN name + (no ``top_n`` cap): the grouped backtest holds the whole bucket. + +Hard boundary preserved (CLAUDE.md invariant #1/#3): nothing here reads a forward +return, a data source, or places an order. The MMP score is computed upstream +(PIT-safe, available_time <= 14:50) and only its CROSS-SECTIONAL RANK is used to +form groups. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd + +from portfolio.base import PortfolioConstructor + + +def assign_quantile_buckets(scores: pd.Series, n_groups: int) -> dict[str, int]: + """Assign each scored symbol to one of ``n_groups`` EQUAL-COUNT rank buckets. + + Args: + scores: symbol-indexed scores for ONE cross-section (one rebalance date). + n_groups: number of quantile groups (e.g. 5). + + Returns: + ``{symbol: label}`` with ``label`` in ``1..n_groups``, where **Q1 (label + 1) is the LOWEST score** and **QN (label n_groups) is the HIGHEST**. + + Semantics: + * NaN / non-finite scores are dropped (never assigned to a group). + * Symbols are ordered by ``(score ascending, symbol ascending)`` and split + BY POSITION into ``n_groups`` contiguous chunks (``np.array_split``), so + the assignment is deterministic even with tied scores (the symbol + tie-break decides) — a value-cut quantile could not split ties cleanly. + * When fewer than ``n_groups`` names are scored, the lower groups fill + first and the high groups are simply empty (no crash). Chunk sizes + differ by at most one; the extra goes to the lower groups (the + ``np.array_split`` convention). + """ + if n_groups < 1: + raise ValueError(f"n_groups must be >= 1; got {n_groups}.") + if scores is None or len(scores) == 0: + return {} + s = pd.Series(scores, dtype=float).dropna() + if s.empty: + return {} + finite_mask = np.isfinite(s.to_numpy(dtype=float)) + s = s[finite_mask] + if s.empty: + return {} + order = sorted(s.index, key=lambda sym: (float(s.loc[sym]), str(sym))) + buckets = np.array_split(np.array(order, dtype=object), n_groups) + out: dict[str, int] = {} + for i, bucket in enumerate(buckets): + for sym in bucket: + out[str(sym)] = i + 1 + return out + + +class GroupScores: + """``ScoresSource`` exposing exactly ONE quantile group to the engine. + + Bridges the precomputed per-date bucket assignment to the ``ScoresSource`` + port :class:`~runtime.backtest.engine.BacktestEngine` depends on. ``get`` marks + a symbol with a constant score ``1.0`` iff it is a member of ``group`` on that + date, else ``NaN`` — so an equal-weight-all constructor selects exactly the + bucket. It only READS a precomputed rank assignment (no forward returns, no + data source), preserving the no-lookahead boundary. + """ + + def __init__( + self, assignments: dict[pd.Timestamp, dict[str, int]], group: int + ) -> None: + # assignments: normalized rebalance date -> {symbol: group label}. + self._assignments = assignments + self._group = int(group) + + def get(self, date: pd.Timestamp, symbols: list[str]) -> pd.Series: + """Return symbol-indexed membership scores (1.0 member / NaN non-member).""" + norm = pd.Timestamp(date).normalize() + members = self._assignments.get(norm, {}) + values = [ + 1.0 if members.get(str(sym)) == self._group else np.nan + for sym in symbols + ] + return pd.Series(values, index=list(symbols), dtype=float) + + +class EqualWeightAll(PortfolioConstructor): + """Equal-weight EVERY non-NaN name in the cross-section (no ``top_n`` cap). + + The grouped backtest holds an entire quantile bucket, so — unlike + :class:`portfolio.construct.TopNEqualWeight` — there is no top-N selection: + every name the :class:`GroupScores` source surfaces (score ``1.0``) is held at + weight ``1/k``. NaN scores are dropped (non-members); an empty cross-section + yields an empty (all-cash) target. Long-only; the input is never mutated. + """ + + def __init__(self, long_only: bool = True) -> None: + self.long_only = bool(long_only) + + def build( + self, + scores: pd.Series, + current_weights: pd.Series | None = None, + ) -> pd.Series: + """Equal-weight all non-NaN names; empty target when there are none.""" + candidates = scores.dropna() + if candidates.empty: + return pd.Series(dtype=float, index=candidates.index[:0], name="weight") + weight = 1.0 / len(candidates) + result = pd.Series(weight, index=candidates.index, name="weight") + result.index.name = scores.index.name + return result From d2ca197dff66c8e3c955ab711f7322be5930dd0d Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Mon, 15 Jun 2026 22:05:22 +0800 Subject: [PATCH 8/8] test: I5d MMP quintile grouping, feasibility, and figures 19 tests over the goal's five groups: equal-count bucket assignment (Q1 low / QN high, NaN/inf excluded, deterministic ties, too-few-names), EqualWeightAll ignores top_n, GroupScores selects one group, fee_rate reaches SimExecution; PIT invariants (grouping uses only the <=14:50 MMP score, post-cutoff and exit bars cannot change the assignment); grouped execution feasibility (raw stk_limit blocks, missing bar blocks without daily-close fallback, per-group counts do not double-count); report H1 names I5d (not stale) and mentions fee_rate=0.001 / analytics.quantiles=5; figures written and non-empty. --- tests/test_i5d_mmp_quintile.py | 387 +++++++++++++++++++++++++++++++++ 1 file changed, 387 insertions(+) create mode 100644 tests/test_i5d_mmp_quintile.py diff --git a/tests/test_i5d_mmp_quintile.py b/tests/test_i5d_mmp_quintile.py new file mode 100644 index 0000000..ea9e516 --- /dev/null +++ b/tests/test_i5d_mmp_quintile.py @@ -0,0 +1,387 @@ +"""I5d: MMP quintile grouped intraday-tail backtest. + +Covers the goal's five test groups: + +1. Bucket assignment — deterministic 5 equal-count rank buckets, Q1 lowest / Q5 + highest, every finite name in exactly one group, NaN/non-finite excluded, ties + deterministic, too-few-names leaves high groups empty (no crash). +2. Group constructor / backtest integration — ``EqualWeightAll`` ignores ``top_n`` + and equal-weights the whole bucket; ``GroupScores`` exposes one group; a fresh + execution/model per group (no cross-group state); ``fee_rate`` reaches + ``SimExecution`` (a nonzero-turnover period has positive cost). +3. PIT / MMP invariants — grouping uses only the PIT MMP score (available_time <= + 14:50); perturbing a post-cutoff bar cannot change the assignment; future + (exit) bars never enter the assignment. +4. Execution feasibility — raw ``stk_limit`` blocking stays active in grouped runs; + a missing execution bar blocks (never daily-close fallback); per-group blocked + counts do not double-count across groups. +5. Report / figures — H1 names I5d (not a stale I5c/I5b label), mentions + ``fee_rate=0.001`` and ``analytics.quantiles=5``; the three required PNGs are + written and non-empty. +""" + +from __future__ import annotations + +from pathlib import Path + +import pandas as pd +import pytest + +from data.clean.intraday_schema import normalize_intraday_bars +from data.clean.intraday_aggregate import asof_daily_features +from data.clean.schema import normalize_panel +from qt.config import load_config +from qt.intraday_group_backtest import ( + GroupRunResult, + I5dResult, + _build_group_assignment, + _group_metrics, + _max_drawdown, +) +from qt.intraday_group_report import _write_figures, _write_report +from qt.intraday_groups import EqualWeightAll, GroupScores, assign_quantile_buckets +from runtime.backtest.engine import BacktestEngine +from runtime.backtest.event_models import IntradayTailEventModel +from runtime.intraday_execution import REASON_NO_BAR, IntradayExecutionConfig +from runtime.backtest.sim_execution import SimExecution +from universe.static import StaticUniverse + +_CONFIG = Path(__file__).resolve().parent.parent / "config" / "phase_i5d_mmp_quintile_5y.yaml" + +# Consecutive month-ends -> >=2 settled monthly periods. +_DATES = ["2024-01-30", "2024-01-31", "2024-02-28", "2024-02-29"] +_JAN, _FEB = pd.Timestamp("2024-01-31"), pd.Timestamp("2024-02-29") + + +# --------------------------------------------------------------------------- # +# fixtures +# --------------------------------------------------------------------------- # +def _daily_panel(closes: dict) -> pd.DataFrame: + rows = [ + {"date": pd.Timestamp(d), "symbol": s, "open": c, "high": c, "low": c, + "close": c, "volume": 1.0, "amount": 1.0, "adj_factor": 1.0} + for (d, s), c in closes.items() + ] + return normalize_panel(pd.DataFrame(rows)) + + +def _grid_panel(symbols: list[str]) -> pd.DataFrame: + return _daily_panel({(d, s): 100.0 for d in _DATES for s in symbols}) + + +def _bars(rows: list[dict]) -> pd.DataFrame: + return normalize_intraday_bars(pd.DataFrame(rows), freq="1min", data_lag="1min") + + +def _exec_rows(specs: list[tuple[str, str, float]]) -> list[dict]: + """Single-bar specs ``(symbol, 'YYYY-MM-DD HH:MM:SS', close)`` -> bar rows.""" + return [ + {"time": pd.Timestamp(t), "symbol": s, "open": c, "high": c, "low": c, + "close": c, "volume": 1.0, "amount": float(c), "source_trade_time": t} + for (s, t, c) in specs + ] + + +def _session_rows(symbol: str, date: str, *, n: int = 24, slope: float) -> list[dict]: + """``n`` in-session 1min bars (from 09:30) with distinct OHLCV -> finite MMP. + + A non-zero ``slope`` makes each symbol's MMP distinct so the rank buckets + differ. ``n=24`` > the 20-bar MMP baseline, so a few valid ``MMP_t`` exist. + """ + day = pd.Timestamp(date) + rows = [] + for i in range(n): + t = day + pd.Timedelta("09:30:00") + pd.Timedelta(minutes=i) + c = 10.0 + slope * i + 0.01 * ((i * 7) % 5) + rows.append({ + "time": t, "symbol": symbol, + "open": c - 0.02, "high": c + 0.05 + 0.01 * (i % 3), + "low": c - 0.05 - 0.01 * (i % 2), "close": c, + "volume": 1000.0 + 50.0 * ((i * 3) % 7), + "amount": c * (1000.0 + 50.0 * ((i * 3) % 7)), + "source_trade_time": str(t), + }) + return rows + + +def _limits(rows: list[tuple]) -> pd.DataFrame: + return pd.DataFrame( + [{"date": pd.Timestamp(d), "symbol": s, "up_limit": up, "down_limit": dn} + for (d, s, up, dn) in rows] + ) + + +# --------------------------------------------------------------------------- # +# 1. Bucket assignment +# --------------------------------------------------------------------------- # +def test_buckets_q1_lowest_qn_highest(): + scores = pd.Series({"A": 1.0, "B": 2.0, "C": 3.0, "D": 4.0, "E": 5.0}) + out = assign_quantile_buckets(scores, 5) + assert out == {"A": 1, "B": 2, "C": 3, "D": 4, "E": 5} # Q1 lowest, Q5 highest + + +def test_buckets_equal_count(): + scores = pd.Series({chr(ord("a") + i): float(i) for i in range(10)}) + out = assign_quantile_buckets(scores, 5) + sizes = [sum(1 for v in out.values() if v == g) for g in range(1, 6)] + assert sizes == [2, 2, 2, 2, 2] + # lowest two scores are group 1, highest two are group 5. + assert out["a"] == 1 and out["b"] == 1 + assert out["i"] == 5 and out["j"] == 5 + + +def test_buckets_all_finite_assigned_exactly_once(): + scores = pd.Series({s: float(ord(s)) for s in "ABCDEFG"}) + out = assign_quantile_buckets(scores, 5) + assert set(out) == set("ABCDEFG") # every finite name assigned + assert all(1 <= g <= 5 for g in out.values()) + + +def test_buckets_drop_nan_and_inf(): + scores = pd.Series({"A": 1.0, "B": float("nan"), "C": float("inf"), + "D": -float("inf"), "E": 2.0}) + out = assign_quantile_buckets(scores, 5) + assert set(out) == {"A", "E"} # NaN / +-inf excluded + + +def test_buckets_ties_deterministic_by_symbol(): + scores = pd.Series({"B": 1.0, "A": 1.0, "D": 1.0, "C": 1.0}) + first = assign_quantile_buckets(scores, 2) + second = assign_quantile_buckets(scores, 2) + assert first == second # deterministic + # all tied -> symbol tie-break (A,B | C,D) splits the lower half to Q1. + assert first["A"] == 1 and first["B"] == 1 + assert first["C"] == 2 and first["D"] == 2 + + +def test_buckets_too_few_names_leaves_high_groups_empty(): + scores = pd.Series({"A": 1.0, "B": 2.0, "C": 3.0}) + out = assign_quantile_buckets(scores, 5) + sizes = [sum(1 for v in out.values() if v == g) for g in range(1, 6)] + assert sizes == [1, 1, 1, 0, 0] # no crash; Q4/Q5 empty + + +def test_buckets_empty_input(): + assert assign_quantile_buckets(pd.Series(dtype=float), 5) == {} + + +# --------------------------------------------------------------------------- # +# 2. Group constructor / scores +# --------------------------------------------------------------------------- # +def test_equal_weight_all_ignores_top_n_and_equal_weights(): + scores = pd.Series({"A": 1.0, "B": 1.0, "C": 1.0, "D": 1.0}) + w = EqualWeightAll().build(scores) + assert len(w) == 4 # NOT capped at any top_n + assert pytest.approx(w.sum()) == 1.0 + assert (w == 0.25).all() + + +def test_equal_weight_all_drops_nan_and_empty(): + w = EqualWeightAll().build(pd.Series({"A": 1.0, "B": float("nan")})) + assert list(w.index) == ["A"] and pytest.approx(w.sum()) == 1.0 + assert EqualWeightAll().build(pd.Series(dtype=float)).empty + + +def test_group_scores_selects_only_group_members(): + assignments = {_JAN: {"A": 1, "B": 2, "C": 1}} + s = GroupScores(assignments, 1).get(_JAN, ["A", "B", "C", "D"]) + assert s["A"] == 1.0 and s["C"] == 1.0 # group-1 members + assert pd.isna(s["B"]) and pd.isna(s["D"]) # non-members -> NaN + + +def test_fee_rate_reaches_simexecution_positive_cost(): + panel = _grid_panel(["A", "B"]) + bars = _bars(_exec_rows([ + ("A", f"{_JAN.date()} 14:51:00", 10.0), ("B", f"{_JAN.date()} 14:51:00", 10.0), + ("A", f"{_FEB.date()} 14:51:00", 11.0), ("B", f"{_FEB.date()} 14:51:00", 9.0), + ])) + assignments = {_JAN: {"A": 1}, _FEB: {"A": 1}} + model = IntradayTailEventModel(calendar_panel=panel, bars=bars, + cfg=IntradayExecutionConfig()) + engine = BacktestEngine( + model=model, universe=StaticUniverse(["A", "B"]), + scores=GroupScores(assignments, 1), constructor=EqualWeightAll(), + execution=SimExecution(fee_rate=0.001), selection_panel=panel, + ) + nav = engine.run() + # first period buys A (turnover 1.0) -> cost = 1.0 * 0.001 > 0. + assert nav["turnover"].iloc[0] > 0 + assert nav["cost"].iloc[0] == pytest.approx(nav["turnover"].iloc[0] * 0.001) + assert nav["cost"].iloc[0] > 0 + + +# --------------------------------------------------------------------------- # +# 3. PIT / MMP invariants (grouping is a function of the PIT score only) +# --------------------------------------------------------------------------- # +def _mmp_assignment(extra_rows: list[dict] | None = None) -> dict: + """Build the MMP score on _JAN for A/B/C and group it into 3 buckets.""" + rows: list[dict] = [] + for sym, slope in (("A", 0.00), ("B", 0.02), ("C", -0.02)): + rows += _session_rows(sym, str(_JAN.date()), slope=slope) + if extra_rows: + rows += extra_rows + bars = _bars(rows) + score = asof_daily_features( + bars, decision_time="14:50:00", session_open="09:30:00", features=["mmp_ew"] + ).iloc[:, 0].rename("score") + panel = _daily_panel({(str(_JAN.date()), s): 10.0 for s in "ABC"}) + assignment = _build_group_assignment( + score, StaticUniverse(["A", "B", "C"]), panel, [_JAN], {"A", "B", "C"}, 3 + ) + return assignment.by_date[_JAN] + + +def test_grouping_uses_pit_score_only(): + base = _mmp_assignment() + # every scored name assigned to exactly one of 3 groups. + assert set(base) == {"A", "B", "C"} + assert sorted(base.values()) == [1, 2, 3] + + +def test_post_cutoff_bar_cannot_change_grouping(): + base = _mmp_assignment() + # a post-14:50 bar with an extreme price must NOT change the assignment. + poison = _exec_rows([("A", f"{_JAN.date()} 14:55:00", 9999.0)]) + perturbed = _mmp_assignment(poison) + assert perturbed == base + + +def test_future_exit_bar_cannot_change_grouping(): + base = _mmp_assignment() + # an exit-date (future) execution bar is irrelevant to the decision-date score. + future = _exec_rows([("A", f"{_FEB.date()} 14:51:00", 0.01)]) + perturbed = _mmp_assignment(future) + assert perturbed == base + + +# --------------------------------------------------------------------------- # +# 4. Execution feasibility in grouped runs +# --------------------------------------------------------------------------- # +def _grouped_engine(group, assignments, bars, panel, limits=None): + model = IntradayTailEventModel( + calendar_panel=panel, bars=bars, cfg=IntradayExecutionConfig(), + price_limits=limits, price_limit_check=limits is not None, + require_price_limit_coverage=False, + ) + engine = BacktestEngine( + model=model, universe=StaticUniverse(sorted({s for a in assignments.values() for s in a})), + scores=GroupScores(assignments, group), constructor=EqualWeightAll(), + execution=SimExecution(fee_rate=0.001), selection_panel=panel, + ) + nav = engine.run() + return model, engine, nav + + +def test_grouped_raw_limit_blocks_and_no_double_count(): + panel = _grid_panel(["A", "B"]) + bars = _bars(_exec_rows([ + ("A", f"{_JAN.date()} 14:51:00", 10.0), ("B", f"{_JAN.date()} 14:51:00", 5.0), + ("A", f"{_FEB.date()} 14:51:00", 10.0), ("B", f"{_FEB.date()} 14:51:00", 5.0), + ])) + # A is at its raw upper limit on _JAN -> a buy must be blocked. + limits = _limits([ + (f"{_JAN.date()}", "A", 10.0, 1.0), (f"{_JAN.date()}", "B", 9.0, 1.0), + (f"{_FEB.date()}", "A", 99.0, 1.0), (f"{_FEB.date()}", "B", 9.0, 1.0), + ]) + assignments = {_JAN: {"A": 1, "B": 2}, _FEB: {"A": 1, "B": 2}} + m1, _, _ = _grouped_engine(1, assignments, bars, panel, limits) # holds A + m2, _, _ = _grouped_engine(2, assignments, bars, panel, limits) # holds B + # the up-limit block is attributed ONLY to the group that wanted to buy A. + assert m1.up_limit_blocked_buys() >= 1 + assert m2.up_limit_blocked_buys() == 0 # no double count across groups + + +def test_grouped_missing_bar_blocks_no_daily_fallback(): + panel = _daily_panel({ # daily close MOVES A 100 -> 200; a fallback would show it + (str(_JAN.date()), "A"): 100.0, (str(_FEB.date()), "A"): 200.0, + (str(_JAN.date()), "B"): 100.0, (str(_FEB.date()), "B"): 100.0, + }) + # A has NO execution bar on _JAN (entry) -> blocked; only B has bars. + bars = _bars(_exec_rows([ + ("B", f"{_JAN.date()} 14:51:00", 10.0), ("B", f"{_FEB.date()} 14:51:00", 10.0), + ("A", f"{_FEB.date()} 14:51:00", 20.0), + ])) + assignments = {_JAN: {"A": 1}, _FEB: {"A": 1}} + model, _, nav = _grouped_engine(1, assignments, bars, panel) + reasons = {f.reason for f in model.blocked_fills()} + assert REASON_NO_BAR in reasons # A blocked by missing bar + # A earned nothing (no daily-close 100->200 fallback): first period stays ~cash. + assert abs(nav["gross_return"].iloc[0]) < 1e-9 + + +# --------------------------------------------------------------------------- # +# 5. Report / figures +# --------------------------------------------------------------------------- # +def _toy_nav(finals: list[float]) -> pd.DataFrame: + dates = [_JAN, _FEB] + prev = 1.0 + rows = [] + for i, d in enumerate(dates): + v = finals[i] + rows.append({"date": d, "nav": v, "gross_return": v / prev - 1.0, + "cost": 0.001, "turnover": 1.0, "net_return": v / prev - 1.0}) + prev = v + return pd.DataFrame(rows).set_index("date") + + +def _toy_group(group: int, finals: list[float]) -> GroupRunResult: + nav = _toy_nav(finals) + return GroupRunResult( + group=group, nav_table=nav, + holdings_log=pd.DataFrame(columns=["date", "symbol", "weight", "rank"]), + metrics=_group_metrics(nav, pd.DataFrame()), + up_limit_blocked_buys=0, down_limit_blocked_sells=0, + missing_limit_rows=0, blocked_fill_reasons={}, + ) + + +def test_figures_written_and_nonempty(tmp_path): + groups = tuple(_toy_group(g, [1.0 + 0.01 * g, 1.0 + 0.02 * g]) for g in range(1, 6)) + per = pd.Series([0.01, 0.01], index=[_JAN, _FEB]) + cum = per.cumsum() + paths = _write_figures(tmp_path, groups, per, cum, 5) + assert set(paths) == {"nav", "spread", "metrics"} + for p in paths.values(): + assert p.exists() and p.stat().st_size > 0 + assert paths["spread"].name == "mmp_q5_minus_q1_spread.png" + + +def test_report_h1_and_mentions(tmp_path): + cfg = load_config(str(_CONFIG)) + groups = tuple(_toy_group(g, [1.0 + 0.01 * g, 1.0 + 0.02 * g]) for g in range(1, 6)) + per = pd.Series([0.01, 0.01], index=[_JAN, _FEB]) + cum = per.cumsum() + fig_dir = tmp_path / "figs" + figure_paths = _write_figures(fig_dir, groups, per, cum, 5) + report_path = tmp_path / "phase_i5d_mmp_quintile_5y.md" + result = I5dResult( + config=cfg, n_groups=5, score_feature="intraday_mmp20_ew_0930_1450", + score_feature_key="mmp_ew", requested_symbols=10, covered_symbols=10, + uncovered_symbols=(), anchor_dates=2, raw_rows=100, normalized_rows=100, + minute_live_calls=0, rebalance_count=2, groups=groups, + spread_per_period=per, spread_cumulative=cum, + spread_summary={"mean_per_period": 0.01, "total": 0.02, "n_periods": 2}, + monotonicity={"annual_spearman": 1.0, "final_nav_spearman": 1.0}, + per_date_rows=({"date": _JAN, "n_scored": 10, "sizes": (2, 2, 2, 2, 2), + "mean": 0.0, "std": 0.0, "p10": 0.0, "p50": 0.0, "p90": 0.0},), + score_coverage={"rows": 20, "valid": 20, "nan": 0}, + price_limit_check=True, limit_coverage={"required": 4, "present": 4, "missing": 0}, + stk_limit_gap_fetches=0, figure_paths=figure_paths, + report_path=report_path, log_path=tmp_path / "x.log", elapsed=1.0, + ) + _write_report(result) + text = report_path.read_text(encoding="utf-8") + h1 = text.splitlines()[0] + assert "I5d" in h1 # not a stale I5c/I5b label + assert "I5c" not in h1 and "I5b" not in h1 + assert "fee_rate=0.001" in text + assert "analytics.quantiles=5" in text + assert "mmp_quintile_nav.png" in text # figures embedded + + +# --------------------------------------------------------------------------- # +# metric helper sanity +# --------------------------------------------------------------------------- # +def test_max_drawdown_includes_initial_baseline(): + nav = _toy_nav([0.9, 1.2]) # drops to 0.9 from the 1.0 start + assert _max_drawdown(nav) == pytest.approx(-0.1)