NexQuant/test/backtesting/test_vbt_backtest.py

"""
Oracle + consistency tests for the unified backtest engine.

Every metric is checked against a value we can reproduce by hand (or via
vectorbt). Same ``(close, signal)`` inputs must yield the same numbers no
matter which call-site is used.
"""
from __future__ import annotations

import numpy as np
import pandas as pd
import pytest

from rdagent.components.backtesting.vbt_backtest import (
    DEFAULT_BARS_PER_YEAR,
    backtest_from_forward_returns,
    backtest_signal,
)


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def flat_close() -> pd.Series:
    """Constant close: any signal should produce zero returns."""
    idx = pd.date_range("2024-01-01", periods=1000, freq="1min")
    return pd.Series(100.0, index=idx)


@pytest.fixture
def trending_close() -> pd.Series:
    """Monotonically increasing close (0.01% per bar)."""
    idx = pd.date_range("2024-01-01", periods=1000, freq="1min")
    return pd.Series(100.0 * (1.0001 ** np.arange(1000)), index=idx)


@pytest.fixture
def random_close() -> pd.Series:
    np.random.seed(42)
    idx = pd.date_range("2024-01-01", periods=5000, freq="1min")
    return pd.Series(100.0 + np.random.randn(5000).cumsum() * 0.05, index=idx)


# ---------------------------------------------------------------------------
# Oracle tests — numbers we can verify by hand
# ---------------------------------------------------------------------------
def test_flat_close_all_long_returns_zero(flat_close):
    """Constant price → strategy returns are exactly −cost per position-change bar."""
    signal = pd.Series(1.0, index=flat_close.index)
    r = backtest_signal(flat_close, signal, txn_cost_bps=1.5)

    # Only one position change (bar 0 → long 1.0): total cost = 1 * 1.5e-4.
    assert r["status"] == "success"
    assert r["total_return"] == pytest.approx(-1.5e-4, abs=1e-8)
    assert r["n_trades"] == 1  # one open trade (still in position at end)
    # Flat price → zero variance → sharpe cannot be computed; returned as 0.
    # But cost introduces a tiny constant return, so std is 0 and sharpe is 0.


def test_trending_close_always_long_matches_price_return(trending_close):
    """
    position=+1 always → strategy return per bar ≈ bar_ret (minus one-time cost).
    total_return should equal (close[-1]/close[0]) - 1 minus the entry cost.
    """
    signal = pd.Series(1.0, index=trending_close.index)
    r = backtest_signal(trending_close, signal, txn_cost_bps=1.5)

    price_tr = trending_close.iloc[-1] / trending_close.iloc[0] - 1
    # Manual: product over (1 + bar_ret - one-time cost at bar 0).
    bar_ret = trending_close.pct_change().fillna(0)
    position = signal.shift(1).fillna(0)
    position_change = position.diff().abs().fillna(position.abs())
    expected_total = float(((1 + position * bar_ret - position_change * 1.5e-4).prod()) - 1)

    assert r["total_return"] == pytest.approx(expected_total, rel=1e-9)
    # Within 1 bp of the raw price trend.
    assert abs(r["total_return"] - price_tr) < 2e-4


def test_always_flat_returns_zero(random_close):
    signal = pd.Series(0.0, index=random_close.index)
    r = backtest_signal(random_close, signal, txn_cost_bps=1.5)

    assert r["total_return"] == 0.0
    assert r["sharpe"] == 0.0
    assert r["max_drawdown"] == 0.0
    assert r["n_trades"] == 0
    assert r["n_position_changes"] == 0


def test_sharpe_annualization_uses_1min_bars(random_close):
    """Sharpe must use √(252*1440), not √252."""
    np.random.seed(0)
    signal = pd.Series(np.random.choice([-1, 0, 1], size=len(random_close)), index=random_close.index)
    r = backtest_signal(random_close, signal, txn_cost_bps=0.0)  # no cost → clean check

    # Reproduce manually.
    bar_ret = random_close.pct_change().fillna(0)
    position = signal.astype(float).shift(1).fillna(0)
    strat_ret = position * bar_ret
    expected_sharpe = strat_ret.mean() / strat_ret.std() * np.sqrt(DEFAULT_BARS_PER_YEAR)

    assert r["sharpe"] == pytest.approx(expected_sharpe, rel=1e-9)
    assert r["bars_per_year"] == 252 * 1440


