spb/crypto-trade-bot

1"""Performance statistics, uncertainty (block bootstrap), multiple-testing (deflated Sharpe), and stress scenarios."""23from __future__ import annotations45import math6from typing import Any78import numpy as np9import pandas as pd10from scipy import stats111213def drawdown_series(equity: pd.Series) -> pd.Series:14    peak = equity.cummax()15    return equity / peak - 1.0161718def compute_metrics(equity: pd.DataFrame, trades: list, interval_min: int, initial_cash: float, fees_paid: float, slippage_paid: float, turnover: float) -> dict[str, Any]:19    eq = equity["equity"].to_numpy(dtype=float)20    n = len(eq)21    if n < 3:22        return {"bars": n, "error": "not enough bars"}23    r = np.diff(np.log(np.maximum(eq, 1e-12)))24    bars_per_year = 365 * 24 * 60 / interval_min25    years = n / bars_per_year26    total_return = eq[-1] / eq[0] - 127    cagr = (eq[-1] / eq[0]) ** (1 / years) - 1 if years > 0 and eq[-1] > 0 else float("nan")28    vol = float(np.std(r, ddof=1) * math.sqrt(bars_per_year)) if n > 2 else float("nan")29    mean_ann = float(np.mean(r) * bars_per_year)30    sharpe = mean_ann / vol if vol and vol > 0 else float("nan")31    downside = r[r < 0]32    dvol = float(np.sqrt(np.mean(downside**2)) * math.sqrt(bars_per_year)) if len(downside) else float("nan")33    sortino = mean_ann / dvol if dvol and dvol > 0 else float("nan")34    dd = drawdown_series(equity["equity"])35    max_dd = float(dd.min())36    calmar = cagr / abs(max_dd) if max_dd < 0 and not math.isnan(cagr) else float("nan")37    exposure = float(equity["exposure"].abs().mean())38    # trade-level (round trips approximated by consecutive buy→sell pairs)39    pnl_trades = _round_trips(trades)40    wins = [p for p in pnl_trades if p > 0]41    losses = [p for p in pnl_trades if p < 0]42    hit = len(wins) / len(pnl_trades) if pnl_trades else float("nan")43    pf = (sum(wins) / abs(sum(losses))) if losses else (float("inf") if wins else float("nan"))44    var95 = float(np.quantile(r, 0.05)) if n > 20 else float("nan")45    cvar95 = float(r[r <= var95].mean()) if n > 20 and np.any(r <= var95) else float("nan")46    fee_drag = fees_paid / initial_cash47    slip_drag = slippage_paid / initial_cash48    underwater = int((dd < 0).sum())49    return {50        "bars": n, "years": years, "total_return": total_return, "cagr": cagr, "volatility": vol, "sharpe": sharpe, "sortino": sortino,51        "max_drawdown": max_dd, "calmar": calmar, "exposure": exposure, "turnover_annual": turnover, "n_trades": len(trades),52        "n_round_trips": len(pnl_trades), "hit_rate": hit, "profit_factor": pf, "avg_trade_pnl": float(np.mean(pnl_trades)) if pnl_trades else float("nan"),53        "var_95_bar": var95, "cvar_95_bar": cvar95, "fees_paid": fees_paid, "fee_drag": fee_drag, "slippage_paid": slippage_paid, "slippage_drag": slip_drag,54        "underwater_bars_pct": underwater / n, "final_equity": float(eq[-1]), "skew": float(stats.skew(r)) if n > 3 else float("nan"), "kurtosis": float(stats.kurtosis(r)) if n > 3 else float("nan"),55    }565758def _round_trips(trades: list) -> list[float]:59    """FIFO round-trip P&L (buy lots closed by sells), fees included."""60    lots: list[list[float]] = []  # [qty, price_incl_fee]61    out: list[float] = []62    for t in trades:63        if t.side == "buy":64            lots.append([t.qty, (t.notional + t.fee) / t.qty if t.qty else t.price])65        else:66            remaining = t.qty67            proceeds_per_unit = (t.notional - t.fee) / t.qty if t.qty else t.price68            pnl = 0.069            while remaining > 1e-12 and lots:70                lot = lots[0]71                take = min(lot[0], remaining)72                pnl += take * (proceeds_per_unit - lot[1])73                lot[0] -= take74                remaining -= take75                if lot[0] <= 1e-12:76                    lots.pop(0)77            out.append(pnl)78    return out798081def rolling_sharpe(equity: pd.DataFrame, interval_min: int, window_bars: int) -> list[dict[str, float]]:82    r = np.log(equity["equity"]).diff()83    bars_per_year = 365 * 24 * 60 / interval_min84    rs = (r.rolling(window_bars).mean() / r.rolling(window_bars).std(ddof=1)) * math.sqrt(bars_per_year)85    return [{"ts": int(t), "sharpe": float(s)} for t, s in zip(equity["ts"], rs, strict=False) if not math.isnan(s)]868788def block_bootstrap_ci(returns: np.ndarray, bars_per_year: float, n_boot: int = 500, block: int | None = None, seed: int = 42) -> dict[str, Any]:89    """Stationary (circular) block bootstrap of the Sharpe ratio and CAGR."""90    rng = np.random.default_rng(seed)91    n = len(returns)92    if n < 30:93        return {"sharpe_ci": [None, None], "cagr_ci": [None, None], "n_boot": 0}94    block = block or max(5, int(round(n ** (1 / 3))))95    sharpes, cagrs = [], []96    nb = int(math.ceil(n / block))97    for _ in range(n_boot):98        starts = rng.integers(0, n, size=nb)99        idx = (starts[:, None] + np.arange(block)[None, :]).ravel() % n100        s = returns[idx[:n]]101        sd = s.std(ddof=1)102        sharpes.append(s.mean() / sd * math.sqrt(bars_per_year) if sd > 0 else 0.0)103        cagrs.append(math.exp(s.sum() * (bars_per_year / n)) - 1)104    return {"sharpe_ci": [float(np.quantile(sharpes, 0.025)), float(np.quantile(sharpes, 0.975))], "cagr_ci": [float(np.quantile(cagrs, 0.025)), float(np.quantile(cagrs, 0.975))], "n_boot": n_boot, "block": block}105106107def deflated_sharpe(sharpe_annual: float, n_obs: int, bars_per_year: float, n_trials: int, skew: float, kurt_excess: float, sharpe_var_trials: float | None = None) -> dict[str, Any]:108    """Bailey & López de Prado (2014) Deflated Sharpe Ratio: probability that the observed SR exceeds the expected max SR of N trials under H0."""109    if n_obs < 10 or not math.isfinite(sharpe_annual):110        return {"dsr": None, "sr0": None}111    sr = sharpe_annual / math.sqrt(bars_per_year)  # per-bar Sharpe112    var_trials = sharpe_var_trials if sharpe_var_trials is not None else (0.5 / bars_per_year)  # conservative default spread of trial SRs (per bar)113    em = 0.5772156649114    n_trials = max(n_trials, 1)115    if n_trials == 1:116        sr0 = 0.0117    else:118        sr0 = math.sqrt(var_trials) * ((1 - em) * stats.norm.ppf(1 - 1 / n_trials) + em * stats.norm.ppf(1 - 1 / (n_trials * math.e)))119    denom = math.sqrt(max(1 - skew * sr + (kurt_excess + 3 - 1) / 4 * sr**2, 1e-12))120    z = (sr - sr0) * math.sqrt(n_obs - 1) / denom121    return {"dsr": float(stats.norm.cdf(z)), "sr0_annual": sr0 * math.sqrt(bars_per_year), "n_trials": n_trials}122123124def stress_scenarios(run_fn, base_cost, equity_df: pd.DataFrame) -> dict[str, Any]:125    """Re-run the backtest under adverse cost/market assumptions. `run_fn(cost_model, df_override=None)` → metrics dict."""126    out: dict[str, Any] = {}127    cm = base_cost128    out["fees_x2"] = run_fn(cm.model_copy(update={"fee_multiplier": 2.0}))129    out["spread_x3"] = run_fn(cm.model_copy(update={"spread_multiplier": 3.0}))130    out["slippage_plus_20bps"] = run_fn(cm.model_copy(update={"slippage_bps": cm.slippage_bps + 20}))131    out["latency_plus_1_bar"] = run_fn(cm.model_copy(update={"latency_bars": cm.latency_bars + 1}))132    out["liquidity_collapse"] = run_fn(cm.model_copy(update={"max_participation": max(cm.max_participation / 10, 1e-4), "impact_coeff": cm.impact_coeff * 5}))133    return out134135136def shuffled_trades_test(trade_pnls: list[float], n: int = 1000, seed: int = 7) -> dict[str, Any]:137    """Monte-Carlo reordering of trade P&Ls: distribution of max drawdown of the cumulative P&L path."""138    if len(trade_pnls) < 5:139        return {"n": 0}140    rng = np.random.default_rng(seed)141    arr = np.array(trade_pnls)142    mdds = []143    for _ in range(n):144        p = rng.permutation(arr).cumsum()145        peak = np.maximum.accumulate(np.concatenate([[0.0], p]))[1:]146        mdds.append(float((p - peak).min()))147    return {"n": n, "max_dd_p50": float(np.median(mdds)), "max_dd_p95": float(np.quantile(mdds, 0.05)), "max_dd_worst": float(min(mdds)), "total_pnl": float(arr.sum())}148