src/metrics.py

"""Performance metrics, computed from the equity curve and trade list.

Every number the UI shows comes from this module, so each one is traceable to
an equity curve or a trade row rather than to a library's internal accounting.
Metrics are pure functions of their inputs -- no globals, no randomness.
"""

from __future__ import annotations

import math
from dataclasses import dataclass, asdict, field

import numpy as np
import pandas as pd

TRADING_DAYS = 252.0


@dataclass
class Metrics:
    """One performance summary. `segment` names the slice it describes."""

    segment: str = "all"
    start_ts: str | None = None
    end_ts: str | None = None
    bars: int = 0

    total_return: float = 0.0
    cagr: float = 0.0
    sharpe: float = 0.0
    sortino: float = 0.0
    max_drawdown: float = 0.0
    volatility: float = 0.0

    win_rate: float = 0.0
    profit_factor: float = 0.0
    exposure: float = 0.0
    trade_count: int = 0

    avg_win: float = 0.0
    avg_loss: float = 0.0
    avg_r: float = 0.0
    best_trade: float = 0.0
    worst_trade: float = 0.0
    costs_paid: float = 0.0
    gross_pnl: float = 0.0
    net_pnl: float = 0.0

    def to_dict(self) -> dict:
        return asdict(self)


def _clean_returns(equity: pd.Series) -> pd.Series:
    r = equity.astype("float64").pct_change()
    return r.replace([np.inf, -np.inf], np.nan).dropna()


def total_return(equity: pd.Series) -> float:
    if len(equity) < 2 or equity.iloc[0] == 0:
        return 0.0
    return float(equity.iloc[-1] / equity.iloc[0] - 1.0)


def cagr(equity: pd.Series, bars_per_year: float) -> float:
    if len(equity) < 2 or equity.iloc[0] <= 0 or equity.iloc[-1] <= 0:
        return 0.0
    years = (len(equity) - 1) / bars_per_year
    if years <= 0:
        return 0.0
    return float((equity.iloc[-1] / equity.iloc[0]) ** (1.0 / years) - 1.0)


def sharpe(equity: pd.Series, bars_per_year: float, rf: float = 0.0) -> float:
    r = _clean_returns(equity)
    if len(r) < 2:
        return 0.0
    excess = r - (rf / bars_per_year)
    sd = float(excess.std(ddof=1))
    if sd == 0 or not math.isfinite(sd):
        return 0.0
    return float(excess.mean() / sd * math.sqrt(bars_per_year))


def sortino(equity: pd.Series, bars_per_year: float, rf: float = 0.0) -> float:
    r = _clean_returns(equity)
    if len(r) < 2:
        return 0.0
    excess = r - (rf / bars_per_year)
    downside = excess[excess < 0]
    if len(downside) == 0:
        return 0.0
    dd = float(np.sqrt((downside ** 2).mean()))
    if dd == 0 or not math.isfinite(dd):
        return 0.0
    return float(excess.mean() / dd * math.sqrt(bars_per_year))


def volatility(equity: pd.Series, bars_per_year: float) -> float:
    r = _clean_returns(equity)
    if len(r) < 2:
        return 0.0
    return float(r.std(ddof=1) * math.sqrt(bars_per_year))


def drawdown_series(equity: pd.Series) -> pd.Series:
    if equity.empty:
        return equity
    peak = equity.cummax()
    return equity / peak - 1.0


def max_drawdown(equity: pd.Series) -> float:
    if len(equity) < 2:
        return 0.0
    dd = drawdown_series(equity)
    return float(dd.min()) if len(dd) else 0.0


def rolling_sharpe(equity: pd.Series, window: int, bars_per_year: float) -> pd.Series:
    r = _clean_returns(equity)
    if len(r) < window:
        return pd.Series(dtype="float64", index=pd.DatetimeIndex([], tz="UTC"))
    mean = r.rolling(window).mean()
    sd = r.rolling(window).std(ddof=1)
    out = (mean / sd.replace(0.0, np.nan)) * math.sqrt(bars_per_year)
    return out.dropna()


def underwater(equity: pd.Series) -> pd.Series:
    return drawdown_series(equity)


def exposure(position: pd.Series) -> float:
    """Fraction of bars holding a non-zero position."""
    if position is None or len(position) == 0:
        return 0.0
    return float((position.abs() > 1e-12).mean())


# --------------------------------------------------------------------------
# Trade-derived statistics
# --------------------------------------------------------------------------