def test_txn_cost_applied_per_position_change(random_close):
    """With 50% of bars flipping, cost ≈ 0.5 * |Δposition|_mean * txn_cost_bps."""
    idx = random_close.index
    # Alternate every bar: -1, 1, -1, 1, ...
    signal = pd.Series([(-1.0) ** i for i in range(len(idx))], index=idx)

    zero_cost = backtest_signal(random_close, signal, txn_cost_bps=0.0)
    with_cost = backtest_signal(random_close, signal, txn_cost_bps=10.0)  # 10 bps

    # Cost difference = |Δposition|.sum() * 10e-4, summed over bars.
    # Alternating ±1 → |Δposition| = 2 per bar (except first: 1).
    bar_ret_diff = zero_cost["total_return"] - with_cost["total_return"]
    assert bar_ret_diff > 0  # with cost must be worse


def test_drawdown_never_clipped():
    """
    A single blow-up bar must not be silently absorbed. Old code clipped
    returns to ±10%; the new engine reports the true drawdown and flags.
    """
    idx = pd.date_range("2024-01-01", periods=500, freq="1min")
    # Prices: gentle rise, then a 20% crash at bar 250, then recovery.
    closes = np.concatenate(
        [np.linspace(100, 101, 250), [80.0], np.linspace(80, 85, 249)]
    )
    close = pd.Series(closes, index=idx)
    signal = pd.Series(1.0, index=idx)

    r = backtest_signal(close, signal, txn_cost_bps=0.0)

    assert r["max_drawdown"] < -0.15  # real DD preserved
    assert "data_quality_flag" in r
    assert "extreme_returns" in r["data_quality_flag"]


def test_forward_returns_ic_computation():
    np.random.seed(7)
    idx = pd.date_range("2024-01-01", periods=2000, freq="1min")
    noise = pd.Series(np.random.randn(2000), index=idx)
    close = pd.Series(100 + noise.cumsum() * 0.01, index=idx)
    fwd = close.pct_change().shift(-1).fillna(0)

    # sign(fwd) correlates with fwd at ~√(2/π) ≈ 0.798 for Gaussian returns.
    sign_signal = pd.Series(np.sign(fwd), index=idx).replace(0, 1)
    r_sign = backtest_signal(close, sign_signal, forward_returns=fwd, txn_cost_bps=0.0)
    assert r_sign["ic"] is not None
    assert 0.7 < r_sign["ic"] < 0.85

    # Passing fwd itself as the signal (clipped to [-1,1]) yields corr = 1.0.
    fwd_signal = fwd.clip(-1, 1)
    r_perfect = backtest_signal(close, fwd_signal, forward_returns=fwd, txn_cost_bps=0.0)
    assert r_perfect["ic"] == pytest.approx(1.0, abs=1e-9)


def test_trade_count_matches_epoch_count():
    """n_trades must equal the number of distinct non-flat position epochs."""
    idx = pd.date_range("2024-01-01", periods=10, freq="1min")
    # Position: 0, 1, 1, 0, -1, -1, 0, 1, 0, 0  → 3 trades
    signal = pd.Series([0, 1, 1, 0, -1, -1, 0, 1, 0, 0], index=idx).astype(float)
    close = pd.Series(np.linspace(100, 101, 10), index=idx)

    r = backtest_signal(close, signal, txn_cost_bps=0.0)
    assert r["n_trades"] == 3


