tests/test_engine.py

"""Phase 1 acceptance: the six known-answer tests, plus engine invariants.

The numbered tests map one-to-one onto the build spec's list. They are the gate
for the whole engine; if any of them fails, nothing downstream is trustworthy.
"""

from __future__ import annotations

import numpy as np
import pandas as pd
import pytest

from src import strategies
from src.engine import (
    BacktestConfig,
    Costs,
    EngineError,
    LookaheadError,
    Sizing,
    Stops,
    StrategyOutput,
    Validation,
    assert_causal,
    build_validation_plan,
    run_backtest,
)

BARS_PER_YEAR = 365.0


# --------------------------------------------------------------------------
# Deterministic synthetic price series
# --------------------------------------------------------------------------


def prices(n=400, seed=7, start="2022-01-01", freq="D", trend=0.0004, vol=0.02):
    rng = np.random.default_rng(seed)
    steps = rng.normal(trend, vol, n)
    close = 100.0 * np.exp(np.cumsum(steps))
    idx = pd.date_range(start, periods=n, freq=freq, tz="UTC")
    close = pd.Series(close, index=idx)
    open_ = close.shift(1).fillna(close.iloc[0] * 0.999)
    high = pd.concat([open_, close], axis=1).max(axis=1) * 1.004
    low = pd.concat([open_, close], axis=1).min(axis=1) * 0.996
    return pd.DataFrame({
        "open": open_, "high": high, "low": low, "close": close,
        "volume": pd.Series(rng.uniform(800, 1200, n), index=idx),
        "source": "synthetic",
    })


def linear_prices(n=100, start="2022-01-01"):
    """Monotone series -- makes hand-computed expectations trivial."""
    idx = pd.date_range(start, periods=n, freq="D", tz="UTC")
    close = pd.Series(np.linspace(100.0, 200.0, n), index=idx)
    open_ = close * 0.995
    return pd.DataFrame({
        "open": open_, "high": close * 1.01, "low": open_ * 0.99,
        "close": close, "volume": 1000.0, "source": "synthetic",
    })


def no_costs() -> Costs:
    return Costs(enabled=False)


def cfg(**kw) -> BacktestConfig:
    base = dict(
        asset="BTC-USD", timeframe="1d", strategy="Buy & Hold (benchmark)",
        costs=no_costs(), validation=Validation(mode="none"), init_cash=100_000.0,
    )
    base.update(kw)
    return BacktestConfig(**base)


# ==========================================================================
# KNOWN-ANSWER TEST 1
# Buy & Hold with zero costs reproduces the asset's return over the period.
# ==========================================================================


@pytest.mark.parametrize("px", [linear_prices(), prices(n=300, seed=3)])
def test_ka1_buy_and_hold_zero_costs_matches_asset_return(px):
    out = strategies.buy_and_hold(px)
    res = run_backtest(px, out, cfg(), bars_per_year=BARS_PER_YEAR)

    # Execution is next-bar-open, so the position is opened at the second bar's
    # open. That fill price is the honest basis for "the asset's return".
    expected = float(px["close"].iloc[-1] / px["open"].iloc[1] - 1.0)
    assert res.metrics_all.total_return == pytest.approx(expected, abs=1e-9)
    assert len(res.trades) == 1
    assert res.trades["costs"].sum() == pytest.approx(0.0, abs=1e-12)


def test_ka1_entry_fill_is_the_second_bars_open_not_the_first():
    px = linear_prices()
    res = run_backtest(px, strategies.buy_and_hold(px), cfg(), bars_per_year=BARS_PER_YEAR)
    entry_px = float(res.trades["entry_px"].iloc[0])
    assert entry_px == pytest.approx(float(px["open"].iloc[1]), abs=1e-9)
    assert entry_px != pytest.approx(float(px["open"].iloc[0]), abs=1e-9)
    assert res.trades["entry_ts"].iloc[0] == px.index[1]


# ==========================================================================
# KNOWN-ANSWER TEST 2
# A deliberately lookahead-biased strategy is caught structurally.
# ==========================================================================


def peeking_strategy(px: pd.DataFrame) -> StrategyOutput:
    """Cheats: decides at bar t using bar t+1's close."""
    future = px["close"].shift(-1)
    entries = (future > px["close"]).fillna(False)
    exits = (future <= px["close"]).fillna(False)
    return StrategyOutput(entries=entries, exits=exits)


