src/transformations/wavelets/backtest.py

"""Walk-forward backtest engine for the MODWT mid-band strategy.

Every signal value is generated using only information available at that point
in time:
  - Price window ends at t-1 (strictly past)
  - Signal applied to return from t to t+1 (via .shift(1))
  - Vol estimator uses 60-day rolling realized vol of PAST returns

Transaction costs: `cost_bps` basis points per side on absolute notional
turnover, deducted from each rebalance return.
"""
from __future__ import annotations
import numpy as np
import pandas as pd

from .signal import compute_signal, volatility_target
from .stats import Stats, perf

# ─────────────────────────── constants ───────────────────────────────────────

LOOKBACK = 1024      # bars (~4 years daily); >> 8× max wavelet scale (128 bars)
STEP = 5             # weekly rebalance
VOL_WINDOW = 60      # rolling realized vol window (bars)
VOL_TARGET = 0.10    # annualized target vol
MAX_LEVERAGE = 2.0
WAVELET = "sym8"
LEVEL = 6
SIG_LEVELS = [4, 5]
SLOPE_WINDOW = 40
COST_BPS = 1.0       # per side in basis points


# ─────────────────────────── SMA benchmark ───────────────────────────────────

def _sma_signal(prices: pd.Series, fast: int = 50, slow: int = 200) -> pd.Series:
    """50/200 SMA crossover signal: +1 above, -1 below, aligned to prices."""
    sma_fast = prices.rolling(fast).mean()
    sma_slow = prices.rolling(slow).mean()
    raw = np.sign(sma_fast - sma_slow).fillna(0)
    return raw


# ─────────────────────────── main loop ───────────────────────────────────────

def run_backtest(
    price_series: pd.Series,
    lookback: int = LOOKBACK,
    step: int = STEP,
    wavelet: str = WAVELET,
    level: int = LEVEL,
    sig_levels: list[int] | None = None,
    slope_window: int = SLOPE_WINDOW,
    vol_window: int = VOL_WINDOW,
    vol_target: float = VOL_TARGET,
    max_leverage: float = MAX_LEVERAGE,
    cost_bps: float = COST_BPS,
) -> dict:
    """Walk-forward MODWT mid-band strategy backtest.

    Args:
        price_series: pd.Series of prices with DatetimeIndex, daily frequency.
        lookback:     bars of history used per MODWT window.
        step:         rebalance frequency in bars.
        wavelet:      PyWavelets wavelet.
        level:        MODWT depth.
        sig_levels:   detail levels to combine into mid-band signal.
        slope_window: safe-series slope estimation window.
        vol_window:   rolling realized vol window for vol targeting.
        cost_bps:     transaction cost per side in basis points.

    Returns:
        dict with keys:
          "strategy_stats":   Stats for MODWT strategy
          "bh_stats":         Stats for buy-and-hold
          "sma_stats":        Stats for 50/200 SMA cross benchmark
          "positions":        pd.Series of daily positions
          "raw_signals":      pd.Series of ±1/0 raw signals (weekly)
          "equity_curve":     pd.Series of strategy cumulative returns
          "bh_equity":        pd.Series of buy-and-hold cumulative returns
          "sma_equity":       pd.Series of SMA cumulative returns
          "daily_returns":    pd.Series of strategy daily P&L
    """
    if sig_levels is None:
        sig_levels = SIG_LEVELS

    prices = price_series.dropna().copy()
    log_prices = np.log(prices)
    daily_ret = log_prices.diff().fillna(0)
    n = len(prices)

    if n < lookback + step:
        raise ValueError(
            f"Price series too short ({n} bars) for lookback={lookback} + step={step}"
        )

    # ── rolling realized vol (past-only) ─────────────────────────────────────
    realized_vol = daily_ret.rolling(vol_window).std() * np.sqrt(252)
    realized_vol = realized_vol.fillna(daily_ret.expanding().std() * np.sqrt(252))

    # ── generate signals on rebalance dates ──────────────────────────────────
    raw_signal = pd.Series(0.0, index=prices.index)
    position = pd.Series(0.0, index=prices.index)

    first_signal_idx = lookback  # earliest bar where we have a full window

    for i in range(first_signal_idx, n, step):
        window_log = log_prices.iloc[i - lookback: i].values

        sig = compute_signal(
            window_log,
            wavelet=wavelet,
            level=level,
            sig_levels=sig_levels,
            slope_window=slope_window,
        )

        vol_today = float(realized_vol.iloc[i - 1])
        sized = volatility_target(sig, vol_today, vol_target=vol_target, max_leverage=max_leverage)

        raw_signal.iloc[i] = sig
        # Hold until next rebalance
        end = min(i + step, n)
        position.iloc[i:end] = sized

    # T+1 execution: position decided at t applied to return t → t+1
    position_lagged = position.shift(1).fillna(0)

    # ── transaction costs ─────────────────────────────────────────────────────
    turnover = position_lagged.diff().abs()
    cost = turnover * (cost_bps / 10_000)

    # ── strategy daily returns ────────────────────────────────────────────────
    strat_ret = position_lagged * daily_ret - cost

    # ── benchmarks ────────────────────────────────────────────────────────────
    # Buy-and-hold (always long 1×)
    bh_ret = daily_ret.copy()

    # SMA 50/200 crossover — same T+1 and cost model
    sma_pos = _sma_signal(prices).shift(1).fillna(0)
    sma_turnover = sma_pos.diff().abs()
    sma_cost = sma_turnover * (cost_bps / 10_000)
    sma_ret = sma_pos * daily_ret - sma_cost

    # ── equity curves ──────────────────────────────────────────────────────────
    # Trim burn-in period (first `lookback` bars had no signals)
    start_idx = first_signal_idx + step
    strat_ret_trim = strat_ret.iloc[start_idx:]
    bh_ret_trim = bh_ret.iloc[start_idx:]
    sma_ret_trim = sma_ret.iloc[start_idx:]
    pos_trim = position_lagged.iloc[start_idx:]

    equity = (1 + strat_ret_trim).cumprod()
    bh_equity = (1 + bh_ret_trim).cumprod()
    sma_equity = (1 + sma_ret_trim).cumprod()

    # ── statistics ────────────────────────────────────────────────────────────
    strategy_stats = perf(
        strat_ret_trim, pos_trim, name="MODWT Mid-Band",
        benchmark=bh_ret_trim,
    )
    bh_stats = perf(
        bh_ret_trim, pd.Series(1.0, index=bh_ret_trim.index),
        name="Buy & Hold", benchmark=bh_ret_trim,
    )
    sma_stats = perf(
        sma_ret_trim, sma_pos.iloc[start_idx:],
        name="50/200 SMA", benchmark=bh_ret_trim,
    )

    return {
        "strategy_stats": strategy_stats,
        "bh_stats": bh_stats,
        "sma_stats": sma_stats,
        "positions": pos_trim,
        "raw_signals": raw_signal.iloc[start_idx:],
        "equity_curve": equity,
        "bh_equity": bh_equity,
        "sma_equity": sma_equity,
        "daily_returns": strat_ret_trim,
    }