def test_win_rate_uses_per_trade_pnl():
    """Win rate must reflect per-trade P&L, not per-bar returns."""
    idx = pd.date_range("2024-01-01", periods=20, freq="1min")
    # Craft a scenario: 2 clearly winning long trades, 1 losing.
    close = pd.Series(
        [100, 101, 102, 103, 102, 101, 100, 99, 98, 99,   # bars 0-9
         100, 101, 102, 103, 104, 103, 102, 101, 100, 100], # bars 10-19
        index=idx,
    ).astype(float)
    # Trade 1: long bars 1..3 (price 101→103 = +2, win)
    # Trade 2: long bars 5..7 (price 101→99  = -2, loss)
    # Trade 3: long bars 11..14 (price 101→104 = +3, win)
    sig = pd.Series(0, index=idx).astype(float)
    sig.iloc[1:4] = 1
    sig.iloc[5:8] = 1
    sig.iloc[11:15] = 1

    r = backtest_signal(close, sig, txn_cost_bps=0.0)
    # Due to the shift(1) lag, actual entry/exit shifts by 1 bar — but the
    # number of epochs and their sign are preserved.
    assert r["n_trades"] == 3
    assert r["win_rate"] == pytest.approx(2 / 3, abs=1e-9)


# ---------------------------------------------------------------------------
# Consistency tests — all four call sites produce identical numbers
# ---------------------------------------------------------------------------
def test_orchestrator_path_matches_direct_call(random_close):
    """Orchestrator's evaluate_strategy should produce the same bt numbers."""
    np.random.seed(11)
    idx = random_close.index
    signal = pd.Series(np.random.choice([-1, 0, 1], size=len(idx)), index=idx).astype(float)

    direct = backtest_signal(random_close, signal, txn_cost_bps=1.5)

    # Reproduce the orchestrator's call signature.
    from rdagent.components.backtesting.vbt_backtest import backtest_signal as orch_bt
    orch = orch_bt(close=random_close.reindex(signal.index).ffill(), signal=signal,
                   txn_cost_bps=1.5, freq="1min")

    for key in ("sharpe", "max_drawdown", "total_return", "n_trades", "win_rate"):
        assert direct[key] == orch[key], f"{key}: {direct[key]} != {orch[key]}"


def test_factor_backtester_wrapper_consistent_with_engine():
    """Legacy FactorBacktester must return the same IC/Sharpe as the unified engine."""
    from rdagent.components.backtesting.backtest_engine import FactorBacktester

    np.random.seed(99)
    n = 500
    idx = pd.date_range("2024-01-01", periods=n, freq="1min")
    factor = pd.Series(np.random.randn(n), index=idx)
    fwd_ret = pd.Series(factor.values * 0.001 + np.random.randn(n) * 0.01, index=idx)

    direct = backtest_from_forward_returns(factor, fwd_ret, txn_cost_bps=1.5)

    fb = FactorBacktester()
    fb.results_path = fb.results_path / "_test_tmp"
    fb.results_path.mkdir(parents=True, exist_ok=True)
    legacy = fb.run_backtest(factor, fwd_ret, "TestFactor", transaction_cost=0.00015)

    assert legacy["sharpe_ratio"] == pytest.approx(direct["sharpe"], rel=1e-9)
    assert legacy["max_drawdown"] == pytest.approx(direct["max_drawdown"], rel=1e-9)
    assert legacy["ic"] == pytest.approx(direct["ic"], rel=1e-9)


# ---------------------------------------------------------------------------
# Cross-check vs vectorbt simulation
# ---------------------------------------------------------------------------
def test_vbt_cross_check_matches_within_tolerance(random_close):
    """
    Our manual total_return and vbt's compounded total_return should agree
    within a few basis points on a realistic 1-min scenario.
    """
    np.random.seed(3)
    idx = random_close.index
    fast = random_close.rolling(10).mean()
    slow = random_close.rolling(50).mean()
    signal = pd.Series(
        np.where(fast > slow, 1.0, np.where(fast < slow, -1.0, 0.0)), index=idx
    )

    r = backtest_signal(random_close, signal, txn_cost_bps=1.5, cross_check=True)
    assert "vbt_total_return" in r
    if r["vbt_total_return"] is not None:
        # Manual: simple (1+position*ret) compounding, one fixed leverage.
        # vbt: target-percent rebalancing on every bar, leverage drifts with equity.
        # The two can differ by O(|return|^2) per bar on flipping strategies;
        # we only require the sign and rough magnitude to agree.
        assert abs(r["total_return"] - r["vbt_total_return"]) < 0.05