"""Single-symbol, long-only backtest engine.
The fill model is the load-bearing decision here: a signal computed from bar
N's close fills at bar N+1's open. Filling at the signal bar's own close would
assume knowledge of a price that had not yet happened, and that lookahead bias
inflates mean-reversion results most — which is most of these strategies.
Every result also carries a **buy-and-hold baseline** over the same bars and the
same costs, because a return reported alone is uninterpretable: `rsi_revert_14`
returning +20.98% reads like a win until the 54.78% it gave up is beside it.
"""
from collections.abc import Sequence
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
from trader.backtest.metrics import (
buy_and_hold,
max_drawdown_pct,
total_return_pct,
win_loss,
)
from trader.domain import Bar, Position
from trader.errors import BacktestError
from trader.strategies.base import Action, Strategy
__all__ = ["BacktestConfig", "BacktestResult", "SimulatedTrade", "run_backtest"]
_BPS = Decimal(10_000)
[docs]
@dataclass(frozen=True, slots=True)
class BacktestConfig:
"""Execution assumptions, identical across strategies so they stay comparable."""
starting_capital: Decimal = Decimal("100000")
slippage_bps: int = 5
position_fraction: Decimal = Decimal("1.0")
[docs]
@dataclass(frozen=True, slots=True)
class SimulatedTrade:
"""One simulated round trip. `exit_at is None` means still open at the end."""
entry_at: datetime
entry_price: Decimal
quantity: int
exit_at: datetime | None = None
exit_price: Decimal | None = None
pnl: Decimal | None = None
[docs]
@dataclass(frozen=True, slots=True)
class BacktestResult:
"""Everything requirements §12 asks a backtest to report."""
symbol: str
strategy_id: str
starting_value: Decimal
ending_value: Decimal
total_return_pct: Decimal
max_drawdown_pct: Decimal
win_count: int
loss_count: int
#: The do-nothing baseline over the same bars and the same costs: buy at the
#: first price the strategy could itself have filled at, hold to the last
#: close. Not optional and without a default on purpose — a result that can
#: be constructed without a baseline is a result someone will report without
#: one, which is the whole failure this field exists to close.
benchmark_ending_value: Decimal
benchmark_return_pct: Decimal
trades: list[SimulatedTrade] = field(default_factory=list)
@property
def excess_return_pct(self) -> Decimal:
"""How much the strategy added over simply holding. Negative is common."""
return self.total_return_pct - self.benchmark_return_pct
[docs]
def run_backtest(
strategy: Strategy,
symbol: str,
bars: Sequence[Bar],
config: BacktestConfig,
) -> BacktestResult:
"""Simulate `strategy` over `bars`, long-only, one position at a time.
Raises:
BacktestError: the window is too short for the strategy to decide once.
"""
# The decision loop is `range(warmup, len(bars) - 1)`, so it runs at least
# once only when `len(bars) >= warmup + 2`. Below that the simulation is
# vacuous, and reporting 0 trades / 0.00% return / 0.00% drawdown would be
# indistinguishable from a strategy that legitimately stood still.
required = strategy.warmup_bars() + 2
if len(bars) < required:
raise BacktestError(
f"{strategy.id} needs at least {required} bars for a decision "
f"(warmup {strategy.warmup_bars()} plus one to decide on and one to "
f"fill against), got {len(bars)}. Widen the date range or check the "
"ticker."
)
slip = Decimal(config.slippage_bps) / _BPS
# The baseline, computed before the simulation so a window that cannot
# produce one fails before any numbers exist to report without it.
#
# It buys at `bars[warmup + 1].open`, which is the *earliest price the
# strategy itself could have filled at*: the first decision happens on
# `bars[warmup]` and fills on the bar after. Starting the baseline at
# `bars[0]` instead would credit it with a move the strategy was
# structurally unable to capture, and `required` above guarantees this index
# exists. It pays the same slippage on entry, and is marked at `bars[-1]`'s
# close rather than sold, exactly as `ending_value` marks a still-open
# position — a baseline that paid an exit cost the strategy did not, or
# dodged one it did, flatters whichever side got the easier deal.
benchmark_buy_price = bars[strategy.warmup_bars() + 1].open * (Decimal(1) + slip)
if benchmark_buy_price <= 0:
raise BacktestError(
f"Cannot price a buy-and-hold baseline for {symbol}: the first "
f"fillable bar ({bars[strategy.warmup_bars() + 1].timestamp:%Y-%m-%d}) "
f"opens at {bars[strategy.warmup_bars() + 1].open}. A strategy return "
"with no baseline beside it is not interpretable, so this refuses "
"rather than reporting one."
)
benchmark_ending_value, benchmark_return = buy_and_hold(
config.starting_capital, benchmark_buy_price, bars[-1].close
)
cash = config.starting_capital
quantity = 0
entry_at: datetime | None = None
entry_price = Decimal(0)
trades: list[SimulatedTrade] = []
# Seed with the opening balance: without a pre-trade point, a decline that
# begins on the very first filled bar has no peak to measure against and
# reports as zero drawdown.
equity_curve: list[Decimal] = [config.starting_capital]
# Stop one short: every decision needs a following bar to fill against. The
# `required` check above guarantees this range is non-empty.
for i in range(strategy.warmup_bars(), len(bars) - 1):
# A real Position, not a flag: a strategy deciding whether to exit needs
# the entry price and the P/L. Priced at THIS bar's close, never the
# next bar's open — the fill price belongs to a bar the decision cannot
# see, and using it would report a P/L the account never held at
# decision time. Same rule as the equity curve below.
decision_bar = bars[i]
held = (
Position(
symbol=decision_bar.symbol,
quantity=Decimal(quantity),
avg_entry_price=entry_price,
current_price=decision_bar.close,
market_value=decision_bar.close * quantity,
unrealized_pl=(decision_bar.close - entry_price) * quantity,
)
if quantity > 0
else None
)
signal = strategy.evaluate(bars[: i + 1], held)
fill_bar = bars[i + 1]
if signal.action is Action.BUY and quantity == 0:
price = fill_bar.open * (Decimal(1) + slip)
affordable = int((cash * config.position_fraction) // price)
if affordable > 0:
quantity = affordable
entry_price = price
entry_at = fill_bar.timestamp
cash -= price * quantity
elif signal.action is Action.SELL and quantity > 0:
price = fill_bar.open * (Decimal(1) - slip)
proceeds = price * quantity
cash += proceeds
trades.append(
SimulatedTrade(
entry_at=entry_at,
entry_price=entry_price,
quantity=quantity,
exit_at=fill_bar.timestamp,
exit_price=price,
pnl=proceeds - entry_price * quantity,
)
)
quantity = 0
entry_at = None
# Mark at the close of the bar the fill happened on, not bar i's close.
# A position bought at bars[i + 1].open never existed at bars[i].close,
# so valuing it there invents a peak (or trough) the account never held
# — and every overnight gap becomes a phantom drawdown.
equity_curve.append(cash + quantity * fill_bar.close)
if quantity > 0 and entry_at is not None:
trades.append(
SimulatedTrade(entry_at=entry_at, entry_price=entry_price, quantity=quantity)
)
ending_value = cash + quantity * bars[-1].close
equity_curve.append(ending_value)
wins, losses = win_loss([t.pnl for t in trades if t.pnl is not None])
return BacktestResult(
symbol=symbol,
strategy_id=strategy.id,
starting_value=config.starting_capital,
ending_value=ending_value,
total_return_pct=total_return_pct(config.starting_capital, ending_value),
max_drawdown_pct=max_drawdown_pct(equity_curve),
win_count=wins,
loss_count=losses,
benchmark_ending_value=benchmark_ending_value,
benchmark_return_pct=benchmark_return,
trades=trades,
)