diff options
Diffstat (limited to 'services/backtester/src')
| -rw-r--r-- | services/backtester/src/backtester/metrics.py | 215 |
1 files changed, 215 insertions, 0 deletions
diff --git a/services/backtester/src/backtester/metrics.py b/services/backtester/src/backtester/metrics.py new file mode 100644 index 0000000..15be0e6 --- /dev/null +++ b/services/backtester/src/backtester/metrics.py @@ -0,0 +1,215 @@ +"""Detailed backtest metrics: Sharpe, Sortino, drawdown, and more.""" +from __future__ import annotations + +import math +from dataclasses import dataclass, field +from datetime import datetime, timedelta +from decimal import Decimal +from typing import Any + + +@dataclass +class TradeRecord: + time: datetime + symbol: str + side: str # "BUY" or "SELL" + price: Decimal + quantity: Decimal + + +@dataclass +class DetailedMetrics: + total_return: float + total_trades: int + winning_trades: int + losing_trades: int + win_rate: float + profit_factor: float + sharpe_ratio: float + sortino_ratio: float + calmar_ratio: float + max_drawdown: float + max_drawdown_duration: timedelta + monthly_returns: dict[str, float] = field(default_factory=dict) + avg_win: float = 0.0 + avg_loss: float = 0.0 + largest_win: float = 0.0 + largest_loss: float = 0.0 + avg_holding_period: timedelta = field(default_factory=timedelta) + trade_pairs: list[dict[str, Any]] = field(default_factory=list) + + +def _pair_trades(trades: list[TradeRecord]) -> list[dict[str, Any]]: + """Pair BUY/SELL trades using FIFO matching.""" + buys: list[TradeRecord] = [] + pairs: list[dict[str, Any]] = [] + + for trade in trades: + if trade.side == "BUY": + buys.append(trade) + elif trade.side == "SELL" and buys: + buy = buys.pop(0) + pnl = float((trade.price - buy.price) * trade.quantity) + pairs.append( + { + "entry_time": buy.time, + "exit_time": trade.time, + "symbol": trade.symbol, + "entry_price": float(buy.price), + "exit_price": float(trade.price), + "quantity": float(trade.quantity), + "pnl": pnl, + "holding_period": trade.time - buy.time, + } + ) + return pairs + + +def compute_detailed_metrics( + trades: list[TradeRecord], + initial_balance: Decimal, + final_balance: Decimal, +) -> DetailedMetrics: + """Compute detailed backtest metrics from a list of trade records.""" + if not trades: + return DetailedMetrics( + total_return=0.0, + total_trades=0, + winning_trades=0, + losing_trades=0, + win_rate=0.0, + profit_factor=0.0, + sharpe_ratio=0.0, + sortino_ratio=0.0, + calmar_ratio=0.0, + max_drawdown=0.0, + max_drawdown_duration=timedelta(0), + ) + + pairs = _pair_trades(trades) + total_return = float(final_balance - initial_balance) / float(initial_balance) if initial_balance else 0.0 + + if not pairs: + return DetailedMetrics( + total_return=total_return, + total_trades=len(trades), + winning_trades=0, + losing_trades=0, + win_rate=0.0, + profit_factor=0.0, + sharpe_ratio=0.0, + sortino_ratio=0.0, + calmar_ratio=0.0, + max_drawdown=0.0, + max_drawdown_duration=timedelta(0), + ) + + # Win/loss stats + wins = [p for p in pairs if p["pnl"] > 0] + losses = [p for p in pairs if p["pnl"] <= 0] + winning_trades = len(wins) + losing_trades = len(losses) + total_pairs = len(pairs) + win_rate = winning_trades / total_pairs * 100 if total_pairs else 0.0 + + gross_profit = sum(p["pnl"] for p in wins) + gross_loss = abs(sum(p["pnl"] for p in losses)) + profit_factor = gross_profit / gross_loss if gross_loss > 0 else float("inf") if gross_profit > 0 else 0.0 + + avg_win = gross_profit / winning_trades if winning_trades else 0.0 + avg_loss = gross_loss / losing_trades if losing_trades else 0.0 + largest_win = max((p["pnl"] for p in wins), default=0.0) + largest_loss = min((p["pnl"] for p in losses), default=0.0) + + # Holding periods + holding_periods = [p["holding_period"] for p in pairs] + avg_holding = sum(holding_periods, timedelta(0)) / len(holding_periods) if holding_periods else timedelta(0) + + # Build equity curve from pairs + init_bal = float(initial_balance) + equity = [init_bal] + for p in pairs: + equity.append(equity[-1] + p["pnl"]) + + # Returns per pair + returns = [] + for i in range(1, len(equity)): + if equity[i - 1] != 0: + returns.append((equity[i] - equity[i - 1]) / equity[i - 1]) + else: + returns.append(0.0) + + # Max drawdown + peak = equity[0] + max_dd = 0.0 + max_dd_duration = timedelta(0) + dd_start_idx = 0 + for i in range(1, len(equity)): + if equity[i] > peak: + peak = equity[i] + dd_start_idx = i + dd = (peak - equity[i]) / peak if peak > 0 else 0.0 + if dd > max_dd: + max_dd = dd + # Duration: use pair exit times + if i <= len(pairs) and dd_start_idx < len(pairs): + start_time = pairs[dd_start_idx]["exit_time"] if dd_start_idx < len(pairs) else pairs[0]["entry_time"] + end_time = pairs[i - 1]["exit_time"] + max_dd_duration = end_time - start_time if end_time > start_time else timedelta(0) + + # Sharpe ratio (annualized, assume 365 trading days) + if len(returns) > 1: + mean_r = sum(returns) / len(returns) + std_r = math.sqrt(sum((r - mean_r) ** 2 for r in returns) / (len(returns) - 1)) + sharpe = (mean_r / std_r * math.sqrt(365)) if std_r > 0 else 0.0 + elif len(returns) == 1: + sharpe = 0.0 + else: + sharpe = 0.0 + + # Sortino ratio (annualized) + if len(returns) > 1: + mean_r = sum(returns) / len(returns) + downside = [min(r, 0.0) for r in returns] + downside_var = sum(d ** 2 for d in downside) / (len(downside) - 1) + downside_std = math.sqrt(downside_var) + sortino = (mean_r / downside_std * math.sqrt(365)) if downside_std > 0 else 0.0 + else: + sortino = 0.0 + + # Calmar ratio = annualized return / max drawdown + # Estimate annualized return from total return and time span + if pairs and max_dd > 0: + time_span = (pairs[-1]["exit_time"] - pairs[0]["entry_time"]).total_seconds() + years = time_span / (365.25 * 86400) if time_span > 0 else 1.0 + annualized_return = (1 + total_return) ** (1 / years) - 1 if years > 0 else total_return + calmar = annualized_return / max_dd + else: + calmar = 0.0 + + # Monthly returns grouped by exit_time month + monthly: dict[str, float] = {} + for p in pairs: + key = p["exit_time"].strftime("%Y-%m") + monthly[key] = monthly.get(key, 0.0) + p["pnl"] + + return DetailedMetrics( + total_return=total_return, + total_trades=len(trades), + winning_trades=winning_trades, + losing_trades=losing_trades, + win_rate=win_rate, + profit_factor=profit_factor, + sharpe_ratio=sharpe, + sortino_ratio=sortino, + calmar_ratio=calmar, + max_drawdown=max_dd, + max_drawdown_duration=max_dd_duration, + monthly_returns=monthly, + avg_win=avg_win, + avg_loss=avg_loss, + largest_win=largest_win, + largest_loss=largest_loss, + avg_holding_period=avg_holding, + trade_pairs=[p for p in pairs], + ) |