You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
468 lines
15 KiB
Python
468 lines
15 KiB
Python
"""Pure-function performance metrics for the backtester.
|
|
|
|
All functions take lists of domain objects (no pandas).
|
|
All ratio helpers return None instead of 0.0 for empty/zero denominators.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import math
|
|
import random
|
|
import statistics
|
|
from collections import defaultdict
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from libs.backtest.domain import DailyPortfolioState, ExitReason, FilledTrade, MetricsBundle
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Trade metrics (7)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def compute_win_rate(trades: list[FilledTrade]) -> float | None:
|
|
if not trades:
|
|
return None
|
|
wins = sum(1 for t in trades if t.net_pnl > 0)
|
|
return wins / len(trades)
|
|
|
|
|
|
def compute_avg_win_pct(trades: list[FilledTrade]) -> float | None:
|
|
wins = [t.pnl_pct for t in trades if t.net_pnl > 0]
|
|
if not wins:
|
|
return None
|
|
return statistics.mean(wins)
|
|
|
|
|
|
def compute_avg_loss_pct(trades: list[FilledTrade]) -> float | None:
|
|
losses = [t.pnl_pct for t in trades if t.net_pnl <= 0]
|
|
if not losses:
|
|
return None
|
|
return statistics.mean(losses)
|
|
|
|
|
|
def compute_profit_factor(trades: list[FilledTrade]) -> float | None:
|
|
gross_profit = sum(t.net_pnl for t in trades if t.net_pnl > 0)
|
|
gross_loss = abs(sum(t.net_pnl for t in trades if t.net_pnl < 0))
|
|
if gross_loss == 0:
|
|
return None
|
|
return gross_profit / gross_loss
|
|
|
|
|
|
def compute_expectancy_r(trades: list[FilledTrade]) -> float | None:
|
|
if not trades:
|
|
return None
|
|
return statistics.mean(t.r_multiple for t in trades)
|
|
|
|
|
|
def compute_avg_r_multiple(trades: list[FilledTrade]) -> float | None:
|
|
if not trades:
|
|
return None
|
|
return statistics.mean(t.r_multiple for t in trades)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Portfolio metrics (8)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def compute_total_return_pct(equity_curve: list[DailyPortfolioState]) -> float | None:
|
|
if len(equity_curve) < 2:
|
|
return None
|
|
start = equity_curve[0].equity
|
|
end = equity_curve[-1].equity
|
|
if start == 0:
|
|
return None
|
|
return (end - start) / start * 100.0
|
|
|
|
|
|
def compute_annualized_return_pct(
|
|
equity_curve: list[DailyPortfolioState],
|
|
) -> float | None:
|
|
if len(equity_curve) < 2:
|
|
return None
|
|
start = equity_curve[0].equity
|
|
end = equity_curve[-1].equity
|
|
if start <= 0:
|
|
return None
|
|
days = (equity_curve[-1].date - equity_curve[0].date).days
|
|
if days <= 0:
|
|
return None
|
|
years = days / 365.25
|
|
return ((end / start) ** (1.0 / years) - 1.0) * 100.0
|
|
|
|
|
|
def compute_max_drawdown_pct(equity_curve: list[DailyPortfolioState]) -> float | None:
|
|
if not equity_curve:
|
|
return None
|
|
peak = equity_curve[0].equity
|
|
max_dd = 0.0
|
|
for state in equity_curve:
|
|
if state.equity > peak:
|
|
peak = state.equity
|
|
if peak > 0:
|
|
dd = (peak - state.equity) / peak * 100.0
|
|
max_dd = max(max_dd, dd)
|
|
return max_dd
|
|
|
|
|
|
def compute_calmar_ratio(
|
|
annualized_return: float | None,
|
|
max_drawdown: float | None,
|
|
) -> float | None:
|
|
if annualized_return is None or max_drawdown is None or max_drawdown == 0:
|
|
return None
|
|
return annualized_return / max_drawdown
|
|
|
|
|
|
def _daily_returns(equity_curve: list[DailyPortfolioState]) -> list[float]:
|
|
returns = []
|
|
for i in range(1, len(equity_curve)):
|
|
prev = equity_curve[i - 1].equity
|
|
curr = equity_curve[i].equity
|
|
if prev > 0:
|
|
returns.append((curr - prev) / prev)
|
|
return returns
|
|
|
|
|
|
def compute_sharpe_ratio(
|
|
equity_curve: list[DailyPortfolioState],
|
|
risk_free_daily: float = 0.0,
|
|
) -> float | None:
|
|
returns = _daily_returns(equity_curve)
|
|
if len(returns) < 2:
|
|
return None
|
|
excess = [r - risk_free_daily for r in returns]
|
|
mean = statistics.mean(excess)
|
|
try:
|
|
std = statistics.stdev(excess)
|
|
except statistics.StatisticsError:
|
|
return None
|
|
if std == 0:
|
|
return None
|
|
return (mean / std) * math.sqrt(252)
|
|
|
|
|
|
def compute_sortino_ratio(
|
|
equity_curve: list[DailyPortfolioState],
|
|
risk_free_daily: float = 0.0,
|
|
) -> float | None:
|
|
returns = _daily_returns(equity_curve)
|
|
if len(returns) < 2:
|
|
return None
|
|
excess = [r - risk_free_daily for r in returns]
|
|
mean = statistics.mean(excess)
|
|
downside = [r for r in excess if r < 0]
|
|
if len(downside) < 2:
|
|
return None
|
|
try:
|
|
downside_std = statistics.stdev(downside)
|
|
except statistics.StatisticsError:
|
|
return None
|
|
if downside_std == 0:
|
|
return None
|
|
return (mean / downside_std) * math.sqrt(252)
|
|
|
|
|
|
def compute_avg_daily_pnl(equity_curve: list[DailyPortfolioState]) -> float | None:
|
|
if len(equity_curve) < 2:
|
|
return None
|
|
daily_pnls = []
|
|
for i in range(1, len(equity_curve)):
|
|
daily_pnls.append(equity_curve[i].equity - equity_curve[i - 1].equity)
|
|
return statistics.mean(daily_pnls)
|
|
|
|
|
|
def compute_avg_positions_held(equity_curve: list[DailyPortfolioState]) -> float | None:
|
|
if not equity_curve:
|
|
return None
|
|
return statistics.mean(len(s.open_positions) for s in equity_curve)
|
|
|
|
|
|
def compute_avg_gross_exposure_pct(
|
|
equity_curve: list[DailyPortfolioState],
|
|
) -> float | None:
|
|
if not equity_curve:
|
|
return None
|
|
exposure_pcts = [
|
|
(state.gross_exposure / state.equity) * 100.0
|
|
for state in equity_curve
|
|
if state.equity > 0
|
|
]
|
|
if not exposure_pcts:
|
|
return None
|
|
return statistics.mean(exposure_pcts)
|
|
|
|
|
|
def compute_avg_net_exposure_pct(
|
|
equity_curve: list[DailyPortfolioState],
|
|
) -> float | None:
|
|
if not equity_curve:
|
|
return None
|
|
exposure_pcts = [
|
|
(state.net_exposure / state.equity) * 100.0
|
|
for state in equity_curve
|
|
if state.equity > 0
|
|
]
|
|
if not exposure_pcts:
|
|
return None
|
|
return statistics.mean(exposure_pcts)
|
|
|
|
|
|
def compute_days_in_market_pct(
|
|
equity_curve: list[DailyPortfolioState],
|
|
) -> float | None:
|
|
if not equity_curve:
|
|
return None
|
|
days_in_market = sum(1 for state in equity_curve if state.gross_exposure > 0)
|
|
return days_in_market / len(equity_curve) * 100.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Stability metrics (4)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def compute_trade_skewness(trades: list[FilledTrade]) -> float | None:
|
|
if len(trades) < 3:
|
|
return None
|
|
pnls = [t.net_pnl for t in trades]
|
|
mean = statistics.mean(pnls)
|
|
try:
|
|
std = statistics.stdev(pnls)
|
|
except statistics.StatisticsError:
|
|
return None
|
|
if std == 0:
|
|
return None
|
|
n = len(pnls)
|
|
skew = sum(((x - mean) / std) ** 3 for x in pnls) * n / ((n - 1) * (n - 2))
|
|
return skew
|
|
|
|
|
|
def compute_trade_kurtosis(trades: list[FilledTrade]) -> float | None:
|
|
if len(trades) < 4:
|
|
return None
|
|
pnls = [t.net_pnl for t in trades]
|
|
mean = statistics.mean(pnls)
|
|
try:
|
|
std = statistics.stdev(pnls)
|
|
except statistics.StatisticsError:
|
|
return None
|
|
if std == 0:
|
|
return None
|
|
n = len(pnls)
|
|
# Excess kurtosis (Fisher's definition)
|
|
kurt = sum(((x - mean) / std) ** 4 for x in pnls) * n * (n + 1) / (
|
|
(n - 1) * (n - 2) * (n - 3)
|
|
) - 3 * (n - 1) ** 2 / ((n - 2) * (n - 3))
|
|
return kurt
|
|
|
|
|
|
def compute_monthly_win_rate(trades: list[FilledTrade]) -> float | None:
|
|
"""Fraction of calendar months with net positive PnL."""
|
|
if not trades:
|
|
return None
|
|
monthly: dict[str, float] = defaultdict(float)
|
|
for t in trades:
|
|
key = t.exit_date.strftime("%Y-%m")
|
|
monthly[key] += t.net_pnl
|
|
if not monthly:
|
|
return None
|
|
wins = sum(1 for v in monthly.values() if v > 0)
|
|
return wins / len(monthly)
|
|
|
|
|
|
def compute_equity_curve_r_squared(equity_curve: list[DailyPortfolioState]) -> float | None:
|
|
"""R² of a linear regression fit to the equity curve (higher = smoother growth)."""
|
|
if len(equity_curve) < 3:
|
|
return None
|
|
n = len(equity_curve)
|
|
xs = list(range(n))
|
|
ys = [s.equity for s in equity_curve]
|
|
x_mean = statistics.mean(xs)
|
|
y_mean = statistics.mean(ys)
|
|
ss_xx = sum((x - x_mean) ** 2 for x in xs)
|
|
ss_xy = sum((x - x_mean) * (y - y_mean) for x, y in zip(xs, ys))
|
|
ss_yy = sum((y - y_mean) ** 2 for y in ys)
|
|
if ss_xx == 0 or ss_yy == 0:
|
|
return None
|
|
r = ss_xy / math.sqrt(ss_xx * ss_yy)
|
|
return r ** 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Practicality metrics (4)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def compute_avg_holding_days(trades: list[FilledTrade]) -> float | None:
|
|
if not trades:
|
|
return None
|
|
return statistics.mean(t.holding_days for t in trades)
|
|
|
|
|
|
def compute_stop_exit_rate(trades: list[FilledTrade]) -> float | None:
|
|
from libs.backtest.domain import ExitReason
|
|
|
|
if not trades:
|
|
return None
|
|
stops = sum(1 for t in trades if t.exit_reason in (ExitReason.STOP, ExitReason.TRAILING))
|
|
return stops / len(trades)
|
|
|
|
|
|
def compute_target_exit_rate(trades: list[FilledTrade]) -> float | None:
|
|
from libs.backtest.domain import ExitReason
|
|
|
|
if not trades:
|
|
return None
|
|
targets = sum(1 for t in trades if t.exit_reason == ExitReason.TARGET)
|
|
return targets / len(trades)
|
|
|
|
|
|
def compute_no_follow_through_rate(trades: list[FilledTrade]) -> float | None:
|
|
from libs.backtest.domain import ExitReason
|
|
|
|
if not trades:
|
|
return None
|
|
nft = sum(1 for t in trades if t.exit_reason == ExitReason.NO_FOLLOW_THROUGH)
|
|
return nft / len(trades)
|
|
|
|
|
|
def compute_score_bucket_hit_rate(trades: list[FilledTrade], candidate_map: dict[str, object]) -> dict[str, float]:
|
|
"""Win rate per score_bucket (uses trade_id -> candidate mapping)."""
|
|
bucket_wins: dict[str, int] = defaultdict(int)
|
|
bucket_total: dict[str, int] = defaultdict(int)
|
|
for t in trades:
|
|
cand = candidate_map.get(t.trade_id)
|
|
if cand is None:
|
|
continue
|
|
bucket = getattr(cand, "score_bucket", "unknown")
|
|
bucket_total[bucket] += 1
|
|
if t.net_pnl > 0:
|
|
bucket_wins[bucket] += 1
|
|
return {
|
|
b: bucket_wins[b] / bucket_total[b]
|
|
for b in bucket_total
|
|
if bucket_total[b] > 0
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Bootstrap confidence intervals
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def bootstrap_ci(
|
|
trades: list[FilledTrade],
|
|
metric_fn: callable,
|
|
n_iterations: int = 1000,
|
|
ci_level: float = 0.95,
|
|
seed: int = 42,
|
|
) -> tuple[float, float] | None:
|
|
"""Compute bootstrap confidence interval for a trade-level metric.
|
|
|
|
Args:
|
|
trades: List of FilledTrade objects.
|
|
metric_fn: Function that takes list[FilledTrade] and returns float | None.
|
|
n_iterations: Number of bootstrap resamples.
|
|
ci_level: Confidence level (default 0.95 for 95% CI).
|
|
seed: Random seed for reproducibility.
|
|
|
|
Returns:
|
|
(lower, upper) bounds or None if metric can't be computed.
|
|
"""
|
|
if len(trades) < 5:
|
|
return None
|
|
|
|
rng = random.Random(seed)
|
|
results = []
|
|
for _ in range(n_iterations):
|
|
sample = rng.choices(trades, k=len(trades))
|
|
val = metric_fn(sample)
|
|
if val is not None:
|
|
results.append(val)
|
|
|
|
if len(results) < n_iterations * 0.5:
|
|
return None
|
|
|
|
results.sort()
|
|
alpha = (1 - ci_level) / 2
|
|
lo_idx = int(alpha * len(results))
|
|
hi_idx = int((1 - alpha) * len(results)) - 1
|
|
return (results[lo_idx], results[hi_idx])
|
|
|
|
|
|
def compute_bootstrap_cis(
|
|
trades: list[FilledTrade],
|
|
n_iterations: int = 1000,
|
|
seed: int = 42,
|
|
) -> dict[str, tuple[float, float] | None]:
|
|
"""Compute 95% bootstrap CIs for key trade metrics."""
|
|
metrics_fns = {
|
|
"win_rate": compute_win_rate,
|
|
"avg_win_pct": compute_avg_win_pct,
|
|
"avg_loss_pct": compute_avg_loss_pct,
|
|
"profit_factor": compute_profit_factor,
|
|
"expectancy_r": compute_expectancy_r,
|
|
}
|
|
return {
|
|
f"{name}_ci_95": bootstrap_ci(trades, fn, n_iterations=n_iterations, seed=seed)
|
|
for name, fn in metrics_fns.items()
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Builder
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def build_metrics_bundle(
|
|
trades: list[FilledTrade],
|
|
equity_curve: list[DailyPortfolioState],
|
|
candidate_map: dict[str, object] | None = None,
|
|
) -> MetricsBundle:
|
|
from libs.backtest.domain import MetricsBundle
|
|
|
|
ann_ret = compute_annualized_return_pct(equity_curve)
|
|
max_dd = compute_max_drawdown_pct(equity_curve)
|
|
|
|
# Bootstrap CIs (only when enough trades)
|
|
cis = compute_bootstrap_cis(trades) if len(trades) >= 5 else {}
|
|
|
|
return MetricsBundle(
|
|
# Trade
|
|
trade_count=len(trades),
|
|
win_rate=compute_win_rate(trades),
|
|
avg_win_pct=compute_avg_win_pct(trades),
|
|
avg_loss_pct=compute_avg_loss_pct(trades),
|
|
profit_factor=compute_profit_factor(trades),
|
|
expectancy_r=compute_expectancy_r(trades),
|
|
avg_r_multiple=compute_avg_r_multiple(trades),
|
|
# Portfolio
|
|
total_return_pct=compute_total_return_pct(equity_curve),
|
|
annualized_return_pct=ann_ret,
|
|
max_drawdown_pct=max_dd,
|
|
calmar_ratio=compute_calmar_ratio(ann_ret, max_dd),
|
|
sharpe_ratio=compute_sharpe_ratio(equity_curve),
|
|
sortino_ratio=compute_sortino_ratio(equity_curve),
|
|
avg_daily_pnl=compute_avg_daily_pnl(equity_curve),
|
|
avg_positions_held=compute_avg_positions_held(equity_curve),
|
|
avg_gross_exposure_pct=compute_avg_gross_exposure_pct(equity_curve),
|
|
avg_net_exposure_pct=compute_avg_net_exposure_pct(equity_curve),
|
|
days_in_market_pct=compute_days_in_market_pct(equity_curve),
|
|
# Stability
|
|
trade_skewness=compute_trade_skewness(trades),
|
|
trade_kurtosis=compute_trade_kurtosis(trades),
|
|
monthly_win_rate=compute_monthly_win_rate(trades),
|
|
equity_curve_r_squared=compute_equity_curve_r_squared(equity_curve),
|
|
# Practicality
|
|
avg_holding_days=compute_avg_holding_days(trades),
|
|
stop_exit_rate=compute_stop_exit_rate(trades),
|
|
target_exit_rate=compute_target_exit_rate(trades),
|
|
no_follow_through_exit_rate=compute_no_follow_through_rate(trades),
|
|
score_bucket_hit_rate=compute_score_bucket_hit_rate(trades, candidate_map or {}),
|
|
# Bootstrap CIs
|
|
bootstrap_cis=cis,
|
|
)
|