backtesting-sim

Skill: Backtesting Sim | Domain: trading | Category: quantitative | Level: intermediate Tags: trading, quant, backtesting, simulation, paper-trading, vectorized

Vectorized Backtester

Vectorized Backtester

import pandas as pd
import numpy as np
from typing import Callable, Optional

# ── Shared helpers ──────────────────────────────────────────────────────────

def _sharpe(r: np.ndarray, rfr: float = 0.04 / 252) -> float:
    std = r.std(ddof=1)
    return float((r.mean() - rfr) / std * np.sqrt(252)) if std > 0 else 0.0

def _sortino(r: np.ndarray, rfr: float = 0.04 / 252) -> float:
    down = r[r < rfr]
    std_dn = down.std(ddof=1) if len(down) > 1 else 0.0
    return float((r.mean() - rfr) / std_dn * np.sqrt(252)) if std_dn > 0 else 0.0

def _calmar(r: np.ndarray) -> float:
    eq      = np.cumprod(1 + r)
    peak    = np.maximum.accumulate(eq)
    max_dd  = abs(((eq - peak) / peak).min())
    ann_ret = eq[-1] ** (252 / len(r)) - 1
    return float(ann_ret / max_dd) if max_dd > 0 else 0.0

def _profit_factor(r: np.ndarray) -> float:
    gains  = r[r > 0].sum()
    losses = abs(r[r < 0].sum())
    return float(gains / losses) if losses > 0 else float("inf")