def subtle_peeking_strategy(px: pd.DataFrame) -> StrategyOutput:
    """Cheats less obviously: a centred rolling mean leaks the future."""
    centred = px["close"].rolling(11, center=True, min_periods=1).mean()
    entries = (px["close"] > centred).fillna(False)
    exits = (px["close"] <= centred).fillna(False)
    return StrategyOutput(entries=entries, exits=exits)


def test_ka2_obvious_lookahead_is_caught():
    px = prices(n=300, seed=11)
    with pytest.raises(LookaheadError, match="after the bar it acts on"):
        assert_causal(peeking_strategy, px)


def test_ka2_subtle_lookahead_is_caught():
    px = prices(n=300, seed=12)
    with pytest.raises(LookaheadError):
        assert_causal(subtle_peeking_strategy, px)


def test_ka2_global_normalisation_lookahead_is_caught():
    """Scaling by the full-sample max leaks the future into every early bar."""
    def global_scaled(px):
        z = px["close"] / px["close"].max()
        entries = (z > 0.8).fillna(False)
        return StrategyOutput(entries=entries, exits=(~entries).fillna(False))

    with pytest.raises(LookaheadError):
        assert_causal(global_scaled, prices(n=300, seed=13))


@pytest.mark.parametrize("name", [
    "Buy & Hold (benchmark)", "SMA Crossover", "RSI Mean Reversion",
    "Bollinger Breakout", "MACD Momentum", "Sentiment-Gated Momentum",
])
def test_ka2_every_shipped_preset_is_causal(name):
    px = prices(n=400, seed=5)
    assert_causal(lambda p: strategies.build(name, p, strategies.defaults_for(name)), px)


def test_ka2_forecast_follower_is_causal_against_stored_signals():
    px = prices(n=300, seed=17)
    sig = pd.DataFrame({
        "q10": px["close"] * 0.97, "q50": px["close"] * 1.01, "q90": px["close"] * 1.05,
    }, index=px.index)
    # Signals are pinned, so only the price path is perturbed -- exactly the
    # situation the store creates when a cached forecast slice is replayed.
    assert_causal(lambda p: strategies.forecast_follower(p, {"threshold": 0.005}, sig), px)


def test_ka2_engine_refuses_same_bar_fills():
    px = linear_prices()
    with pytest.raises(EngineError, match="non-negotiable"):
        run_backtest(px, strategies.buy_and_hold(px), cfg(fill="same_bar_close"),
                     bars_per_year=BARS_PER_YEAR)


def test_ka2_decisions_cannot_be_shifted_twice():
    from src.engine import _shift_decisions

    px = linear_prices()
    once = _shift_decisions(strategies.buy_and_hold(px))
    with pytest.raises(EngineError, match="already shifted"):
        _shift_decisions(once)


# ==========================================================================
# KNOWN-ANSWER TEST 3
# Zero-cost vs costed runs differ by exactly the modeled costs on the trades.
# ==========================================================================


def alternating(px: pd.DataFrame, hold=7, gap=5) -> StrategyOutput:
    """Deterministic in-and-out signal, independent of price."""
    entries = pd.Series(False, index=px.index)
    exits = pd.Series(False, index=px.index)
    i = 10
    while i + hold < len(px) - 2:
        entries.iloc[i] = True
        exits.iloc[i + hold] = True
        i += hold + gap
    return StrategyOutput(entries=entries, exits=exits)


def test_ka3_cost_difference_equals_sum_of_trade_costs():
    px = prices(n=300, seed=21)
    out = alternating(px)

    # Fixed unit sizing keeps costs additive: with percent sizing, fees change
    # position size and the difference compounds instead of summing.
    sizing = Sizing(mode="fixed_units", units=1.0)
    free = run_backtest(px, alternating(px),
                        cfg(costs=no_costs(), sizing=sizing), bars_per_year=BARS_PER_YEAR)
    paid = run_backtest(px, out,
                        cfg(costs=Costs(enabled=True, commission_bps=10.0,
                                        slippage_bps=5.0, slippage_model="fixed"),
                            sizing=sizing), bars_per_year=BARS_PER_YEAR)

    assert len(free.trades) == len(paid.trades) > 3

    pnl_gap = float(free.trades["net_pnl"].sum() - paid.trades["net_pnl"].sum())
    modeled = float(paid.trades["costs"].sum())
    assert pnl_gap == pytest.approx(modeled, rel=1e-9, abs=1e-6)


