walk_forward.py

"""
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}