""" Walk-forward validation harness. This is the ONLY code in the project that is allowed to call calibration.fit_calibration() or pick DecisionConfig parameters -- everywhere else (app.py, backtest_engine.py in normal use) takes an already-fit calibrator / already-chosen config as a given, frozen input. The structural leakage guarantee this module exists to provide: data is split chronologically once, train+validation are used for everything that "learns" anything (calibration fitting, threshold selection), and the test split is touched exactly once, at the very end, by evaluate_frozen_strategy() -- which takes a config that was ALREADY decided before it is ever called and returns a report, not a score select_parameters() could react to. There is no code path in this file that feeds a test-split result back into parameter selection; that is what "frozen" means here, not just a comment saying so. This module also runs two simple, causal baseline strategies (item 14) through the exact same WalkForwardBacktester / cost model / timestamps as the real Moirai forecaster, so "did Moirai actually help" is a real, apples-to-apples comparison rather than a claim. What this module CANNOT do from here: produce a genuine verdict without real market data. Every function below is fully exercised by this project's test suite on synthetic data (proving the splitting, fitting, freezing, and acceptance-checking logic is each individually correct), but a synthetic-data "PASS" from check_acceptance() is not evidence of real trading edge -- only a real data run is. See README.md's "Profitability reset" section. """ 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 backtest_engine import WalkForwardBacktester from calibration import ProbabilityCalibrator, fit_calibration from decision_engine import decide from metrics import trading_summary from moirai_model import QuantileForecast def chronological_split(feature_df: pd.DataFrame, wf: cfg.WalkForwardConfig = None): """Splits by TIME ORDER ONLY -- no shuffling, no random sampling (shuffling a time series before splitting is itself a leakage bug: it lets information from the future sit in what's supposed to be the past). Returns (train_df, val_df, test_df); each retains full row content (not just an index range) so callers can run compute_indicators-derived features directly on each split.""" wf = wf or cfg.WalkForwardConfig() n = len(feature_df) train_end = int(n * wf.train_frac) val_end = int(n * (wf.train_frac + wf.val_frac)) if train_end < 10 or (val_end - train_end) < 10 or (n - val_end) < 10: raise ValueError( f"frame of {n} bars is too short to split into three usable " f"train/val/test windows with train_frac={wf.train_frac}, " f"val_frac={wf.val_frac} -- widen the History window" ) return feature_df.iloc[:train_end], feature_df.iloc[train_end:val_end], feature_df.iloc[val_end:] def collect_raw_probability_outcome_pairs(forecaster, feature_df: pd.DataFrame, timeframe: str, asset_class: str, horizon: int, context_length: int) -> tuple: """Walks every eligible origin in feature_df (same warm-up/context rules as WalkForwardBacktester, deliberately reusing it in a zero-risk, no-threshold-gating configuration rather than duplicating the origin-walking logic a second time) and records, for BOTH the buy-side and sell-side event at each origin, the raw (uncalibrated) probability decide() computed and whether that event actually happened by the horizon exit. Pooling buy-side and sell-side pairs into one calibration dataset is a deliberate simplification (documented, not hidden): it roughly doubles the usable sample size and is reasonable unless forex/crypto returns are strongly directionally asymmetric, which nothing in this project's indicator set currently detects either way. Returns (raw_probs, outcomes, period_start, period_end) -- the exact inputs fit_calibration() needs, plus the window they came from for the calibrator's own record-keeping. """ from decision_engine import _prob_exceeds # internal helper, deliberately not part of the public decide() surface cost_frac = cfg.round_trip_cost_frac(cfg.COST_PROFILES[asset_class]) delay = cfg.COST_PROFILES[asset_class].execution_delay_bars from features import warm_up_valid_from warm = warm_up_valid_from(feature_df) n = len(feature_df) start_idx = max(warm, warm + context_length) last_origin = n - horizon - delay - 1 freq = cfg.TIMEFRAME_TO_PANDAS_FREQ[timeframe] raw_probs, outcomes = [], [] for i in range(start_idx, max(start_idx, last_origin)): context = feature_df.iloc[i - context_length + 1: i + 1] try: forecast = 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=horizon, freq=freq, ) except Exception: continue price_now = float(feature_df["close"].iloc[i]) levels = list(forecast.quantile_levels) step_returns = np.asarray(forecast.quantiles)[-1] / price_now - 1.0 raw_p_buy = _prob_exceeds(levels, step_returns, threshold=cost_frac) raw_p_sell = 1.0 - _prob_exceeds(levels, step_returns, threshold=-cost_frac) exit_idx = i + delay + horizon entry_price = float(feature_df["close"].iloc[i + delay]) exit_price = float(feature_df["close"].iloc[exit_idx]) realized_return = exit_price / entry_price - 1.0 raw_probs.append(raw_p_buy) outcomes.append(1.0 if realized_return > cost_frac else 0.0) raw_probs.append(raw_p_sell) outcomes.append(1.0 if realized_return < -cost_frac else 0.0) period_start = str(feature_df.index[start_idx]) if len(feature_df) else "" period_end = str(feature_df.index[min(last_origin, n - 1)]) if last_origin > 0 else "" return np.array(raw_probs), np.array(outcomes), period_start, period_end class ZeroSkillForecaster: """Baseline: always predicts no change at all (flat quantiles at the current price). Under normal thresholds this always HOLDs -- the "can't possibly have a real edge, and shouldn't claim one" reference point (audit item 14). Never reachable from app.py.""" def predict(self, context_df, target_col, past_covariate_cols, prediction_length, freq): last_close = float(context_df[target_col].iloc[-1]) prices = [last_close] * 5 return QuantileForecast(horizon=prediction_length, quantile_levels=[0.1, 0.25, 0.5, 0.75, 0.9], quantiles=np.array([prices]), samples=None, context_length_used=len(context_df), prediction_length=prediction_length) class NaiveMomentumForecaster: """Baseline: predicts continuation of the most recent realized bar's return, with a fixed, non-overconfident spread (a naive rule has no real uncertainty estimate, so this doesn't manufacture one narrow enough to win on confidence grounds alone). Run through the exact same WalkForwardBacktester as the real forecaster for a fair, same-costs, same-timestamps comparison (audit item 14). Never reachable from app.py -- this is a validation tool, not a second predictive model, and does not compete with or fall back from Salesforce/moirai-moe-1.0-R-small.""" def predict(self, context_df, target_col, past_covariate_cols, prediction_length, freq): last_close = float(context_df[target_col].iloc[-1]) prev_close = float(context_df[target_col].iloc[-2]) if len(context_df) > 1 else last_close last_return = last_close / prev_close - 1.0 if prev_close else 0.0 p50_price = last_close * (1 + last_return) spread = abs(last_return) * 2 + 1e-6 prices = [p50_price * (1 - spread), p50_price * (1 - spread / 2), p50_price, p50_price * (1 + spread / 2), p50_price * (1 + spread)] return QuantileForecast(horizon=prediction_length, quantile_levels=[0.1, 0.25, 0.5, 0.75, 0.9], quantiles=np.array([prices]), samples=None, context_length_used=len(context_df), prediction_length=prediction_length) def select_frozen_config(forecaster, train_df: pd.DataFrame, val_df: pd.DataFrame, timeframe: str, asset_class: str, horizon: int, context_length: int, wf: cfg.WalkForwardConfig = None): """The only function allowed to CHOOSE calibration + threshold parameters. Fits calibration on train_df, then grid-searches min_probability x buy_threshold (bounded by wf's own grids -- at most len(min_probability_grid)*len(buy_threshold_grid) backtest runs, deliberately small) evaluated ONLY on val_df, picking the combination with the best validation net_pnl among those meeting wf.min_trades_for_acceptance. Returns (calibrator, frozen_dc, selection_report) -- selection_report is for transparency only, never fed back into this function. test_df is not a parameter here on purpose: this function must be structurally unable to see it.""" wf = wf or cfg.WalkForwardConfig() raw_probs, outcomes, p_start, p_end = collect_raw_probability_outcome_pairs( forecaster, train_df, timeframe, asset_class, horizon, context_length) if len(raw_probs) < wf.min_trades_for_calibration: raise ValueError( f"only {len(raw_probs)} calibration pairs from the training " f"window -- need at least {wf.min_trades_for_calibration}; " f"widen the History window or the train_frac" ) calibrator = fit_calibration(raw_probs, outcomes, period_start=p_start, period_end=p_end) best = None trials = [] for min_prob in wf.min_probability_grid: for buy_th in wf.buy_threshold_grid: dc = cfg.DecisionConfig(min_probability=min_prob, buy_threshold=buy_th, sell_threshold=-buy_th) bt = WalkForwardBacktester(forecaster, timeframe=timeframe, asset_class=asset_class, horizon=horizon, context_length=context_length, decision_cfg=dc, calibrator=calibrator) result = bt.run(val_df) trades_df = pd.DataFrame([t.__dict__ for t in result["trades"]]) summary = trading_summary(trades_df, result["equity_curve"], cfg.TIMEFRAME_MINUTES[timeframe]) n_trades = summary.get("num_trades", 0) net_pnl = summary.get("net_pnl", float("-inf")) if n_trades >= wf.min_trades_for_acceptance else float("-inf") trials.append({"min_probability": min_prob, "buy_threshold": buy_th, "val_num_trades": n_trades, "val_net_pnl": summary.get("net_pnl")}) if best is None or net_pnl > best[0]: best = (net_pnl, dc, min_prob, buy_th) if best is None or best[0] == float("-inf"): raise ValueError( f"no (min_probability, buy_threshold) combination in the grid produced " f"{wf.min_trades_for_acceptance}+ trades on the validation window -- " f"this is itself a legitimate result (the signal may not be strong " f"enough at any tested threshold on this data), not a bug to work around" ) _, frozen_dc, chosen_min_prob, chosen_buy_th = best selection_report = { "chosen_min_probability": chosen_min_prob, "chosen_buy_threshold": chosen_buy_th, "n_calibration_pairs": len(raw_probs), "trials": trials, } return calibrator, frozen_dc, selection_report def run_baseline(forecaster_cls, test_df: pd.DataFrame, timeframe: str, asset_class: str, horizon: int, context_length: int, dc: cfg.DecisionConfig, calibrator=None) -> dict: """Runs a baseline forecaster through the identical WalkForwardBacktester/cost/timestamp machinery as the real model, so baseline vs. Moirai is a fair comparison (audit item 14).""" bt = WalkForwardBacktester(forecaster_cls(), timeframe=timeframe, asset_class=asset_class, horizon=horizon, context_length=context_length, decision_cfg=dc, calibrator=calibrator) result = bt.run(test_df) trades_df = pd.DataFrame([t.__dict__ for t in result["trades"]]) return trading_summary(trades_df, result["equity_curve"], cfg.TIMEFRAME_MINUTES[timeframe]) def check_acceptance(test_summary: dict, baseline_summaries: dict, test_equity: pd.Series, wf: cfg.WalkForwardConfig = None) -> dict: """Item 18's checklist, applied mechanically to already-computed results -- this function makes no trading decisions and touches no raw data, it only judges numbers it's handed. Returns a dict of named boolean checks plus an overall `accepted`; every check is named so a REJECT is explainable, not just asserted.""" wf = wf or cfg.WalkForwardConfig() n_trades = test_summary.get("num_trades", 0) checks = { "sufficient_trade_count": n_trades >= wf.min_trades_for_acceptance, "positive_net_return": test_summary.get("net_pnl", -1) is not None and test_summary.get("net_pnl", -1) > 0, "profit_factor_above_1": (test_summary.get("profit_factor") or 0) > 1.0, "acceptable_drawdown": (test_summary.get("max_drawdown") or -1.0) > -0.25, } for name, baseline_summary in baseline_summaries.items(): checks[f"beats_baseline_{name}"] = ( (test_summary.get("net_pnl") or float("-inf")) > (baseline_summary.get("net_pnl") or float("-inf")) ) # No single period should explain most of the profit. if n_trades > 0 and len(test_equity) > 1 and isinstance(test_equity.index, pd.DatetimeIndex): monthly = test_equity.resample("30D").last().dropna() period_returns = monthly.pct_change().dropna() total_gain = period_returns[period_returns > 0].sum() max_single = period_returns[period_returns > 0].max() if not period_returns.empty else 0.0 share = (max_single / total_gain) if total_gain > 0 else 0.0 checks["not_single_period_dependent"] = share <= wf.max_single_period_profit_share else: checks["not_single_period_dependent"] = False accepted = n_trades > 0 and all(checks.values()) checks = {k: bool(v) for k, v in checks.items()} # numpy bool_ vs Python bool: `is False` must work for callers return {"accepted": bool(accepted), "checks": checks, "num_trades": n_trades}