def test_ka3_gross_minus_costs_equals_net_on_every_trade():
    px = prices(n=300, seed=22)
    res = run_backtest(px, alternating(px),
                       cfg(costs=Costs(enabled=True, commission_bps=8.0, slippage_bps=4.0),
                           sizing=Sizing(mode="fixed_units", units=1.0)),
                       bars_per_year=BARS_PER_YEAR)
    residual = res.trades["gross_pnl"] - res.trades["costs"] - res.trades["net_pnl"]
    assert np.allclose(residual.to_numpy(), 0.0, atol=1e-9)


def test_ka3_costs_are_on_by_default():
    c = Costs()
    assert c.enabled
    assert c.commission_rate > 0 and c.slippage_rate > 0
    assert BacktestConfig().costs.enabled


def test_ka3_more_slippage_never_helps():
    px = prices(n=300, seed=23)
    sizing = Sizing(mode="fixed_units", units=1.0)
    returns = []
    for bps in (0.0, 5.0, 20.0):
        r = run_backtest(px, alternating(px),
                         cfg(costs=Costs(enabled=True, commission_bps=0.0, slippage_bps=bps),
                             sizing=sizing), bars_per_year=BARS_PER_YEAR)
        returns.append(float(r.trades["net_pnl"].sum()))
    assert returns[0] > returns[1] > returns[2]


# ==========================================================================
# KNOWN-ANSWER TEST 4
# A strategy with no signals produces flat equity and zero trades.
# ==========================================================================


def test_ka4_no_signals_gives_flat_equity_and_no_trades():
    px = prices(n=200, seed=31)
    silent = StrategyOutput(
        entries=pd.Series(False, index=px.index),
        exits=pd.Series(False, index=px.index),
    )
    res = run_backtest(px, silent, cfg(), bars_per_year=BARS_PER_YEAR)

    assert len(res.trades) == 0
    assert res.metrics_all.trade_count == 0
    assert res.equity.nunique() == 1
    assert float(res.equity.iloc[-1]) == pytest.approx(100_000.0, abs=1e-9)
    assert res.metrics_all.total_return == pytest.approx(0.0, abs=1e-12)
    assert res.metrics_all.max_drawdown == pytest.approx(0.0, abs=1e-12)
    assert res.metrics_all.sharpe == 0.0
    assert res.metrics_all.win_rate == 0.0
    assert res.metrics_all.profit_factor == 0.0
    assert res.costs_paid == 0.0


def test_ka4_flat_run_with_costs_still_pays_nothing():
    px = prices(n=200, seed=32)
    silent = StrategyOutput(entries=pd.Series(False, index=px.index),
                            exits=pd.Series(False, index=px.index))
    res = run_backtest(px, silent, cfg(costs=Costs(enabled=True)), bars_per_year=BARS_PER_YEAR)
    assert res.costs_paid == 0.0
    assert float(res.equity.iloc[-1]) == pytest.approx(100_000.0, abs=1e-9)


# ==========================================================================
# KNOWN-ANSWER TEST 5
# Walk-forward boundaries never overlap; the holdout is never selectable.
# ==========================================================================


def test_ka5_walk_forward_windows_never_overlap_train_and_test():
    idx = pd.date_range("2021-01-01", "2026-01-01", freq="D", tz="UTC")
    plan = build_validation_plan(idx, Validation(
        mode="walk_forward", train_months=12, test_months=3, roll_months=3,
    ))
    assert len(plan.windows) >= 4
    for w in plan.windows:
        assert w.train_start < w.train_end
        assert w.test_start > w.train_end, f"window {w.idx} overlaps"
        assert w.test_start <= w.test_end


def test_ka5_consecutive_test_windows_do_not_overlap_each_other():
    idx = pd.date_range("2021-01-01", "2026-01-01", freq="D", tz="UTC")
    plan = build_validation_plan(idx, Validation(
        mode="walk_forward", train_months=12, test_months=3, roll_months=3,
    ))
    for a, b in zip(plan.windows, plan.windows[1:]):
        assert b.test_start > a.test_end


