From cd3a06c7788ad8a747b1b4579fb6c45b6c43008e Mon Sep 17 00:00:00 2001 From: TheSiahxyz <164138827+TheSiahxyz@users.noreply.github.com> Date: Wed, 1 Apr 2026 18:23:39 +0900 Subject: feat(backtester): improve metrics with daily Sharpe, recovery factor, consecutive stats --- services/backtester/src/backtester/metrics.py | 79 +++++++++++++++++++++------ 1 file changed, 63 insertions(+), 16 deletions(-) (limited to 'services/backtester/src') diff --git a/services/backtester/src/backtester/metrics.py b/services/backtester/src/backtester/metrics.py index caf8477..5b43afd 100644 --- a/services/backtester/src/backtester/metrics.py +++ b/services/backtester/src/backtester/metrics.py @@ -16,6 +16,7 @@ class TradeRecord: side: str # "BUY" or "SELL" price: Decimal quantity: Decimal + fee: Decimal = Decimal("0") @dataclass @@ -38,6 +39,11 @@ class DetailedMetrics: largest_loss: float = 0.0 avg_holding_period: timedelta = field(default_factory=timedelta) trade_pairs: list[dict[str, Any]] = field(default_factory=list) + risk_free_rate: float = 0.0 + recovery_factor: float = 0.0 + max_consecutive_losses: int = 0 + max_consecutive_wins: int = 0 + daily_returns: list[float] = field(default_factory=list) def _pair_trades(trades: list[TradeRecord]) -> list[dict[str, Any]]: @@ -50,7 +56,8 @@ def _pair_trades(trades: list[TradeRecord]) -> list[dict[str, Any]]: buys.append(trade) elif trade.side == "SELL" and buys: buy = buys.pop(0) - pnl = float((trade.price - buy.price) * trade.quantity) + total_fees = float(buy.fee + trade.fee) + pnl = float((trade.price - buy.price) * trade.quantity) - total_fees pairs.append( { "entry_time": buy.time, @@ -60,6 +67,7 @@ def _pair_trades(trades: list[TradeRecord]) -> list[dict[str, Any]]: "exit_price": float(trade.price), "quantity": float(trade.quantity), "pnl": pnl, + "fees": total_fees, "holding_period": trade.time - buy.time, } ) @@ -70,6 +78,7 @@ def compute_detailed_metrics( trades: list[TradeRecord], initial_balance: Decimal, final_balance: Decimal, + risk_free_rate: float = 0.05, ) -> DetailedMetrics: """Compute detailed backtest metrics from a list of trade records.""" if not trades: @@ -170,26 +179,59 @@ def compute_detailed_metrics( 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 + # Build daily equity curve for Sharpe/Sortino + from collections import defaultdict + + daily_equity: dict[str, float] = defaultdict(float) + for p in pairs: + day_key = p["exit_time"].strftime("%Y-%m-%d") + daily_equity[day_key] = daily_equity.get(day_key, 0.0) + p["pnl"] + + sorted_days = sorted(daily_equity.keys()) + daily_returns: list[float] = [] + balance = float(initial_balance) + for day in sorted_days: + pnl = daily_equity[day] + if balance > 0: + daily_returns.append(pnl / balance) + balance += pnl - # 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) + # Sharpe = (mean_daily - daily_rf) / std_daily * sqrt(365) + daily_rf = risk_free_rate / 365 + if len(daily_returns) > 1: + mean_r = sum(daily_returns) / len(daily_returns) + excess_returns = [r - daily_rf for r in daily_returns] + mean_excess = sum(excess_returns) / len(excess_returns) + std_r = math.sqrt(sum((r - mean_r) ** 2 for r in daily_returns) / (len(daily_returns) - 1)) + sharpe = (mean_excess / std_r * math.sqrt(365)) if std_r > 0 else 0.0 + + # Sortino (downside deviation of excess returns) + downside = [min(r - daily_rf, 0.0) for r in daily_returns] + downside_var = sum(d ** 2 for d in downside) / len(downside) downside_std = math.sqrt(downside_var) - sortino = (mean_r / downside_std * math.sqrt(365)) if downside_std > 0 else 0.0 + sortino = (mean_excess / downside_std * math.sqrt(365)) if downside_std > 0 else 0.0 else: + sharpe = 0.0 sortino = 0.0 + # Recovery factor + recovery_factor = abs(total_return / max_dd) if max_dd > 0 else 0.0 + + # Consecutive wins/losses + max_consec_wins = 0 + max_consec_losses = 0 + current_wins = 0 + current_losses = 0 + for p in pairs: + if p["pnl"] > 0: + current_wins += 1 + current_losses = 0 + max_consec_wins = max(max_consec_wins, current_wins) + else: + current_losses += 1 + current_wins = 0 + max_consec_losses = max(max_consec_losses, current_losses) + # Calmar ratio = annualized return / max drawdown # Estimate annualized return from total return and time span if pairs and max_dd > 0: @@ -225,4 +267,9 @@ def compute_detailed_metrics( largest_loss=largest_loss, avg_holding_period=avg_holding, trade_pairs=[p for p in pairs], + risk_free_rate=risk_free_rate, + recovery_factor=recovery_factor, + max_consecutive_losses=max_consec_losses, + max_consecutive_wins=max_consec_wins, + daily_returns=daily_returns, ) -- cgit v1.2.3