| """ |
| 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 |
|
|
| 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")) |
| ) |
| |
| 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()} |
| return {"accepted": bool(accepted), "checks": checks, "num_trades": n_trades} |
|
|
