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/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/data/clean/intraday_aggregate.py b/data/clean/intraday_aggregate.py index 3d62308..d3058a9 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,12 @@ 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, + 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) @@ -143,12 +220,54 @@ 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, trade_date, session_open) + return out + + +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``; + 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( + 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)] + return float(np.mean(valid)) if valid.size else float("nan") def asof_daily_features( @@ -158,6 +277,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 +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) + 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]) @@ -198,6 +320,61 @@ 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, + 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 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( + [], 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): + 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( + 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))) + 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``. 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/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): 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 diff --git a/qt/intraday_tail_framework.py b/qt/intraday_tail_framework.py index b048ffd..cef262c 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,125 @@ 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, session_open=ic.session_open + ) + 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 +456,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 +487,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 +521,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 +606,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 +625,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 +809,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") 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 ) 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..23966ce --- /dev/null +++ b/tests/test_i5c_mmp_minute_factor.py @@ -0,0 +1,307 @@ +"""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_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) + 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") 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)