""" Walk-forward, look-ahead-safe backtest engine (spec sections 28-30, 53-54, 92). Decisions at time t only ever see data <= t; future closes are used strictly afterward, only to score the already-fixed decision (section 53: "Only AFTER execution/decision is fixed may future data be used to calculate realized PnL"). Fill-price convention (documented per section 25's "all assumptions must be explicit, configurable, and auditable"): entry fills at the CLOSE of the bar `execution_delay_bars` after the decision origin; exit fills at the close `horizon` bars after that. Swap for next-bar-OPEN if you want a stricter convention — it's one line, marked below. Single-position convention: this simulates ONE account with no margin or leverage model, so a new trade is never opened while a previously opened trade hasn't exited yet (decide() has no notion of position state, so this is enforced here). Without this, a fresh full-notional trade would be opened at every origin regardless of what's still open -- during a run of same-direction signals that's several positions stacked on the same equity curve at once, which silently multiplies exposure the cost model and risk metrics know nothing about. """ from __future__ import annotations from dataclasses import dataclass from typing import List, Optional import numpy as np import pandas as pd import config as cfg from calibration import ProbabilityCalibrator from decision_engine import Decision, decide from features import warm_up_valid_from from leakage_audit import LeakageError, audit_feature_frame class InsufficientDataError(Exception): """Raised when the feature frame has fewer bars than this backtest configuration needs for even a single origin -> exit sequence. Distinct from the ValueError warm_up_valid_from raises (which means indicator warm-up is impossible at all): this means warm-up succeeded but there isn't enough data left over for context_length plus at least one tradeable origin. Previously this case wasn't checked for at all -- feature_df.index[start_idx] would raise a bare, confusing IndexError once start_idx ran past the end of the frame. Checking explicitly means the message can say what failed, not just that something did (spec section 98).""" @dataclass class Trade: decision_ts: pd.Timestamp entry_ts: pd.Timestamp exit_ts: pd.Timestamp action: str entry_price: float exit_price: float forecast_return: float gross_pnl: float net_pnl: float costs: float exit_reason: str = "horizon" # "horizon" | "stop_loss" | "take_profit" prob_favorable: float = float("nan") # decision.prob_favorable at open, for post-hoc entry-filter analysis interval_width_pct: float = float("nan") # decision.interval_width_pct at open, same reason def _scan_for_early_exit(feature_df: pd.DataFrame, exec_idx: int, horizon_exit_idx: int, action: str, entry_price: float, risk: Optional[cfg.RiskManagement]) -> tuple: """First bar strictly after entry whose high/low breaches a stop-loss or take-profit level, or horizon_exit_idx if neither ever triggers (or risk management is off). Returns (exit_idx, exit_reason). The exit PRICE is still that bar's CLOSE, not the stop/TP level itself -- consistent with the rest of this engine's documented close-only fill convention (module docstring), and because intrabar high/low only tells us a threshold was crossed, not what price you'd actually have been filled at; inventing an exact fill price from that would be manufacturing data we don't have (spec section 88). This is conservative in both directions: a stop may report a smaller loss than a real gap-through would produce, and a take-profit may report less gain than the intrabar peak. If both levels are breached within the same bar, we can't tell from OHLC alone which was touched first, so this assumes the stop -- the conservative read when direction of causation is unknown.""" if risk is None or (risk.stop_loss_frac is None and risk.take_profit_frac is None): return horizon_exit_idx, "horizon" if action == "BUY": stop_level = entry_price * (1 - risk.stop_loss_frac) if risk.stop_loss_frac else None tp_level = entry_price * (1 + risk.take_profit_frac) if risk.take_profit_frac else None else: stop_level = entry_price * (1 + risk.stop_loss_frac) if risk.stop_loss_frac else None tp_level = entry_price * (1 - risk.take_profit_frac) if risk.take_profit_frac else None for j in range(exec_idx + 1, horizon_exit_idx + 1): lo = float(feature_df["low"].iloc[j]) hi = float(feature_df["high"].iloc[j]) hit_stop = stop_level is not None and ( (action == "BUY" and lo <= stop_level) or (action == "SELL" and hi >= stop_level) ) hit_tp = tp_level is not None and ( (action == "BUY" and hi >= tp_level) or (action == "SELL" and lo <= tp_level) ) if hit_stop: return j, "stop_loss" if hit_tp: return j, "take_profit" return horizon_exit_idx, "horizon" def _build_bar_by_bar_equity(feature_df: pd.DataFrame, start_idx: int, end_idx: int, trades: List["Trade"]) -> pd.Series: """TRUE mark-to-market equity: one point per BAR from start_idx to end_idx inclusive (not one point per origin/trade, and never a gap during an open position). While flat, equity is unchanged. While a position is open, each bar's value reflects that bar's UNREALIZED gross return relative to entry -- capturing genuine interim drawdown/runup a trade-event-only curve hides entirely (a position that spikes 2% against you mid-hold before recovering to a 0.3% loss at exit previously showed as a single flat jump straight to -0.3%, with the 2% excursion invisible to max_drawdown/Sharpe/ Sortino/Calmar). The round-trip cost is applied once, at the realized exit bar (matching net_pnl's own definition), not smeared across intermediate marks -- you haven't paid the exit-side cost until you've actually exited.""" idx = feature_df.index[start_idx:end_idx + 1] closes = feature_df["close"].iloc[start_idx:end_idx + 1].to_numpy() n = len(idx) equity = np.empty(n) entry_map = {t.entry_ts: t for t in trades} base = 1.0 active = None entry_base = 1.0 for k in range(n): ts = idx[k] if active is None and ts in entry_map: active = entry_map[ts] entry_base = base if active is not None: close_k = float(closes[k]) unreal = (close_k / active.entry_price - 1.0) if active.action == "BUY" \ else (active.entry_price / close_k - 1.0) if ts == active.exit_ts: equity[k] = entry_base * (1.0 + active.net_pnl) base = equity[k] active = None else: equity[k] = entry_base * (1.0 + unreal) else: equity[k] = base return pd.Series(equity, index=idx) class WalkForwardBacktester: """`forecaster` must expose .predict(context_df, target_col, past_covariate_cols, prediction_length, freq) -> an object with .quantiles / .quantile_levels (see moirai_model.QuantileForecast). In production this is always a MoiraiMoEForecaster; tests may inject a deterministic stub to verify the pipeline's causal behavior — that stub is never imported by app.py (see tests/fixtures.py). This is the ONLY place that turns a Moirai forecast into a simulated trade: app.py's Live tab calls decide() directly with the same DecisionConfig/cost/calibrator a backtest run would use, so there is exactly one decision pipeline, not two independently- maintained ones (spec: "backtest profitable but live behaves differently" is a structural risk this project treats as a real bug class, not just a documentation gap).""" def __init__(self, forecaster, timeframe: str, asset_class: str = "forex", horizon: int = 5, context_length: int = 256, decision_cfg: Optional[cfg.DecisionConfig] = None, risk: Optional[cfg.RiskManagement] = None, calibrator: Optional["ProbabilityCalibrator"] = None): self.forecaster = forecaster self.timeframe = timeframe self.freq = cfg.TIMEFRAME_TO_PANDAS_FREQ[timeframe] self.cost_profile = cfg.COST_PROFILES[asset_class] self.horizon = horizon self.context_length = context_length self.decision_cfg = decision_cfg or cfg.DecisionConfig() self.risk = risk # None => no early exit, identical to every pre-existing backtest self.calibrator = calibrator # None => raw probabilities, identical to every pre-existing backtest def run(self, feature_df: pd.DataFrame, start_idx: Optional[int] = None, step: int = 1) -> dict: cost_frac = cfg.round_trip_cost_frac(self.cost_profile) warm = warm_up_valid_from(feature_df) # raises ValueError if warm-up is impossible at all n = len(feature_df) delay = self.cost_profile.execution_delay_bars min_start = max(start_idx or 0, warm, warm + self.context_length) last_origin = n - self.horizon - delay - 1 if min_start > last_origin: bars_needed = min_start + self.horizon + delay + 1 raise InsufficientDataError( f"Not enough data for a single backtest origin: {n} bar(s) " f"available, but this configuration needs at least " f"{bars_needed} (warm-up {warm} + context_length " f"{self.context_length} + horizon {self.horizon} + " f"execution_delay {delay} + 1). Widen the History window " f"by at least {bars_needed - n} bar(s), or reduce context " f"length / horizon." ) start_idx = min_start trades: List[Trade] = [] decisions_log: List[dict] = [] errors: List[dict] = [] n_origins = 0 open_until_idx = -1 # bar index the currently-open trade exits at; -1 == flat end_idx = start_idx # highest bar index any origin's exit reaches; grown below # Diagnostics (spec section 98 / audit item 18): every origin ends # up in exactly one of these buckets, so they must sum to # n_origins -- asserted at the end rather than just hoped for. diag = { "signals_buy": 0, "signals_sell": 0, "hold_wide_interval": 0, "hold_no_edge": 0, "hold_low_confidence_buy": 0, "hold_low_confidence_sell": 0, "model_errors": 0, "skipped_tail_insufficient_future_bars": 0, "skipped_overlap_position_open": 0, "trades_opened_buy": 0, "trades_opened_sell": 0, "exit_horizon": 0, "exit_stop_loss": 0, "exit_take_profit": 0, } for i in range(start_idx, max(start_idx, last_origin), step): n_origins += 1 origin_ts = feature_df.index[i] context = feature_df.iloc[i - self.context_length + 1: i + 1] audit = audit_feature_frame(context, origin_ts) if not audit["passed"]: # Fail loudly rather than silently dropping the bad origin # (spec section 26). raise LeakageError(f"Backtest halted at {origin_ts}: {audit['violations']}") try: forecast = self.forecaster.predict( context_df=context, target_col=cfg.TARGET_FEATURE, past_covariate_cols=[c for c in cfg.PAST_DYNAMIC_FEATURES if c in context.columns], prediction_length=self.horizon, freq=self.freq, ) except Exception as e: errors.append({"origin_ts": str(origin_ts), "error": str(e), "error_type": type(e).__name__}) diag["model_errors"] += 1 continue price_now = float(feature_df["close"].iloc[i]) decision: Decision = decide(forecast, price_now, self.decision_cfg, cost_estimate_frac=cost_frac, calibrator=self.calibrator) decisions_log.append({"origin_ts": origin_ts, "action": decision.action, "reject_stage": decision.reject_stage}) if decision.action == "BUY": diag["signals_buy"] += 1 elif decision.action == "SELL": diag["signals_sell"] += 1 elif f"hold_{decision.reject_stage}" in diag: diag[f"hold_{decision.reject_stage}"] += 1 else: # Should be unreachable -- decide() only ever sets # reject_stage to one of the 4 keys above when action is # HOLD. Surfacing this loudly rather than silently # dropping the origin from the count is the whole point # of asserting `accounted == n_origins` below. raise AssertionError( f"decide() returned action={decision.action!r} with an " f"unrecognized reject_stage={decision.reject_stage!r} — " f"diagnostics cannot account for this origin" ) exec_idx = i + delay exit_idx = i + delay + self.horizon if exit_idx >= n: diag["skipped_tail_insufficient_future_bars"] += 1 continue end_idx = max(end_idx, exit_idx) if i < open_until_idx: # Still holding a position opened at an earlier origin -- # don't stack a second full-notional trade on top of it # (see "Single-position convention" in the module # docstring). The already-open trade's own exit will # still post its P&L to the equity curve when its turn # comes; this origin just isn't tradable right now. if decision.action in ("BUY", "SELL"): diag["skipped_overlap_position_open"] += 1 continue entry_price = float(feature_df["close"].iloc[exec_idx]) # fill convention, see module docstring if decision.action == "HOLD": continue actual_exit_idx, exit_reason = _scan_for_early_exit( feature_df, exec_idx, exit_idx, decision.action, entry_price, self.risk ) exit_price = float(feature_df["close"].iloc[actual_exit_idx]) # future data — used only to SCORE, decision already fixed if decision.action == "BUY": gross = exit_price / entry_price - 1.0 else: # SELL gross = entry_price / exit_price - 1.0 net = gross - cost_frac trades.append(Trade( decision_ts=origin_ts, entry_ts=feature_df.index[exec_idx], exit_ts=feature_df.index[actual_exit_idx], action=decision.action, entry_price=entry_price, exit_price=exit_price, forecast_return=decision.expected_return, gross_pnl=gross, net_pnl=net, costs=cost_frac, exit_reason=exit_reason, prob_favorable=decision.prob_favorable, interval_width_pct=decision.interval_width_pct, )) diag[f"trades_opened_{decision.action.lower()}"] += 1 diag[f"exit_{exit_reason}"] += 1 open_until_idx = actual_exit_idx # Every origin must land in exactly one bucket -- a mismatch here # would mean the diagnostics themselves are lying, which is worse # than not having them (audit item 18 exists so counts can be # trusted, not just displayed). accounted = (diag["signals_buy"] + diag["signals_sell"] + diag["hold_wide_interval"] + diag["hold_no_edge"] + diag["hold_low_confidence_buy"] + diag["hold_low_confidence_sell"] + diag["model_errors"] + diag["skipped_tail_insufficient_future_bars"]) # (BUY/SELL signals still count once here even when later skipped # for overlap -- skipped_overlap_position_open is a sub-count of # signals_buy/signals_sell, not an additional bucket, since the # decision itself was still a real BUY/SELL signal.) assert accounted == n_origins, ( f"diagnostic bucket mismatch: {accounted} accounted vs {n_origins} origins — " f"diagnostics do not fully explain every origin's outcome" ) equity_curve = _build_bar_by_bar_equity(feature_df, start_idx, end_idx, trades) return { "trades": trades, "decisions_log": decisions_log, "equity_curve": equity_curve, "num_origins": n_origins, "errors": errors, "diagnostics": diag, "requested_risk": {"stop_loss_frac": self.risk.stop_loss_frac if self.risk else None, "take_profit_frac": self.risk.take_profit_frac if self.risk else None}, }