class VectorizedBacktester:
    """
    Production-quality vectorised backtester with full performance metrics.

    Signals must be pre-shifted (no look-ahead bias).
    Supports transaction costs, position sizing, and walk-forward validation.

    Example
    -------
    >>> df["signal"] = np.sign(df["close"].pct_change(5))  # 5-bar momentum
    >>> result = VectorizedBacktester.backtest(df, spread_bps=2.0)
    >>> print(result["sharpe"], result["max_drawdown_pct"])
    1.34  -12.5
    """

    @staticmethod
    def backtest(
        df: pd.DataFrame,
        signal_col: str = "signal",
        spread_bps: float = 2.0,
        slippage_bps: float = 1.0,
        initial_capital: float = 10_000.0,
        position_size: float = 1.0,
        risk_free_annual: float = 0.04,
    ) -> dict:
        """
        Vectorised backtest engine.

        Parameters
        ----------
        df              : OHLCV DataFrame with a signal column
        signal_col      : column name for signal (1=long, -1=short, 0=flat)
        spread_bps      : round-trip spread cost in basis points
        slippage_bps    : round-trip slippage estimate in basis points
        initial_capital : starting capital
        position_size   : fraction of capital at risk (1.0 = fully invested)
        risk_free_annual: used for Sharpe / Sortino calculations

        Returns
        -------
        Comprehensive dict including Sharpe, Sortino, Calmar, max DD,
        win rate, profit factor, equity curve, and full annotated DataFrame.

        Notes
        -----
        ALL signals are shifted by 1 bar to prevent look-ahead.
        Costs are charged on position changes (not every bar).
        """
        if signal_col not in df.columns:
            raise KeyError(f"Column '{signal_col}' not found in DataFrame")
        if len(df) < 10:
            raise ValueError("Need at least 10 bars to backtest")

        out = df.copy()
        cost_rt = (spread_bps + slippage_bps) / 10_000.0   # Round-trip cost fraction

        out["returns"]      = out["close"].pct_change()
        out["position"]     = out[signal_col].shift(1).fillna(0) * position_size
        out["trade"]        = out["position"].diff().abs().fillna(0)
        out["gross_return"] = out["position"] * out["returns"]
        out["cost"]         = out["trade"] * cost_rt
        out["net_return"]   = out["gross_return"] - out["cost"]

        # Equity curve (multiplicative)
        out["equity"]       = initial_capital * (1 + out["net_return"]).cumprod()
        out["peak"]         = out["equity"].cummax()
        out["drawdown"]     = (out["equity"] - out["peak"]) / out["peak"]

        # Trade-level stats
        out["trade_entry"]  = (out["position"] != 0) & (out["position"].shift(1) == 0)
        out["trade_exit"]   = (out["position"] == 0) & (out["position"].shift(1) != 0)
        n_trades            = int(out["trade_entry"].sum())

        net     = out["net_return"].dropna().values
        rfr_d   = risk_free_annual / 252

        # Compute comprehensive metrics from net returns
        total_return = float((out["equity"].iloc[-1] / initial_capital - 1) * 100)
        max_dd       = float(out["drawdown"].min() * 100)
        sharpe       = _sharpe(net, rfr_d)
        sortino      = _sortino(net, rfr_d)
        calmar       = _calmar(net)
        pf           = _profit_factor(net)

        # Win rate on closed trades (more accurate than bar-level)
        exit_returns = out.loc[out["trade_exit"], "net_return"]
        win_rate     = float((exit_returns > 0).mean() * 100) if len(exit_returns) > 0 else 0.0

        # Consecutive loss streak
        signs = np.sign(net)
        streak = 0
        max_streak = 0
        for s in signs:
            if s < 0:
                streak += 1
                max_streak = max(max_streak, streak)
            else:
                streak = 0

        return {
            "total_return_pct":   round(total_return, 2),
            "ann_return_pct":     round(float((out["equity"].iloc[-1] / initial_capital)
                                              ** (252 / max(len(net), 1)) - 1) * 100, 2),
            "sharpe":             round(sharpe, 3),
            "sortino":            round(sortino, 3),
            "calmar":             round(calmar, 3),
            "max_drawdown_pct":   round(max_dd, 2),
            "profit_factor":      round(pf, 3),
            "n_trades":           n_trades,
            "win_rate":           round(win_rate, 1),
            "max_consec_losses":  int(max_streak),
            "total_costs_pct":    round(float(out["cost"].sum() * 100), 2),
            "equity_curve":       out["equity"],
            "drawdown_series":    out["drawdown"],
            "df":                 out,
        }

    @staticmethod
    def walk_forward_backtest(
        df: pd.DataFrame,
        signal_fn: Callable,
        optimize_fn: Callable,
        train_bars: int = 500,
        test_bars: int = 100,
        min_folds: int = 3,
        **kwargs,
    ) -> dict:
        """
        Anchored walk-forward validation.

        Parameters
        ----------
        signal_fn   : callable(df, params) → signal Series
        optimize_fn : callable(train_df) → params dict
        train_bars  : in-sample training window
        test_bars   : out-of-sample test window per fold
        min_folds   : minimum folds required for a valid result

        Returns
        -------
        dict with per-fold and aggregate OOS statistics.
        """
        results:    list[dict] = []
        all_equity: list[pd.Series] = []

        for start in range(0, len(df) - train_bars - test_bars, test_bars):
            train = df.iloc[start: start + train_bars]
            test  = df.iloc[start + train_bars: start + train_bars + test_bars].copy()

            try:
                params          = optimize_fn(train)
                test["signal"]  = signal_fn(test, params)
                bt              = VectorizedBacktester.backtest(test, **kwargs)
                results.append({
                    "fold":    len(results),
                    "sharpe":  bt["sharpe"],
                    "sortino": bt["sortino"],
                    "return":  bt["total_return_pct"],
                    "max_dd":  bt["max_drawdown_pct"],
                    "params":  params,
                })
                all_equity.append(bt["equity_curve"])
            except Exception as e:
                results.append({"fold": len(results), "error": str(e)})

        valid = [r for r in results if "sharpe" in r]
        if len(valid) < min_folds:
            return {
                "method":  "walk_forward",
                "error":   f"Only {len(valid)} valid folds (need {min_folds})",
                "n_folds": len(results),
            }

        sharpes  = [r["sharpe"]  for r in valid]
        sortinos = [r["sortino"] for r in valid]
        returns  = [r["return"]  for r in valid]

        # Concatenate OOS equity curves for a continuous equity line
        oos_equity = pd.concat(all_equity).sort_index() if all_equity else pd.Series(dtype=float)

        return {
            "method":            "walk_forward",
            "n_folds":           len(results),
            "n_valid_folds":     len(valid),
            "avg_sharpe":        round(float(np.mean(sharpes)), 3),
            "median_sharpe":     round(float(np.median(sharpes)), 3),
            "std_sharpe":        round(float(np.std(sharpes, ddof=1)), 3),
            "pct_folds_positive": round(float(np.mean([r > 0 for r in returns])) * 100, 1),
            "avg_return":        round(float(np.mean(returns)), 2),
            "avg_sortino":       round(float(np.mean(sortinos)), 3),
            "fold_results":      valid,
            "oos_equity":        oos_equity,
            "WARNING":           "Past performance ≠ future results. OOS validation required.",
        }

    @staticmethod
    def compare_strategies(
        df: pd.DataFrame,
        strategies: dict[str, pd.Series],
        spread_bps: float = 2.0,
        slippage_bps: float = 1.0,
        initial_capital: float = 10_000.0,
    ) -> pd.DataFrame:
        """
        Run multiple strategies on the same data and compare side-by-side.

        Parameters
        ----------
        strategies : {"name": signal_series, ...}

        Returns
        -------
        DataFrame ranked by Sharpe ratio.
        """
        results: list[dict] = []
        for name, signal_series in strategies.items():
            df_copy = df.copy()
            df_copy["signal"] = signal_series
            try:
                bt = VectorizedBacktester.backtest(
                    df_copy,
                    spread_bps=spread_bps,
                    slippage_bps=slippage_bps,
                    initial_capital=initial_capital,
                )
                results.append({
                    "strategy":     name,
                    "return":       bt["total_return_pct"],
                    "sharpe":       bt["sharpe"],
                    "sortino":      bt["sortino"],
                    "calmar":       bt["calmar"],
                    "max_dd":       bt["max_drawdown_pct"],
                    "n_trades":     bt["n_trades"],
                    "win_rate":     bt["win_rate"],
                    "profit_factor": bt["profit_factor"],
                })
            except Exception as e:
                results.append({"strategy": name, "error": str(e)})

        return pd.DataFrame(results).sort_values("sharpe", ascending=False)