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