From cfa56de3d6b1effa2406ebdd11592e8bd3dd52ce Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Mon, 15 Jun 2026 11:31:36 +0800 Subject: [PATCH 1/4] refactor(runtime): shared event-driven backtest engine + event models Extract the backtest ledger from BacktestDriver into a reusable BacktestEngine driven by an EventModel (schedule + pricing + feasibility). DailyCloseEventModel reproduces the accepted monthly close-to-close behaviour byte-for-byte; IntradayTailEventModel prices fills at 1min execution bars (next_minute_close) with exec-to-exec holding returns and explicit blocks (never a daily-close fallback). BacktestDriver is now a thin import-compatible wrapper around the engine + DailyCloseEventModel, so qt.pipeline / qt.oos_stability / qt.phase2_baseline are untouched. events.py: HoldingPeriod + the shared monthly rebalance schedule. engine.py: the single loop (universe -> scores -> constructor -> feasible fill via SimExecution/simulate_fills -> settle -> NAV/feasibility/holdings/ event logs). No fill/cash/settlement logic is duplicated. --- runtime/backtest/driver.py | 285 ++++++------------------------- runtime/backtest/engine.py | 234 +++++++++++++++++++++++++ runtime/backtest/event_models.py | 217 +++++++++++++++++++++++ runtime/backtest/events.py | 93 ++++++++++ 4 files changed, 593 insertions(+), 236 deletions(-) create mode 100644 runtime/backtest/engine.py create mode 100644 runtime/backtest/event_models.py create mode 100644 runtime/backtest/events.py 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..2d325bc --- /dev/null +++ b/runtime/backtest/engine.py @@ -0,0 +1,234 @@ +"""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", "next_decision_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, + } + ) + next_decision = ( + periods[i + 1].decision_ts if i + 1 < len(periods) else pd.NaT + ) + self._record_event(period, next_decision) + 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 + ) -> None: + self._event_log.append( + { + "date": period.date, + "decision_ts": period.decision_ts, + "execution_ts": period.execution_ts, + "exit_date": period.exit_date, + "next_decision_ts": next_decision_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..e88f61b --- /dev/null +++ b/runtime/backtest/event_models.py @@ -0,0 +1,217 @@ +"""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, + ) + 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() + out.append( + HoldingPeriod( + date=date, + entry_date=date, + exit_date=exit_date, + decision_ts=day + decision, + execution_ts=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..7d1572a --- /dev/null +++ b/runtime/backtest/events.py @@ -0,0 +1,93 @@ +"""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 + timestamps recorded for auditability (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 + + +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 From 21c1a4996b304d8d4b8d8a0110bf92e4bef44780 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Mon, 15 Jun 2026 11:31:54 +0800 Subject: [PATCH 2/4] feat(config): opt-in intraday tail-rebalance event model + validation Add an optional IntradayCfg (enabled/decision_time/data_lag/session_open/ execution_model/execution_window/require_cache_coverage/missing_execution) and constrain backtest.event_order to a Literal. A RootConfig validator requires intraday.enabled=true when event_order='intraday_tail_rebalance'; the daily default needs no intraday section, so every existing config validates unchanged. Unsupported execution models and malformed/ordered execution windows fail with readable errors. --- qt/config.py | 89 +++++++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 88 insertions(+), 1 deletion(-) 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: From 340d627c89186e490a708b85e2fcda7c9ec85d12 Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Mon, 15 Jun 2026 11:31:54 +0800 Subject: [PATCH 3/4] feat(qt): run-phase-i5a-intraday architecture smoke + tests New CLI run mode + qt/intraday_tail_framework.py: build the daily panel/ universe via the existing P4 cache, read required 1min bars from the EXISTING intraday cache only (a cache miss is a hard blocker, never a silent warm -> zero stk_mins live calls), score with the PIT-safe I3 intraday_ret_0930_1450 feature, and run the engine with the intraday tail event model. Writes an auditable report (event timestamps, exec-to-exec statement, blocked fills, achieved holdings, turnover/cost/cash, limitations). config/phase_i5a_intraday_tail_framework.yaml is a SH/SZ SSE50 smoke. Tests cover the daily golden (engine == driver), event-model units, intraday PIT (exec!=close-to-close; perturb post-cutoff bars -> feature unchanged; perturb exec bar -> returns move; perturb daily close -> intraday unchanged; missing bar blocks; holdings/turnover are achieved), and config validation. --- config/phase_i5a_intraday_tail_framework.yaml | 108 +++++ qt/cli.py | 30 ++ qt/intraday_tail_framework.py | 396 ++++++++++++++++++ tests/test_i5a_event_backtest.py | 357 ++++++++++++++++ 4 files changed, 891 insertions(+) create mode 100644 config/phase_i5a_intraday_tail_framework.yaml create mode 100644 qt/intraday_tail_framework.py create mode 100644 tests/test_i5a_event_backtest.py 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/intraday_tail_framework.py b/qt/intraday_tail_framework.py new file mode 100644 index 0000000..95d7510 --- /dev/null +++ b/qt/intraday_tail_framework.py @@ -0,0 +1,396 @@ +"""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] + 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 not covered: + raise ValueError( + "intraday smoke blocked: NONE of the requested symbols are fully " + f"covered in the minute cache for [{cfg.data.start}, {cfg.data.end}]. " + "Shrink/move the window to covered SH/SZ data — this runner refuses to " + "warm missing minute history." + ) + if uncovered: + logger.info( + "intraday cache: %d/%d symbols covered; %d excluded (uncovered)", + len(covered), len(symbols), len(uncovered), + ) + + 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 + + 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, + 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("") + if result.event_log.empty: + lines.append("_no settled periods_") + else: + lines.append("| date | decision_ts | execution_ts | exit_date | next_decision_ts |") + lines.append("|---|---|---|---|---|") + for date, row in result.event_log.iterrows(): + lines.append( + f"| {pd.Timestamp(date).date()} | {row['decision_ts']} | " + f"{row['execution_ts']} | {pd.Timestamp(row['exit_date']).date()} | " + f"{row['next_decision_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/tests/test_i5a_event_backtest.py b/tests/test_i5a_event_backtest.py new file mode 100644 index 0000000..4c9a9f7 --- /dev/null +++ b/tests/test_i5a_event_backtest.py @@ -0,0 +1,357 @@ +"""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 + +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.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) From d4e952edac32b3b298d8970fd28b705bbf671fdd Mon Sep 17 00:00:00 2001 From: shaofl <2899218482@qq.com> Date: Mon, 15 Jun 2026 15:03:11 +0800 Subject: [PATCH 4/4] fix(i5a): fail on partial minute coverage; surface actual exec bars Review fixes: - require_cache_coverage=true now fails loudly when ANY requested symbol is uncovered (previously it only failed when NONE were covered, silently dropping uncovered names and biasing the realized universe). Lenient mode (false) still drops + discloses, and an all-uncovered set is a clear blocker. Adds partial-coverage tests (required-fails / lenient-drops). - Make the holding period's exit execution anchor explicit (HoldingPeriod.exit_execution_ts) so audit timing no longer has to be inferred from the next period (and the final period shows a real exit anchor, not NaT). Engine event_log adds exit_execution_ts + next_execution_ts; the I5a report shows planned vs ACTUAL execution-bar time per period (a 14:52 fill when 14:51 is missing is now visible). Daily behaviour unchanged (phase0 ic 0.9600/annual 0.8408). --- qt/intraday_tail_framework.py | 72 +++++++++++++++++++++++---- runtime/backtest/engine.py | 19 +++++-- runtime/backtest/event_models.py | 5 ++ runtime/backtest/events.py | 9 +++- tests/test_i5a_event_backtest.py | 85 ++++++++++++++++++++++++++++++++ 5 files changed, 173 insertions(+), 17 deletions(-) diff --git a/qt/intraday_tail_framework.py b/qt/intraday_tail_framework.py index 95d7510..064baa3 100644 --- a/qt/intraday_tail_framework.py +++ b/qt/intraday_tail_framework.py @@ -70,6 +70,10 @@ class I5aResult: 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 @@ -113,18 +117,32 @@ def _load_minute_bars_cache_only( (uncovered if gaps else covered).append(sym) require_cov = cfg.intraday is not None and cfg.intraday.require_cache_coverage - if require_cov and not covered: + 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: NONE of the requested symbols are fully " - f"covered in the minute cache for [{cfg.data.start}, {cfg.data.end}]. " - "Shrink/move the window to covered SH/SZ data — this runner refuses to " - "warm missing minute history." + "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)", + "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( @@ -210,6 +228,21 @@ def run_phase_i5a_intraday(config_path: str) -> I5aResult: 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, @@ -225,6 +258,7 @@ def run_phase_i5a_intraday(config_path: str) -> I5aResult: 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, ) @@ -309,16 +343,34 @@ def _write_report(result: I5aResult, *, elapsed: float) -> None: 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 | execution_ts | exit_date | next_decision_ts |") - lines.append("|---|---|---|---|---|") + 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']} | {pd.Timestamp(row['exit_date']).date()} | " - f"{row['next_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") diff --git a/runtime/backtest/engine.py b/runtime/backtest/engine.py index 2d325bc..8a95334 100644 --- a/runtime/backtest/engine.py +++ b/runtime/backtest/engine.py @@ -35,7 +35,8 @@ _NAV_COLUMNS = ["nav", "gross_return", "cost", "turnover", "net_return"] _EVENT_COLUMNS = [ - "date", "decision_ts", "execution_ts", "exit_date", "next_decision_ts", + "date", "decision_ts", "execution_ts", "exit_date", "exit_execution_ts", + "next_decision_ts", "next_execution_ts", ] @@ -126,10 +127,12 @@ def run(self) -> pd.DataFrame: "net_return": net, } ) - next_decision = ( - periods[i + 1].decision_ts if i + 1 < len(periods) else pd.NaT + 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, ) - self._record_event(period, next_decision) out = pd.DataFrame(rows, columns=["date", *_NAV_COLUMNS]) return out.set_index("date") @@ -215,7 +218,11 @@ def holdings_log(self) -> pd.DataFrame: # -- event log (I5a auditability) ------------------------------------- # def _record_event( - self, period: HoldingPeriod, next_decision_ts: pd.Timestamp + self, + period: HoldingPeriod, + *, + next_decision_ts: pd.Timestamp, + next_execution_ts: pd.Timestamp, ) -> None: self._event_log.append( { @@ -223,7 +230,9 @@ def _record_event( "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, } ) diff --git a/runtime/backtest/event_models.py b/runtime/backtest/event_models.py index e88f61b..aca1ba5 100644 --- a/runtime/backtest/event_models.py +++ b/runtime/backtest/event_models.py @@ -56,6 +56,7 @@ def holding_periods(self) -> list[HoldingPeriod]: 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 ] @@ -155,6 +156,7 @@ def holding_periods(self) -> list[HoldingPeriod]: 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, @@ -162,6 +164,9 @@ def holding_periods(self) -> list[HoldingPeriod]: 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 diff --git a/runtime/backtest/events.py b/runtime/backtest/events.py index 7d1572a..053b330 100644 --- a/runtime/backtest/events.py +++ b/runtime/backtest/events.py @@ -35,8 +35,12 @@ 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 - timestamps recorded for auditability (they may carry intra-day precision). + ``(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``. """ @@ -46,6 +50,7 @@ class HoldingPeriod: 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: diff --git a/tests/test_i5a_event_backtest.py b/tests/test_i5a_event_backtest.py index 4c9a9f7..d0cd0b2 100644 --- a/tests/test_i5a_event_backtest.py +++ b/tests/test_i5a_event_backtest.py @@ -17,6 +17,7 @@ from __future__ import annotations +import logging from pathlib import Path import pandas as pd @@ -25,6 +26,9 @@ 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 @@ -355,3 +359,84 @@ def test_invalid_execution_window_fails_readably(): 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