def trade_stats(trades: pd.DataFrame) -> dict:
    """Win rate, profit factor and friends from the trade list."""
    empty = {
        "trade_count": 0, "win_rate": 0.0, "profit_factor": 0.0,
        "avg_win": 0.0, "avg_loss": 0.0, "avg_r": 0.0,
        "best_trade": 0.0, "worst_trade": 0.0,
        "costs_paid": 0.0, "gross_pnl": 0.0, "net_pnl": 0.0,
    }
    if trades is None or trades.empty:
        return empty

    net = trades["net_pnl"].astype("float64")
    wins = net[net > 0]
    losses = net[net < 0]
    gross_profit = float(wins.sum())
    gross_loss = float(-losses.sum())

    if gross_loss > 0:
        pf = gross_profit / gross_loss
    elif gross_profit > 0:
        pf = float("inf")
    else:
        pf = 0.0

    r_vals = trades["r_multiple"].replace([np.inf, -np.inf], np.nan).dropna() \
        if "r_multiple" in trades.columns else pd.Series(dtype="float64")

    return {
        "trade_count": int(len(trades)),
        "win_rate": float(len(wins) / len(net)) if len(net) else 0.0,
        "profit_factor": float(pf),
        "avg_win": float(wins.mean()) if len(wins) else 0.0,
        "avg_loss": float(losses.mean()) if len(losses) else 0.0,
        "avg_r": float(r_vals.mean()) if len(r_vals) else 0.0,
        "best_trade": float(net.max()),
        "worst_trade": float(net.min()),
        "costs_paid": float(trades["costs"].sum()) if "costs" in trades.columns else 0.0,
        "gross_pnl": float(trades["gross_pnl"].sum()) if "gross_pnl" in trades.columns else 0.0,
        "net_pnl": float(net.sum()),
    }


def compute_metrics(
    equity: pd.Series,
    trades: pd.DataFrame | None,
    bars_per_year: float,
    *,
    segment: str = "all",
    position: pd.Series | None = None,
) -> Metrics:
    """Assemble the full metric set for one equity slice."""
    equity = equity.dropna()
    m = Metrics(
        segment=segment,
        start_ts=str(equity.index[0]) if len(equity) else None,
        end_ts=str(equity.index[-1]) if len(equity) else None,
        bars=int(len(equity)),
        total_return=total_return(equity),
        cagr=cagr(equity, bars_per_year),
        sharpe=sharpe(equity, bars_per_year),
        sortino=sortino(equity, bars_per_year),
        max_drawdown=max_drawdown(equity),
        volatility=volatility(equity, bars_per_year),
        exposure=exposure(position) if position is not None else 0.0,
    )
    for k, v in trade_stats(trades).items():
        setattr(m, k, v)
    return m


# --------------------------------------------------------------------------
# Forecast quality (used by comparisons/ in Phase 2)
# --------------------------------------------------------------------------


def calibration_coverage(
    actual: pd.Series, lower: pd.Series, upper: pd.Series
) -> float:
    """Empirical coverage: share of actuals inside [lower, upper].

    A well-calibrated q10-q90 band should cover ~0.80 of outcomes.
    """
    df = pd.concat([actual, lower, upper], axis=1).dropna()
    if df.empty:
        return float("nan")
    a, lo, hi = df.iloc[:, 0], df.iloc[:, 1], df.iloc[:, 2]
    return float(((a >= lo) & (a <= hi)).mean())


def calibration_error(coverage: float, nominal: float = 0.80) -> float:
    """Signed miss against the nominal band width. 0.0 is perfect."""
    if not math.isfinite(coverage):
        return float("nan")
    return float(coverage - nominal)


def directional_accuracy(actual_next: pd.Series, predicted_next: pd.Series,
                         reference: pd.Series) -> float:
    """Share of *directional calls* that matched the realised direction.

    Bars where the forecast is exactly flat are excluded, not counted as
    misses. A random-walk forecast predicts "no change" every bar; it is never
    wrong about direction because it never claims one. Scoring it 0% would say
    it is always wrong, which is a different and false statement. When a model
    never takes a side, its directional accuracy is undefined and returns NaN.
    """
    df = pd.concat([actual_next, predicted_next, reference], axis=1).dropna()
    if df.empty:
        return float("nan")
    a, p, ref = df.iloc[:, 0], df.iloc[:, 1], df.iloc[:, 2]
    actual_dir = np.sign(a - ref)
    pred_dir = np.sign(p - ref)
    mask = (actual_dir != 0) & (pred_dir != 0)
    if not mask.any():
        return float("nan")
    return float((actual_dir[mask] == pred_dir[mask]).mean())


def pinball_loss(actual: pd.Series, pred: pd.Series, q: float) -> float:
    """Quantile (pinball) loss -- lower is better."""
    df = pd.concat([actual, pred], axis=1).dropna()
    if df.empty:
        return float("nan")
    a, p = df.iloc[:, 0], df.iloc[:, 1]
    diff = a - p
    return float(np.maximum(q * diff, (q - 1) * diff).mean())