Spaces:

ResearchEngineering
/

financial_news_bot

Running

financial_news_bot / src /transformations /wavelets /backtest.py

Dmitry Beresnev

add wavelet analysis

0821f38 6 months ago

7.82 kB

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

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