def test_ka5_holdout_is_excluded_from_the_parameter_selection_index():
    idx = pd.date_range("2021-01-01", "2026-01-01", freq="D", tz="UTC")
    plan = build_validation_plan(idx, Validation(mode="holdout", holdout_months=6))
    assert plan.holdout_start is not None

    selectable = plan.selectable_index()
    holdout = idx[idx >= plan.holdout_start]

    assert len(holdout) > 0
    assert len(set(selectable) & set(holdout)) == 0
    assert selectable.max() < holdout.min()


def test_ka5_walk_forward_windows_never_reach_into_the_holdout():
    idx = pd.date_range("2021-01-01", "2026-01-01", freq="D", tz="UTC")
    v = Validation(mode="walk_forward", train_months=12, test_months=3,
                   roll_months=3, holdout_months=6)
    plan = build_validation_plan(idx, v)
    assert plan.holdout_start is not None
    assert len(plan.windows) > 0
    for w in plan.windows:
        assert w.test_end < plan.holdout_start
        assert w.train_end < plan.holdout_start


def test_ka5_holdout_bars_are_labelled_holdout_not_oos():
    px = prices(n=900, seed=41, start="2022-01-01")
    res = run_backtest(px, strategies.buy_and_hold(px),
                       cfg(validation=Validation(mode="holdout", holdout_months=6)),
                       bars_per_year=BARS_PER_YEAR)
    assert res.metrics_holdout is not None
    assert res.metrics_holdout.bars > 0
    assert res.plan.segment_of(px.index[-1]) == "holdout"
    assert res.plan.segment_of(px.index[0]) == "IS"


def test_ka5_split_mode_partitions_every_bar_exactly_once():
    """With no holdout reserved, IS and OOS must tile the whole period."""
    px = prices(n=400, seed=42)
    res = run_backtest(px, strategies.buy_and_hold(px),
                       cfg(validation=Validation(mode="split", split_frac=0.7,
                                                 holdout_months=0)),
                       bars_per_year=BARS_PER_YEAR)
    assert res.metrics_holdout is None
    assert res.metrics_is.bars + res.metrics_oos.bars == len(px)
    assert res.metrics_is.bars == pytest.approx(280, abs=1)


def test_ka5_holdout_is_carved_out_of_split_mode_too():
    """A reserved holdout is honoured regardless of how the rest is divided."""
    px = prices(n=900, seed=43, start="2022-01-01")
    res = run_backtest(px, strategies.buy_and_hold(px),
                       cfg(validation=Validation(mode="split", split_frac=0.7,
                                                 holdout_months=6)),
                       bars_per_year=BARS_PER_YEAR)
    assert res.metrics_holdout is not None and res.metrics_holdout.bars > 0
    total = res.metrics_is.bars + res.metrics_oos.bars + res.metrics_holdout.bars
    assert total == len(px)


# ==========================================================================
# KNOWN-ANSWER TEST 6
# Same config + same data => bit-identical results.
# ==========================================================================


def _signature(res) -> tuple:
    return (
        tuple(np.round(res.equity.to_numpy(), 12)),
        tuple(np.round(res.trades["net_pnl"].to_numpy(), 12)),
        tuple(np.round(res.trades["costs"].to_numpy(), 12)),
        res.metrics_all.total_return,
        res.metrics_all.sharpe,
        res.metrics_oos.sharpe,
    )


def test_ka6_repeated_runs_are_bit_identical():
    px = prices(n=400, seed=51)
    c = cfg(strategy="SMA Crossover", costs=Costs(enabled=True),
            validation=Validation(mode="walk_forward"))
    params = {"fast_ma": 20, "slow_ma": 50}

    runs = [
        run_backtest(px, strategies.build("SMA Crossover", px, params), c,
                     bars_per_year=BARS_PER_YEAR)
        for _ in range(3)
    ]
    assert _signature(runs[0]) == _signature(runs[1]) == _signature(runs[2])


def test_ka6_fingerprint_is_stable_and_config_sensitive():
    a = cfg(strategy="SMA Crossover", params={"fast_ma": 20})
    b = cfg(strategy="SMA Crossover", params={"fast_ma": 20})
    c = cfg(strategy="SMA Crossover", params={"fast_ma": 21})
    assert a.fingerprint() == b.fingerprint()
    assert a.fingerprint() != c.fingerprint()


def test_ka6_cost_change_changes_the_fingerprint():
    a = cfg(costs=Costs(enabled=True, commission_bps=10.0))
    b = cfg(costs=Costs(enabled=True, commission_bps=11.0))
    assert a.fingerprint() != b.fingerprint()


