diff --git a/config/phase_i5d_mmp_quintile_5y.yaml b/config/phase_i5d_mmp_quintile_5y.yaml new file mode 100644 index 0000000..cf42d12 --- /dev/null +++ b/config/phase_i5d_mmp_quintile_5y.yaml @@ -0,0 +1,129 @@ +# Phase I5d — MMP Quintile 5Y Group Backtest (EXPLORATORY, NOT a performance claim). +# +# Ranks the I5c Minute Microstructure Pressure (MMP) daily score +# (intraday_mmp20_ew_0930_1450) on each monthly rebalance date and splits the +# cross-section into analytics.quantiles EQUAL-COUNT groups (Q1 = lowest score, +# Q5 = highest). Each group runs as its own long-only equal-weight portfolio +# through the SAME I5a/I5b intraday tail event model (14:50 decision / 14:51 +# execution, exec-to-exec returns) with raw stk_limit execution feasibility ON and +# trading cost fee_rate=0.001. Factor math / event model / execution feasibility +# are UNCHANGED — only the grouping + per-group orchestration is new. No tuning. +# +# Window = 5 complete years (2021-06-01 .. 2026-05-31). Primary universe = CSI500 +# (000905.SH): enough cross-section for 5 groups. Minute bars are read from the +# EXISTING intraday cache ONLY on the monthly anchor dates (read-only; a miss is a +# hard blocker -> zero stk_mins live calls). require_cache_coverage=false drops the +# minute-uncovered constituents (DISCLOSED in the report; the WINDOW is not +# shortened). Raw stk_limit flows through the same P4 read-through cache as the +# daily endpoints. + +project: + name: quantitative_trading_phase_i5d_mmp_quintile_5y + timezone: Asia/Shanghai +data: + source: tushare + freq: D + start: '2021-06-01' + end: '2026-05-31' + external_secret_file: /home/shaofl/Projects/financial_projects/.config.json + tushare_token_key: tushare.token + output_name: i5d_daily + cache: + enabled: true + root_dir: artifacts/cache/tushare/v1 + refresh_recent_days: 14 + refresh_dimension_days: 30 + force_refresh: [] +universe: + type: index + index_code: 000905.SH + symbols: [] + min_listing_days: 60 + filters: + # Selection-level daily tradability flags stay DISABLED (as in I5b/I5c): the + # daily close limit flag is an EOD value not known at 14:50. The honest + # direction-aware blocking is execution-time raw stk_limit vs the raw + # execution-minute close. + missing_close: true + suspended: false + st: false + limit_up_down: false +# factors is required by the schema but the runner ignores it: the score is the I3 +# mmp_ew feature (selected via intraday.score_feature), not a config factor. +factors: +- name: momentum_20 + enabled: true + params: + window: 20 + price_col: close +processing: + drop_missing: true + standardize: + enabled: true + method: zscore + winsorize: + enabled: false + method: mad + n: 3.0 + neutralize: + enabled: false + industry_col: industry + size_col: market_cap + industry_level: L1 +alpha: + model: equal_weight + params: {} +portfolio: + # top_n is required by the schema but IGNORED by the grouped backtest: each + # quantile group holds ALL its bucket members (EqualWeightAll), not a top-N cut. + constructor: topn_equal_weight + top_n: 10 + long_only: true + max_weight: null + turnover_cap: null +backtest: + initial_nav: 1.0 + rebalance: monthly + event_order: intraday_tail_rebalance + cash_return: 0.0 +intraday: + enabled: true + decision_time: '14:50:00' + data_lag: 1min + session_open: '09:30:00' + execution_model: next_minute_close + execution_window: + - '14:51:00' + - '14:56:59' + # A 5-year CSI500 minute history is not fully cached for every constituent; drop + # the uncovered names (disclosed in the report) instead of failing the whole run. + # The window is NOT shortened. + require_cache_coverage: false + missing_execution: block + # I5b execution-time raw price-limit feasibility stays ON. + price_limit_check: true + require_price_limit_coverage: true + limit_tolerance: 1.0e-06 + # I5c: select the exploratory MMP minute factor as the PIT-safe daily score. + score_feature: mmp_ew +cost: + fee_rate: 0.001 + slippage_rate: 0.0 + turnover_formula: l1 +analytics: + forward_return_periods: + - 1 + - 5 + - 20 + # number of equal-count quantile groups for the I5d grouped backtest. + quantiles: 5 + benchmark: null +output: + root_dir: artifacts + data_dir: artifacts/data + factor_dir: artifacts/factors + report_dir: artifacts/reports + log_dir: artifacts/logs + overwrite: true + intraday_report_name: phase_i5d_mmp_quintile_5y + intraday_report_title: Phase I5d — MMP Quintile 5Y Group Backtest diff --git a/qt/cli.py b/qt/cli.py index af6b865..c43b85a 100644 --- a/qt/cli.py +++ b/qt/cli.py @@ -179,6 +179,30 @@ def _cmd_run_phase_i5a_intraday(args: argparse.Namespace) -> int: return 0 +def _cmd_run_phase_i5d_intraday_groups(args: argparse.Namespace) -> int: + """Run the I5d MMP quintile grouped intraday-tail backtest + report/figures.""" + from qt.intraday_group_backtest import run_phase_i5d_intraday_groups + + try: + result = run_phase_i5d_intraday_groups(args.config) + except (ConfigError, ValueError, FileNotFoundError, RuntimeError) as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 1 + navs = " ".join( + f"Q{g.group}={g.metrics['final_nav']:.4f}" for g in result.groups + ) + print( + f"OK run-phase-i5d-intraday-groups: groups={result.n_groups}, " + f"rebalances={result.rebalance_count}, " + f"covered={result.covered_symbols}/{result.requested_symbols}, " + f"stk_mins_live_calls={result.minute_live_calls}, " + f"fee_rate={result.config.cost.fee_rate}\n" + f"final_nav: {navs}\n" + f"report: {result.report_path}" + ) + return 0 + + def _cmd_data_update(args: argparse.Namespace) -> int: """Warm/update the tushare caches (P4-3); never runs a backtest.""" from qt.data_updater import format_summary, run_data_update @@ -269,6 +293,13 @@ def build_parser() -> argparse.ArgumentParser: p_i5a.add_argument("--config", required=True, help="Path to the YAML config.") p_i5a.set_defaults(func=_cmd_run_phase_i5a_intraday) + p_i5d = sub.add_parser( + "run-phase-i5d-intraday-groups", + help="Run the I5d MMP quintile grouped intraday-tail backtest (5 groups).", + ) + p_i5d.add_argument("--config", required=True, help="Path to the YAML config.") + p_i5d.set_defaults(func=_cmd_run_phase_i5d_intraday_groups) + for name, func, help_text in ( ("fetch-data", _cmd_fetch_data, "Run the spine, report data fetch."), ("compute-factors", _cmd_compute_factors, "Run the spine, report factor compute."), diff --git a/qt/intraday_group_backtest.py b/qt/intraday_group_backtest.py new file mode 100644 index 0000000..bee0e9a --- /dev/null +++ b/qt/intraday_group_backtest.py @@ -0,0 +1,611 @@ +"""run-phase-i5d-intraday-groups: a 5-quantile grouped intraday-tail backtest of +the I5c MMP daily score (EXPLORATORY factor analysis — NOT a performance claim). + +On each monthly rebalance date the cross-section is ranked by the PIT-safe daily +score ``intraday_mmp20_ew_0930_1450`` (the I5c Minute Microstructure Pressure +factor, visible at the 14:50 cutoff) and split into ``analytics.quantiles`` +EQUAL-COUNT groups (Q1 = lowest score, QN = highest). Each group is run as its own +long-only equal-weight portfolio through the SAME event-driven machinery the daily +and I5a/I5b intraday paths use — a fresh :class:`BacktestEngine` + +:class:`IntradayTailEventModel` + ``SimExecution(fee_rate=...)`` per group — so +14:51 execution pricing, exec-to-exec holding returns, raw ``stk_limit`` execution +feasibility (I5b), turnover, cost and cash are modelled consistently and NEVER +fall back to a daily close-to-close return. + +Two engineering constraints drive the design: + + * **memory** — a 5-year CSI500 minute history is enormous, but the monthly event + model only needs minute bars on the ANCHOR dates (each rebalance/exit date). + :func:`_load_anchor_minute_bars` reads ONLY those days from the persistent + intraday cache (read-only; a miss never warms the cache → zero ``stk_mins`` + live calls), so the run never materialises the full-window minute panel. + * **compute** — the execution-price matrix is a pure function of + ``(bars, anchors, symbols, cfg)``; it is built ONCE and shared (immutable) + across the N fresh per-group models, while each model keeps its OWN mutable + feasibility diagnostics (no cross-group contamination). + +The MMP score, the factor math, the daily backtest, and the execution feasibility +are all UNCHANGED — only the grouping + per-group orchestration is new. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from pathlib import Path + +import numpy as np +import pandas as pd + +from data.cache.intervals import subtract_intervals +from data.cache.intraday_cache import ENDPOINT as INTRADAY_ENDPOINT +from data.cache.intraday_cache import READ_COLUMNS +from data.cache.intraday_coverage import IntradayCoverageLedger +from data.cache.intraday_parquet_store import IntradayParquetStore +from data.clean.intraday_schema import RAW_INTRADAY_FREQ, normalize_intraday_bars +from portfolio.base import PortfolioConstructor +from qt.config import RootConfig, load_config + +# Reuse the stable I5a/I5b/I5c helpers rather than copy-paste them (goal §1: avoid +# drift). These are import-stable private helpers: run preconditions, the exec +# config builder, the configured-feature daily score, and the raw stk_limit loader. +from qt.intraday_groups import EqualWeightAll, GroupScores, assign_quantile_buckets +from qt.intraday_tail_framework import ( + _check_i5a_preconditions, + _exec_cfg_from, + _load_price_limits, + _score_panel, +) +from qt.pipeline import _build_cache, _build_universe, _load_panel, _make_logger +from runtime.backtest.engine import BacktestEngine +from runtime.backtest.event_models import IntradayTailEventModel +from runtime.backtest.events import monthly_anchor_pairs, trading_calendar +from runtime.backtest.sim_execution import SimExecution +from runtime.intraday_execution import IntradayExecutionConfig + +_LOGGER_NAME = "qt.intraday_group_backtest" +_DEFAULT_REPORT_NAME = "phase_i5d_mmp_quintile_5y" +_DEFAULT_REPORT_TITLE = "Phase I5d — MMP Quintile 5Y Group Backtest" + +# Monthly rebalance -> 12 periods/year for annualization. +_PERIODS_PER_YEAR = 12.0 + + +# --------------------------------------------------------------------------- # +# Per-group performance metrics (monthly net-return series -> annualized stats) +# --------------------------------------------------------------------------- # +def _annualized_return(nav: pd.DataFrame) -> float: + """CAGR from the settled NAV path (monthly compounding).""" + if nav.empty: + return float("nan") + final = float(nav["nav"].iloc[-1]) + n = len(nav) + if n <= 0 or final <= 0: + return float("nan") + return final ** (_PERIODS_PER_YEAR / n) - 1.0 + + +def _annualized_vol(nav: pd.DataFrame) -> float: + """Annualized volatility of the monthly net returns.""" + if len(nav) < 2: + return float("nan") + return float(nav["net_return"].std(ddof=1) * np.sqrt(_PERIODS_PER_YEAR)) + + +def _sharpe(nav: pd.DataFrame) -> float: + """Annualized Sharpe (rf=0) = mean(net)*12 / (std(net)*sqrt(12)).""" + sigma = _annualized_vol(nav) + if not np.isfinite(sigma) or sigma == 0.0: + return float("nan") + mu = float(nav["net_return"].mean()) * _PERIODS_PER_YEAR + return mu / sigma + + +def _max_drawdown(nav: pd.DataFrame) -> float: + """Worst peak-to-trough drawdown of the NAV path (incl. the 1.0 start).""" + if nav.empty: + return float("nan") + path = np.concatenate([[1.0], nav["nav"].to_numpy(dtype=float)]) + running_max = np.maximum.accumulate(path) + return float(np.min(path / running_max - 1.0)) + + +def _avg_holdings(holdings_log: pd.DataFrame) -> float: + """Mean number of achieved holdings per settled rebalance date.""" + if holdings_log is None or holdings_log.empty: + return 0.0 + return float(holdings_log.groupby("date").size().mean()) + + +def _group_metrics(nav: pd.DataFrame, holdings_log: pd.DataFrame) -> dict[str, float]: + """Headline per-group metrics from one group's NAV table + holdings log.""" + final_nav = float(nav["nav"].iloc[-1]) if not nav.empty else float("nan") + return { + "final_nav": final_nav, + "annual_return": _annualized_return(nav), + "volatility": _annualized_vol(nav), + "sharpe": _sharpe(nav), + "max_drawdown": _max_drawdown(nav), + "mean_turnover": float(nav["turnover"].mean()) if not nav.empty else float("nan"), + "total_cost": float(nav["cost"].sum()) if not nav.empty else 0.0, + "avg_holdings": _avg_holdings(holdings_log), + "n_periods": int(len(nav)), + } + + +# --------------------------------------------------------------------------- # +# Minute loading (cache-only, anchor-date-sliced -> memory-conscious) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class _MinuteLoad: + """Diagnostics from the cache-only, anchor-date-sliced minute read.""" + + bars: pd.DataFrame + covered: list[str] + uncovered: list[str] + raw_rows: int + normalized_rows: int + anchor_dates: int + live_calls: int + + +def _coverage_partition( + cfg: RootConfig, symbols: list[str], root: str +) -> tuple[list[str], list[str]]: + """Split ``symbols`` into (fully covered, uncovered) over the data window. + + A symbol is covered only if the intraday ledger leaves NO uncovered trading-day + gap across ``[data.start, data.end]`` — exactly the I5a rule, so the realized + universe is auditable and never silently shrunk by a partial gap. + """ + ledger = IntradayCoverageLedger(root) + req_start = pd.Timestamp(cfg.data.start).normalize() + req_end = pd.Timestamp(cfg.data.end).normalize() + covered: list[str] = [] + uncovered: list[str] = [] + for sym in symbols: + gaps = subtract_intervals( + req_start, + req_end, + ledger.covered_day_intervals(INTRADAY_ENDPOINT, sym, RAW_INTRADAY_FREQ), + ) + (uncovered if gaps else covered).append(sym) + return covered, uncovered + + +def _load_anchor_minute_bars( + cfg: RootConfig, + symbols: list[str], + anchor_dates: list[pd.Timestamp], + logger, +) -> _MinuteLoad: + """Read 1min bars for ONLY the ``anchor_dates`` from the cache (read-only). + + Memory-conscious: the monthly event model prices only the rebalance/exit + anchors, so we read one trading day at a time from the month-partitioned store + and keep just those rows — the full multi-year minute history is NEVER + materialised. No fetch closure exists here, so ``stk_mins`` live calls are + provably zero (a miss simply yields no rows for that day/symbol). Uncovered + symbols are handled exactly as I5a: in ``require_cache_coverage`` mode a single + uncovered name is a loud blocker; otherwise they are excluded and disclosed. + """ + root = cfg.data.cache.root_dir + covered, uncovered = _coverage_partition(cfg, symbols, root) + + require_cov = cfg.intraday is not None and cfg.intraday.require_cache_coverage + if require_cov and uncovered: + shown = ", ".join(uncovered[:10]) + more = "" if len(uncovered) <= 10 else f" (+{len(uncovered) - 10} more)" + raise ValueError( + "i5d grouped backtest blocked: require_cache_coverage=true but " + f"{len(uncovered)}/{len(symbols)} requested symbols are NOT fully " + f"covered in the minute cache for [{cfg.data.start}, {cfg.data.end}]: " + f"{shown}{more}. Set intraday.require_cache_coverage=false to drop the " + "uncovered names (disclosed), or shrink the window to fully-covered " + "data. This runner refuses to warm missing minute history." + ) + if uncovered: + logger.info( + "intraday cache: %d/%d symbols fully covered; %d excluded (uncovered, " + "require_cache_coverage=false)", + len(covered), len(symbols), len(uncovered), + ) + if not covered: + raise ValueError( + "i5d grouped backtest blocked: no requested symbol is fully covered in " + f"the minute cache for [{cfg.data.start}, {cfg.data.end}]." + ) + + store = IntradayParquetStore(root) + frames: list[pd.DataFrame] = [] + raw_rows = 0 + for anchor in anchor_dates: + day = pd.Timestamp(anchor).normalize() + day_end = day + pd.Timedelta("23:59:59") + for sym in covered: + part = store.read_range( + INTRADAY_ENDPOINT, sym, RAW_INTRADAY_FREQ, day, day_end + ) + if not part.empty: + hit = part.rename(columns={"bar_end": "time"}) + frames.append(hit[READ_COLUMNS]) + raw_rows += len(part) + if not frames: + raise ValueError( + "i5d grouped backtest blocked: the covered symbols returned no cached " + "1min bars on the anchor dates (unexpected — check the coverage ledger)." + ) + read = pd.concat(frames, ignore_index=True) + bars = normalize_intraday_bars( + read, freq=RAW_INTRADAY_FREQ, data_lag=cfg.intraday.data_lag + ) + logger.info( + "intraday minute load (anchor-sliced): %d anchor dates, %d covered symbols, " + "%d raw rows, %d normalized rows, stk_mins_live_calls=0", + len(anchor_dates), len(covered), raw_rows, len(read), + ) + return _MinuteLoad( + bars=bars, + covered=covered, + uncovered=uncovered, + raw_rows=raw_rows, + normalized_rows=int(len(read)), + anchor_dates=len(anchor_dates), + live_calls=0, + ) + + +# --------------------------------------------------------------------------- # +# Group assignment (single source of truth, shared by all N group engines) +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class _GroupAssignment: + """Per-date bucket assignment + per-date diagnostics (immutable).""" + + by_date: dict[pd.Timestamp, dict[str, int]] + per_date_rows: tuple[dict, ...] + + +def _build_group_assignment( + score_series: pd.Series, + universe, + panel: pd.DataFrame, + rebalance_dates: list[pd.Timestamp], + covered: set[str], + n_groups: int, +) -> _GroupAssignment: + """Rank the MMP cross-section into N equal-count buckets on each rebalance date. + + Uses ONLY the daily PIT universe (``universe.tradable``) ∩ covered names ∩ + valid MMP scores — no forward return ever enters. Returns the per-date + ``{symbol: group}`` map and the per-date diagnostics (scored count, group + sizes, score distribution). + """ + by_date: dict[pd.Timestamp, dict[str, int]] = {} + rows: list[dict] = [] + for d in rebalance_dates: + norm = pd.Timestamp(d).normalize() + tradable = [s for s in universe.tradable(d, panel) if str(s) in covered] + try: + cross = score_series.xs(norm, level="date") + except KeyError: + cross = pd.Series(dtype=float) + cross = cross.reindex([str(s) for s in tradable]).dropna() + assignment = assign_quantile_buckets(cross, n_groups) + by_date[norm] = assignment + sizes = [0] * n_groups + for g in assignment.values(): + sizes[g - 1] += 1 + if len(cross) > 0: + arr = cross.to_numpy(dtype=float) + stats = { + "mean": float(np.mean(arr)), + "std": float(np.std(arr)), + "p10": float(np.percentile(arr, 10)), + "p50": float(np.percentile(arr, 50)), + "p90": float(np.percentile(arr, 90)), + } + else: + stats = {k: float("nan") for k in ("mean", "std", "p10", "p50", "p90")} + rows.append( + {"date": norm, "n_scored": int(len(cross)), "sizes": tuple(sizes), **stats} + ) + return _GroupAssignment(by_date=by_date, per_date_rows=tuple(rows)) + + +# --------------------------------------------------------------------------- # +# Result container +# --------------------------------------------------------------------------- # +@dataclass(frozen=True) +class GroupRunResult: + """One quantile group's settled run + diagnostics (immutable).""" + + group: int + nav_table: pd.DataFrame + holdings_log: pd.DataFrame + metrics: dict[str, float] + up_limit_blocked_buys: int + down_limit_blocked_sells: int + missing_limit_rows: int + blocked_fill_reasons: dict[str, int] + + +@dataclass(frozen=True) +class I5dResult: + """Immutable summary of the I5d grouped backtest run.""" + + config: RootConfig + n_groups: int + score_feature: str + score_feature_key: str + requested_symbols: int + covered_symbols: int + uncovered_symbols: tuple[str, ...] + anchor_dates: int + raw_rows: int + normalized_rows: int + minute_live_calls: int + rebalance_count: int + groups: tuple[GroupRunResult, ...] + spread_per_period: pd.Series + spread_cumulative: pd.Series + spread_summary: dict[str, float] + monotonicity: dict[str, float] + per_date_rows: tuple[dict, ...] + score_coverage: dict[str, int] + price_limit_check: bool + limit_coverage: dict[str, int] + stk_limit_gap_fetches: int + figure_paths: dict[str, Path] + report_path: Path + log_path: Path + elapsed: float + + +# --------------------------------------------------------------------------- # +# Runner +# --------------------------------------------------------------------------- # +def _report_basename(cfg: RootConfig) -> str: + return cfg.output.intraday_report_name or _DEFAULT_REPORT_NAME + + +def _run_one_group( + group: int, + assignment: _GroupAssignment, + *, + cfg: RootConfig, + panel: pd.DataFrame, + bars: pd.DataFrame, + universe, + exec_cfg: IntradayExecutionConfig, + price_limits: pd.DataFrame | None, + shared_prices: tuple[pd.DataFrame, list] | None, + constructor: PortfolioConstructor, +) -> tuple[GroupRunResult, tuple[pd.DataFrame, list]]: + """Run ONE quantile group through a fresh engine/model/execution. + + A FRESH :class:`IntradayTailEventModel` and ``SimExecution`` per group (no + shared mutable state); the immutable execution-price matrix is built by the + first group and reused by the rest (``shared_prices``). Returns the group + result and the shared prices (so the caller can thread them to the next group). + """ + ic = cfg.intraday + assert ic is not None + model = IntradayTailEventModel( + calendar_panel=panel, + bars=bars, + cfg=exec_cfg, + price_limits=price_limits, + price_limit_check=ic.price_limit_check, + limit_tolerance=ic.limit_tolerance, + require_price_limit_coverage=ic.require_price_limit_coverage, + precomputed_prices=shared_prices, + ) + if shared_prices is None: + # First group builds the matrix; reuse it (immutable) for the rest. + shared_prices = (model.execution_prices(), model.fills()) + + execution = SimExecution(fee_rate=cfg.cost.fee_rate) + engine = BacktestEngine( + model=model, + universe=universe, + scores=GroupScores(assignment.by_date, group), + constructor=constructor, + execution=execution, + selection_panel=panel, + initial_nav=cfg.backtest.initial_nav, + cash_return=cfg.backtest.cash_return, + ) + nav = engine.run() + holdings = engine.holdings_log() + blocked: dict[str, int] = {} + for f in model.blocked_fills(): + key = f.reason or "unknown" + blocked[key] = blocked.get(key, 0) + 1 + result = GroupRunResult( + group=group, + nav_table=nav, + holdings_log=holdings, + metrics=_group_metrics(nav, holdings), + up_limit_blocked_buys=model.up_limit_blocked_buys(), + down_limit_blocked_sells=model.down_limit_blocked_sells(), + missing_limit_rows=model.missing_limit_rows(), + blocked_fill_reasons=blocked, + ) + return result, shared_prices + + +def _spread_and_monotonicity( + groups: tuple[GroupRunResult, ...], n_groups: int +) -> tuple[pd.Series, pd.Series, dict[str, float], dict[str, float]]: + """Synthetic QN−Q1 spread series + group monotonicity diagnostics (report-only). + + The spread is the per-period difference of the high-group and low-group NET + returns and its CUMULATIVE SUM — a synthetic long-only leg difference, NOT a + separately executed dollar-neutral portfolio. Monotonicity is the Spearman + rank correlation of the group index (1..N) vs the group annual return / final + NAV. + """ + by_group = {g.group: g for g in groups} + low, high = by_group.get(1), by_group.get(n_groups) + if low is None or high is None or low.nav_table.empty or high.nav_table.empty: + empty = pd.Series(dtype=float) + return empty, empty, {}, {} + aligned = pd.concat( + [high.nav_table["net_return"].rename("high"), + low.nav_table["net_return"].rename("low")], + axis=1, + ).dropna() + per_period = (aligned["high"] - aligned["low"]).rename("spread") + cumulative = per_period.cumsum().rename("cum_spread") + summary = { + "mean_per_period": float(per_period.mean()) if len(per_period) else float("nan"), + "total": float(cumulative.iloc[-1]) if len(cumulative) else float("nan"), + "n_periods": int(len(per_period)), + } + idx = pd.Series( + [g for g in sorted(by_group)], dtype=float, index=sorted(by_group) + ) + annual = pd.Series( + {g: by_group[g].metrics["annual_return"] for g in sorted(by_group)} + ) + final_nav = pd.Series( + {g: by_group[g].metrics["final_nav"] for g in sorted(by_group)} + ) + mono = { + "annual_spearman": float(idx.corr(annual, method="spearman")), + "final_nav_spearman": float(idx.corr(final_nav, method="spearman")), + } + return per_period, cumulative, summary, mono + + +def run_phase_i5d_intraday_groups(config_path: str) -> I5dResult: + """Run the I5d MMP quintile grouped intraday-tail backtest and write its report.""" + cfg = load_config(config_path) + _check_i5a_preconditions(cfg) # real tushare, intraday enabled, event order, cache + ic = cfg.intraday + assert ic is not None + if ic.score_feature != "mmp_ew": + raise ValueError( + "run-phase-i5d-intraday-groups is the MMP quintile study and requires " + f"intraday.score_feature='mmp_ew' (got {ic.score_feature!r})." + ) + n_groups = int(cfg.analytics.quantiles) + if n_groups < 2: + raise ValueError( + f"analytics.quantiles must be >= 2 for a grouped backtest; got {n_groups}." + ) + + basename = _report_basename(cfg) + log_dir = Path(cfg.output.log_dir) + log_dir.mkdir(parents=True, exist_ok=True) + log_path = log_dir / f"{basename}.log" + logger = _make_logger(log_path, name=_LOGGER_NAME) + started = time.monotonic() + + cache = _build_cache(cfg) + universe, symbols = _build_universe(cfg, logger, cache) + panel = _load_panel(cfg, symbols, logger, cache) # daily qfq selection panel + calendar + + # Monthly anchors from the daily calendar; minute bars only on those days. + pairs = monthly_anchor_pairs(trading_calendar(panel)) + rebalance_dates = [pd.Timestamp(p[0]).normalize() for p in pairs] + anchor_dates = sorted({pd.Timestamp(d).normalize() for pair in pairs for d in pair}) + logger.info( + "schedule: %d monthly rebalances, %d anchor dates", len(pairs), len(anchor_dates) + ) + + load = _load_anchor_minute_bars(cfg, symbols, anchor_dates, logger) + score_series, score_feature = _score_panel(cfg, load.bars, logger) + score_coverage = { + "rows": int(score_series.shape[0]), + "valid": int(score_series.notna().sum()), + "nan": int(score_series.isna().sum()), + } + + price_limits, stk_limit_gap_fetches = _load_price_limits( + cfg, load.covered, cache, logger + ) + exec_cfg = _exec_cfg_from(cfg) + + assignment = _build_group_assignment( + score_series, universe, panel, rebalance_dates, set(load.covered), n_groups + ) + + constructor = EqualWeightAll(long_only=cfg.portfolio.long_only) + group_results: list[GroupRunResult] = [] + shared_prices: tuple[pd.DataFrame, list] | None = None + limit_coverage: dict[str, int] = {} + for group in range(1, n_groups + 1): + result, shared_prices = _run_one_group( + group, + assignment, + cfg=cfg, + panel=panel, + bars=load.bars, + universe=universe, + exec_cfg=exec_cfg, + price_limits=price_limits, + shared_prices=shared_prices, + constructor=constructor, + ) + group_results.append(result) + logger.info( + "group Q%d: periods=%d final_nav=%.6f up_blocked=%d down_blocked=%d", + group, result.metrics["n_periods"], result.metrics["final_nav"], + result.up_limit_blocked_buys, result.down_limit_blocked_sells, + ) + # limit coverage is identical across groups (derived from the shared prices); + # read it once from a fresh model so the diagnostic is reported a single time. + if ic.price_limit_check: + cov_model = IntradayTailEventModel( + calendar_panel=panel, bars=load.bars, cfg=exec_cfg, + price_limits=price_limits, price_limit_check=True, + limit_tolerance=ic.limit_tolerance, + require_price_limit_coverage=ic.require_price_limit_coverage, + precomputed_prices=shared_prices, + ) + limit_coverage = cov_model.limit_coverage() + + per_period, cumulative, spread_summary, monotonicity = _spread_and_monotonicity( + tuple(group_results), n_groups + ) + + report_path = Path(cfg.output.report_dir) / f"{basename}.md" + figure_dir = Path(cfg.output.report_dir) / f"{basename}_figures" + result = I5dResult( + config=cfg, + n_groups=n_groups, + score_feature=score_feature, + score_feature_key=ic.score_feature, + requested_symbols=len(symbols), + covered_symbols=len(load.covered), + uncovered_symbols=tuple(load.uncovered), + anchor_dates=load.anchor_dates, + raw_rows=load.raw_rows, + normalized_rows=load.normalized_rows, + minute_live_calls=load.live_calls, + rebalance_count=len(pairs), + groups=tuple(group_results), + spread_per_period=per_period, + spread_cumulative=cumulative, + spread_summary=spread_summary, + monotonicity=monotonicity, + per_date_rows=assignment.per_date_rows, + score_coverage=score_coverage, + price_limit_check=ic.price_limit_check, + limit_coverage=limit_coverage, + stk_limit_gap_fetches=stk_limit_gap_fetches, + figure_paths=_write_figures(figure_dir, tuple(group_results), per_period, + cumulative, n_groups), + report_path=report_path, + log_path=log_path, + elapsed=time.monotonic() - started, + ) + _write_report(result) + logger.info("report: %s", report_path) + return result + + +# Report + figure rendering live in a sibling module for cohesion. +from qt.intraday_group_report import _write_figures, _write_report # noqa: E402 diff --git a/qt/intraday_group_figures.py b/qt/intraday_group_figures.py new file mode 100644 index 0000000..e2443f4 --- /dev/null +++ b/qt/intraday_group_figures.py @@ -0,0 +1,139 @@ +"""Matplotlib (Agg) figures for the I5d MMP quintile grouped backtest. + +Pure plotting helpers: each takes already-computed frames/series and an output +path, renders ONE PNG with a non-interactive ``Agg`` backend (no display needed), +and returns the written path. They are deliberately decoupled from the run logic +so they can be unit-tested with toy NAV frames (goal §Tests 5). + +Direction convention is made explicit on every figure: **Q1 = lowest MMP score, +QN = highest** — the legend/labels say "low"/"high" so the quantile direction can +never be misread. +""" + +from __future__ import annotations + +from pathlib import Path + +import matplotlib + +matplotlib.use("Agg") # headless: render to PNG, never open a display. + +import matplotlib.pyplot as plt # noqa: E402 (must follow the use("Agg") call) +import pandas as pd # noqa: E402 + +# Required metric columns for the grouped-bar figure (one bar group per metric). +METRIC_COLUMNS: tuple[str, ...] = ( + "annual_return", + "max_drawdown", + "mean_turnover", + "total_cost", +) +_METRIC_TITLES = { + "annual_return": "Annualized return", + "max_drawdown": "Max drawdown", + "mean_turnover": "Mean turnover", + "total_cost": "Total cost", +} + + +def _group_label(group: int, n_groups: int) -> str: + """``Q{g}`` with a low/high tag on the extremes (direction is unmissable).""" + if group == 1: + return f"Q{group} (low)" + if group == n_groups: + return f"Q{group} (high)" + return f"Q{group}" + + +def plot_quintile_nav( + nav_by_group: dict[int, pd.DataFrame], out_path: str | Path, n_groups: int +) -> Path: + """NAV curves for Q1..QN on one chart (x = rebalance date, y = NAV).""" + out = Path(out_path) + out.parent.mkdir(parents=True, exist_ok=True) + fig, ax = plt.subplots(figsize=(10, 6)) + for group in sorted(nav_by_group): + nav = nav_by_group[group] + if nav is None or nav.empty or "nav" not in nav.columns: + continue + ax.plot( + [pd.Timestamp(d) for d in nav.index], + nav["nav"].to_numpy(dtype=float), + marker="o", + markersize=3, + label=_group_label(group, n_groups), + ) + ax.set_title("MMP quintile group NAV (Q1 = lowest score, QN = highest)") + ax.set_xlabel("Rebalance date") + ax.set_ylabel("NAV (start = 1.0)") + ax.axhline(1.0, color="grey", linewidth=0.8, linestyle="--") + ax.legend(loc="best", fontsize=9) + ax.grid(True, alpha=0.3) + fig.autofmt_xdate() + fig.tight_layout() + fig.savefig(out, dpi=120) + plt.close(fig) + return out + + +def plot_spread_curve( + spread_curve: pd.Series, + out_path: str | Path, + *, + low_label: str = "Q1", + high_label: str = "Q5", +) -> Path: + """Cumulative synthetic ``high − low`` group spread curve (long-only legs).""" + out = Path(out_path) + out.parent.mkdir(parents=True, exist_ok=True) + fig, ax = plt.subplots(figsize=(10, 5)) + if spread_curve is not None and len(spread_curve) > 0: + ax.plot( + [pd.Timestamp(d) for d in spread_curve.index], + spread_curve.to_numpy(dtype=float), + color="tab:purple", + marker="o", + markersize=3, + ) + ax.axhline(0.0, color="grey", linewidth=0.8, linestyle="--") + ax.set_title( + f"Cumulative synthetic {high_label}−{low_label} spread " + "(long-only leg difference, NOT a dollar-neutral book)" + ) + ax.set_xlabel("Rebalance date") + ax.set_ylabel(f"Cumulative {high_label}−{low_label} net return") + ax.grid(True, alpha=0.3) + fig.autofmt_xdate() + fig.tight_layout() + fig.savefig(out, dpi=120) + plt.close(fig) + return out + + +def plot_group_metrics( + metrics: pd.DataFrame, out_path: str | Path, n_groups: int +) -> Path: + """Compact 2x2 grouped-bar panel: annual return / maxDD / turnover / cost.""" + out = Path(out_path) + out.parent.mkdir(parents=True, exist_ok=True) + groups = sorted(metrics.index) + labels = [_group_label(int(g), n_groups) for g in groups] + fig, axes = plt.subplots(2, 2, figsize=(11, 8)) + for ax, col in zip(axes.flat, METRIC_COLUMNS): + if col in metrics.columns: + values = [float(metrics.loc[g, col]) for g in groups] + else: + values = [float("nan")] * len(groups) + ax.bar(labels, values, color="tab:blue") + ax.set_title(_METRIC_TITLES.get(col, col)) + ax.axhline(0.0, color="grey", linewidth=0.8) + ax.grid(True, axis="y", alpha=0.3) + ax.tick_params(axis="x", labelrotation=0, labelsize=9) + fig.suptitle( + "MMP quintile group metrics (Q1 = lowest score, QN = highest)", + fontsize=12, + ) + fig.tight_layout(rect=(0, 0, 1, 0.97)) + fig.savefig(out, dpi=120) + plt.close(fig) + return out diff --git a/qt/intraday_group_report.py b/qt/intraday_group_report.py new file mode 100644 index 0000000..21d1dd1 --- /dev/null +++ b/qt/intraday_group_report.py @@ -0,0 +1,345 @@ +"""Markdown report + PNG figures for the I5d MMP quintile grouped backtest. + +Kept in a sibling module (cohesion / small files) and decoupled from the run +logic: every function takes already-computed results and only READS them, so there +is no import cycle with :mod:`qt.intraday_group_backtest`. The report makes the +exploratory framing, the Q1/Q5 direction, the ``fee_rate`` and the cache-only +minute provenance explicit, and embeds the three required figures. +""" + +from __future__ import annotations + +from pathlib import Path + +import pandas as pd + +from qt.intraday_group_figures import ( + METRIC_COLUMNS, + plot_group_metrics, + plot_quintile_nav, + plot_spread_curve, +) + + +def _write_figures( + figure_dir: Path, + groups: tuple, + per_period: pd.Series, + cumulative: pd.Series, + n_groups: int, +) -> dict[str, Path]: + """Render the three required PNGs (NAV curves / spread / metric bars).""" + figure_dir.mkdir(parents=True, exist_ok=True) + nav_by_group = {g.group: g.nav_table for g in groups} + metrics = pd.DataFrame( + {g.group: {c: g.metrics.get(c, float("nan")) for c in METRIC_COLUMNS} + for g in groups} + ).T + nav_path = plot_quintile_nav( + nav_by_group, figure_dir / "mmp_quintile_nav.png", n_groups + ) + spread_path = plot_spread_curve( + cumulative, + figure_dir / f"mmp_q{n_groups}_minus_q1_spread.png", + low_label="Q1", + high_label=f"Q{n_groups}", + ) + metrics_path = plot_group_metrics( + metrics, figure_dir / "mmp_quintile_metrics.png", n_groups + ) + return {"nav": nav_path, "spread": spread_path, "metrics": metrics_path} + + +def _fmt(v: float, pct: bool = False, nd: int = 4) -> str: + """Format a float, NaN-safe; ``pct`` renders as a percentage.""" + if v is None or (isinstance(v, float) and pd.isna(v)): + return "NaN" + if pct: + return f"{v * 100:.2f}%" + return f"{v:.{nd}f}" + + +def _intro_lines(result) -> list[str]: + """Title + exploratory framing + study design.""" + cfg = result.config + title = cfg.output.intraday_report_title or "Phase I5d — MMP Quintile 5Y Group Backtest" + t = title.strip() + h1 = t if t.startswith("#") else f"# {t}" + lines = [ + h1, + "", + "**This is an EXPLORATORY grouped factor analysis of the I5c MMP " + "minute-derived daily score — NOT a performance claim and NOT parameter " + "tuning.** On each monthly rebalance date the cross-section is ranked by the " + "PIT-safe daily score and split into equal-count quantile groups; each group " + "is run as its own long-only equal-weight portfolio through the SAME " + "event-driven intraday-tail model (14:50 decision / 14:51 execution, " + "exec-to-exec returns) with I5b raw `stk_limit` execution feasibility ON and " + "trading cost `fee_rate` applied. No parameter, group count, cost, universe, " + "or window was tuned from results.", + "", + "## Study design", + "", + f"- window: `{cfg.data.start}` → `{cfg.data.end}`", + f"- universe: `{cfg.universe.type}` `{cfg.universe.index_code or ''}` " + f"(requested {result.requested_symbols} distinct constituents)", + f"- rebalance: monthly — {result.rebalance_count} settled periods", + f"- score column: `{result.score_feature}` (key=`{result.score_feature_key}`)", + f"- groups: `analytics.quantiles={result.n_groups}` equal-count rank buckets, " + "**Q1 = lowest MMP score, " + f"Q{result.n_groups} = highest**", + f"- decision_time `{cfg.intraday.decision_time}` / execution_window " + f"`[{cfg.intraday.execution_window[0]}, {cfg.intraday.execution_window[1]}]`; " + "returns are execution-to-execution, NEVER close-to-close", + f"- trading cost: `cost.fee_rate={cfg.cost.fee_rate}`, " + f"`slippage_rate={cfg.cost.slippage_rate}` (cost line = turnover × fee_rate " + "inside SimExecution; no extra ad-hoc cost layer)", + "", + ] + return lines + + +def _mmp_and_grouping_lines(result) -> list[str]: + return [ + "## MMP factor & group assignment", + "", + "Per 1min bar `t`: `mid=(high+low)/2`; `S=(close-mid)/mid`; " + "`V=sqrt(volume/median(vol[t-20:t]))`; `B=|close-open|/(high-low+eps)`; " + "`R=(high-low)/(mean(hl[t-20:t])+eps)`; **`MMP_t = S*V*B*R`** (`eps=1e-6`). " + "The daily score `intraday_mmp20_ew_0930_1450` is the EQUAL-WEIGHT mean of " + "valid `MMP_t` over the in-session bars visible at the 14:50 cutoff " + "(`[session_open, decision_time]`); rolling baselines use only the prior 20 " + "in-session bars (first 20 are NaN). The MMP math is UNCHANGED from I5c.", + "- **group rule**: drop NaN/non-finite scores, sort by `(score asc, symbol " + "asc)`, split BY RANK into equal-count buckets. `Q1` = lowest score, " + f"`Q{result.n_groups}` = highest. Equal-count (not value-cut) buckets keep " + "tied/degenerate scores deterministic; a date with too few valid names " + "leaves the high groups empty (no crash).", + "- **no lookahead**: grouping uses only the daily PIT universe ∩ " + "minute-covered names ∩ valid MMP score; forward returns NEVER enter the " + "assignment.", + "", + ] + + +def _provenance_lines(result) -> list[str]: + lines = [ + "## Minute-cache coverage & data provenance", + "", + f"- requested symbols: {result.requested_symbols}; minute-cache fully " + f"covered: {result.covered_symbols}; excluded (uncovered): " + f"{len(result.uncovered_symbols)}", + f"- anchor dates requiring minute bars: {result.anchor_dates} (rebalance ∪ " + "exit dates only — the full multi-year minute history is never loaded)", + f"- raw rows loaded: {result.raw_rows}; normalized rows used: " + f"{result.normalized_rows}", + f"- **stk_mins live API calls during this run: {result.minute_live_calls}** " + "(cache-only, read-only; a miss is never a silent warm/backfill)", + f"- elapsed: {result.elapsed:.1f}s", + ] + if result.uncovered_symbols: + shown = ", ".join(result.uncovered_symbols[:20]) + more = ( + "" if len(result.uncovered_symbols) <= 20 + else f" (+{len(result.uncovered_symbols) - 20} more)" + ) + lines.append( + f"- **excluded (uncovered) symbols** ({len(result.uncovered_symbols)}): " + f"{shown}{more}" + ) + lines.append( + " - ⚠️ dropping uncovered names trades full-window completion for a " + "potential coverage bias (the realized cross-section omits names with " + "no cached minute history). Disclosed, not silent; the 5-year WINDOW is " + "NOT shortened." + ) + sc = result.score_coverage + lines.append( + f"- daily MMP score panel: {sc['rows']} (date,symbol) rows; valid " + f"{sc['valid']}; NaN {sc['nan']}." + ) + lines.append("") + return lines + + +def _feasibility_lines(result) -> list[str]: + cfg = result.config + lines = ["## Execution-time price-limit feasibility (I5b)", ""] + if not result.price_limit_check: + lines.append( + "- **disabled**: feasibility is the base bar-exists rule only " + "(missing/NaN execution bar blocks both directions)." + ) + lines.append("") + return lines + cov = result.limit_coverage + lines.extend([ + "- **enabled** (`intraday.price_limit_check=true`): a buy is blocked at the " + "raw upper limit and a sell at the raw lower limit, comparing the selected " + "execution-minute **raw** 1min close to the raw `stk_limit` band " + "(RAW-vs-RAW; never qfq / daily close / a daily-close-derived flag).", + f"- limit tolerance: `{cfg.intraday.limit_tolerance}`; " + f"require_price_limit_coverage: `{cfg.intraday.require_price_limit_coverage}`", + f"- limit coverage over rebalance anchors (shared across groups): required " + f"{cov.get('required', 0)} (date, symbol) pairs; present " + f"{cov.get('present', 0)}; missing {cov.get('missing', 0)}", + f"- **stk_limit cache gap-fetches this run: {result.stk_limit_gap_fetches}** " + "(read through the existing P4 daily cache — never a minute/stk_mins fetch).", + "", + "Per-group blocked buy/sell counts (a fresh model per group, so counts do " + "NOT double-count across groups):", + "", + "| group | up-limit blocked buys | down-limit blocked sells | unchecked limit rows |", + "|---|---|---|---|", + ]) + for g in result.groups: + lines.append( + f"| Q{g.group} | {g.up_limit_blocked_buys} | " + f"{g.down_limit_blocked_sells} | {g.missing_limit_rows} |" + ) + lines.append("") + return lines + + +def _performance_lines(result) -> list[str]: + lines = [ + "## Per-group NAV / performance", + "", + "| group | final NAV | annual | vol | Sharpe | maxDD | mean turnover | " + "total cost | avg holdings |", + "|---|---|---|---|---|---|---|---|---|", + ] + for g in result.groups: + m = g.metrics + tag = " (low)" if g.group == 1 else (" (high)" if g.group == result.n_groups else "") + lines.append( + f"| Q{g.group}{tag} | {_fmt(m['final_nav'])} | " + f"{_fmt(m['annual_return'], pct=True)} | {_fmt(m['volatility'], pct=True)} " + f"| {_fmt(m['sharpe'])} | {_fmt(m['max_drawdown'], pct=True)} | " + f"{_fmt(m['mean_turnover'])} | {_fmt(m['total_cost'])} | " + f"{_fmt(m['avg_holdings'], nd=1)} |" + ) + lines.append("") + lines.append( + "- turnover/cost count the ACHIEVED book after feasible fills (price-limit " + "blocked + missing-bar blocked names earn nothing, never a daily-close " + "fallback); idle cash earns `cash_return`." + ) + lines.append("") + return lines + + +def _spread_lines(result) -> list[str]: + n = result.n_groups + s = result.spread_summary + mono = result.monotonicity + lines = [ + f"## Q{n}−Q1 synthetic spread & monotonicity (report-only)", + "", + f"The Q{n}−Q1 spread is the per-period difference of the Q{n} (high) and Q1 " + "(low) group NET returns; the cumulative curve is their running sum. It is a " + "**synthetic long-only leg difference, NOT a separately executed " + "dollar-neutral portfolio** (no long-short execution model is run).", + ] + if s: + lines.append( + f"- mean per-period Q{n}−Q1 net return: {_fmt(s.get('mean_per_period'), pct=True)} " + f"over {int(s.get('n_periods', 0))} periods; cumulative (sum): " + f"{_fmt(s.get('total'), pct=True)}." + ) + if mono: + lines.append( + f"- group monotonicity (Spearman of group index 1..{n} vs metric): " + f"annual return {_fmt(mono.get('annual_spearman'))}, final NAV " + f"{_fmt(mono.get('final_nav_spearman'))} " + "(+1 = perfectly increasing Q1→QN, −1 = perfectly decreasing)." + ) + lines.append( + "- **report-only**: returns are read here for analytics only and never feed " + "the factor/alpha; a single overlapping window is far too little to infer " + "factor quality." + ) + lines.append("") + return lines + + +def _per_date_lines(result) -> list[str]: + n = result.n_groups + header = "| date | n_scored | " + " | ".join(f"Q{i}" for i in range(1, n + 1)) + \ + " | score mean | std | p10 | p50 | p90 |" + sep = "|---|---|" + "---|" * n + "---|---|---|---|---|" + lines = [ + "## Per-rebalance group sizes & score distribution", + "", + header, + sep, + ] + for r in result.per_date_rows: + sizes = " | ".join(str(x) for x in r["sizes"]) + lines.append( + f"| {pd.Timestamp(r['date']).date()} | {r['n_scored']} | {sizes} | " + f"{_fmt(r['mean'], nd=6)} | {_fmt(r['std'], nd=6)} | {_fmt(r['p10'], nd=6)} " + f"| {_fmt(r['p50'], nd=6)} | {_fmt(r['p90'], nd=6)} |" + ) + lines.append("") + return lines + + +def _figure_lines(result) -> list[str]: + report_dir = result.report_path.parent + lines = ["## Figures", ""] + captions = { + "nav": "NAV curves for each quantile group (Q1 = lowest MMP score, " + f"Q{result.n_groups} = highest).", + "spread": f"Cumulative synthetic Q{result.n_groups}−Q1 net-return spread " + "(long-only leg difference).", + "metrics": "Grouped bars: annualized return, max drawdown, mean turnover, " + "total cost per group.", + } + for key in ("nav", "spread", "metrics"): + path = result.figure_paths.get(key) + if path is None: + continue + rel = Path(path).relative_to(report_dir).as_posix() + lines.append(f"**{captions[key]}**") + lines.append("") + lines.append(f"![{key}]({rel})") + lines.append("") + return lines + + +def _limitations_lines(result) -> list[str]: + return [ + "## Limitations (explicit)", + "", + "- **EXPLORATORY grouped factor analysis, NOT a performance claim**: one " + "factor, one overlapping 5-year window, one universe. No tuning, no " + "robustness matrix, no learned / IC-weighted alpha.", + "- **coverage bias**: uncovered minute names are dropped (disclosed above); " + "the realized cross-section is the covered subset.", + "- **execution feasibility** models price-limit + bar-existence only; no " + "partial-fill / liquidity / volume cap at the execution minute. Suspended " + "names have no minute bar and are blocked by the missing-bar rule; explicit " + "ST status is not consulted at 14:50.", + f"- **Q{result.n_groups}−Q1 spread is synthetic** (a long-only leg " + "difference), not a separately executed dollar-neutral book.", + "", + ] + + +def _write_report(result) -> None: + """Write the I5d grouped-backtest markdown report (with embedded figures).""" + path = result.report_path + path.parent.mkdir(parents=True, exist_ok=True) + lines: list[str] = [] + lines += _intro_lines(result) + lines += _mmp_and_grouping_lines(result) + lines += _provenance_lines(result) + lines += _performance_lines(result) + lines += _spread_lines(result) + lines += _feasibility_lines(result) + lines += _per_date_lines(result) + lines += _figure_lines(result) + lines += _limitations_lines(result) + lines.append(f"_elapsed: {result.elapsed:.1f}s_") + path.write_text("\n".join(lines) + "\n", encoding="utf-8") diff --git a/qt/intraday_groups.py b/qt/intraday_groups.py new file mode 100644 index 0000000..9b5f6b4 --- /dev/null +++ b/qt/intraday_groups.py @@ -0,0 +1,131 @@ +"""Cross-sectional quantile grouping primitives for the I5d grouped backtest. + +These are the small, pure pieces the grouped intraday-tail backtest +(:mod:`qt.intraday_group_backtest`) layers on TOP of the existing event-driven +machinery — they decide *which names land in which group* and present one group +to the shared :class:`~runtime.backtest.engine.BacktestEngine` as an equal-weight +target, WITHOUT touching the engine, the execution feasibility, or the factor +math: + + * :func:`assign_quantile_buckets` — split one scored cross-section into N + EQUAL-COUNT rank buckets (Q1 = lowest score, QN = highest). Rank/position + buckets (not value cuts) so tied or degenerate scores still produce a + deterministic, auditable assignment; too few names simply leave high groups + empty instead of crashing. + * :class:`GroupScores` — a ``ScoresSource`` that exposes ONE group's members to + the engine (1.0 for a member, NaN otherwise), so the engine's selection picks + exactly that bucket. + * :class:`EqualWeightAll` — a constructor that equal-weights EVERY non-NaN name + (no ``top_n`` cap): the grouped backtest holds the whole bucket. + +Hard boundary preserved (CLAUDE.md invariant #1/#3): nothing here reads a forward +return, a data source, or places an order. The MMP score is computed upstream +(PIT-safe, available_time <= 14:50) and only its CROSS-SECTIONAL RANK is used to +form groups. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd + +from portfolio.base import PortfolioConstructor + + +def assign_quantile_buckets(scores: pd.Series, n_groups: int) -> dict[str, int]: + """Assign each scored symbol to one of ``n_groups`` EQUAL-COUNT rank buckets. + + Args: + scores: symbol-indexed scores for ONE cross-section (one rebalance date). + n_groups: number of quantile groups (e.g. 5). + + Returns: + ``{symbol: label}`` with ``label`` in ``1..n_groups``, where **Q1 (label + 1) is the LOWEST score** and **QN (label n_groups) is the HIGHEST**. + + Semantics: + * NaN / non-finite scores are dropped (never assigned to a group). + * Symbols are ordered by ``(score ascending, symbol ascending)`` and split + BY POSITION into ``n_groups`` contiguous chunks (``np.array_split``), so + the assignment is deterministic even with tied scores (the symbol + tie-break decides) — a value-cut quantile could not split ties cleanly. + * When fewer than ``n_groups`` names are scored, the lower groups fill + first and the high groups are simply empty (no crash). Chunk sizes + differ by at most one; the extra goes to the lower groups (the + ``np.array_split`` convention). + """ + if n_groups < 1: + raise ValueError(f"n_groups must be >= 1; got {n_groups}.") + if scores is None or len(scores) == 0: + return {} + s = pd.Series(scores, dtype=float).dropna() + if s.empty: + return {} + finite_mask = np.isfinite(s.to_numpy(dtype=float)) + s = s[finite_mask] + if s.empty: + return {} + order = sorted(s.index, key=lambda sym: (float(s.loc[sym]), str(sym))) + buckets = np.array_split(np.array(order, dtype=object), n_groups) + out: dict[str, int] = {} + for i, bucket in enumerate(buckets): + for sym in bucket: + out[str(sym)] = i + 1 + return out + + +class GroupScores: + """``ScoresSource`` exposing exactly ONE quantile group to the engine. + + Bridges the precomputed per-date bucket assignment to the ``ScoresSource`` + port :class:`~runtime.backtest.engine.BacktestEngine` depends on. ``get`` marks + a symbol with a constant score ``1.0`` iff it is a member of ``group`` on that + date, else ``NaN`` — so an equal-weight-all constructor selects exactly the + bucket. It only READS a precomputed rank assignment (no forward returns, no + data source), preserving the no-lookahead boundary. + """ + + def __init__( + self, assignments: dict[pd.Timestamp, dict[str, int]], group: int + ) -> None: + # assignments: normalized rebalance date -> {symbol: group label}. + self._assignments = assignments + self._group = int(group) + + def get(self, date: pd.Timestamp, symbols: list[str]) -> pd.Series: + """Return symbol-indexed membership scores (1.0 member / NaN non-member).""" + norm = pd.Timestamp(date).normalize() + members = self._assignments.get(norm, {}) + values = [ + 1.0 if members.get(str(sym)) == self._group else np.nan + for sym in symbols + ] + return pd.Series(values, index=list(symbols), dtype=float) + + +class EqualWeightAll(PortfolioConstructor): + """Equal-weight EVERY non-NaN name in the cross-section (no ``top_n`` cap). + + The grouped backtest holds an entire quantile bucket, so — unlike + :class:`portfolio.construct.TopNEqualWeight` — there is no top-N selection: + every name the :class:`GroupScores` source surfaces (score ``1.0``) is held at + weight ``1/k``. NaN scores are dropped (non-members); an empty cross-section + yields an empty (all-cash) target. Long-only; the input is never mutated. + """ + + def __init__(self, long_only: bool = True) -> None: + self.long_only = bool(long_only) + + def build( + self, + scores: pd.Series, + current_weights: pd.Series | None = None, + ) -> pd.Series: + """Equal-weight all non-NaN names; empty target when there are none.""" + candidates = scores.dropna() + if candidates.empty: + return pd.Series(dtype=float, index=candidates.index[:0], name="weight") + weight = 1.0 / len(candidates) + result = pd.Series(weight, index=candidates.index, name="weight") + result.index.name = scores.index.name + return result 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_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)