| """
|
| 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
|
|
|
|
|
| @dataclass
|
| class DriftAlert:
|
| """Alert when live performance diverges from backtest expectations."""
|
| metric: str
|
| expected: float
|
| observed: float
|
| deviation_pct: float
|
| severity: str
|
|
|
|
|
| class PerformanceMonitor:
|
| """Track and alert on live trading performance."""
|
|
|
| def __init__(
|
| self,
|
| rolling_window: int = 50,
|
| sharpe_alert_floor: float = 0.5,
|
| winrate_alert_floor: float = 0.35,
|
| drift_threshold_pct: float = 30.0,
|
| ):
|
| 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] = []
|
| self._outcomes: list[int] = []
|
| self._equity_curve: list[float] = []
|
| self._snapshots: list[PerformanceSnapshot] = []
|
| self._alerts: list[DriftAlert] = []
|
|
|
|
|
| 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))
|
|
|
|
|
| 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:])
|
|
|
|
|
| if np.std(recent) > 0:
|
| sharpe = (np.mean(recent) / np.std(recent)) * np.sqrt(252)
|
| else:
|
| sharpe = 0.0
|
|
|
|
|
| win_rate = float(np.mean(recent_outcomes))
|
|
|
|
|
| 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
|
|
|
|
|
| roll_ret = float(np.prod(1 + recent) - 1)
|
|
|
|
|
| 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
|
|
|
|
|
| recent_preds = np.array(self._predictions[-window:])
|
| recent_outs = np.array(self._outcomes[-window:])
|
|
|
| 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
|
|
|
|
|
| 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",
|
| ))
|
|
|
|
|
| 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",
|
| ))
|
|
|
|
|
| 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",
|
| ))
|
|
|
|
|
| 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)
|
|
|
|
|
| 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
|
|
|
|
|
| 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}")
|
|
|
|
|
| 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}%)")
|
|
|
|
|
| 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)
|
|
|
