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/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/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_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/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")