feat: backtesting engine — historical replay with shared strategies

backtester/metrics.py

@@ -0,0 +1,280 @@
+"""Performance metrics for backtesting results.
+
+Computes standard risk and return metrics from the trade log and equity
+curve produced by a backtest run.
+"""
+
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta
+from typing import Any
+
+from shared.schemas.trading import OrderSide, TradeExecution
+
+
+@dataclass
+class BacktestResult:
+    """Container for all computed backtest metrics.
+
+    Attributes
+    ----------
+    total_return:
+        ``(final - initial) / initial * 100`` as a percentage.
+    annualized_return:
+        Total return annualized using 252 trading days.
+    sharpe_ratio:
+        ``mean(daily_returns) / std(daily_returns) * sqrt(252)``.
+    sortino_ratio:
+        Like Sharpe but using only downside deviation.
+    max_drawdown_pct:
+        Maximum peak-to-trough decline as a percentage.
+    max_drawdown_duration_days:
+        Duration (in calendar days) of the longest drawdown.
+    win_rate:
+        Percentage of winning trades.
+    avg_win_loss_ratio:
+        ``avg(winning_pnl) / abs(avg(losing_pnl))``.
+    trade_count:
+        Total number of round-trip trades.
+    avg_hold_duration:
+        Mean hold duration across all round-trip trades.
+    equity_curve:
+        List of ``(timestamp, equity)`` snapshots.
+    trade_log:
+        Raw list of :class:`TradeExecution` objects.
+    """
+
+    total_return: float = 0.0
+    annualized_return: float = 0.0
+    sharpe_ratio: float = 0.0
+    sortino_ratio: float = 0.0
+    max_drawdown_pct: float = 0.0
+    max_drawdown_duration_days: float = 0.0
+    win_rate: float = 0.0
+    avg_win_loss_ratio: float = 0.0
+    trade_count: int = 0
+    avg_hold_duration: timedelta = field(default_factory=lambda: timedelta(0))
+    equity_curve: list[tuple[datetime, float]] = field(default_factory=list)
+    trade_log: list[TradeExecution] = field(default_factory=list)
+
+
+def compute_metrics(
+    trade_log: list[TradeExecution],
+    equity_curve: list[tuple[datetime, float]],
+    initial_capital: float = 100_000.0,
+) -> BacktestResult:
+    """Compute all performance metrics from a backtest run.
+
+    Parameters
+    ----------
+    trade_log:
+        Chronological list of every executed trade (buys and sells).
+    equity_curve:
+        List of ``(timestamp, portfolio_equity)`` snapshots.
+    initial_capital:
+        Starting capital used to compute total return.
+
+    Returns
+    -------
+    BacktestResult
+        Populated metrics dataclass.
+    """
+    result = BacktestResult(
+        equity_curve=equity_curve,
+        trade_log=trade_log,
+    )
+
+    if not equity_curve:
+        return result
+
+    # ----- Total return -----
+    final_equity = equity_curve[-1][1]
+    result.total_return = (final_equity - initial_capital) / initial_capital * 100.0
+
+    # ----- Annualized return -----
+    if len(equity_curve) >= 2:
+        total_days = (equity_curve[-1][0] - equity_curve[0][0]).days
+        if total_days > 0:
+            trading_years = total_days / 365.25
+            growth_factor = final_equity / initial_capital
+            if growth_factor > 0:
+                result.annualized_return = (
+                    (growth_factor ** (1.0 / trading_years)) - 1.0
+                ) * 100.0
+
+    # ----- Daily returns -----
+    daily_returns = _compute_daily_returns(equity_curve)
+
+    # ----- Sharpe ratio -----
+    result.sharpe_ratio = _compute_sharpe(daily_returns)
+
+    # ----- Sortino ratio -----
+    result.sortino_ratio = _compute_sortino(daily_returns)
+
+    # ----- Max drawdown -----
+    dd_pct, dd_duration = _compute_max_drawdown(equity_curve)
+    result.max_drawdown_pct = dd_pct
+    result.max_drawdown_duration_days = dd_duration
+
+    # ----- Round-trip trade analysis -----
+    round_trips = _build_round_trips(trade_log)
+    result.trade_count = len(round_trips)
+
+    if round_trips:
+        pnls = [rt["pnl"] for rt in round_trips]
+        wins = [p for p in pnls if p > 0]
+        losses = [p for p in pnls if p <= 0]
+
+        result.win_rate = (len(wins) / len(pnls)) * 100.0 if pnls else 0.0
+
+        avg_win = sum(wins) / len(wins) if wins else 0.0
+        avg_loss = sum(losses) / len(losses) if losses else 0.0
+        if avg_loss != 0:
+            result.avg_win_loss_ratio = abs(avg_win / avg_loss)
+        elif avg_win > 0:
+            result.avg_win_loss_ratio = float("inf")
+
+        durations = [rt["duration"] for rt in round_trips]
+        result.avg_hold_duration = sum(durations, timedelta()) / len(durations)
+
+    return result
+
+
+# ------------------------------------------------------------------
+# Internal helpers
+# ------------------------------------------------------------------
+
+
+def _compute_daily_returns(equity_curve: list[tuple[datetime, float]]) -> list[float]:
+    """Compute simple daily returns from the equity curve."""
+    if len(equity_curve) < 2:
+        return []
+    returns: list[float] = []
+    for i in range(1, len(equity_curve)):
+        prev = equity_curve[i - 1][1]
+        curr = equity_curve[i][1]
+        if prev != 0:
+            returns.append((curr - prev) / prev)
+        else:
+            returns.append(0.0)
+    return returns
+
+
+def _compute_sharpe(daily_returns: list[float]) -> float:
+    """Sharpe ratio: mean / std * sqrt(252)."""
+    if len(daily_returns) < 2:
+        return 0.0
+
+    mean_ret = sum(daily_returns) / len(daily_returns)
+    variance = sum((r - mean_ret) ** 2 for r in daily_returns) / (len(daily_returns) - 1)
+    std_ret = math.sqrt(variance)
+
+    if std_ret == 0:
+        return 0.0
+
+    return (mean_ret / std_ret) * math.sqrt(252)
+
+
+def _compute_sortino(daily_returns: list[float]) -> float:
+    """Sortino ratio: mean / downside_deviation * sqrt(252)."""
+    if len(daily_returns) < 2:
+        return 0.0
+
+    mean_ret = sum(daily_returns) / len(daily_returns)
+    downside = [r for r in daily_returns if r < 0]
+
+    if not downside:
+        return 0.0 if mean_ret == 0 else float("inf")
+
+    downside_variance = sum(r ** 2 for r in downside) / len(downside)
+    downside_dev = math.sqrt(downside_variance)
+
+    if downside_dev == 0:
+        return 0.0
+
+    return (mean_ret / downside_dev) * math.sqrt(252)
+
+
+def _compute_max_drawdown(
+    equity_curve: list[tuple[datetime, float]],
+) -> tuple[float, float]:
+    """Compute max drawdown percentage and duration in days.
+
+    Returns
+    -------
+    tuple[float, float]
+        ``(max_drawdown_pct, max_drawdown_duration_days)``
+    """
+    if len(equity_curve) < 2:
+        return 0.0, 0.0
+
+    peak = equity_curve[0][1]
+    peak_ts = equity_curve[0][0]
+    max_dd = 0.0
+    max_dd_duration = 0.0
+
+    for ts, equity in equity_curve[1:]:
+        if equity >= peak:
+            peak = equity
+            peak_ts = ts
+        else:
+            dd = (peak - equity) / peak * 100.0 if peak > 0 else 0.0
+            duration = (ts - peak_ts).days
+            if dd > max_dd:
+                max_dd = dd
+                max_dd_duration = duration
+
+    return max_dd, max_dd_duration
+
+
+def _build_round_trips(
+    trade_log: list[TradeExecution],
+) -> list[dict[str, Any]]:
+    """Match buys with sells to produce round-trip P&L and duration.
+
+    Uses a simple FIFO approach: each BUY opens (or adds to) a
+    position; each SELL closes (reduces) it.
+    """
+    # ticker -> list of {"qty": float, "price": float, "timestamp": datetime}
+    open_positions: dict[str, list[dict[str, Any]]] = {}
+    round_trips: list[dict[str, Any]] = []
+
+    for trade in trade_log:
+        ticker = trade.ticker
+        if trade.side == OrderSide.BUY:
+            if ticker not in open_positions:
+                open_positions[ticker] = []
+            open_positions[ticker].append({
+                "qty": trade.qty,
+                "price": trade.price,
+                "timestamp": trade.timestamp,
+            })
+        elif trade.side == OrderSide.SELL:
+            if ticker not in open_positions or not open_positions[ticker]:
+                continue
+            remaining_sell_qty = trade.qty
+            while remaining_sell_qty > 0 and open_positions.get(ticker):
+                entry = open_positions[ticker][0]
+                matched_qty = min(remaining_sell_qty, entry["qty"])
+
+                pnl = (trade.price - entry["price"]) * matched_qty
+                duration = trade.timestamp - entry["timestamp"]
+
+                round_trips.append({
+                    "ticker": ticker,
+                    "qty": matched_qty,
+                    "entry_price": entry["price"],
+                    "exit_price": trade.price,
+                    "pnl": pnl,
+                    "duration": duration,
+                })
+
+                entry["qty"] -= matched_qty
+                remaining_sell_qty -= matched_qty
+
+                if entry["qty"] <= 0:
+                    open_positions[ticker].pop(0)
+
+    return round_trips