TradingBot-NewBot/engine/backtest.py

"""
Backtest engine.  Strategy-agnostic — the Strategy object owns all signal
and position-sizing logic; this engine handles order simulation and P&L
accounting.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, Protocol

import numpy as np
import pandas as pd


PIP_SIZE: dict[str, float] = {
    "JPY": 0.01,   # e.g. EURJPY
    "default": 0.0001,
}


def pip_size(symbol: str) -> float:
    return PIP_SIZE["JPY"] if symbol.endswith("JPY") else PIP_SIZE["default"]


# ── trade record ─────────────────────────────────────────────────────────────

@dataclass
class Trade:
    direction:   str      # 'long' | 'short'
    entry_time:  datetime
    entry_price: float
    sl:          float
    tp:          float
    risk_pct:    float    # fraction of balance risked (e.g. 0.01)
    exit_time:   Optional[datetime] = None
    exit_price:  Optional[float]    = None
    exit_reason: str                = ""
    pnl_pips:    float              = 0.0
    pnl_r:       float              = 0.0    # P&L in units of risk (R)

    @property
    def closed(self) -> bool:
        return self.exit_time is not None

    @property
    def duration_hours(self) -> float:
        if not self.closed:
            return 0.0
        return (self.exit_time - self.entry_time).total_seconds() / 3600  # type: ignore[operator]


# ── strategy protocol ─────────────────────────────────────────────────────────

class Strategy(Protocol):
    """Minimal interface every strategy must satisfy."""

    def prepare(self, df: pd.DataFrame) -> pd.DataFrame:
        """Compute and attach all indicator columns needed by `signal()`."""
        ...

    def signal(
        self,
        df: pd.DataFrame,
        i: int,
    ) -> Optional[tuple[str, float, float]]:
        """
        Called on each bar when no position is open.
        Returns (direction, sl_price, tp_price) or None.
        direction is 'long' or 'short'.
        Prices are for the bar at index i+1 (next open).
        """
        ...


# ── engine ────────────────────────────────────────────────────────────────────

class BacktestEngine:
    def __init__(
        self,
        symbol: str,
        initial_balance: float = 10_000.0,
        risk_per_trade: float  = 0.01,      # 1 % per trade
        spread_pips: float     = 0.5,        # half-spread each side
    ):
        self.symbol          = symbol
        self.initial_balance = initial_balance
        self.risk_per_trade  = risk_per_trade
        self.spread          = spread_pips * pip_size(symbol)

    def run(self, df: pd.DataFrame, strategy: Strategy) -> tuple[list[Trade], np.ndarray]:
        """
        Run the strategy over `df`.
        Returns (trades, equity_curve) where equity_curve has one value per bar.
        """
        df = strategy.prepare(df.copy())

        trades:   list[Trade] = []
        equity:   list[float] = [self.initial_balance]
        balance               = self.initial_balance
        position: Optional[Trade] = None

        for i in range(len(df) - 1):
            row      = df.iloc[i]
            next_bar = df.iloc[i + 1]

            # ── manage open position ─────────────────────────────────────────
            if position is not None:
                hi, lo = row["high"], row["low"]
                sl, tp = position.sl, position.tp
                ps     = pip_size(self.symbol)

                hit_sl = (position.direction == "long"  and lo <= sl) or \
                         (position.direction == "short" and hi >= sl)
                hit_tp = (position.direction == "long"  and hi >= tp) or \
                         (position.direction == "short" and lo <= tp)

                if hit_sl or hit_tp:
                    exit_px = sl if hit_sl else tp
                    reason  = "SL" if hit_sl else "TP"
                    raw     = exit_px - position.entry_price
                    pips    = (raw if position.direction == "long" else -raw) / ps
                    sl_dist = abs(position.entry_price - sl) / ps
                    r_mult  = pips / sl_dist if sl_dist > 0 else 0
                    pnl_pct = r_mult * position.risk_pct

                    position.exit_time  = row.name
                    position.exit_price = exit_px
                    position.exit_reason = reason
                    position.pnl_pips   = pips
                    position.pnl_r      = r_mult
                    balance            *= (1 + pnl_pct)
                    position = None

                equity.append(balance)
                continue

            # ── check for entry ──────────────────────────────────────────────
            result = strategy.signal(df, i)
            if result is None:
                equity.append(balance)
                continue

            direction, sl_px, tp_px = result
            entry_px = next_bar["open"]
            # apply spread (widen SL, narrow TP)
            if direction == "long":
                entry_px += self.spread
                sl_px    -= self.spread
                tp_px    -= self.spread
            else:
                entry_px -= self.spread
                sl_px    += self.spread
                tp_px    += self.spread

            trade = Trade(
                direction   = direction,
                entry_time  = next_bar.name,
                entry_price = entry_px,
                sl          = sl_px,
                tp          = tp_px,
                risk_pct    = self.risk_per_trade,
            )
            trades.append(trade)
            position = trade
            equity.append(balance)

        return trades, np.array(equity)