Blame for tests/test_walk_forward.py

"""
walk_forward.py tests. test_frozen_selection_is_unaffected_by_anything_
in_the_test_window is the single most important test in this file --
it's a direct, executable proof of the leakage-safety claim in that
module's docstring, not just a check that functions run without error.
"""
import numpy as np
import pandas as pd

import config as cfg
import walk_forward as wf
from features import compute_indicators
from fixtures import make_synthetic_ohlcv
from moirai_model import QuantileForecast


def test_chronological_split_preserves_time_order_and_proportions():
    df = make_synthetic_ohlcv(n=1000, seed=70, freq="1h")
    train, val, test = wf.chronological_split(df, cfg.WalkForwardConfig(train_frac=0.5, val_frac=0.3))
    assert len(train) + len(val) + len(test) == len(df)
    assert train.index[-1] < val.index[0] <= val.index[-1] < test.index[0]
    assert abs(len(train) / len(df) - 0.5) < 0.01
    assert abs(len(val) / len(df) - 0.3) < 0.01


def test_chronological_split_never_shuffles():
    df = make_synthetic_ohlcv(n=500, seed=71, freq="1h")
    train, val, test = wf.chronological_split(df)
    for part in (train, val, test):
        assert part.index.is_monotonic_increasing


def test_chronological_split_rejects_too_short_data():
    df = make_synthetic_ohlcv(n=20, seed=72, freq="1h")
    try:
        wf.chronological_split(df)
        raise AssertionError("expected ValueError for too-short data")
    except ValueError:
        pass