# ==========================================================================
# Engine invariants beyond the six
# ==========================================================================


def test_trade_list_carries_every_documented_column():
    px = prices(n=300, seed=61)
    res = run_backtest(px, alternating(px), cfg(costs=Costs(enabled=True)),
                       bars_per_year=BARS_PER_YEAR)
    for col in ("id", "entry_ts", "exit_ts", "side", "entry_px", "exit_px", "size",
                "gross_pnl", "costs", "net_pnl", "r_multiple", "mae", "mfe",
                "duration_bars", "trigger", "segment"):
        assert col in res.trades.columns


def test_mae_is_never_positive_and_mfe_never_negative_for_longs():
    px = prices(n=300, seed=62)
    res = run_backtest(px, alternating(px), cfg(), bars_per_year=BARS_PER_YEAR)
    longs = res.trades[res.trades["side"] == "long"]
    assert (longs["mae"] <= 1e-12).all()
    assert (longs["mfe"] >= -1e-12).all()


def test_trigger_reason_is_populated_from_the_strategy():
    px = prices(n=400, seed=63)
    out = strategies.build("SMA Crossover", px, {"fast_ma": 10, "slow_ma": 30})
    res = run_backtest(px, out, cfg(strategy="SMA Crossover"), bars_per_year=BARS_PER_YEAR)
    assert len(res.trades) > 0
    assert (res.trades["trigger"].str.len() > 0).all()
    assert "crossed above" in res.trades["trigger"].iloc[0]


def test_index_mismatch_is_rejected():
    px = prices(n=100, seed=64)
    bad = StrategyOutput(
        entries=pd.Series(False, index=px.index[:50]),
        exits=pd.Series(False, index=px.index[:50]),
    )
    with pytest.raises(EngineError, match="index does not match"):
        run_backtest(px, bad, cfg(), bars_per_year=BARS_PER_YEAR)


def test_empty_prices_are_rejected():
    empty = pd.DataFrame(columns=["open", "high", "low", "close", "volume"])
    out = StrategyOutput(entries=pd.Series(dtype=bool), exits=pd.Series(dtype=bool))
    with pytest.raises(EngineError, match="no price data"):
        run_backtest(empty, out, cfg(), bars_per_year=BARS_PER_YEAR)


def test_stops_are_applied_when_configured():
    px = prices(n=400, seed=65, trend=-0.002, vol=0.03)
    with_stop = run_backtest(px, alternating(px),
                             cfg(stops=Stops(sl_pct=0.02)), bars_per_year=BARS_PER_YEAR)
    without = run_backtest(px, alternating(px), cfg(), bars_per_year=BARS_PER_YEAR)
    assert with_stop.trades["net_pnl"].min() > without.trades["net_pnl"].min()


def test_r_multiple_uses_the_configured_stop_distance():
    px = prices(n=300, seed=66)
    res = run_backtest(px, alternating(px), cfg(stops=Stops(sl_pct=0.05)),
                       bars_per_year=BARS_PER_YEAR)
    t = res.trades.iloc[0]
    expected = t["net_pnl"] / (0.05 * t["entry_px"] * t["size"])
    assert t["r_multiple"] == pytest.approx(expected, rel=1e-9)


def test_unavailable_presets_refuse_to_run():
    px = prices(n=100, seed=67)
    with pytest.raises(ValueError, match="never executes untrusted code"):
        strategies.build("Custom (code)", px)
    with pytest.raises(ValueError, match="second leg"):
        strategies.build("Pairs Trading", px)


def test_forecast_follower_requires_signals():
    px = prices(n=100, seed=68)
    with pytest.raises(ValueError, match="needs stored model signals"):
        strategies.build("Chronos Forecast Follower", px, {}, signals=None)


def test_default_run_completes_well_under_two_seconds():
    """Perf budget: a 3-year daily run must leave room for chart building."""
    px = prices(n=365 * 3, seed=69)
    out = strategies.build("SMA Crossover", px, {"fast_ma": 20, "slow_ma": 50})
    res = run_backtest(px, out, cfg(strategy="SMA Crossover", costs=Costs(enabled=True),
                                    validation=Validation(mode="walk_forward")),
                       bars_per_year=BARS_PER_YEAR)
    assert res.elapsed_s < 2.0, f"engine took {res.elapsed_s:.2f}s"