Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .claude/commands/release.md
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,6 @@ Create release: $ARGUMENTS
1. Read @README.md and @CONTRIBUTING.md file to understand how to publish this release.
2. Bump version in pyproject.toml
3. Perform the git cli commands
4. Use gh to create the release and publish
4. Use gh to create the release and publish (allow gh to generate notes)

NOTE: The PR will run a few guardrail checks, so you'll need to wait for these to complete before merging.
74 changes: 4 additions & 70 deletions notebooks/02-idc_and_metrics_walkthrough.ipynb

Large diffs are not rendered by default.

76 changes: 65 additions & 11 deletions src/nexa_backtest/analysis/metrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@
import math
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import date
from datetime import date, timedelta
from decimal import Decimal
from pathlib import Path
from typing import TYPE_CHECKING
Expand Down Expand Up @@ -100,7 +100,8 @@ def compute_sharpe(equity_snapshots: list[EquitySnapshot]) -> Decimal | None:

Uses percentage returns between consecutive equity snapshots.
Annualisation factor is derived from the actual average observation
frequency using ``sqrt(252 * periods_per_day)``.
frequency using ``sqrt(365 * periods_per_day)`` (energy markets
operate every day of the year, unlike equities' 252-day convention).

Args:
equity_snapshots: Chronological list of equity snapshots.
Expand Down Expand Up @@ -129,7 +130,7 @@ def compute_sharpe(equity_snapshots: list[EquitySnapshot]) -> Decimal | None:
total_days = max(total_seconds / 86400.0, 1.0)
n_obs = len(returns)
periods_per_day = n_obs / total_days
annualisation = math.sqrt(252.0 * periods_per_day)
annualisation = math.sqrt(365.0 * periods_per_day)

sharpe = (mean_ret / std_ret) * annualisation
return Decimal(str(round(sharpe, 6)))
Expand Down Expand Up @@ -167,6 +168,38 @@ def compute_max_drawdown(
return max_dd, max_dd_pct


def compute_time_in_drawdown(
equity_snapshots: list[EquitySnapshot],
) -> timedelta:
"""Compute total time spent in drawdown (equity below its running peak).

Walks the equity curve and accumulates the duration of periods where
equity is strictly below the running peak.

Args:
equity_snapshots: Chronological list of equity snapshots.

Returns:
Total time in drawdown as a ``timedelta``. Zero if fewer than two
snapshots or the curve is monotonically increasing.
"""
if len(equity_snapshots) < 2:
return timedelta(0)

peak = equity_snapshots[0].total_equity
total = timedelta(0)

for i in range(1, len(equity_snapshots)):
prev = equity_snapshots[i - 1]
curr = equity_snapshots[i]
if prev.total_equity < peak:
total += curr.timestamp - prev.timestamp
if curr.total_equity > peak:
peak = curr.total_equity

return total


def compute_profit_factor(
fills: list[Fill],
market_vwap: Decimal,
Expand Down Expand Up @@ -217,6 +250,7 @@ class BacktestResult:
sharpe_ratio: Annualised Sharpe ratio, or ``None`` if < 2 snapshots.
max_drawdown: Largest peak-to-trough drawdown in EUR.
max_drawdown_pct: Max drawdown as a fraction of peak equity.
time_in_drawdown: Total time the equity curve spent below its peak.
profit_factor: Sum of gains / sum of losses, or ``None`` if no losses.
avg_trade_pnl: Mean per-fill PnL in EUR.
best_trade: Fill with the highest individual PnL, or ``None``.
Expand All @@ -242,6 +276,7 @@ class BacktestResult:
sharpe_ratio: Decimal | None = None
max_drawdown: Decimal = Decimal("0")
max_drawdown_pct: Decimal = Decimal("0")
time_in_drawdown: timedelta = field(default_factory=lambda: timedelta(0))
profit_factor: Decimal | None = None
avg_trade_pnl: Decimal = Decimal("0")
best_trade: Fill | None = None
Expand Down Expand Up @@ -271,13 +306,16 @@ def summary(self) -> str:
total_pnl = p.total_alpha_eur
final_equity = self.initial_capital + total_pnl

total_volume = sum((f.volume for f in self.fills), Decimal("0"))

lines: list[str] = [
sep,
f" Backtest Results: {self.start} to {self.end} ({self.duration_days} days)",
f" Exchange: {self.exchange} | Algo: {self.algo_name}",
sep,
"",
f" Total PnL: {total_pnl:>+14,.2f} EUR",
f" PnL/MWh: {float(p.pnl_per_mwh):>+14.2f} EUR/MWh",
f" Market VWAP: {p.market_vwap:>14.2f} EUR/MWh",
]

Expand All @@ -291,22 +329,26 @@ def summary(self) -> str:
f" Max Drawdown: {-float(self.max_drawdown):>+14,.2f} EUR ({-dd_pct:.1f}%)"
)

dd_hours = self.time_in_drawdown.total_seconds() / 3600
if dd_hours >= 24:
dd_days = self.time_in_drawdown.days
dd_rem_hours = self.time_in_drawdown.seconds // 3600
lines.append(f" Time in Drawdown: {dd_days:>10d}d {dd_rem_hours:02d}h")
else:
lines.append(f" Time in Drawdown: {dd_hours:>13.1f}h")

if self.profit_factor is not None:
lines.append(f" Profit Factor: {float(self.profit_factor):>14.2f}")
else:
lines.append(f" Profit Factor: {'N/A (no losses)':>14}")

total_volume = sum((f.volume for f in self.fills), Decimal("0"))
win_rate = (
(p.buys.win_rate * p.buys.count + p.sells.win_rate * p.sells.count) / len(self.fills)
if self.fills
else 0.0
)

lines += [
"",
f" Trades: {len(self.fills):>14,d}",
f" Win Rate: {win_rate:>14.1%}",
f" Win Rate: {p.win_rate:>14.1%}",
f" Loss Rate: {p.loss_rate:>14.1%}",
f" Longs: {p.long_pct:>14.1%}",
f" Shorts: {p.short_pct:>14.1%}",
f" Avg Trade PnL: {float(self.avg_trade_pnl):>+14,.2f} EUR",
]

Expand Down Expand Up @@ -584,7 +626,13 @@ def _snap_to_dict(s: EquitySnapshot) -> dict[str, str]:
"sharpe_ratio": str(self.sharpe_ratio) if self.sharpe_ratio is not None else None,
"max_drawdown": str(self.max_drawdown),
"max_drawdown_pct": str(self.max_drawdown_pct),
"time_in_drawdown_seconds": self.time_in_drawdown.total_seconds(),
"profit_factor": str(self.profit_factor) if self.profit_factor is not None else None,
"pnl_per_mwh": str(self.pnl.pnl_per_mwh),
"win_rate": self.pnl.win_rate,
"loss_rate": self.pnl.loss_rate,
"long_pct": self.pnl.long_pct,
"short_pct": self.pnl.short_pct,
"avg_trade_pnl": str(self.avg_trade_pnl),
"trade_count": len(self.fills),
"equity_curve": [_snap_to_dict(s) for s in self.equity_curve],
Expand Down Expand Up @@ -624,7 +672,13 @@ def to_parquet(self, path: str) -> None:
"sharpe_ratio": str(self.sharpe_ratio) if self.sharpe_ratio is not None else None,
"max_drawdown": str(self.max_drawdown),
"max_drawdown_pct": str(self.max_drawdown_pct),
"time_in_drawdown_seconds": self.time_in_drawdown.total_seconds(),
"profit_factor": str(self.profit_factor) if self.profit_factor is not None else None,
"pnl_per_mwh": str(self.pnl.pnl_per_mwh),
"win_rate": self.pnl.win_rate,
"loss_rate": self.pnl.loss_rate,
"long_pct": self.pnl.long_pct,
"short_pct": self.pnl.short_pct,
"avg_trade_pnl": str(self.avg_trade_pnl),
"trade_count": len(self.fills),
}
Expand Down
26 changes: 26 additions & 0 deletions src/nexa_backtest/analysis/pnl.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,12 +51,22 @@ class PnlSummary:
buys: Statistics for the algo's buy fills.
sells: Statistics for the algo's sell fills.
total_alpha_eur: Combined VWAP alpha across buys and sells in EUR.
pnl_per_mwh: Alpha per MWh traded (``total_alpha_eur / total_volume``).
win_rate: Combined win rate across buys and sells (fill-weighted).
loss_rate: ``1 - win_rate``.
long_pct: Fraction of fills that were buys.
short_pct: Fraction of fills that were sells.
"""

market_vwap: Decimal
buys: SideSummary
sells: SideSummary
total_alpha_eur: Decimal
pnl_per_mwh: Decimal = Decimal("0")
win_rate: float = 0.0
loss_rate: float = 1.0
long_pct: float = 0.0
short_pct: float = 0.0


def _side_summary(
Expand Down Expand Up @@ -157,10 +167,26 @@ def compute_pnl(
sells = _side_summary(fills, market_vwap, Side.SELL, product_vwaps)

total_alpha = buys.total_alpha_eur + sells.total_alpha_eur
total_volume = buys.volume_mwh + sells.volume_mwh
total_count = buys.count + sells.count

pnl_per_mwh = total_alpha / total_volume if total_volume > 0 else Decimal("0")
win_rate = (
(buys.win_rate * buys.count + sells.win_rate * sells.count) / total_count
if total_count > 0
else 0.0
)
long_pct = buys.count / total_count if total_count > 0 else 0.0
short_pct = sells.count / total_count if total_count > 0 else 0.0

return PnlSummary(
market_vwap=market_vwap,
buys=buys,
sells=sells,
total_alpha_eur=total_alpha,
pnl_per_mwh=pnl_per_mwh,
win_rate=win_rate,
loss_rate=1.0 - win_rate,
long_pct=long_pct,
short_pct=short_pct,
)
14 changes: 14 additions & 0 deletions src/nexa_backtest/analysis/report.py
Original file line number Diff line number Diff line change
Expand Up @@ -194,11 +194,20 @@ def _metric_rows(result: BacktestResult) -> list[tuple[str, str]]:
final_equity = result.initial_capital + total_pnl
total_volume = sum((f.volume for f in result.fills), Decimal("0"))

dd_hours = result.time_in_drawdown.total_seconds() / 3600
if dd_hours >= 24:
dd_days = result.time_in_drawdown.days
dd_rem_hours = result.time_in_drawdown.seconds // 3600
dd_str = f"{dd_days}d {dd_rem_hours:02d}h"
else:
dd_str = f"{dd_hours:.1f}h"

rows: list[tuple[str, str]] = [
("Period", f"{result.start} — {result.end} ({result.duration_days} days)"),
("Exchange", result.exchange),
("Algo", result.algo_name),
("Total PnL", f"{total_pnl:+,.2f} EUR"),
("PnL/MWh", f"{float(p.pnl_per_mwh):+.2f} EUR/MWh"),
("Market VWAP", f"{p.market_vwap:.2f} EUR/MWh"),
(
"Sharpe Ratio",
Expand All @@ -211,11 +220,16 @@ def _metric_rows(result: BacktestResult) -> list[tuple[str, str]]:
f" ({-float(result.max_drawdown_pct) * 100:.1f}%)"
),
),
("Time in Drawdown", dd_str),
(
"Profit Factor",
f"{float(result.profit_factor):.2f}" if result.profit_factor is not None else "N/A",
),
("Trade Count", f"{len(result.fills):,d}"),
("Win Rate", f"{p.win_rate:.1%}"),
("Loss Rate", f"{p.loss_rate:.1%}"),
("Longs", f"{p.long_pct:.1%}"),
("Shorts", f"{p.short_pct:.1%}"),
("Total Volume", f"{float(total_volume):,.1f} MW"),
("Avg Trade PnL", f"{float(result.avg_trade_pnl):+,.2f} EUR"),
("Initial Capital", f"{float(result.initial_capital):,.2f} EUR"),
Expand Down
3 changes: 3 additions & 0 deletions src/nexa_backtest/engines/backtest.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@
compute_max_drawdown,
compute_profit_factor,
compute_sharpe,
compute_time_in_drawdown,
)
from nexa_backtest.analysis.pnl import compute_pnl
from nexa_backtest.analysis.vwap import compute_idc_vwaps, compute_market_vwap
Expand Down Expand Up @@ -888,6 +889,7 @@ def run(self) -> BacktestResult:

sharpe = compute_sharpe(equity_snapshots)
max_dd, max_dd_pct = compute_max_drawdown(equity_snapshots)
dd_time = compute_time_in_drawdown(equity_snapshots)
product_vwaps_for_metrics = per_product_vwap or {}
profit_fac = compute_profit_factor(
all_fills,
Expand Down Expand Up @@ -930,6 +932,7 @@ def run(self) -> BacktestResult:
sharpe_ratio=sharpe,
max_drawdown=max_dd,
max_drawdown_pct=max_dd_pct,
time_in_drawdown=dd_time,
profit_factor=profit_fac,
avg_trade_pnl=avg_trade,
best_trade=best_fill,
Expand Down
3 changes: 3 additions & 0 deletions src/nexa_backtest/engines/shared.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,6 +45,7 @@
compute_max_drawdown,
compute_profit_factor,
compute_sharpe,
compute_time_in_drawdown,
)
from nexa_backtest.analysis.pnl import compute_pnl
from nexa_backtest.analysis.vwap import compute_idc_vwaps, compute_market_vwap
Expand Down Expand Up @@ -837,6 +838,7 @@ def _build_result(self, runner: _AlgoRunner, zone: str) -> BacktestResult:

sharpe = compute_sharpe(equity_snapshots)
max_dd, max_dd_pct = compute_max_drawdown(equity_snapshots)
dd_time = compute_time_in_drawdown(equity_snapshots)
product_vwaps_for_metrics = per_product_vwap or {}
profit_fac = compute_profit_factor(
all_fills, market_vwap, product_vwaps_for_metrics or None
Expand Down Expand Up @@ -881,6 +883,7 @@ def _build_result(self, runner: _AlgoRunner, zone: str) -> BacktestResult:
sharpe_ratio=sharpe,
max_drawdown=max_dd,
max_drawdown_pct=max_dd_pct,
time_in_drawdown=dd_time,
profit_factor=profit_fac,
avg_trade_pnl=avg_trade,
best_trade=best_fill,
Expand Down
72 changes: 72 additions & 0 deletions tests/test_analysis/test_metrics.py
Original file line number Diff line number Diff line change
Expand Up @@ -201,3 +201,75 @@ def test_uses_product_vwaps_consistently(self) -> None:

product_result = compute_profit_factor([fill], Decimal("50"), product_vwaps)
assert product_result is None # gain vs product VWAP → no losses


class TestComputeTimeInDrawdown:
"""Tests for compute_time_in_drawdown."""

def test_empty_snapshots_returns_zero(self) -> None:
from datetime import timedelta

from nexa_backtest.analysis.metrics import compute_time_in_drawdown

assert compute_time_in_drawdown([]) == timedelta(0)

def test_single_snapshot_returns_zero(self) -> None:
from datetime import timedelta

from nexa_backtest.analysis.metrics import compute_time_in_drawdown
from nexa_backtest.types import EquitySnapshot

snap = EquitySnapshot(
timestamp=datetime(2026, 3, 1, 10, 0, tzinfo=UTC),
realised_pnl=Decimal("100"),
unrealised_pnl=Decimal("0"),
total_equity=Decimal("1100"),
cash=Decimal("1100"),
net_position_mw=Decimal("0"),
)
assert compute_time_in_drawdown([snap]) == timedelta(0)

def test_monotonically_increasing_no_drawdown(self) -> None:
from datetime import timedelta

from nexa_backtest.analysis.metrics import compute_time_in_drawdown
from nexa_backtest.types import EquitySnapshot

snaps = [
EquitySnapshot(
timestamp=datetime(2026, 3, 1, h, 0, tzinfo=UTC),
realised_pnl=Decimal(str(h * 10)),
unrealised_pnl=Decimal("0"),
total_equity=Decimal(str(1000 + h * 10)),
cash=Decimal(str(1000 + h * 10)),
net_position_mw=Decimal("0"),
)
for h in range(10, 15)
]
assert compute_time_in_drawdown(snaps) == timedelta(0)

def test_drawdown_period_measured(self) -> None:
from datetime import timedelta

from nexa_backtest.analysis.metrics import compute_time_in_drawdown
from nexa_backtest.types import EquitySnapshot

def _snap(hour: int, equity: int) -> EquitySnapshot:
return EquitySnapshot(
timestamp=datetime(2026, 3, 1, hour, 0, tzinfo=UTC),
realised_pnl=Decimal(str(equity - 1000)),
unrealised_pnl=Decimal("0"),
total_equity=Decimal(str(equity)),
cash=Decimal(str(equity)),
net_position_mw=Decimal("0"),
)

# Peak at h=10 (1100), dips at h=11 (1050), h=12 (1080), recovers h=13 (1100)
snaps = [
_snap(10, 1100), # peak
_snap(11, 1050), # below peak → drawdown starts
_snap(12, 1080), # still below peak
_snap(13, 1100), # recovered to peak
]
# Drawdown time: h11→h12 (1h) + h12→h13 (1h) = 2h
assert compute_time_in_drawdown(snaps) == timedelta(hours=2)
Loading
Loading