backtesting-sim

> **Skill:** Backtesting Sim | **Domain:** trading | **Category:** quantitative | **Level:** intermediate > **Tags:** `trading`, `quant`, `backtesting`, `simulation`, `paper-trading`, `vectorized` ## Vectorized Backtester # Vectorized Backtester ```python 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) ``` --- ## Trade Simulator Paper # Trade Simulator Paper ```python import pandas as pd import numpy as np from datetime import datetime from typing import Optional class PaperTradeSimulator: """ Realistic paper trading simulator with proper spread/slippage modelling, margin tracking, and full trade history for post-session analysis. Uses a reproducible RNG (np.random.default_rng) for slippage simulation. Example ------- >>> sim = PaperTradeSimulator(initial_balance=10_000, seed=42) >>> sim.open_trade("EURUSD", "buy", lots=0.1, price=1.0850, sl=1.0810, tp=1.0920) >>> closed = sim.check_positions({"EURUSD": 1.0920}) >>> print(closed[0]["reason"], closed[0]["pnl_usd"]) 'TP' 70.0 """