ml/monitoring.py

"""

ml/monitoring.py — Performance monitoring & drift detection.



Capabilities:

  1. Rolling Sharpe / win rate / profit factor tracking

  2. Live vs backtest drift detection

  3. Model prediction quality monitoring (calibration drift)

  4. Drawdown curve analysis with recovery time

  5. Summary dashboard metrics

"""
from __future__ import annotations

import logging
from dataclasses import dataclass, field

import numpy as np

logger = logging.getLogger("trading_system.ml.monitoring")


@dataclass
class PerformanceSnapshot:
    """Point-in-time performance metrics."""
    timestamp_idx: int
    rolling_sharpe: float
    rolling_win_rate: float
    rolling_profit_factor: float
    rolling_return: float
    drawdown: float
    n_trades: int
    avg_trade_return: float
    predicted_accuracy: float     # how well predictions match outcomes


@dataclass
class DriftAlert:
    """Alert when live performance diverges from backtest expectations."""
    metric: str
    expected: float
    observed: float
    deviation_pct: float
    severity: str   # "info", "warning", "critical"


class PerformanceMonitor:
    """Track and alert on live trading performance."""

    def __init__(

        self,

        rolling_window: int = 50,        # trades for rolling metrics

        sharpe_alert_floor: float = 0.5,  # alert if rolling Sharpe < this

        winrate_alert_floor: float = 0.35,

        drift_threshold_pct: float = 30.0,  # alert if 30% worse than backtest

    ):
        self.rolling_window = rolling_window
        self.sharpe_alert_floor = sharpe_alert_floor
        self.winrate_alert_floor = winrate_alert_floor
        self.drift_threshold_pct = drift_threshold_pct

        self._trade_returns: list[float] = []
        self._predictions: list[float] = []  # predicted P(win) or confidence
        self._outcomes: list[int] = []         # 1=win, 0=loss
        self._equity_curve: list[float] = []
        self._snapshots: list[PerformanceSnapshot] = []
        self._alerts: list[DriftAlert] = []

        # Backtest baselines (set externally after backtest)
        self.backtest_sharpe: float = 0.0
        self.backtest_winrate: float = 0.0
        self.backtest_return: float = 0.0
        self.backtest_max_dd: float = 0.0

    def set_backtest_baseline(

        self,

        sharpe: float,

        winrate: float,

        total_return: float,

        max_dd: float,

    ):
        """Set backtest performance as drift comparison baseline."""
        self.backtest_sharpe = sharpe
        self.backtest_winrate = winrate
        self.backtest_return = total_return
        self.backtest_max_dd = max_dd

    def record_trade(

        self,

        ret: float,

        predicted_confidence: float = 0.5,

        won: bool | None = None,

    ):
        """Record a completed trade."""
        self._trade_returns.append(ret)
        self._predictions.append(predicted_confidence)

        if won is None:
            won = ret > 0
        self._outcomes.append(int(won))

        # Update equity
        if self._equity_curve:
            new_eq = self._equity_curve[-1] * (1 + ret)
        else:
            new_eq = 1.0 * (1 + ret)
        self._equity_curve.append(new_eq)

    def get_snapshot(self) -> PerformanceSnapshot | None:
        """Compute current rolling performance snapshot."""
        n = len(self._trade_returns)
        if n < 5:
            return None

        window = min(n, self.rolling_window)
        recent = np.array(self._trade_returns[-window:])
        recent_outcomes = np.array(self._outcomes[-window:])

        # Rolling Sharpe
        if np.std(recent) > 0:
            sharpe = (np.mean(recent) / np.std(recent)) * np.sqrt(252)
        else:
            sharpe = 0.0

        # Win rate
        win_rate = float(np.mean(recent_outcomes))

        # Profit factor
        gains = recent[recent > 0].sum()
        losses = abs(recent[recent < 0].sum())
        pf = gains / losses if losses > 0 else float('inf') if gains > 0 else 0.0

        # Rolling return
        roll_ret = float(np.prod(1 + recent) - 1)

        # Current drawdown
        equity = np.array(self._equity_curve)
        peak = np.maximum.accumulate(equity)
        dd = float((peak[-1] - equity[-1]) / peak[-1]) if peak[-1] > 0 else 0.0

        # Prediction accuracy (calibration)
        recent_preds = np.array(self._predictions[-window:])
        recent_outs = np.array(self._outcomes[-window:])
        # Bucket accuracy: for predicted > 0.5, check if outcome was 1
        high_conf = recent_preds > 0.5
        if high_conf.sum() > 0:
            pred_acc = float(np.mean(recent_outs[high_conf]))
        else:
            pred_acc = 0.5

        snapshot = PerformanceSnapshot(
            timestamp_idx=n,
            rolling_sharpe=float(sharpe),
            rolling_win_rate=win_rate,
            rolling_profit_factor=float(pf),
            rolling_return=roll_ret,
            drawdown=dd,
            n_trades=n,
            avg_trade_return=float(np.mean(recent)),
            predicted_accuracy=pred_acc,
        )
        self._snapshots.append(snapshot)
        return snapshot

    def check_drift(self) -> list[DriftAlert]:
        """Check for drift between live and backtest performance."""
        alerts = []
        snap = self.get_snapshot()
        if snap is None:
            return alerts

        # Sharpe drift
        if self.backtest_sharpe > 0:
            dev = ((snap.rolling_sharpe - self.backtest_sharpe) / self.backtest_sharpe) * 100
            if dev < -self.drift_threshold_pct:
                alerts.append(DriftAlert(
                    metric="Sharpe",
                    expected=self.backtest_sharpe,
                    observed=snap.rolling_sharpe,
                    deviation_pct=dev,
                    severity="critical" if dev < -50 else "warning",
                ))

        # Win rate drift
        if self.backtest_winrate > 0:
            dev = ((snap.rolling_win_rate - self.backtest_winrate) / self.backtest_winrate) * 100
            if dev < -self.drift_threshold_pct:
                alerts.append(DriftAlert(
                    metric="Win Rate",
                    expected=self.backtest_winrate,
                    observed=snap.rolling_win_rate,
                    deviation_pct=dev,
                    severity="warning",
                ))

        # Sharpe floor
        if snap.rolling_sharpe < self.sharpe_alert_floor and len(self._trade_returns) > 30:
            alerts.append(DriftAlert(
                metric="Sharpe Floor",
                expected=self.sharpe_alert_floor,
                observed=snap.rolling_sharpe,
                deviation_pct=0,
                severity="warning",
            ))

        # Drawdown alert (if significantly worse than backtest)
        if self.backtest_max_dd > 0 and snap.drawdown > self.backtest_max_dd * 1.5:
            alerts.append(DriftAlert(
                metric="Drawdown",
                expected=self.backtest_max_dd,
                observed=snap.drawdown,
                deviation_pct=((snap.drawdown - self.backtest_max_dd) / self.backtest_max_dd) * 100,
                severity="critical",
            ))

        self._alerts.extend(alerts)
        return alerts

    def get_drawdown_analysis(self) -> dict:
        """Analyze drawdown periods: depth, duration, recovery."""
        if len(self._equity_curve) < 10:
            return {}

        equity = np.array(self._equity_curve)
        peak = np.maximum.accumulate(equity)
        dd = (peak - equity) / (peak + 1e-9)

        # Find drawdown periods
        in_dd = dd > 0.001
        periods = []
        start = None

        for i in range(len(in_dd)):
            if in_dd[i] and start is None:
                start = i
            elif not in_dd[i] and start is not None:
                max_dd = float(dd[start:i].max())
                duration = i - start
                periods.append({
                    "start": start,
                    "end": i,
                    "duration": duration,
                    "max_depth": max_dd,
                })
                start = None

        # Handle ongoing drawdown
        if start is not None:
            max_dd = float(dd[start:].max())
            periods.append({
                "start": start,
                "end": len(dd),
                "duration": len(dd) - start,
                "max_depth": max_dd,
                "ongoing": True,
            })

        return {
            "current_dd": float(dd[-1]),
            "max_dd": float(dd.max()),
            "n_drawdown_periods": len(periods),
            "avg_dd_duration": float(np.mean([p["duration"] for p in periods])) if periods else 0,
            "worst_periods": sorted(periods, key=lambda p: -p["max_depth"])[:5],
        }

    def summary(self) -> dict:
        """Dashboard summary metrics."""
        if not self._trade_returns:
            return {}

        returns = np.array(self._trade_returns)
        equity = np.array(self._equity_curve) if self._equity_curve else np.cumprod(1 + returns)

        total_ret = float(equity[-1] / equity[0] - 1) if len(equity) > 0 else 0
        wins = np.sum(returns > 0)
        losses = np.sum(returns < 0)
        win_rate = wins / (wins + losses) if (wins + losses) > 0 else 0

        gains = returns[returns > 0].sum()
        loss_sum = abs(returns[returns < 0].sum())
        pf = gains / loss_sum if loss_sum > 0 else float('inf') if gains > 0 else 0

        peak = np.maximum.accumulate(equity)
        dd = (peak - equity) / (peak + 1e-9)

        return {
            "total_return": total_ret,
            "total_trades": len(returns),
            "win_rate": float(win_rate),
            "profit_factor": float(pf),
            "sharpe": float((np.mean(returns) / np.std(returns)) * np.sqrt(252)) if np.std(returns) > 0 else 0,
            "max_drawdown": float(dd.max()),
            "avg_trade": float(np.mean(returns)),
            "best_trade": float(returns.max()),
            "worst_trade": float(returns.min()),
            "n_alerts": len(self._alerts),
        }


