""" 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)