ForexBot2/metrics.py

"""
ForexBot2 — Backtest Metrics
==============================
Pure functions over Trade records & equity curves.
"""

from dataclasses import dataclass, field, asdict
from typing import Dict, List, Optional

import numpy as np
import pandas as pd


@dataclass
class Trade:
    """A single closed trade."""
    entry_time: pd.Timestamp
    exit_time: pd.Timestamp
    direction: str            # "BUY" / "SELL"
    entry_price: float
    exit_price: float
    stop_loss: float
    take_profit: float
    exit_reason: str          # "tp" / "sl" / "end_of_data"
    risk_pips: float
    pnl_pips: float
    pnl_pct: float            # P&L as % of equity at trade open
    pnl_currency: float       # P&L in account currency (units = balance)
    duration_bars: int
    grade: str                # signal grade at entry
    score_at_entry: float
    confidence_at_entry: float
    # Diagnostic fields captured at entry — used to mine for loss-pattern
    # filters when a window underperforms.
    entry_hour: int = -1            # UTC hour 0-23
    entry_dow: int = -1             # 0=Mon, 6=Sun
    adx_at_entry: float = 0.0
    atr_pct_at_entry: float = 0.0
    market_regime: str = ""         # trending_up / trending_down / ranging
    volatility_regime: str = ""     # low / medium / high
    mtf_agreement: bool = False
    m15_score: float = 0.0
    h1_score: float = 0.0
    h4_score: float = 0.0
    rsi_at_entry: float = 50.0
    ema_distance_pct: float = 0.0    # (price - ema_trend) / ema_trend * 100


@dataclass
class BacktestResult:
    """Complete backtest output."""
    trades: List[Trade] = field(default_factory=list)
    equity_curve: pd.Series = field(default_factory=lambda: pd.Series(dtype=float))
    initial_balance: float = 0.0
    final_balance: float = 0.0
    metrics: Dict[str, float] = field(default_factory=dict)
    params: Dict[str, float] = field(default_factory=dict)


def compute_metrics(
    trades: List[Trade],
    equity_curve: pd.Series,
    initial_balance: float,
    elapsed_days: Optional[float] = None,
) -> Dict[str, float]:
    """All summary metrics from a list of closed trades + equity curve."""
    if not trades:
        return {
            "total_trades": 0,
            "win_rate": 0.0,
            "profit_factor": 0.0,
            "expectancy_per_trade": 0.0,
            "expectancy_pips": 0.0,
            "total_return_pct": 0.0,
            "cagr": 0.0,
            "max_drawdown_pct": 0.0,
            "sharpe": 0.0,
            "sortino": 0.0,
            "avg_trade_duration_bars": 0.0,
            "longest_winning_streak": 0,
            "longest_losing_streak": 0,
            "final_balance": initial_balance,
        }

    pnl_currency = np.array([t.pnl_currency for t in trades])
    pnl_pips = np.array([t.pnl_pips for t in trades])

    wins = pnl_currency > 0
    losses = pnl_currency < 0

    total_wins = pnl_currency[wins].sum() if wins.any() else 0.0
    total_losses = -pnl_currency[losses].sum() if losses.any() else 0.0

    win_rate = wins.sum() / len(trades) * 100
    profit_factor = (total_wins / total_losses) if total_losses > 0 else float("inf")
    expectancy_per_trade = pnl_currency.mean()
    expectancy_pips = pnl_pips.mean()

    final_balance = float(equity_curve.iloc[-1]) if len(equity_curve) else initial_balance
    total_return_pct = (final_balance / initial_balance - 1) * 100

    # CAGR
    if elapsed_days and elapsed_days > 0 and final_balance > 0:
        years = elapsed_days / 365.25
        cagr = ((final_balance / initial_balance) ** (1 / years) - 1) * 100 if years > 0 else 0.0
    else:
        cagr = 0.0

    # Max drawdown
    if len(equity_curve):
        running_max = equity_curve.cummax()
        drawdown = (equity_curve - running_max) / running_max
        max_dd = float(drawdown.min()) * 100
    else:
        max_dd = 0.0

    # Sharpe / Sortino — daily returns from equity curve
    sharpe, sortino = 0.0, 0.0
    if len(equity_curve) > 2:
        daily = equity_curve.resample("1D").last().ffill().pct_change().dropna()
        if len(daily) > 1 and daily.std() > 0:
            sharpe = float(daily.mean() / daily.std() * np.sqrt(252))
        downside = daily[daily < 0]
        if len(downside) > 1 and downside.std() > 0:
            sortino = float(daily.mean() / downside.std() * np.sqrt(252))

    # Streaks
    win_loss_seq = [1 if t.pnl_currency > 0 else (-1 if t.pnl_currency < 0 else 0) for t in trades]
    longest_win = _longest_streak(win_loss_seq, 1)
    longest_loss = _longest_streak(win_loss_seq, -1)

    avg_duration = float(np.mean([t.duration_bars for t in trades]))

    return {
        "total_trades": len(trades),
        "wins": int(wins.sum()),
        "losses": int(losses.sum()),
        "win_rate": float(win_rate),
        "profit_factor": float(profit_factor),
        "expectancy_per_trade": float(expectancy_per_trade),
        "expectancy_pips": float(expectancy_pips),
        "total_return_pct": float(total_return_pct),
        "cagr": float(cagr),
        "max_drawdown_pct": float(max_dd),
        "sharpe": float(sharpe),
        "sortino": float(sortino),
        "avg_trade_duration_bars": avg_duration,
        "longest_winning_streak": longest_win,
        "longest_losing_streak": longest_loss,
        "final_balance": final_balance,
    }


def _longest_streak(seq: List[int], target: int) -> int:
    best = current = 0
    for x in seq:
        if x == target:
            current += 1
            best = max(best, current)
        else:
            current = 0
    return best


def trades_to_dataframe(trades: List[Trade]) -> pd.DataFrame:
    """Convert trade journal to a DataFrame for CSV/JSON export."""
    if not trades:
        return pd.DataFrame()
    return pd.DataFrame([asdict(t) for t in trades])