def print_monitoring_dashboard(monitor: PerformanceMonitor):
    """Print performance monitoring dashboard."""
    s = monitor.summary()
    if not s:
        print("No trades recorded yet.")
        return

    print("\n" + "=" * 60)
    print("  PERFORMANCE MONITORING DASHBOARD")
    print("=" * 60)
    print(f"  Total Return:    {s['total_return']:>8.2%}")
    print(f"  Total Trades:    {s['total_trades']:>8d}")
    print(f"  Win Rate:        {s['win_rate']:>8.1%}")
    print(f"  Profit Factor:   {s['profit_factor']:>8.2f}")
    print(f"  Sharpe Ratio:    {s['sharpe']:>8.2f}")
    print(f"  Max Drawdown:    {s['max_drawdown']:>8.2%}")
    print(f"  Avg Trade:       {s['avg_trade']:>8.4%}")
    print(f"  Best Trade:      {s['best_trade']:>8.4%}")
    print(f"  Worst Trade:     {s['worst_trade']:>8.4%}")
    print(f"  Active Alerts:   {s['n_alerts']:>8d}")

    # Drift alerts
    alerts = monitor.check_drift()
    if alerts:
        print("\n  --- DRIFT ALERTS ---")
        for a in alerts:
            icon = "⚠" if a.severity == "warning" else "🔴"
            print(f"  {icon} {a.metric}: expected {a.expected:.2f}, observed {a.observed:.2f} ({a.deviation_pct:+.1f}%)")

    # Drawdown analysis
    dd = monitor.get_drawdown_analysis()
    if dd:
        print(f"\n  --- DRAWDOWN ANALYSIS ---")
        print(f"  Current DD:      {dd['current_dd']:>8.2%}")
        print(f"  Max DD:          {dd['max_dd']:>8.2%}")
        print(f"  DD Periods:      {dd['n_drawdown_periods']:>8d}")
        print(f"  Avg DD Duration: {dd['avg_dd_duration']:>8.0f} trades")

    print("=" * 60)