diff --git a/config/phase_i5a_intraday_tail_framework.yaml b/config/phase_i5a_intraday_tail_framework.yaml new file mode 100644 index 0000000..fdc9bc7 --- /dev/null +++ b/config/phase_i5a_intraday_tail_framework.yaml @@ -0,0 +1,108 @@ +# Phase I5a — intraday tail-rebalance event-framework smoke (architecture, NOT research). +# +# Proves the shared event-driven backtest engine drives an IntradayTailEventModel +# end-to-end on REAL SH/SZ minute data: decision at 14:50, execution at the first +# valid 1min close in [14:51, 14:56:59], exec-to-exec holding returns. The score is +# a single PIT-safe I3 feature (intraday_ret_0930_1450), not a research alpha. +# +# Minute bars are read from the EXISTING intraday cache only (read-only; a cache +# miss is a hard blocker, never a silent warm -> zero stk_mins live calls). The +# daily panel + PIT index membership come through the existing P4 read-through cache. +# SSE50 (000016.SH) is SH/SZ-only; window is short + recent (covered by the cache). + +project: + name: quantitative_trading_phase_i5a_intraday_tail + timezone: Asia/Shanghai +data: + source: tushare + freq: D + start: '2026-03-03' + end: '2026-06-12' + external_secret_file: /home/shaofl/Projects/financial_projects/.config.json + tushare_token_key: tushare.token + output_name: i5a_daily + cache: + enabled: true + root_dir: artifacts/cache/tushare/v1 + refresh_recent_days: 14 + refresh_dimension_days: 30 + force_refresh: [] +universe: + type: index + index_code: 000016.SH + symbols: [] + min_listing_days: 60 + filters: + # Selection-level daily tradability flags are DISABLED for this smoke: at the + # 14:50 decision the daily-close-derived limit flag is an EOD value, and the + # honest blocking happens at EXECUTION via minute bars (a suspended/halted name + # has no minute bar and is blocked there). Disclosed in the report. + missing_close: true + suspended: false + st: false + limit_up_down: false +# factors is required by the schema but the I5a runner ignores it: the score is the +# I3 intraday_ret feature, not a config factor. Kept as a minimal valid placeholder. +factors: +- name: momentum_20 + enabled: true + params: + window: 20 + price_col: close +processing: + drop_missing: true + standardize: + enabled: true + method: zscore + winsorize: + enabled: false + method: mad + n: 3.0 + neutralize: + enabled: false + industry_col: industry + size_col: market_cap + industry_level: L1 +alpha: + model: equal_weight + params: {} +portfolio: + constructor: topn_equal_weight + top_n: 10 + long_only: true + max_weight: null + turnover_cap: null +backtest: + initial_nav: 1.0 + rebalance: monthly + event_order: intraday_tail_rebalance + cash_return: 0.0 +intraday: + enabled: true + decision_time: '14:50:00' + data_lag: 1min + session_open: '09:30:00' + execution_model: next_minute_close + execution_window: + - '14:51:00' + - '14:56:59' + require_cache_coverage: true + missing_execution: block +cost: + fee_rate: 0.001 + slippage_rate: 0.0 + turnover_formula: l1 +analytics: + forward_return_periods: + - 1 + - 5 + - 20 + quantiles: 5 + benchmark: null +output: + root_dir: artifacts + data_dir: artifacts/data + factor_dir: artifacts/factors + report_dir: artifacts/reports + log_dir: artifacts/logs + overwrite: true diff --git a/qt/cli.py b/qt/cli.py index 5b2aace..af6b865 100644 --- a/qt/cli.py +++ b/qt/cli.py @@ -156,6 +156,29 @@ def _cmd_run_phase3_subset(args: argparse.Namespace) -> int: return 0 +def _cmd_run_phase_i5a_intraday(args: argparse.Namespace) -> int: + """Run the I5a intraday tail-rebalance architecture smoke + report.""" + from qt.intraday_tail_framework import run_phase_i5a_intraday + + try: + result = run_phase_i5a_intraday(args.config) + except (ConfigError, ValueError, FileNotFoundError, RuntimeError) as exc: + print(f"ERROR: {exc}", file=sys.stderr) + return 1 + final_nav = ( + result.nav_table["nav"].iloc[-1] if not result.nav_table.empty else float("nan") + ) + n_blocked = sum(result.blocked_fill_counts.values()) + print( + f"OK run-phase-i5a-intraday: periods={len(result.nav_table)}, " + f"covered={result.covered_symbols}/{result.requested_symbols}, " + f"stk_mins_live_calls={result.minute_live_calls}, blocked_fills={n_blocked}, " + f"final_nav={final_nav:.6f}\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 @@ -239,6 +262,13 @@ def build_parser() -> argparse.ArgumentParser: p_du.add_argument("--config", required=True, help="Path to the YAML config.") p_du.set_defaults(func=_cmd_data_update) + p_i5a = sub.add_parser( + "run-phase-i5a-intraday", + help="Run the I5a intraday tail-rebalance architecture smoke (minute cache).", + ) + p_i5a.add_argument("--config", required=True, help="Path to the YAML config.") + p_i5a.set_defaults(func=_cmd_run_phase_i5a_intraday) + for name, func, help_text in ( ("fetch-data", _cmd_fetch_data, "Run the spine, report data fetch."), ("compute-factors", _cmd_compute_factors, "Run the spine, report factor compute."), diff --git a/qt/config.py b/qt/config.py index 464a01d..e800899 100644 --- a/qt/config.py +++ b/qt/config.py @@ -216,10 +216,76 @@ def _check_top_n(cls, v: int) -> int: class BacktestCfg(_Strict): initial_nav: float = 1.0 rebalance: Literal["monthly"] = "monthly" - event_order: str = "close_to_next_period" + # ``close_to_next_period`` = the daily close-to-close model (default, unchanged). + # ``intraday_tail_rebalance`` = the I5a 14:50-decision / 14:51-execution model, + # which requires ``intraday.enabled=true`` (enforced on RootConfig). + event_order: Literal["close_to_next_period", "intraday_tail_rebalance"] = ( + "close_to_next_period" + ) cash_return: float = 0.0 +# Execution models the intraday tail event model can price (I5a: just the +# conservative first one). Coarser/auction proxies are future work and rejected +# readably so a config can never silently fall back to a wrong model. +_SUPPORTED_INTRADAY_EXECUTION_MODELS: tuple[str, ...] = ("next_minute_close",) + + +def _parse_hms(value: str) -> int: + """Parse an ``HH:MM:SS`` clock string to seconds-since-midnight (validation only).""" + parts = str(value).split(":") + if len(parts) != 3: + raise ValueError(f"expected HH:MM:SS, got {value!r}") + h, m, s = (int(p) for p in parts) + if not (0 <= h < 24 and 0 <= m < 60 and 0 <= s < 60): + raise ValueError(f"out-of-range clock value {value!r}") + return h * 3600 + m * 60 + s + + +class IntradayCfg(_Strict): + """Opt-in intraday tail-rebalance event model declaration (I5a). + + Off by default; all daily configs validate unchanged. When the backtest's + ``event_order`` is ``intraday_tail_rebalance`` this section must have + ``enabled=true`` (enforced on RootConfig). ``decision_time`` is the signal + cutoff (features must satisfy ``available_time <= decision_time``); + ``execution_window`` is where the fill bar is taken; the window must start + strictly after the decision and be non-empty. + """ + + enabled: bool = False + decision_time: str = "14:50:00" + data_lag: str = "1min" + session_open: str = "09:30:00" + execution_model: str = "next_minute_close" + execution_window: tuple[str, str] = ("14:51:00", "14:56:59") + require_cache_coverage: bool = True + missing_execution: Literal["block"] = "block" + + @model_validator(mode="after") + def _check_execution(self) -> "IntradayCfg": + if self.execution_model not in _SUPPORTED_INTRADAY_EXECUTION_MODELS: + raise ValueError( + f"intraday.execution_model {self.execution_model!r} is not " + f"supported; choose one of {_SUPPORTED_INTRADAY_EXECUTION_MODELS}." + ) + try: + decision = _parse_hms(self.decision_time) + start = _parse_hms(self.execution_window[0]) + end = _parse_hms(self.execution_window[1]) + except ValueError as exc: + raise ValueError( + f"intraday time fields must be HH:MM:SS clock strings: {exc}" + ) from exc + if not (start > decision and start <= end): + raise ValueError( + "intraday.execution_window must satisfy " + "decision_time < window_start <= window_end (got " + f"decision={self.decision_time}, window={list(self.execution_window)})." + ) + return self + + class CostCfg(_Strict): fee_rate: float = 0.001 slippage_rate: float = 0.0 @@ -559,6 +625,9 @@ class RootConfig(_Strict): cost: CostCfg analytics: AnalyticsCfg = Field(default_factory=AnalyticsCfg) output: OutputCfg + # P-I5a intraday tail event model (consumed only by run-phase-i5a-intraday and + # any backtest with event_order='intraday_tail_rebalance'). Off by default. + intraday: IntradayCfg | None = None oos: OOSCfg | None = None # P3-4 robustness matrix (consumed only by run-phase3-robustness). robustness: RobustnessCfg | None = None @@ -567,6 +636,24 @@ class RootConfig(_Strict): # P4-3 data updater (consumed only by data-update). data_update: DataUpdateCfg | None = None + @model_validator(mode="after") + def _check_intraday_event_order(self) -> "RootConfig": + """``intraday_tail_rebalance`` requires an enabled ``intraday`` section. + + The daily default (``close_to_next_period``) needs no intraday section, so + every existing config validates unchanged. Selecting the intraday event + model without ``intraday.enabled=true`` is a configuration error (it would + otherwise silently run the daily model). + """ + if self.backtest.event_order == "intraday_tail_rebalance": + if self.intraday is None or not self.intraday.enabled: + raise ValueError( + "backtest.event_order='intraday_tail_rebalance' requires an " + "'intraday' section with enabled=true; got " + f"intraday={'None' if self.intraday is None else 'enabled=false'}." + ) + return self + @model_validator(mode="after") def _check_subset_groups_reference_enabled_factors(self) -> "RootConfig": if self.subset_validation is None: diff --git a/qt/intraday_tail_framework.py b/qt/intraday_tail_framework.py new file mode 100644 index 0000000..064baa3 --- /dev/null +++ b/qt/intraday_tail_framework.py @@ -0,0 +1,448 @@ +"""run-phase-i5a-intraday: an architecture smoke for the intraday tail event model. + +This runner proves the shared event-driven backtest engine +(:class:`runtime.backtest.engine.BacktestEngine`) can drive an +:class:`runtime.backtest.event_models.IntradayTailEventModel` end-to-end on REAL +SH/SZ data — without a new research alpha and without touching the daily path. + +Pipeline: + + build daily panel + universe (existing P4 read-through cache) + -> load required 1min bars from the EXISTING intraday cache (read-only; a + cache miss is a loud blocker, NEVER a silent warm -> zero stk_mins calls) + -> deterministic PIT-safe score = the I3 ``intraday_ret_0930_1450`` feature + (only bars with available_time <= 14:50 enter it) + -> engine + IntradayTailEventModel: decision 14:50, execute at the first valid + 1min close in [14:51, 14:56:59], exec-to-exec holding returns + -> NAV / feasibility / holdings / event logs -> markdown report. + +It is intentionally small: the score is a single already-PIT-safe feature solely +to exercise the event framework, not a performance claim. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from pathlib import Path + +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 TushareIntradayCache +from data.cache.intraday_coverage import IntradayCoverageLedger +from data.cache.intraday_parquet_store import IntradayParquetStore +from data.clean.intraday_aggregate import asof_daily_features +from data.clean.intraday_schema import RAW_INTRADAY_FREQ, normalize_intraday_bars +from portfolio.construct import TopNEqualWeight +from qt.config import RootConfig, load_config +from qt.pipeline import ( + _FrameScores, + _build_cache, + _build_universe, + _load_panel, + _make_logger, +) +from runtime.backtest.engine import BacktestEngine +from runtime.backtest.event_models import IntradayTailEventModel +from runtime.backtest.sim_execution import SimExecution +from runtime.intraday_execution import IntradayExecutionConfig + +_SCORE_FEATURE = "intraday_ret" # the I3 feature family used as the smoke score +_LOGGER_NAME = "qt.intraday_tail_framework" + + +@dataclass(frozen=True) +class I5aResult: + """Immutable summary of one I5a intraday-tail smoke run.""" + + config: RootConfig + event_order: str + exec_cfg: IntradayExecutionConfig + score_feature: str + requested_symbols: int + covered_symbols: int + uncovered_symbols: tuple[str, ...] + minute_live_calls: int + nav_table: pd.DataFrame + event_log: pd.DataFrame + feasibility_log: pd.DataFrame + holdings_log: pd.DataFrame + blocked_fill_counts: dict[str, int] + # per rebalance date -> (earliest, latest) ACTUAL execution-bar time among the + # non-blocked fills (so a 14:52 fill when 14:51 was missing is auditable, vs + # the planned execution_ts). Empty tuple key absent -> no fill that date. + actual_exec_by_date: dict[pd.Timestamp, tuple[pd.Timestamp, pd.Timestamp]] + report_path: Path + log_path: Path + + +def _exec_cfg_from(cfg: RootConfig) -> IntradayExecutionConfig: + ic = cfg.intraday + assert ic is not None # guarded by the caller + return IntradayExecutionConfig( + decision_time=ic.decision_time, + data_lag=ic.data_lag, + execution_model=ic.execution_model, + execution_window=tuple(ic.execution_window), + ) + + +def _load_minute_bars_cache_only( + cfg: RootConfig, symbols: list[str], logger +) -> tuple[pd.DataFrame, list[str], list[str], int]: + """Read 1min bars for ``symbols`` over the window from the cache ONLY. + + Returns ``(bars, covered_symbols, uncovered_symbols, live_calls)``. A symbol + whose window is not fully covered by the intraday ledger is EXCLUDED (never + warmed); the read uses a fetch closure that raises, so any planned gap is a + loud error and ``live_calls`` is provably zero on the happy path. + """ + root = cfg.data.cache.root_dir + ledger = IntradayCoverageLedger(root) + store = IntradayParquetStore(root) + cache = TushareIntradayCache(store, ledger) + + 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) + + require_cov = cfg.intraday is not None and cfg.intraday.require_cache_coverage + if require_cov and uncovered: + # require_cache_coverage=true must NOT silently drop uncovered names: that + # would bias the realized universe. Fail loudly so the operator shrinks the + # window (or opts into lenient mode). Matches the goal: coverage missing -> + # change the window or stop, never warm. + shown = ", ".join(uncovered[:10]) + more = "" if len(uncovered) <= 10 else f" (+{len(uncovered) - 10} more)" + raise ValueError( + "intraday smoke blocked: require_cache_coverage=true but " + f"{len(uncovered)}/{len(symbols)} requested symbols are NOT fully covered " + f"in the minute cache for [{cfg.data.start}, {cfg.data.end}]: {shown}{more}. " + "Shrink/move the window to fully-covered SH/SZ data, or set " + "intraday.require_cache_coverage=false to drop the uncovered names " + "(disclosed). This runner refuses to warm missing minute history." + ) + if uncovered: + logger.info( + "intraday cache: %d/%d symbols covered; %d excluded (uncovered, " + "require_cache_coverage=false)", + len(covered), len(symbols), len(uncovered), + ) + if not covered: + raise ValueError( + "intraday smoke blocked: no requested symbol is covered in the minute " + f"cache for [{cfg.data.start}, {cfg.data.end}]." + ) + + def _no_warm(sym: str, start_dt: str, end_dt: str): + raise RuntimeError( + f"minute cache miss for {sym} [{start_dt}, {end_dt}]; the I5a runner " + "is read-only and refuses to warm the intraday cache." + ) + + start_dt = f"{cfg.data.start} 00:00:00" + end_dt = f"{cfg.data.end} 23:59:59" + read = cache.stk_mins_1min(covered, start_dt, end_dt, _no_warm, freq=RAW_INTRADAY_FREQ) + live_calls = int(cache.stats().get(INTRADAY_ENDPOINT, 0)) + if read.empty: + raise ValueError( + "intraday smoke blocked: the covered symbols returned no cached 1min " + "bars for the window (unexpected — check the coverage ledger)." + ) + bars = normalize_intraday_bars( + read, freq=RAW_INTRADAY_FREQ, data_lag=cfg.intraday.data_lag + ) + return bars, covered, uncovered, live_calls + + +def _score_panel(cfg: RootConfig, bars: pd.DataFrame, logger) -> tuple[pd.Series, str]: + """Deterministic PIT-safe score = the I3 intraday return feature. + + ``asof_daily_features`` keeps only bars with ``available_time <= decision_time`` + before aggregating to (date, symbol), so the score for date T is known at T's + 14:50 cutoff — exactly the information a tail decision may use. Returns the + score Series (named ``score``) and the source feature column name. + """ + ic = cfg.intraday + assert ic is not None + feats = asof_daily_features( + bars, decision_time=ic.decision_time, session_open=ic.session_open + ) + col = next((c for c in feats.columns if c.startswith(_SCORE_FEATURE)), None) + if col is None: + raise ValueError( + f"no {_SCORE_FEATURE!r} feature column produced by asof_daily_features " + f"(got {list(feats.columns)})." + ) + logger.info("intraday score: feature=%s, %d (date,symbol) rows", col, len(feats)) + return feats[col].rename("score"), col + + +def run_phase_i5a_intraday(config_path: str) -> I5aResult: + """Run the I5a intraday-tail architecture smoke and write its report.""" + cfg = load_config(config_path) + _check_i5a_preconditions(cfg) + + log_dir = Path(cfg.output.log_dir) + log_dir.mkdir(parents=True, exist_ok=True) + log_path = log_dir / "phase_i5a_intraday_tail_framework.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 panel: selection + calendar + + exec_cfg = _exec_cfg_from(cfg) + bars, covered, uncovered, live_calls = _load_minute_bars_cache_only(cfg, symbols, logger) + score_series, score_feature = _score_panel(cfg, bars, logger) + scores = _FrameScores(score_series) + + model = IntradayTailEventModel(calendar_panel=panel, bars=bars, cfg=exec_cfg) + execution = SimExecution(fee_rate=cfg.cost.fee_rate) + engine = BacktestEngine( + model=model, + universe=universe, + scores=scores, + constructor=TopNEqualWeight(cfg.portfolio.top_n, long_only=cfg.portfolio.long_only), + execution=execution, + selection_panel=panel, + initial_nav=cfg.backtest.initial_nav, + cash_return=cfg.backtest.cash_return, + ) + nav_table = engine.run() + logger.info("intraday backtest: %d settled periods", len(nav_table)) + + blocked = model.blocked_fills() + blocked_counts: dict[str, int] = {} + for f in blocked: + blocked_counts[f.reason or "unknown"] = blocked_counts.get(f.reason or "unknown", 0) + 1 + + # ACTUAL execution-bar time range per rebalance date (non-blocked fills): a fill + # later than the planned execution_ts (e.g. 14:52 when 14:51 was missing) is + # auditable here, distinct from the planned anchor in the event log. + actual_exec: dict[pd.Timestamp, tuple[pd.Timestamp, pd.Timestamp]] = {} + rebalance_dates = {pd.Timestamp(d).normalize() for d in nav_table.index} + for f in model.fills(): + if f.blocked or f.exec_time is None: + continue + d = pd.Timestamp(f.date).normalize() + if d not in rebalance_dates: + continue # exit-only anchors are not rebalance rows + t = pd.Timestamp(f.exec_time) + lo, hi = actual_exec.get(d, (t, t)) + actual_exec[d] = (min(lo, t), max(hi, t)) + + report_path = Path(cfg.output.report_dir) / "phase_i5a_intraday_tail_framework.md" + result = I5aResult( + config=cfg, + event_order=cfg.backtest.event_order, + exec_cfg=exec_cfg, + score_feature=score_feature, + requested_symbols=len(symbols), + covered_symbols=len(covered), + uncovered_symbols=tuple(uncovered), + minute_live_calls=live_calls, + nav_table=nav_table, + event_log=engine.event_log(), + feasibility_log=engine.feasibility_log(), + holdings_log=engine.holdings_log(), + blocked_fill_counts=blocked_counts, + actual_exec_by_date=actual_exec, + report_path=report_path, + log_path=log_path, + ) + _write_report(result, elapsed=time.monotonic() - started) + logger.info("report: %s", report_path) + return result + + +def _check_i5a_preconditions(cfg: RootConfig) -> None: + """Guards: real tushare source, intraday enabled, intraday event order, cache on.""" + if cfg.data.source != "tushare": + raise ValueError( + "run-phase-i5a-intraday is a REAL-data smoke and requires " + f"data.source='tushare' (got {cfg.data.source!r})." + ) + if cfg.intraday is None or not cfg.intraday.enabled: + raise ValueError( + "run-phase-i5a-intraday requires an 'intraday' section with enabled=true." + ) + if cfg.backtest.event_order != "intraday_tail_rebalance": + raise ValueError( + "run-phase-i5a-intraday requires backtest.event_order=" + f"'intraday_tail_rebalance' (got {cfg.backtest.event_order!r})." + ) + if not cfg.data.cache.enabled: + raise ValueError( + "run-phase-i5a-intraday requires data.cache.enabled=true (the minute " + "bars are read from the persistent intraday cache)." + ) + + +def _write_report(result: I5aResult, *, elapsed: float) -> None: + """Write the I5a architecture-smoke markdown report (auditable event basis).""" + cfg = result.config + ec = result.exec_cfg + path = result.report_path + path.parent.mkdir(parents=True, exist_ok=True) + + nav = result.nav_table + lines: list[str] = [] + lines.append("# Phase I5a — Intraday Tail-Rebalance Event Framework (architecture smoke)") + lines.append("") + lines.append( + "**This is an architecture/framework run, NOT a research result.** The " + "score is a single PIT-safe I3 feature used solely to exercise the shared " + "event-driven backtest engine with an intraday tail event model." + ) + lines.append("") + lines.append("## Event model") + lines.append("") + lines.append("- event model: `IntradayTailEventModel`") + lines.append(f"- backtest.event_order: `{result.event_order}`") + lines.append(f"- decision_time (signal cutoff): `{ec.decision_time}`") + lines.append(f"- data_lag (available_time = bar_end + lag): `{ec.data_lag}`") + lines.append(f"- execution_model: `{ec.execution_model}`") + lines.append( + f"- execution_window: `[{ec.execution_window[0]}, {ec.execution_window[1]}]`" + ) + lines.append(f"- score feature: `{result.score_feature}`") + lines.append("") + lines.append("**Returns are execution-to-execution, NOT close-to-close.** Holding " + "return of a period = exec_price(next) / exec_price(this) - 1, priced " + "at the first valid 1min close in the execution window.") + lines.append("") + lines.append("## Minute-cache coverage & data provenance") + lines.append("") + lines.append(f"- window: `{cfg.data.start}` → `{cfg.data.end}`") + lines.append(f"- universe: `{cfg.universe.type}` `{cfg.universe.index_code or ''}`") + lines.append( + f"- requested symbols: {result.requested_symbols}; " + f"minute-cache covered: {result.covered_symbols}; " + f"excluded (uncovered): {len(result.uncovered_symbols)}" + ) + lines.append( + f"- **stk_mins live API calls during this run: {result.minute_live_calls}** " + "(read-only; a cache miss is a hard blocker, never a silent warm)" + ) + if result.uncovered_symbols: + shown = ", ".join(result.uncovered_symbols[:10]) + more = "" if len(result.uncovered_symbols) <= 10 else f" (+{len(result.uncovered_symbols)-10} more)" + lines.append(f"- excluded symbols: {shown}{more}") + lines.append("") + lines.append("## Event table (decision / execution / exit anchors)") + lines.append("") + lines.append( + "`exec_ts(planned)` is the window start; `exec_bar(actual)` is the actual " + "fill-bar time range used across the held names (e.g. `14:52:00` would show " + "if 14:51 was missing). `exit_exec_ts` is the planned exit fill time at " + "`exit_date` — the holding period runs `exec_ts -> exit_exec_ts`." + ) + lines.append("") + if result.event_log.empty: + lines.append("_no settled periods_") + else: + lines.append( + "| date | decision_ts | exec_ts(planned) | exec_bar(actual) | " + "exit_date | exit_exec_ts |" + ) + lines.append("|---|---|---|---|---|---|") + for date, row in result.event_log.iterrows(): + actual = result.actual_exec_by_date.get(pd.Timestamp(date).normalize()) + if actual is None: + actual_str = "—" + else: + lo, hi = actual + actual_str = ( + f"{lo.time()}" if lo == hi else f"{lo.time()}–{hi.time()}" + ) + lines.append( + f"| {pd.Timestamp(date).date()} | {row['decision_ts']} | " + f"{row['execution_ts']} | {actual_str} | " + f"{pd.Timestamp(row['exit_date']).date()} | {row['exit_execution_ts']} |" + ) + lines.append("") + lines.append("## NAV / turnover / cost / cash") + lines.append("") + if nav.empty: + lines.append("_no settled periods_") + else: + lines.append("| date | nav | net_return | gross_return | cost | turnover |") + lines.append("|---|---|---|---|---|---|") + for date, row in nav.iterrows(): + lines.append( + f"| {pd.Timestamp(date).date()} | {row['nav']:.6f} | " + f"{row['net_return']:.6f} | {row['gross_return']:.6f} | " + f"{row['cost']:.6f} | {row['turnover']:.6f} |" + ) + lines.append("") + lines.append( + f"- final NAV: {nav['nav'].iloc[-1]:.6f}; " + f"avg turnover: {nav['turnover'].mean():.6f}; " + f"total cost: {nav['cost'].sum():.6f}; cash_return: {cfg.backtest.cash_return}" + ) + lines.append( + "- turnover/cost count the ACHIEVED book after feasible fills, not the " + "desired target." + ) + lines.append("") + lines.append("## Blocked fills (by reason)") + lines.append("") + if not result.blocked_fill_counts: + lines.append("_none_") + else: + for reason, n in sorted(result.blocked_fill_counts.items()): + lines.append(f"- `{reason}`: {n}") + lines.append( + "\nA blocked entry/exit bar excludes that symbol from the period (it earns " + "nothing) and is NEVER replaced by a daily close." + ) + lines.append("") + lines.append("## Achieved holdings (sample)") + lines.append("") + h = result.holdings_log + if h.empty: + lines.append("_none_") + else: + lines.append("| date | symbol | weight | rank |") + lines.append("|---|---|---|---|") + for _, row in h.head(15).iterrows(): + lines.append( + f"| {pd.Timestamp(row['date']).date()} | {row['symbol']} | " + f"{row['weight']:.6f} | {int(row['rank'])} |" + ) + if len(h) > 15: + lines.append(f"| … | … | … | … | ({len(h)} rows total) |") + lines.append("") + lines.append("## Limitations (explicit)") + lines.append("") + lines.append( + "- **Execution-time feasibility is the minimum I5a rule**: a missing/NaN " + "execution bar blocks BOTH directions. Price-limit feasibility at execution " + "time is NOT applied — the daily panel carries only daily-close-derived " + "limit flags, which the I5a contract forbids using for execution. Raw " + "`stk_limit` vs execution-minute-close is future work." + ) + lines.append( + "- **ST / suspension availability**: suspended stocks have no minute bars, " + "so they are blocked by the missing-bar rule; explicit ST status is NOT " + "consulted at execution time in this smoke." + ) + lines.append( + "- The score is a single PIT-safe intraday feature to prove the framework; " + "this is not a performance claim and no parameters were tuned." + ) + lines.append("") + lines.append(f"_elapsed: {elapsed:.1f}s_") + path.write_text("\n".join(lines) + "\n", encoding="utf-8") diff --git a/runtime/backtest/driver.py b/runtime/backtest/driver.py index c8d8967..9f01756 100644 --- a/runtime/backtest/driver.py +++ b/runtime/backtest/driver.py @@ -1,60 +1,46 @@ -"""BacktestDriver: wires the strategy layers through the fixed event order. +"""BacktestDriver: the daily close-to-close backtest (compatibility wrapper). -The driver owns no strategy logic of its own; it is pure orchestration over -injected collaborators (Ports): +Since I5a the backtest ledger lives in :class:`runtime.backtest.engine.BacktestEngine` +and the daily time-basis in :class:`runtime.backtest.event_models.DailyCloseEventModel`. +``BacktestDriver`` is now a thin import-compatible wrapper around that pair: same +constructor, same ``run`` / ``rebalance_dates`` / ``feasibility_log`` / +``holdings_log`` surface, byte-identical output (locked by the daily golden tests +and the phase0 regression). New code can target the engine directly; existing +callers (``qt.pipeline``, ``qt.oos_stability``, ``qt.phase2_baseline``) are +untouched. - universe -> .members / .tradable(date, panel) - scores -> .get(date, symbols) -> symbol-indexed scores - constructor -> .build(scores, current_weights) -> target weights - execution -> BacktestExecution port (Execution + settle / last_cost / - last_turnover; the backtest-only sub-port) - prices -> canonical (date, symbol) panel with a ``close`` column - -Fixed event order (CONTRACTS §6): +Fixed event order (CONTRACTS §6, preserved by ``DailyCloseEventModel``): compute factor at close of t -> rebalance after close of t -> hold from t+1 -So each rebalance is settled against the NEXT holding period's close-to-close -return (BT-003), never the same-day return the factor already saw. Empty -tradable universe -> the book is cash and earns ``cash_return`` (BT-007). +so each rebalance is settled against the NEXT holding period's close-to-close +return (BT-003). Empty tradable universe -> the book is cash and earns +``cash_return`` (BT-007). -Output (BT-005/006, framework_settings §7.9): a date-indexed DataFrame with -columns ``[nav, gross_return, cost, turnover, net_return]``. +Output (BT-005/006): a date-indexed DataFrame with columns +``[nav, gross_return, cost, turnover, net_return]``. """ from __future__ import annotations -from typing import Protocol - import pandas as pd +from runtime.backtest.engine import ( + BacktestEngine, + ConstructorPort, + ScoresSource, + UniversePort, +) +from runtime.backtest.event_models import DailyCloseEventModel +from runtime.backtest.events import monthly_rebalance_dates, trading_calendar from runtime.execution import BacktestExecution -from runtime.fills import feasibility_from_cross - - -class ScoresSource(Protocol): - """Minimal scores port the driver depends on.""" - - def get(self, date: pd.Timestamp, symbols: list[str]) -> pd.Series: ... - - -class UniversePort(Protocol): - """Minimal universe port the driver depends on.""" - - def tradable(self, date: pd.Timestamp, panel: pd.DataFrame) -> list[str]: ... - - -class ConstructorPort(Protocol): - """Minimal portfolio-constructor port the driver depends on.""" - - def build(self, scores: pd.Series, current_weights: pd.Series | None = ...) -> pd.Series: ... - - -_NAV_COLUMNS = ["nav", "gross_return", "cost", "turnover", "net_return"] class BacktestDriver: - """Monthly-rebalanced, single-period-compounding backtest over a price panel.""" + """Monthly-rebalanced, single-period-compounding daily backtest. + + Compatibility wrapper over :class:`BacktestEngine` + :class:`DailyCloseEventModel`. + """ def __init__( self, @@ -75,210 +61,37 @@ def __init__( ) if "close" not in prices.columns: raise ValueError("price panel must have a 'close' column") - self._universe = universe - self._scores = scores - self._constructor = constructor - self._execution = execution self._prices = prices - self._rebalance = rebalance + # fee_rate is carried for signature compatibility; the trading cost is + # owned by the execution adapter (SimExecution.fee_rate), as before. self._fee_rate = float(fee_rate) - self._initial_nav = float(initial_nav) - self._cash_return = float(cash_return) - self._feasibility_log: list[dict] = [] - self._holdings_log: list[dict] = [] - - # -- calendar --------------------------------------------------------- # - def _calendar(self) -> pd.DatetimeIndex: - """Sorted unique trading dates present in the price panel.""" - dates = self._prices.index.get_level_values("date").unique() - return pd.DatetimeIndex(sorted(dates)) + # Keep the execution adapter reachable on the driver (the engine holds the + # same object), preserving the legacy attribute surface for callers that + # inspect the post-run book. + self._execution = execution + self._engine = BacktestEngine( + model=DailyCloseEventModel(prices), + universe=universe, + scores=scores, + constructor=constructor, + execution=execution, + selection_panel=prices, + initial_nav=initial_nav, + cash_return=cash_return, + ) def rebalance_dates(self) -> list[pd.Timestamp]: """Last trading day of each month in the panel calendar (BT-001).""" - cal = self._calendar() - if len(cal) == 0: - return [] - frame = pd.DataFrame({"date": cal}) - frame["ym"] = frame["date"].dt.to_period("M") - last = frame.groupby("ym")["date"].max().sort_values() - return list(last) + return monthly_rebalance_dates(trading_calendar(self._prices)) - # -- pricing ---------------------------------------------------------- # - def _close_at(self, date: pd.Timestamp, symbol: str) -> float: - """Close of ``symbol`` on ``date`` (NaN if missing).""" - try: - return float(self._prices.loc[(date, symbol), "close"]) - except KeyError: - return float("nan") - - def _holding_returns( - self, start: pd.Timestamp, end: pd.Timestamp, symbols: list[str] - ) -> pd.Series: - """Close-to-close gross return per symbol over the FORWARD window. - - ``start`` is the rebalance (close of t); ``end`` is the next rebalance - date (or the final trading day). The position is held over (start, end], - so this is the next holding period's return — never a window ending on - ``start`` itself (BT-003). Symbols with a missing/zero start price get a - flat (0.0) return rather than NaN, so the book stays well-defined. - """ - out: dict[str, float] = {} - for sym in symbols: - start_px = self._close_at(start, sym) - end_px = self._close_at(end, sym) - if ( - start_px is None - or end_px is None - or pd.isna(start_px) - or pd.isna(end_px) - or start_px == 0.0 - ): - out[sym] = 0.0 - else: - out[sym] = end_px / start_px - 1.0 - return pd.Series(out, dtype=float) - - # -- run -------------------------------------------------------------- # def run(self) -> pd.DataFrame: """Run the backtest and return the NAV table (BT-005/006).""" - reb = self.rebalance_dates() - cal = self._calendar() - rows: list[dict] = [] - nav = self._initial_nav - self._feasibility_log = [] # re-entrant: fresh per run - self._holdings_log = [] - for i, date in enumerate(reb): - end = reb[i + 1] if i + 1 < len(reb) else cal[-1] - # A rebalance on the final trading day has no forward holding window - # (end <= start) — settling it would emit a spurious zero-length - # period (close[d]/close[d]-1 == 0). Skip it (BT-003 honesty). - if end <= date: - continue - turnover, cost, gross, net = self._step(date, end) - nav = nav * (1.0 + net) - rows.append( - { - "date": date, - "nav": nav, - "gross_return": gross, - "cost": cost, - "turnover": turnover, - "net_return": net, - } - ) - out = pd.DataFrame(rows, columns=["date", *_NAV_COLUMNS]) - return out.set_index("date") - - def _step( - self, date: pd.Timestamp, end: pd.Timestamp - ) -> tuple[float, float, float, float]: - """One rebalance: universe -> scores -> build -> feasible fill -> settle. - - Selection (``universe.tradable``) decides what the strategy WANTS; the - execution-feasibility flags (limits / suspension / missing, read from the - date's cross-section) decide what it can actually trade. Blocked trades - carry forward and turnover/cost count only executed trades (no forced - impossible trades). An empty tradable universe still routes through the - fill so held names that turned untradeable are carried, not force-sold. - - Returns ``(turnover, cost, gross_return, net_return)`` for the period - held from ``date`` (exclusive) to ``end`` (inclusive). - """ - current = self._execution.positions() - tradable = list(self._universe.tradable(date, self._prices)) - if tradable: - scores = self._scores.get(date, tradable) - target = self._constructor.build(scores, current) - else: - target = pd.Series(dtype=float) # nothing to hold -> exit what we can - - symbols = sorted(set(current.index) | set(target.index)) - can_buy, can_sell = self._feasibility(date, symbols) - self._execution.rebalance_to(target, date, can_buy=can_buy, can_sell=can_sell) - - achieved = self._execution.positions() - holding = self._holding_returns(date, end, list(achieved.index)) - invested_net = self._execution.settle(holding) - # The driver owns cash semantics (BT-007): the uninvested fraction — - # nonzero when blocked/cash-starved buys left cash idle — earns the - # driver's cash_return, even if the execution adapter's differs. - idle = 1.0 - float(achieved.sum()) - net = invested_net + idle * self._cash_return - turnover = self._execution.last_turnover - cost = self._execution.last_cost - gross = net + cost - self._record_feasibility(date, achieved) - self._record_holdings(date, achieved) - return (turnover, cost, gross, net) - - # -- execution feasibility ------------------------------------------- # - def _feasibility( - self, date: pd.Timestamp, symbols: list[str] - ) -> tuple[dict, dict]: - """Per-symbol (can_buy, can_sell) from the date's cross-section.""" - if not symbols: - return {}, {} - try: - cross = self._prices.xs(pd.Timestamp(date).normalize(), level="date") - except KeyError: - cross = self._prices.iloc[0:0] - return feasibility_from_cross(cross, symbols) - - def _record_feasibility(self, date: pd.Timestamp, achieved: pd.Series) -> None: - """Append the most recent fill's feasibility diagnostics to the log.""" - fill = getattr(self._execution, "last_fill", None) - if fill is None: - return - self._feasibility_log.append( - { - "date": date, - "blocked_buys": len(fill.blocked_buys), - "blocked_sells": len(fill.blocked_sells), - "cash_constrained_buys": len(fill.cash_constrained_buys), - "carried": len(fill.carried), - "executed_turnover": float(fill.executed_turnover), - "invested": float(achieved.sum()), - } - ) + return self._engine.run() def feasibility_log(self) -> pd.DataFrame: - """Per-settled-rebalance execution-feasibility diagnostics (date-indexed). - - Columns: ``blocked_buys``, ``blocked_sells``, ``cash_constrained_buys``, - ``carried`` (counts), ``executed_turnover``, ``invested`` (sum of achieved - weights; ``1 - invested`` is idle cash). Aligns 1:1 with the NAV table's - settled rebalance dates. - """ - cols = [ - "blocked_buys", "blocked_sells", "cash_constrained_buys", - "carried", "executed_turnover", "invested", - ] - if not self._feasibility_log: - return pd.DataFrame(columns=cols, index=pd.Index([], name="date")) - return pd.DataFrame(self._feasibility_log).set_index("date") - - def _record_holdings(self, date: pd.Timestamp, achieved: pd.Series) -> None: - """Record the ACHIEVED book (post-feasibility) held for this period.""" - for sym, weight in achieved.items(): - self._holdings_log.append( - {"date": date, "symbol": str(sym), "weight": float(weight)} - ) + """Per-settled-rebalance execution-feasibility diagnostics (date-indexed).""" + return self._engine.feasibility_log() def holdings_log(self) -> pd.DataFrame: - """Per-settled-rebalance ACHIEVED holdings (long-form date,symbol,weight,rank). - - These are the ACTUAL positions held after execution feasibility — a name - whose sell was blocked appears here carried at its old weight, and a name - whose buy was blocked is absent. This is the auditable book, NOT the - constructor's desired target (which can differ once a trade is blocked). - Ranked by weight desc then symbol for determinism; aligns with the NAV / - feasibility log's settled dates. - """ - cols = ["date", "symbol", "weight", "rank"] - if not self._holdings_log: - return pd.DataFrame(columns=cols) - df = pd.DataFrame(self._holdings_log).sort_values( - ["date", "weight", "symbol"], ascending=[True, False, True] - ) - df["rank"] = df.groupby("date").cumcount() + 1 - return df.reset_index(drop=True)[cols] + """Per-settled-rebalance ACHIEVED holdings (long-form date,symbol,weight,rank).""" + return self._engine.holdings_log() diff --git a/runtime/backtest/engine.py b/runtime/backtest/engine.py new file mode 100644 index 0000000..8a95334 --- /dev/null +++ b/runtime/backtest/engine.py @@ -0,0 +1,243 @@ +"""BacktestEngine: the shared event-driven backtest loop (I5a). + +One loop, two event models. The engine owns the *ledger* — universe selection, +feasible fills, settlement, cash, NAV compounding, and the feasibility / holdings +/ event logs — and is agnostic to whether a period is priced daily close-to-close +or intraday execution-to-execution. The time-basis differences live entirely in +the injected :class:`EventModel`: + + model.holding_periods() -> the rebalance schedule (anchors); + model.holding_returns(period, symbols) -> per-symbol gross entry->exit return; + model.feasibility(period, symbols) -> per-symbol (can_buy, can_sell). + +This is a behaviour-preserving extraction of the legacy ``BacktestDriver`` loop: +with a :class:`~runtime.backtest.event_models.DailyCloseEventModel` the engine +reproduces the accepted monthly close-to-close NAV / cost / turnover / feasibility +/ holdings byte-for-byte (locked by tests). The same loop, given an +``IntradayTailEventModel``, prices fills at 1min execution bars and measures +exec-to-exec returns — without duplicating any fill/cash/settlement logic. + +Selection vs execution are deliberately separated (the I5a contract): +``universe.tradable(date, panel)`` decides what the strategy WANTS from the daily +selection panel; ``model.feasibility`` decides what it can actually trade at the +period's execution time. The daily panel is used ONLY for selection here; the +intraday model never prices off it. +""" + +from __future__ import annotations + +from typing import Protocol + +import pandas as pd + +from runtime.backtest.events import HoldingPeriod +from runtime.execution import BacktestExecution + +_NAV_COLUMNS = ["nav", "gross_return", "cost", "turnover", "net_return"] +_EVENT_COLUMNS = [ + "date", "decision_ts", "execution_ts", "exit_date", "exit_execution_ts", + "next_decision_ts", "next_execution_ts", +] + + +class ScoresSource(Protocol): + """Minimal scores port the engine depends on.""" + + def get(self, date: pd.Timestamp, symbols: list[str]) -> pd.Series: ... + + +class UniversePort(Protocol): + """Minimal universe port the engine depends on.""" + + def tradable(self, date: pd.Timestamp, panel: pd.DataFrame) -> list[str]: ... + + +class ConstructorPort(Protocol): + """Minimal portfolio-constructor port the engine depends on.""" + + def build( + self, scores: pd.Series, current_weights: pd.Series | None = ... + ) -> pd.Series: ... + + +class EventModel(Protocol): + """Time-basis strategy: schedule + pricing + execution feasibility.""" + + def holding_periods(self) -> list[HoldingPeriod]: ... + + def holding_returns( + self, period: HoldingPeriod, symbols: list[str] + ) -> pd.Series: ... + + def feasibility( + self, period: HoldingPeriod, symbols: list[str] + ) -> tuple[dict, dict]: ... + + +class BacktestEngine: + """Single-period-compounding backtest over a sequence of holding periods.""" + + def __init__( + self, + *, + model: EventModel, + universe: UniversePort, + scores: ScoresSource, + constructor: ConstructorPort, + execution: BacktestExecution, + selection_panel: pd.DataFrame, + initial_nav: float = 1.0, + cash_return: float = 0.0, + ) -> None: + self._model = model + self._universe = universe + self._scores = scores + self._constructor = constructor + self._execution = execution + self._panel = selection_panel + self._initial_nav = float(initial_nav) + self._cash_return = float(cash_return) + self._feasibility_log: list[dict] = [] + self._holdings_log: list[dict] = [] + self._event_log: list[dict] = [] + + # -- run -------------------------------------------------------------- # + def run(self) -> pd.DataFrame: + """Run the backtest and return the date-indexed NAV table. + + Columns ``[nav, gross_return, cost, turnover, net_return]`` — identical + to the legacy driver. Re-entrant: each run rebuilds the logs. + """ + periods = self._model.holding_periods() + rows: list[dict] = [] + nav = self._initial_nav + self._feasibility_log = [] + self._holdings_log = [] + self._event_log = [] + for i, period in enumerate(periods): + turnover, cost, gross, net = self._step(period) + nav = nav * (1.0 + net) + rows.append( + { + "date": period.date, + "nav": nav, + "gross_return": gross, + "cost": cost, + "turnover": turnover, + "net_return": net, + } + ) + nxt = periods[i + 1] if i + 1 < len(periods) else None + self._record_event( + period, + next_decision_ts=nxt.decision_ts if nxt else pd.NaT, + next_execution_ts=nxt.execution_ts if nxt else pd.NaT, + ) + out = pd.DataFrame(rows, columns=["date", *_NAV_COLUMNS]) + return out.set_index("date") + + def _step( + self, period: HoldingPeriod + ) -> tuple[float, float, float, float]: + """One rebalance: universe -> scores -> build -> feasible fill -> settle. + + Returns ``(turnover, cost, gross_return, net_return)`` for ``period``. + """ + current = self._execution.positions() + tradable = list(self._universe.tradable(period.date, self._panel)) + if tradable: + scores = self._scores.get(period.date, tradable) + target = self._constructor.build(scores, current) + else: + target = pd.Series(dtype=float) # nothing to hold -> exit what we can + + symbols = sorted(set(current.index) | set(target.index)) + can_buy, can_sell = self._model.feasibility(period, symbols) + self._execution.rebalance_to( + target, period.date, can_buy=can_buy, can_sell=can_sell + ) + + achieved = self._execution.positions() + holding = self._model.holding_returns(period, list(achieved.index)) + invested_net = self._execution.settle(holding) + # The engine owns cash semantics (BT-007): the uninvested fraction earns + # the configured cash_return, even if the execution adapter's differs. + idle = 1.0 - float(achieved.sum()) + net = invested_net + idle * self._cash_return + turnover = self._execution.last_turnover + cost = self._execution.last_cost + gross = net + cost + self._record_feasibility(period.date, achieved) + self._record_holdings(period.date, achieved) + return (turnover, cost, gross, net) + + # -- feasibility log -------------------------------------------------- # + def _record_feasibility(self, date: pd.Timestamp, achieved: pd.Series) -> None: + fill = getattr(self._execution, "last_fill", None) + if fill is None: + return + self._feasibility_log.append( + { + "date": date, + "blocked_buys": len(fill.blocked_buys), + "blocked_sells": len(fill.blocked_sells), + "cash_constrained_buys": len(fill.cash_constrained_buys), + "carried": len(fill.carried), + "executed_turnover": float(fill.executed_turnover), + "invested": float(achieved.sum()), + } + ) + + def feasibility_log(self) -> pd.DataFrame: + """Per-settled-rebalance execution-feasibility diagnostics (date-indexed).""" + cols = [ + "blocked_buys", "blocked_sells", "cash_constrained_buys", + "carried", "executed_turnover", "invested", + ] + if not self._feasibility_log: + return pd.DataFrame(columns=cols, index=pd.Index([], name="date")) + return pd.DataFrame(self._feasibility_log).set_index("date") + + # -- holdings log ----------------------------------------------------- # + def _record_holdings(self, date: pd.Timestamp, achieved: pd.Series) -> None: + for sym, weight in achieved.items(): + self._holdings_log.append( + {"date": date, "symbol": str(sym), "weight": float(weight)} + ) + + def holdings_log(self) -> pd.DataFrame: + """Per-settled-rebalance ACHIEVED holdings (long-form date,symbol,weight,rank).""" + cols = ["date", "symbol", "weight", "rank"] + if not self._holdings_log: + return pd.DataFrame(columns=cols) + df = pd.DataFrame(self._holdings_log).sort_values( + ["date", "weight", "symbol"], ascending=[True, False, True] + ) + df["rank"] = df.groupby("date").cumcount() + 1 + return df.reset_index(drop=True)[cols] + + # -- event log (I5a auditability) ------------------------------------- # + def _record_event( + self, + period: HoldingPeriod, + *, + next_decision_ts: pd.Timestamp, + next_execution_ts: pd.Timestamp, + ) -> None: + self._event_log.append( + { + "date": period.date, + "decision_ts": period.decision_ts, + "execution_ts": period.execution_ts, + "exit_date": period.exit_date, + "exit_execution_ts": period.exit_execution_ts, + "next_decision_ts": next_decision_ts, + "next_execution_ts": next_execution_ts, + } + ) + + def event_log(self) -> pd.DataFrame: + """Per-settled-period event timing (decision/execution/exit anchors).""" + if not self._event_log: + return pd.DataFrame(columns=_EVENT_COLUMNS).set_index("date") + return pd.DataFrame(self._event_log).set_index("date") diff --git a/runtime/backtest/event_models.py b/runtime/backtest/event_models.py new file mode 100644 index 0000000..aca1ba5 --- /dev/null +++ b/runtime/backtest/event_models.py @@ -0,0 +1,222 @@ +"""Event models for the shared backtest engine (I5a). + +Each model bundles a *schedule* with a *pricing* and *feasibility* time-basis: + + * :class:`DailyCloseEventModel` — the accepted monthly close-to-close model. + Pricing and feasibility both read the daily panel; reproduces the legacy + ``BacktestDriver`` ledger byte-for-byte. + * :class:`IntradayTailEventModel` — a 14:50-decision / 14:51-execution tail + rebalance. Pricing is the 1min execution-bar price (reusing + :func:`runtime.intraday_execution.build_execution_prices`); returns are + exec-to-exec; a missing/NaN execution bar is an explicit block, NEVER a + daily-close fallback. + +Both schedules come from :func:`runtime.backtest.events.monthly_anchor_pairs`, so +daily and intraday rebalance on the same calendar; only the time basis differs. +The intraday model never prices off the daily panel. +""" + +from __future__ import annotations + +import pandas as pd + +from runtime.backtest.events import ( + HoldingPeriod, + monthly_anchor_pairs, + trading_calendar, +) +from runtime.fills import feasibility_from_cross +from runtime.intraday_execution import ( + ExecutionFill, + IntradayExecutionConfig, + build_execution_prices, +) + + +class DailyCloseEventModel: + """Monthly close-to-close model — the accepted daily behaviour. + + decision = execution = entry = the rebalance date (close of T); exit = the + next rebalance date (close of T_next). Holding returns and execution + feasibility both read the daily ``prices`` panel, exactly as the legacy + driver did. + """ + + def __init__(self, prices: pd.DataFrame) -> None: + if "close" not in prices.columns: + raise ValueError("price panel must have a 'close' column") + self._prices = prices + + def holding_periods(self) -> list[HoldingPeriod]: + pairs = monthly_anchor_pairs(trading_calendar(self._prices)) + return [ + HoldingPeriod( + date=date, + entry_date=date, + exit_date=exit_date, + decision_ts=date, + execution_ts=date, + exit_execution_ts=exit_date, # priced at the exit date's close + ) + for date, exit_date in pairs + ] + + def _close_at(self, date: pd.Timestamp, symbol: str) -> float: + try: + return float(self._prices.loc[(date, symbol), "close"]) + except KeyError: + return float("nan") + + def holding_returns( + self, period: HoldingPeriod, symbols: list[str] + ) -> pd.Series: + """Close-to-close gross return per symbol over (entry, exit]. + + Symbols with a missing/zero start price get a flat (0.0) return rather + than NaN, so the book stays well-defined (matches the legacy driver). + """ + start, end = period.entry_date, period.exit_date + out: dict[str, float] = {} + for sym in symbols: + start_px = self._close_at(start, sym) + end_px = self._close_at(end, sym) + if ( + pd.isna(start_px) + or pd.isna(end_px) + or start_px == 0.0 + ): + out[sym] = 0.0 + else: + out[sym] = end_px / start_px - 1.0 + return pd.Series(out, dtype=float) + + def feasibility( + self, period: HoldingPeriod, symbols: list[str] + ) -> tuple[dict, dict]: + """Per-symbol (can_buy, can_sell) from the rebalance date's cross-section.""" + if not symbols: + return {}, {} + try: + cross = self._prices.xs( + pd.Timestamp(period.date).normalize(), level="date" + ) + except KeyError: + cross = self._prices.iloc[0:0] + return feasibility_from_cross(cross, symbols) + + +class IntradayTailEventModel: + """14:50-decision / 14:51-execution tail rebalance over minute bars (I5a). + + The rebalance schedule is the same monthly calendar as the daily model, but + each period's decision is timestamped at ``decision_time`` and its execution + at the start of the execution window. Per-symbol entry/exit prices are the + earliest valid 1min close in the execution window (``next_minute_close``); + holding returns are ``exec_price(exit) / exec_price(entry) - 1``. + + Minimum I5a feasibility rule (goal §5): a symbol can be traded at the + rebalance ONLY if it has a valid (non-NaN) execution bar at the entry anchor; + a missing/NaN execution bar blocks BOTH directions. A suspended stock has no + minute bars and is blocked by this rule. Price-limit feasibility at execution + time is deferred (the daily panel carries only daily-close-derived limit + flags, which §5 forbids using for execution) and disclosed in the report. + """ + + def __init__( + self, + *, + calendar_panel: pd.DataFrame, + bars: pd.DataFrame, + cfg: IntradayExecutionConfig | None = None, + ) -> None: + self._calendar_panel = calendar_panel + self._cfg = cfg or IntradayExecutionConfig() + pairs = monthly_anchor_pairs(trading_calendar(calendar_panel)) + self._pairs = pairs + # Anchor dates we must price: every rebalance date AND every exit date. + anchor_dates = sorted( + {d for pair in pairs for d in pair}, + key=lambda t: pd.Timestamp(t), + ) + symbols = sorted({str(s) for s in bars.index.get_level_values("symbol")}) + self._symbols = symbols + if anchor_dates and symbols: + prices, fills = build_execution_prices( + bars, anchor_dates, symbols, self._cfg + ) + else: + prices = pd.DataFrame() + fills = [] + self._exec_prices = prices + self._fills = fills + + def holding_periods(self) -> list[HoldingPeriod]: + decision = pd.Timedelta(self._cfg.decision_time) + execution = pd.Timedelta(self._cfg.execution_window[0]) + out: list[HoldingPeriod] = [] + for date, exit_date in self._pairs: + day = pd.Timestamp(date).normalize() + exit_day = pd.Timestamp(exit_date).normalize() + out.append( + HoldingPeriod( + date=date, + entry_date=date, + exit_date=exit_date, + decision_ts=day + decision, + execution_ts=day + execution, + # the book is priced out at the exit date's execution-window + # start (the planned exit fill time), NOT close-to-close. + exit_execution_ts=exit_day + execution, + ) + ) + return out + + def _exec_price(self, date: pd.Timestamp, symbol: str) -> float: + try: + return float(self._exec_prices.loc[pd.Timestamp(date).normalize(), str(symbol)]) + except (KeyError, TypeError): + return float("nan") + + def holding_returns( + self, period: HoldingPeriod, symbols: list[str] + ) -> pd.Series: + """exec-to-exec gross return per symbol over (entry, exit]. + + A symbol missing an execution price at EITHER anchor is omitted (so the + engine's ``settle`` treats it as flat — it earns nothing for the period). + Never substitutes a daily close. + """ + entry, exit_ = period.entry_date, period.exit_date + out: dict[str, float] = {} + for sym in symbols: + a = self._exec_price(entry, sym) + b = self._exec_price(exit_, sym) + if pd.notna(a) and pd.notna(b) and a != 0.0: + out[str(sym)] = b / a - 1.0 + # else: blocked at an anchor -> omitted -> flat via settle. + return pd.Series(out, dtype=float) + + def feasibility( + self, period: HoldingPeriod, symbols: list[str] + ) -> tuple[dict, dict]: + """Per-symbol (can_buy, can_sell): tradable iff a valid entry exec bar.""" + can_buy: dict[str, bool] = {} + can_sell: dict[str, bool] = {} + for sym in symbols: + ok = pd.notna(self._exec_price(period.entry_date, sym)) + can_buy[str(sym)] = bool(ok) + can_sell[str(sym)] = bool(ok) + return can_buy, can_sell + + # -- diagnostics ------------------------------------------------------ # + def execution_prices(self) -> pd.DataFrame: + """The (date x symbol) execution-price matrix (NaN = blocked).""" + return self._exec_prices + + def fills(self) -> list[ExecutionFill]: + """Per-(date, symbol) execution fill log (incl. blocked reasons).""" + return list(self._fills) + + def blocked_fills(self) -> list[ExecutionFill]: + """Only the blocked fills (no execution bar / NaN price).""" + return [f for f in self._fills if f.blocked] diff --git a/runtime/backtest/events.py b/runtime/backtest/events.py new file mode 100644 index 0000000..053b330 --- /dev/null +++ b/runtime/backtest/events.py @@ -0,0 +1,98 @@ +"""Event timing primitives for the shared backtest engine (I5a). + +A backtest is a sequence of *holding periods*. Each period makes the time basis +EXPLICIT and auditable so the engine never has to assume "daily close-to-close": + + decision_ts when the strategy decides the target weights; + execution_ts when the (planned) fill happens; + entry_date the anchor date whose price enters the book; + exit_date the anchor date whose price closes the book; + date the label/rebalance date (scores, universe, NAV index). + +Two event models populate these fields differently: + + * DailyCloseEventModel: decision = execution = entry = ``date`` (close of T), + exit = the next rebalance date (close of T_next). Reproduces the accepted + monthly close-to-close behaviour exactly. + * IntradayTailEventModel: decision = ``date`` 14:50, execution = ``date`` 14:51, + entry/exit anchors are still keyed by date but priced from 1min execution + bars (exec-to-exec), NEVER daily close. + +The monthly schedule helper here is the single source of truth for "which dates +are rebalance dates", shared by both models so the schedule can never silently +diverge between daily and intraday. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +import pandas as pd + + +@dataclass(frozen=True) +class HoldingPeriod: + """One settled holding period with an explicit, auditable time basis. + + ``entry_date`` / ``exit_date`` are the price anchors (a pricing provider maps + ``(anchor_date, symbol) -> price``); ``decision_ts`` / ``execution_ts`` are the + entry timestamps recorded for auditability, and ``exit_execution_ts`` is the + (planned) timestamp at which the book is priced out at ``exit_date`` — so the + holding period's time basis is fully explicit (``execution_ts`` -> + ``exit_execution_ts``), with no need to infer it from the next period (which + does not exist for the final period). They may carry intra-day precision. + ``date`` is the rebalance label used for scores, universe selection, and the + NAV index — for the daily model it equals ``entry_date``. + """ + + date: pd.Timestamp + entry_date: pd.Timestamp + exit_date: pd.Timestamp + decision_ts: pd.Timestamp + execution_ts: pd.Timestamp + exit_execution_ts: pd.Timestamp | None = None + + +def trading_calendar(prices: pd.DataFrame) -> pd.DatetimeIndex: + """Sorted unique trading dates present in a canonical (date, symbol) panel.""" + dates = prices.index.get_level_values("date").unique() + return pd.DatetimeIndex(sorted(dates)) + + +def monthly_rebalance_dates(calendar: pd.DatetimeIndex) -> list[pd.Timestamp]: + """Last trading day of each month in ``calendar`` (BT-001), sorted. + + This is exactly the schedule the legacy ``BacktestDriver.rebalance_dates`` + produced; both event models build their holding periods from it so daily and + intraday share one rebalance calendar. + """ + if len(calendar) == 0: + return [] + frame = pd.DataFrame({"date": calendar}) + frame["ym"] = frame["date"].dt.to_period("M") + last = frame.groupby("ym")["date"].max().sort_values() + return list(last) + + +def monthly_anchor_pairs( + calendar: pd.DatetimeIndex, +) -> list[tuple[pd.Timestamp, pd.Timestamp]]: + """``(rebalance_date, exit_date)`` pairs for the monthly schedule. + + For each monthly rebalance date the exit anchor is the NEXT rebalance date, + except the final rebalance whose exit is the last trading day in the + calendar. A pair whose exit ``<=`` its rebalance date (a rebalance on the + final trading day → a zero-length forward window) is dropped, matching the + legacy driver's ``if end <= date: continue`` (BT-003 honesty). + """ + reb = monthly_rebalance_dates(calendar) + if not reb: + return [] + last_day = calendar[-1] + pairs: list[tuple[pd.Timestamp, pd.Timestamp]] = [] + for i, date in enumerate(reb): + exit_date = reb[i + 1] if i + 1 < len(reb) else last_day + if exit_date <= date: + continue + pairs.append((date, exit_date)) + return pairs diff --git a/tests/test_i5a_event_backtest.py b/tests/test_i5a_event_backtest.py new file mode 100644 index 0000000..d0cd0b2 --- /dev/null +++ b/tests/test_i5a_event_backtest.py @@ -0,0 +1,442 @@ +"""I5a: shared event-driven backtest engine + intraday tail event model. + +Covers the goal's four test groups: + +1. Daily golden — the engine with ``DailyCloseEventModel`` reproduces the legacy + ``BacktestDriver`` ledger (NAV / cost / turnover / feasibility / holdings) and + the wrapper stays import-compatible. +2. Event-model unit — daily schedule == the monthly rebalance schedule; intraday + anchors carry the configured decision/execution times and block missing bars. +3. Intraday PIT — exec-to-exec differs from close-to-close; perturbing post-cutoff + bars leaves the decision feature unchanged; perturbing the execution bar moves + returns; perturbing daily close does NOT move intraday returns; a missing + execution bar blocks (never daily-fallbacks); holdings/turnover are achieved. +4. Config — existing configs validate; ``intraday_tail_rebalance`` needs + ``intraday.enabled``; invalid window/model fail readably. +""" + +from __future__ import annotations + +import logging +from pathlib import Path + +import pandas as pd +import pytest +import yaml +from pandas.testing import assert_frame_equal +from pydantic import ValidationError + +from data.cache.intraday_cache import TushareIntradayCache +from data.cache.intraday_coverage import IntradayCoverageLedger +from data.cache.intraday_parquet_store import IntradayParquetStore +from data.clean.intraday_schema import normalize_intraday_bars +from data.clean.schema import normalize_panel +from portfolio.construct import TopNEqualWeight +from qt.config import RootConfig, load_config +from qt.pipeline import _FrameScores +from runtime.backtest.driver import BacktestDriver +from runtime.backtest.engine import BacktestEngine +from runtime.backtest.event_models import DailyCloseEventModel, IntradayTailEventModel +from runtime.backtest.events import ( + HoldingPeriod, + monthly_anchor_pairs, + monthly_rebalance_dates, + trading_calendar, +) +from runtime.backtest.sim_execution import SimExecution +from runtime.intraday_execution import ( + REASON_NO_BAR, + IntradayExecutionConfig, +) +from universe.static import StaticUniverse + +_CONFIG_DIR = Path(__file__).resolve().parent.parent / "config" +_I5A_CONFIG = _CONFIG_DIR / "phase_i5a_intraday_tail_framework.yaml" + +# Month-end calendar with consecutive month-ends so the monthly schedule yields +# >=2 settled holding periods. +_DATES = [ + "2024-01-30", "2024-01-31", + "2024-02-28", "2024-02-29", + "2024-03-28", "2024-03-29", +] +_JAN, _FEB, _MAR = pd.Timestamp("2024-01-31"), pd.Timestamp("2024-02-29"), pd.Timestamp("2024-03-29") + + +# --------------------------------------------------------------------------- # +# fixtures / builders +# --------------------------------------------------------------------------- # +def _daily_panel(closes: dict) -> pd.DataFrame: + """Canonical (date, symbol) panel from ``{(date_str, symbol): close}``.""" + rows = [] + for (d, s), c in closes.items(): + rows.append( + { + "date": pd.Timestamp(d), "symbol": s, + "open": c, "high": c, "low": c, "close": c, + "volume": 1.0, "amount": 1.0, "adj_factor": 1.0, + } + ) + return normalize_panel(pd.DataFrame(rows)) + + +def _grid_panel(symbols: list[str], price_fn) -> pd.DataFrame: + closes = {} + for d in _DATES: + for s in symbols: + closes[(d, s)] = price_fn(d, s) + return _daily_panel(closes) + + +def _scores(panel_map: dict) -> _FrameScores: + """`_FrameScores` over a ``{(date, symbol): score}`` map.""" + idx = pd.MultiIndex.from_tuples( + [(pd.Timestamp(d), s) for (d, s) in panel_map], names=["date", "symbol"] + ) + return _FrameScores(pd.Series(list(panel_map.values()), index=idx, name="score")) + + +def _minute_bars(specs: list[tuple[str, str, float]]) -> pd.DataFrame: + """Canonical 1min bars from ``(symbol, 'YYYY-MM-DD HH:MM:SS', close)`` specs.""" + rows = [ + { + "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 + ] + return normalize_intraday_bars(pd.DataFrame(rows), freq="1min", data_lag="1min") + + +# --------------------------------------------------------------------------- # +# 1. Daily golden — engine(DailyCloseEventModel) == legacy BacktestDriver +# --------------------------------------------------------------------------- # +def _daily_inputs(): + panel = _grid_panel(["A", "B", "C"], lambda d, s: 100.0 + hash((d, s)) % 7) + universe = StaticUniverse(["A", "B", "C"]) + scores = _scores({(d, s): {"A": 3.0, "B": 2.0, "C": 1.0}[s] for d in _DATES for s in "ABC"}) + return panel, universe, scores + + +def test_engine_daily_equals_driver_wrapper(): + panel, universe, scores = _daily_inputs() + + driver = BacktestDriver( + universe=universe, scores=scores, + constructor=TopNEqualWeight(2, long_only=True), + execution=SimExecution(fee_rate=0.001), prices=panel, + rebalance="monthly", fee_rate=0.001, initial_nav=1.0, cash_return=0.0, + ) + driver_nav = driver.run() + + engine = BacktestEngine( + model=DailyCloseEventModel(panel), universe=universe, scores=scores, + constructor=TopNEqualWeight(2, long_only=True), + execution=SimExecution(fee_rate=0.001), selection_panel=panel, + initial_nav=1.0, cash_return=0.0, + ) + engine_nav = engine.run() + + assert_frame_equal(driver_nav, engine_nav) + assert_frame_equal(driver.feasibility_log(), engine.feasibility_log()) + assert_frame_equal(driver.holdings_log(), engine.holdings_log()) + + +def test_backtest_driver_import_compatible_surface(): + panel, universe, scores = _daily_inputs() + driver = BacktestDriver( + universe=universe, scores=scores, + constructor=TopNEqualWeight(2), execution=SimExecution(), prices=panel, + ) + # Legacy public surface still present. + assert callable(driver.run) + assert callable(driver.rebalance_dates) + assert callable(driver.feasibility_log) + assert callable(driver.holdings_log) + assert driver.rebalance_dates() == monthly_rebalance_dates(trading_calendar(panel)) + + +# --------------------------------------------------------------------------- # +# 2. Event-model unit +# --------------------------------------------------------------------------- # +def test_daily_event_model_schedule_matches_monthly_pairs(): + panel, _, _ = _daily_inputs() + model = DailyCloseEventModel(panel) + periods = model.holding_periods() + pairs = monthly_anchor_pairs(trading_calendar(panel)) + assert [(p.date, p.exit_date) for p in periods] == pairs + # daily model: decision == execution == entry == the rebalance date. + for p in periods: + assert p.decision_ts == p.date == p.entry_date == p.execution_ts + + +def test_intraday_event_model_anchors_and_blocks_missing(): + panel = _grid_panel(["A", "B"], lambda d, s: 100.0) + bars = _minute_bars( + # A has an execution bar on both month-ends; B is missing Feb's. + [("A", f"{_JAN.date()} 14:51:00", 10.0), ("A", f"{_FEB.date()} 14:51:00", 11.0), + ("A", f"{_MAR.date()} 14:51:00", 12.0), + ("B", f"{_JAN.date()} 14:51:00", 20.0), ("B", f"{_MAR.date()} 14:51:00", 22.0)] + ) + cfg = IntradayExecutionConfig() + model = IntradayTailEventModel(calendar_panel=panel, bars=bars, cfg=cfg) + + periods = model.holding_periods() + p0 = periods[0] + assert p0.decision_ts == pd.Timestamp(f"{p0.date.date()} 14:50:00") + assert p0.execution_ts == pd.Timestamp(f"{p0.date.date()} 14:51:00") + + # Feasibility at Feb (entry) — A has a bar (tradable), B does not (blocked both). + feb_period = next(p for p in periods if p.date == _FEB) + can_buy, can_sell = model.feasibility(feb_period, ["A", "B"]) + assert can_buy["A"] is True and can_sell["A"] is True + assert can_buy["B"] is False and can_sell["B"] is False + assert any(f.blocked and f.reason == REASON_NO_BAR for f in model.blocked_fills()) + + +# --------------------------------------------------------------------------- # +# 3. Intraday PIT +# --------------------------------------------------------------------------- # +def test_intraday_exec_to_exec_differs_from_close_to_close(): + # Daily close falls; minute execution price rises -> opposite-signed returns. + panel = _daily_panel( + {(str(_JAN.date()), "A"): 100.0, (str(_FEB.date()), "A"): 50.0, + (str(_JAN.date()), "B"): 100.0, (str(_FEB.date()), "B"): 50.0} + ) + bars = _minute_bars( + [("A", f"{_JAN.date()} 14:51:00", 10.0), ("A", f"{_FEB.date()} 14:51:00", 11.0)] + ) + cfg = IntradayExecutionConfig() + intra = IntradayTailEventModel(calendar_panel=panel, bars=bars, cfg=cfg) + daily = DailyCloseEventModel(panel) + period = HoldingPeriod(_JAN, _JAN, _FEB, _JAN, _JAN) + + intra_r = intra.holding_returns(period, ["A"]) + daily_r = daily.holding_returns(period, ["A"]) + assert intra_r["A"] == pytest.approx(11.0 / 10.0 - 1.0) # +0.10 exec-to-exec + assert daily_r["A"] == pytest.approx(50.0 / 100.0 - 1.0) # -0.50 close-to-close + assert intra_r["A"] != pytest.approx(daily_r["A"]) + + +def test_perturb_execution_bar_changes_intraday_returns(): + panel = _daily_panel( + {(str(_JAN.date()), "A"): 100.0, (str(_FEB.date()), "A"): 100.0} + ) + period = HoldingPeriod(_JAN, _JAN, _FEB, _JAN, _JAN) + base = _minute_bars( + [("A", f"{_JAN.date()} 14:51:00", 10.0), ("A", f"{_FEB.date()} 14:51:00", 11.0)] + ) + bumped = _minute_bars( + [("A", f"{_JAN.date()} 14:51:00", 10.0), ("A", f"{_FEB.date()} 14:51:00", 20.0)] + ) + r0 = IntradayTailEventModel(calendar_panel=panel, bars=base, cfg=IntradayExecutionConfig()) + r1 = IntradayTailEventModel(calendar_panel=panel, bars=bumped, cfg=IntradayExecutionConfig()) + assert r0.holding_returns(period, ["A"])["A"] != pytest.approx( + r1.holding_returns(period, ["A"])["A"] + ) + + +def test_perturb_daily_close_does_not_change_intraday_returns(): + bars = _minute_bars( + [("A", f"{_JAN.date()} 14:51:00", 10.0), ("A", f"{_FEB.date()} 14:51:00", 11.0)] + ) + period = HoldingPeriod(_JAN, _JAN, _FEB, _JAN, _JAN) + panel_a = _daily_panel( + {(str(_JAN.date()), "A"): 100.0, (str(_FEB.date()), "A"): 50.0} + ) + panel_b = _daily_panel( + {(str(_JAN.date()), "A"): 100.0, (str(_FEB.date()), "A"): 999.0} + ) + m_a = IntradayTailEventModel(calendar_panel=panel_a, bars=bars, cfg=IntradayExecutionConfig()) + m_b = IntradayTailEventModel(calendar_panel=panel_b, bars=bars, cfg=IntradayExecutionConfig()) + assert m_a.holding_returns(period, ["A"])["A"] == pytest.approx( + m_b.holding_returns(period, ["A"])["A"] + ) + + +def test_missing_execution_bar_blocks_never_daily_fallback(): + # A held into Feb but Feb has NO execution bar -> omitted from returns (flat), + # never the daily close (which would be a -0.5 fallback). + panel = _daily_panel( + {(str(_JAN.date()), "A"): 100.0, (str(_FEB.date()), "A"): 50.0} + ) + bars = _minute_bars([("A", f"{_JAN.date()} 14:51:00", 10.0)]) # no Feb bar + period = HoldingPeriod(_JAN, _JAN, _FEB, _JAN, _JAN) + model = IntradayTailEventModel(calendar_panel=panel, bars=bars, cfg=IntradayExecutionConfig()) + r = model.holding_returns(period, ["A"]) + assert "A" not in r.index # omitted (flat), NOT -0.5 + assert any(f.blocked for f in model.blocked_fills()) + + +def test_perturb_post_decision_bars_leave_decision_feature_unchanged(): + # The score = the I3 intraday_ret feature; perturbing a post-14:50 bar must not + # change it (only available_time <= 14:50 bars enter). + from data.clean.intraday_aggregate import asof_daily_features + + base = _minute_bars( + [("A", f"{_JAN.date()} 09:30:00", 10.0), + ("A", f"{_JAN.date()} 14:49:00", 12.0), # available 14:50 -> included + ("A", f"{_JAN.date()} 14:55:00", 99.0)] # available 14:56 -> excluded + ) + perturbed = _minute_bars( + [("A", f"{_JAN.date()} 09:30:00", 10.0), + ("A", f"{_JAN.date()} 14:49:00", 12.0), + ("A", f"{_JAN.date()} 14:55:00", 777.0)] # perturb the excluded bar + ) + f0 = asof_daily_features(base, decision_time="14:50:00", session_open="09:30:00") + f1 = asof_daily_features(perturbed, decision_time="14:50:00", session_open="09:30:00") + assert_frame_equal(f0, f1) + + +def test_intraday_engine_turnover_and_holdings_are_achieved(): + # Full engine run with the intraday model: turnover/holdings reflect the + # ACHIEVED book after a blocked entry, not the desired target. + panel = _grid_panel(["A", "B"], lambda d, s: 100.0) + bars = _minute_bars( + # A executes on every month-end; B can never be entered (no exec bars). + [("A", f"{_JAN.date()} 14:51:00", 10.0), ("A", f"{_FEB.date()} 14:51:00", 11.0), + ("A", f"{_MAR.date()} 14:51:00", 12.0)] + ) + universe = StaticUniverse(["A", "B"]) + # B scores highest, but it has no execution bar -> blocked -> A is held. + scores = _scores({(d, s): {"A": 1.0, "B": 9.0}[s] for d in _DATES for s in "AB"}) + engine = BacktestEngine( + model=IntradayTailEventModel(calendar_panel=panel, bars=bars, cfg=IntradayExecutionConfig()), + universe=universe, scores=scores, constructor=TopNEqualWeight(2), + execution=SimExecution(fee_rate=0.0), selection_panel=panel, + ) + nav = engine.run() + holdings = engine.holdings_log() + assert not nav.empty + # B never appears in the achieved book (its buy was always blocked). + assert "B" not in set(holdings["symbol"]) + assert set(holdings["symbol"]) <= {"A"} + + +# --------------------------------------------------------------------------- # +# 4. Config +# --------------------------------------------------------------------------- # +def test_i5a_config_validates(): + cfg = load_config(str(_I5A_CONFIG)) + assert cfg.backtest.event_order == "intraday_tail_rebalance" + assert cfg.intraday is not None and cfg.intraday.enabled is True + + +def _i5a_dict() -> dict: + return yaml.safe_load(_I5A_CONFIG.read_text()) + + +def test_intraday_event_order_requires_enabled(): + d = _i5a_dict() + d["intraday"]["enabled"] = False + with pytest.raises(ValidationError, match="requires an 'intraday' section"): + RootConfig(**d) + + d2 = _i5a_dict() + d2.pop("intraday") + with pytest.raises(ValidationError, match="requires an 'intraday' section"): + RootConfig(**d2) + + +def test_default_daily_config_needs_no_intraday_section(): + d = _i5a_dict() + d["backtest"]["event_order"] = "close_to_next_period" + d.pop("intraday") + cfg = RootConfig(**d) # validates fine without an intraday section + assert cfg.intraday is None + + +def test_invalid_execution_model_fails_readably(): + d = _i5a_dict() + d["intraday"]["execution_model"] = "tail_vwap" + with pytest.raises(ValidationError, match="execution_model"): + RootConfig(**d) + + +def test_invalid_execution_window_fails_readably(): + d = _i5a_dict() + d["intraday"]["execution_window"] = ["14:49:00", "14:56:59"] # before decision + with pytest.raises(ValidationError, match="execution_window"): + RootConfig(**d) + + +# --------------------------------------------------------------------------- # +# 5. Partial cache coverage (require_cache_coverage semantics) + event audit +# --------------------------------------------------------------------------- # +def _synthetic_stk_mins(symbol, start_dt, end_dt): + """Raw stk_mins-shaped 1min bars for [start_dt, end_dt] (one bar at 14:51/day).""" + rows = [] + day = pd.Timestamp(start_dt).normalize() + end = pd.Timestamp(end_dt).normalize() + while day <= end: + for clock in ("09:31:00", "14:51:00"): + ts = day + pd.Timedelta(clock) + rows.append( + {"ts_code": symbol, "trade_time": str(ts), "open": 10.0, + "high": 10.0, "low": 10.0, "close": 10.0, "vol": 1.0, "amount": 10.0} + ) + day += pd.Timedelta(days=1) + return pd.DataFrame(rows) + + +def _cfg_with_cache(tmp_path, start, end, *, require: bool) -> RootConfig: + d = _i5a_dict() + d["data"]["cache"]["root_dir"] = str(tmp_path) + d["data"]["start"] = start + d["data"]["end"] = end + d["intraday"]["require_cache_coverage"] = require + return RootConfig(**d) + + +def _warm_one_symbol(tmp_path, symbol, start, end): + cache = TushareIntradayCache(IntradayParquetStore(str(tmp_path)), IntradayCoverageLedger(str(tmp_path))) + cache.stk_mins_1min( + [symbol], f"{start} 00:00:00", f"{end} 23:59:59", _synthetic_stk_mins, freq="1min" + ) + + +def test_partial_coverage_required_fails(tmp_path): + from qt.intraday_tail_framework import _load_minute_bars_cache_only + + start, end = "2024-01-02", "2024-01-03" + _warm_one_symbol(tmp_path, "AAA.SH", start, end) # only AAA is covered + cfg = _cfg_with_cache(tmp_path, start, end, require=True) + log = logging.getLogger("test.i5a") + # require_cache_coverage=true + one uncovered symbol -> hard blocker (no silent drop). + with pytest.raises(ValueError, match="require_cache_coverage=true"): + _load_minute_bars_cache_only(cfg, ["AAA.SH", "BBB.SH"], log) + + +def test_partial_coverage_lenient_drops_uncovered(tmp_path): + from qt.intraday_tail_framework import _load_minute_bars_cache_only + + start, end = "2024-01-02", "2024-01-03" + _warm_one_symbol(tmp_path, "AAA.SH", start, end) + cfg = _cfg_with_cache(tmp_path, start, end, require=False) + log = logging.getLogger("test.i5a") + bars, covered, uncovered, live = _load_minute_bars_cache_only(cfg, ["AAA.SH", "BBB.SH"], log) + assert covered == ["AAA.SH"] + assert uncovered == ["BBB.SH"] + assert live == 0 # read-only: zero live stk_mins calls + assert not bars.empty + + +def test_event_log_exposes_exit_and_next_execution_anchors(): + panel, universe, scores = _daily_inputs() + engine = BacktestEngine( + model=DailyCloseEventModel(panel), universe=universe, scores=scores, + constructor=TopNEqualWeight(2), execution=SimExecution(), selection_panel=panel, + ) + engine.run() + ev = engine.event_log() + for col in ("exit_execution_ts", "next_decision_ts", "next_execution_ts"): + assert col in ev.columns + # daily: the exit execution anchor is the exit date's close. + assert (ev["exit_execution_ts"] == ev["exit_date"]).all() + # consecutive periods chain: next_execution_ts == the following row's execution_ts. + execs = list(ev["execution_ts"]) + nexts = list(ev["next_execution_ts"]) + for i in range(len(execs) - 1): + assert nexts[i] == execs[i + 1] + assert pd.isna(nexts[-1]) # last settled period has no successor