class _NoisyStub:
    """Raw probability is a noisy but genuinely informative function of
    the actual forward return -- used to build a calibration dataset
    that isn't pure garbage, without needing the real model."""
    def __init__(self, seed=0):
        self.rng = np.random.default_rng(seed)

    def predict(self, context_df, target_col, past_covariate_cols, prediction_length, freq):
        last_close = float(context_df[target_col].iloc[-1])
        drift = self.rng.normal(0, 0.003)
        p50 = last_close * (1 + drift)
        spread = 0.004
        prices = [p50 * (1 - spread), p50 * (1 - spread / 2), p50, p50 * (1 + spread / 2), p50 * (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 test_frozen_selection_is_unaffected_by_anything_in_the_test_window():
    """THE core leakage-safety test: two frames identical through the
    train/val boundary but with completely different (differently
    seeded, differently distributed) data after it must produce an
    IDENTICAL calibrator and frozen DecisionConfig from
    select_frozen_config -- since that function is never given the test
    split at all, this is a structural guarantee, and this test proves
    it holds in practice, not just in the code's own claim about
    itself."""
    base = make_synthetic_ohlcv(n=900, seed=73, freq="1h")
    # Loose grid values here are deliberate: this test isolates the
    # leakage-safety question ("is the SELECTED config independent of
    # test-region data") from the strategy-quality question (already
    # covered by test_select_frozen_config_rejects_grid_points_below_
    # min_trade_count) -- a stub with genuinely zero real skill gets
    # correctly calibrated toward the base rate by design, so a
    # realistic min_probability like 0.55-0.70 would legitimately find
    # no acceptable config here, which is correct behavior but leaves
    # nothing to compare across variants A/B.
    wf_cfg = cfg.WalkForwardConfig(train_frac=0.5, val_frac=0.25,
                                    min_probability_grid=(0.40, 0.50), buy_threshold_grid=(0.0003, 0.0006),
                                    min_trades_for_calibration=10, min_trades_for_acceptance=1)

    # Variant A: base data, unmodified.
    feat_a = compute_indicators(base)
    train_a, val_a, test_a = wf.chronological_split(feat_a, wf_cfg)

    # Variant B: identical through train+val, then a totally different
    # (differently-seeded) tail spliced in as the "test" region.
    other_tail = make_synthetic_ohlcv(n=len(base), seed=999, freq="1h")
    other_tail.index = base.index  # align so concat/slicing stays simple
    combined = base.copy()
    val_end = int(len(base) * (wf_cfg.train_frac + wf_cfg.val_frac))
    combined.iloc[val_end:] = other_tail.iloc[val_end:]
    feat_b = compute_indicators(combined)
    train_b, val_b, test_b = wf.chronological_split(feat_b, wf_cfg)

    assert train_a.equals(train_b), "train split should be identical by construction"
    assert val_a.equals(val_b), "val split should be identical by construction"
    assert not test_a.equals(test_b), "test split MUST differ for this to be a meaningful test"

    calib_a, dc_a, report_a = wf.select_frozen_config(
        _NoisyStub(seed=5), train_a, val_a, timeframe="1h", asset_class="forex",
        horizon=5, context_length=32, wf=wf_cfg)
    calib_b, dc_b, report_b = wf.select_frozen_config(
        _NoisyStub(seed=5), train_b, val_b, timeframe="1h", asset_class="forex",
        horizon=5, context_length=32, wf=wf_cfg)

    assert np.array_equal(calib_a.bin_edges, calib_b.bin_edges)
    assert np.array_equal(calib_a.calibrated_values, calib_b.calibrated_values)
    assert dc_a.min_probability == dc_b.min_probability
    assert dc_a.buy_threshold == dc_b.buy_threshold
    assert report_a["trials"] == report_b["trials"]


def test_select_frozen_config_grid_is_bounded_not_exhaustive_search():
    """Item 12: "do not search thousands of combinations" -- the number
    of backtest runs select_frozen_config performs must equal exactly
    len(min_probability_grid) * len(buy_threshold_grid), not something
    open-ended."""
    df = make_synthetic_ohlcv(n=700, seed=74, freq="1h")
    feat = compute_indicators(df)
    wf_cfg = cfg.WalkForwardConfig(train_frac=0.5, val_frac=0.3,
                                    min_probability_grid=(0.55, 0.60, 0.65),
                                    buy_threshold_grid=(0.0015, 0.0025),
                                    min_trades_for_calibration=10, min_trades_for_acceptance=1)
    train, val, _ = wf.chronological_split(feat, wf_cfg)
    _, _, report = wf.select_frozen_config(_NoisyStub(seed=1), train, val, timeframe="1h",
                                             asset_class="forex", horizon=5, context_length=32, wf=wf_cfg)
    assert len(report["trials"]) == 3 * 2


def test_select_frozen_config_rejects_grid_points_below_min_trade_count():
    """Item 13: a combination profitable on a handful of trades must not
    be selectable -- verified by setting min_trades_for_acceptance
    absurdly high and confirming selection then fails outright rather
    than quietly accepting a thin result."""
    df = make_synthetic_ohlcv(n=700, seed=75, freq="1h")
    feat = compute_indicators(df)
    wf_cfg = cfg.WalkForwardConfig(train_frac=0.5, val_frac=0.3,
                                    min_probability_grid=(0.55,), buy_threshold_grid=(0.0015,),
                                    min_trades_for_calibration=10, min_trades_for_acceptance=100_000)
    train, val, _ = wf.chronological_split(feat, wf_cfg)
    try:
        wf.select_frozen_config(_NoisyStub(seed=2), train, val, timeframe="1h",
                                 asset_class="forex", horizon=5, context_length=32, wf=wf_cfg)
        raise AssertionError("expected ValueError: no grid point can reach 100,000 trades")
    except ValueError as e:
        assert "no (min_probability, buy_threshold)" in str(e)


def test_baseline_forecasters_run_through_the_real_backtest_engine():
    """Both baselines must be usable exactly like the real forecaster --
    same WalkForwardBacktester, same trading_summary -- for the
    baseline comparison to mean anything."""
    df = make_synthetic_ohlcv(n=300, seed=76, freq="1h")
    feat = compute_indicators(df)
    dc = cfg.DecisionConfig(min_probability=0.55, buy_threshold=0.0005, sell_threshold=-0.0005)
    zero_summary = wf.run_baseline(wf.ZeroSkillForecaster, feat, "1h", "forex", 5, 32, dc)
    assert zero_summary.get("num_trades", 0) == 0, "zero-skill baseline predicts no change -> never clears any edge threshold -> always HOLD"

    momentum_summary = wf.run_baseline(wf.NaiveMomentumForecaster, feat, "1h", "forex", 5, 32, dc)
    assert "num_trades" in momentum_summary  # may or may not trade depending on synthetic noise; just must run cleanly


def test_check_acceptance_rejects_on_insufficient_trades():
    summary = {"num_trades": 5, "net_pnl": 0.05, "profit_factor": 1.5, "max_drawdown": -0.02}
    result = wf.check_acceptance(summary, {}, pd.Series([1.0, 1.01, 1.05]),
                                  wf=cfg.WalkForwardConfig(min_trades_for_acceptance=30))
    assert result["accepted"] is False
    assert result["checks"]["sufficient_trade_count"] is False


def test_check_acceptance_rejects_negative_net_return_even_with_many_trades():
    summary = {"num_trades": 100, "net_pnl": -0.05, "profit_factor": 0.8, "max_drawdown": -0.02}
    result = wf.check_acceptance(summary, {}, pd.Series([1.0, 0.99, 0.95]),
                                  wf=cfg.WalkForwardConfig(min_trades_for_acceptance=30))
    assert result["accepted"] is False
    assert result["checks"]["positive_net_return"] is False
    assert result["checks"]["profit_factor_above_1"] is False


def test_check_acceptance_rejects_when_baseline_wins():
    summary = {"num_trades": 50, "net_pnl": 0.01, "profit_factor": 1.1, "max_drawdown": -0.02}
    baselines = {"momentum": {"net_pnl": 0.05}}
    result = wf.check_acceptance(summary, baselines, pd.Series([1.0, 1.005, 1.01]),
                                  wf=cfg.WalkForwardConfig(min_trades_for_acceptance=30))
    assert result["checks"]["beats_baseline_momentum"] is False
    assert result["accepted"] is False


def test_check_acceptance_can_accept_a_genuinely_good_result():
    idx = pd.date_range("2024-01-01", periods=200, freq="1D")  # ~6.7 months -> several distinct 30D bins
    equity = pd.Series(np.linspace(1.0, 1.15, 200), index=idx)  # smooth, steady growth -- no single-period dependency
    summary = {"num_trades": 50, "net_pnl": 0.15, "profit_factor": 1.8, "max_drawdown": -0.03}
    baselines = {"zero_skill": {"net_pnl": 0.0}, "momentum": {"net_pnl": 0.02}}
    result = wf.check_acceptance(summary, baselines, equity, wf=cfg.WalkForwardConfig(min_trades_for_acceptance=30))
    assert result["accepted"] is True, result["checks"]


def test_check_acceptance_rejects_when_one_period_explains_most_profit():
    idx = pd.date_range("2024-01-01", periods=200, freq="1D")  # ~6.7 months -> several distinct 30D bins
    vals = np.ones(200)
    vals[150:] = 1.30  # one sharp jump near the end, flat everywhere else
    equity = pd.Series(vals, index=idx)
    summary = {"num_trades": 50, "net_pnl": 0.30, "profit_factor": 5.0, "max_drawdown": -0.01}
    result = wf.check_acceptance(summary, {}, equity, wf=cfg.WalkForwardConfig(min_trades_for_acceptance=30,
                                                                                 max_single_period_profit_share=0.70))
    assert result["checks"]["not_single_period_dependent"] is False
    assert result["accepted"] is False