280 lines
8.6 KiB
Python
280 lines
8.6 KiB
Python
"""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
|