feat(backtester): extend compute_metrics with alpha/beta/winners/best
Some checks failed
ci/woodpecker/push/woodpecker Pipeline was canceled

In-place extension (no fork). Existing tests still pass; new fields are
optional and None when no benchmark is supplied.
This commit is contained in:
Viktor Barzin 2026-05-24 00:57:42 +00:00
parent 23ce45a4f2
commit cd75c4ab7e
2 changed files with 121 additions and 3 deletions

View file

@ -185,9 +185,7 @@ def _populate_dict_trade_aggregates(
if not trade_log:
return
closed = [
t for t in trade_log if t.get("pnl_pct") is not None
]
closed = [t for t in trade_log if t.get("pnl_pct") is not None]
if not closed:
return
@ -203,6 +201,27 @@ def _populate_dict_trade_aggregates(
elif avg_win > 0:
result.avg_win_loss_ratio = float("inf")
# Kevin extensions: winners / losers / best / worst (use Decimal pnl_pct if present)
winners_d = [t for t in closed if Decimal(str(t["pnl_pct"])) > 0]
losers_d = [t for t in closed if Decimal(str(t["pnl_pct"])) <= 0]
if winners_d:
total = sum(Decimal(str(t["pnl_pct"])) for t in winners_d)
result.avg_winner_pct = total / Decimal(len(winners_d))
if losers_d:
total_l = sum(Decimal(str(t["pnl_pct"])) for t in losers_d)
result.avg_loser_pct = total_l / Decimal(len(losers_d))
if closed:
best = max(closed, key=lambda t: Decimal(str(t["pnl_pct"])))
worst = min(closed, key=lambda t: Decimal(str(t["pnl_pct"])))
result.best_trade = {
"symbol": best["symbol"],
"pnl_pct": Decimal(str(best["pnl_pct"])),
}
result.worst_trade = {
"symbol": worst["symbol"],
"pnl_pct": Decimal(str(worst["pnl_pct"])),
}
def _populate_benchmark_metrics(
result: BacktestResult,

View file

@ -0,0 +1,99 @@
"""Kevin-extension tests for compute_metrics (alpha/beta/winners/losers/best/worst)."""
from datetime import datetime, timedelta, timezone
from decimal import Decimal
import pandas as pd
from backtester.metrics import compute_metrics
def _ts(day):
return datetime(2026, 5, 15, tzinfo=timezone.utc) + timedelta(days=day)
def test_compute_metrics_returns_alpha_and_beta_with_benchmark():
spy = pd.DataFrame(
{
"open": [500, 500, 510, 510, 520],
"high": [500] * 5,
"low": [500] * 5,
"close": [500, 500, 510, 510, 520],
"volume": [0] * 5,
},
index=pd.date_range("2026-05-15", periods=5, freq="D", tz="UTC"),
)
trades = [
{
"symbol": "NVDA",
"source_mention_id": 1,
"entry_at": _ts(0),
"entry_price": Decimal("100"),
"exit_at": _ts(4),
"exit_price": Decimal("110"),
"qty": Decimal("10"),
"pnl_usd": Decimal("100"),
"pnl_pct": Decimal("10"),
"holding_days_actual": 4,
}
]
curve = [(_ts(i), Decimal(str(100000 + 25 * i))) for i in range(5)]
result = compute_metrics(
trade_log=trades,
equity_curve=curve,
initial_capital=Decimal("100000"),
benchmark_bars=spy,
)
assert result.alpha_vs_spy_pct is not None
assert result.beta_vs_spy is not None
def test_compute_metrics_returns_winners_losers_best_worst():
trades = [
{
"symbol": "NVDA",
"source_mention_id": 1,
"entry_at": _ts(0),
"entry_price": Decimal("100"),
"exit_at": _ts(5),
"exit_price": Decimal("110"),
"qty": Decimal("10"),
"pnl_usd": Decimal("100"),
"pnl_pct": Decimal("10"),
"holding_days_actual": 5,
},
{
"symbol": "INTC",
"source_mention_id": 2,
"entry_at": _ts(0),
"entry_price": Decimal("50"),
"exit_at": _ts(5),
"exit_price": Decimal("45"),
"qty": Decimal("20"),
"pnl_usd": Decimal("-100"),
"pnl_pct": Decimal("-10"),
"holding_days_actual": 5,
},
]
result = compute_metrics(
trade_log=trades,
equity_curve=[(_ts(0), Decimal("100000"))],
initial_capital=Decimal("100000"),
)
assert result.avg_winner_pct == Decimal("10")
assert result.avg_loser_pct == Decimal("-10")
assert result.best_trade == {"symbol": "NVDA", "pnl_pct": Decimal("10")}
assert result.worst_trade == {"symbol": "INTC", "pnl_pct": Decimal("-10")}
def test_compute_metrics_no_benchmark_omits_alpha_beta():
result = compute_metrics(
trade_log=[],
equity_curve=[(_ts(0), Decimal("100000"))],
initial_capital=Decimal("100000"),
)
assert result.alpha_vs_spy_pct is None
assert result.beta_vs_spy is None