crypto_trading/backend/app/backtest/backtester.py

from __future__ import annotations

from collections.abc import Sequence

from app.backtest.metrics import compute_metrics
from app.broker.paper import PaperBroker
from app.engine.portfolio import (
    OpenPosition,
    account_equity,
    advance_trailing_stop,
    apply_close,
    apply_open,
    order_qty,
    realized_pnl,
    risk_exit,
)
from app.engine.strategies.base import Strategy, StrategyContext
from app.models import OrderSide, PositionSide
from app.schemas import BacktestPointOut, BacktestResult, BacktestTradeOut, Candle


def _open_side(side: PositionSide) -> OrderSide:
    return OrderSide.BUY if side == PositionSide.LONG else OrderSide.SELL


def _close_side(side: PositionSide) -> OrderSide:
    return OrderSide.SELL if side == PositionSide.LONG else OrderSide.BUY


class Backtester:
    """Runs a strategy bar-by-bar over historical candles.

    Reuses the very same ``Strategy``/``StrategyContext`` interface and the same
    portfolio math as the live engine, so results carry over to paper/live.
    """

    def __init__(
        self,
        *,
        fee_bps: float = 10.0,
        slippage_bps: float = 5.0,
        starting_balance: float = 10_000.0,
    ) -> None:
        self.broker = PaperBroker(fee_bps=fee_bps, slippage_bps=slippage_bps)
        self.starting_balance = starting_balance

    def run(
        self,
        candles: Sequence[Candle],
        strategy: Strategy,
        symbol: str,
        timeframe: str,
    ) -> BacktestResult:
        starting = self.starting_balance
        cash = starting
        position: OpenPosition | None = None
        trades: list[BacktestTradeOut] = []
        curve: list[BacktestPointOut] = []
        error: str | None = None
        first_close = candles[0].close if candles else 0.0

        def close_at(level: float, ts: int, reason: str) -> None:
            nonlocal cash, position
            assert position is not None
            fill = self.broker.fill(symbol, _close_side(position.side), position.qty, level)
            cash = apply_close(cash, position.side, fill.qty, fill.price, fill.fee)
            pnl = realized_pnl(
                position.side,
                position.entry_price,
                fill.price,
                position.qty,
                position.entry_fee,
                fill.fee,
            )
            cost = position.entry_price * position.qty
            trades.append(
                BacktestTradeOut(
                    side=position.side.value,
                    entry_ts=position.entry_ts,
                    exit_ts=ts,
                    entry_price=position.entry_price,
                    exit_price=fill.price,
                    qty=position.qty,
                    pnl=pnl,
                    return_pct=(pnl / cost * 100) if cost else 0.0,
                    reason=reason,
                )
            )
            position = None

        try:
            for i, candle in enumerate(candles):
                mark = candle.close
                if position is not None:
                    if position.trailing_stop_pct > 0:
                        position.stop_loss = advance_trailing_stop(
                            position.side,
                            position.stop_loss,
                            position.trailing_stop_pct,
                            candle.high,
                            candle.low,
                        )
                    hit = risk_exit(
                        position.side,
                        position.stop_loss,
                        position.take_profit,
                        candle.high,
                        candle.low,
                    )
                    if hit is not None:
                        close_at(hit[1], candle.ts, hit[0])

                equity = (
                    account_equity(cash, position.side, position.qty, mark) if position else cash
                )
                ctx = StrategyContext(candles[: i + 1], position, equity, cash)
                strategy.on_candle(ctx)

                for intent in ctx.take_intents():
                    if intent.action == "close":
                        if position is not None:
                            close_at(mark, candle.ts, intent.reason or "strategy close")
                    elif intent.action in {"enter_long", "enter_short"} and position is None:
                        side = (
                            PositionSide.LONG
                            if intent.action == "enter_long"
                            else PositionSide.SHORT
                        )
                        size = intent.size if intent.size is not None else 1.0
                        qty = order_qty(cash, size, side, cash, mark)
                        if qty <= 0:
                            continue
                        fill = self.broker.fill(symbol, _open_side(side), qty, mark)
                        cash = apply_open(cash, side, fill.qty, fill.price, fill.fee)
                        position = OpenPosition(
                            side=side,
                            qty=fill.qty,
                            entry_price=fill.price,
                            entry_fee=fill.fee,
                            stop_loss=intent.stop_loss,
                            take_profit=intent.take_profit,
                            trailing_stop_pct=intent.trailing_stop_pct,
                            entry_ts=candle.ts,
                        )

                equity = (
                    account_equity(cash, position.side, position.qty, mark) if position else cash
                )
                curve.append(
                    BacktestPointOut(
                        ts=candle.ts,
                        equity=equity,
                        buy_hold=starting * (mark / first_close) if first_close else starting,
                        price=mark,
                    )
                )
        except Exception as exc:  # surface strategy/config errors in the result
            error = f"{type(exc).__name__}: {exc}"

        # Force-close any open position at the last close so trades are realized.
        if position is not None and candles and error is None:
            last = candles[-1]
            close_at(last.close, last.ts, "end of data")
            if curve:
                curve[-1] = BacktestPointOut(
                    ts=last.ts,
                    equity=cash,
                    buy_hold=curve[-1].buy_hold,
                    price=last.close,
                )

        return BacktestResult(
            strategy_key=strategy.key,
            strategy_name=strategy.name or strategy.key,
            symbol=symbol,
            timeframe=timeframe,
            starting_balance=starting,
            metrics=compute_metrics(curve, trades, starting, timeframe),
            equity_curve=curve,
            trades=trades,
            error=error,
        )