import json import time import logging import threading import numpy as np from pathlib import Path from datetime import datetime, timedelta from dataclasses import dataclass, asdict from collections import deque, defaultdict from typing import Dict, List, Optional, Any, Tuple logger = logging.getLogger(__name__) @dataclass class PredictionRecord: """Individual prediction record with metadata""" timestamp: str text_hash: str prediction: str confidence: float processing_time: float model_version: str text_length: int word_count: int client_id: Optional[str] = None user_agent: Optional[str] = None session_id: Optional[str] = None @dataclass class MonitoringMetrics: """Aggregated monitoring metrics""" timestamp: str total_predictions: int predictions_per_minute: float avg_confidence: float avg_processing_time: float confidence_distribution: Dict[str, int] prediction_distribution: Dict[str, int] error_rate: float response_time_percentiles: Dict[str, float] anomaly_score: float class PredictionMonitor: """Real-time prediction monitoring system""" def __init__(self, base_dir: Path): self.base_dir = Path(base_dir) self.monitor_dir = self.base_dir / "monitor" self.monitor_dir.mkdir(parents=True, exist_ok=True) # Storage paths self.predictions_log_path = self.monitor_dir / "predictions.json" self.metrics_log_path = self.monitor_dir / "metrics.json" self.alerts_log_path = self.monitor_dir / "alerts.json" # In-memory storage for real-time analysis self.recent_predictions = deque(maxlen=10000) # Last 10k predictions self.prediction_buffer = deque(maxlen=1000) # Buffer for batch processing # Metrics tracking self.metrics_history = deque(maxlen=1440) # 24 hours of minute-level metrics self.error_count = 0 self.total_predictions = 0 # Configuration self.confidence_thresholds = { 'very_low': 0.5, 'low': 0.7, 'medium': 0.8, 'high': 0.9 } self.performance_thresholds = { 'response_time_warning': 5.0, # seconds 'response_time_critical': 10.0, 'confidence_warning': 0.6, # average confidence below this 'error_rate_warning': 0.05, # 5% error rate 'error_rate_critical': 0.10 # 10% error rate } # Background processing self.monitoring_active = False self.monitoring_thread = None # Load existing data self.load_historical_data() def start_monitoring(self): """Start background monitoring thread""" if not self.monitoring_active: self.monitoring_active = True self.monitoring_thread = threading.Thread(target=self._monitoring_loop, daemon=True) self.monitoring_thread.start() logger.info("Prediction monitoring started") def stop_monitoring(self): """Stop background monitoring""" self.monitoring_active = False if self.monitoring_thread: self.monitoring_thread.join(timeout=5) logger.info("Prediction monitoring stopped") def record_prediction(self, prediction: str, confidence: float, processing_time: float, text: str, model_version: str = "unknown", client_id: Optional[str] = None, user_agent: Optional[str] = None, session_id: Optional[str] = None) -> str: """Record a new prediction with comprehensive metadata""" # Create prediction record text_hash = self._hash_text(text) record = PredictionRecord( timestamp=datetime.now().isoformat(), text_hash=text_hash, prediction=prediction, confidence=confidence, processing_time=processing_time, model_version=model_version, text_length=len(text), word_count=len(text.split()), client_id=client_id, user_agent=user_agent, session_id=session_id ) # Add to in-memory storage self.recent_predictions.append(record) self.prediction_buffer.append(record) self.total_predictions += 1 # Trigger batch processing if buffer is full if len(self.prediction_buffer) >= 100: self._process_prediction_batch() return text_hash def record_error(self, error_type: str, error_message: str, context: Dict = None): """Record prediction error""" self.error_count += 1 error_record = { 'timestamp': datetime.now().isoformat(), 'error_type': error_type, 'error_message': error_message, 'context': context or {}, 'total_errors': self.error_count, 'error_rate': self.get_current_error_rate() } # Save error to alerts log self._append_to_log(self.alerts_log_path, error_record) # Check if error rate exceeds thresholds self._check_error_rate_alerts() def get_current_metrics(self) -> MonitoringMetrics: """Get current real-time metrics""" now = datetime.now() recent_predictions = self._get_recent_predictions(minutes=5) if not recent_predictions: return MonitoringMetrics( timestamp=now.isoformat(), total_predictions=self.total_predictions, predictions_per_minute=0.0, avg_confidence=0.0, avg_processing_time=0.0, confidence_distribution={}, prediction_distribution={}, error_rate=0.0, response_time_percentiles={}, anomaly_score=0.0 ) # Calculate metrics confidences = [p.confidence for p in recent_predictions] processing_times = [p.processing_time for p in recent_predictions] predictions = [p.prediction for p in recent_predictions] return MonitoringMetrics( timestamp=now.isoformat(), total_predictions=self.total_predictions, predictions_per_minute=len(recent_predictions) / 5.0, avg_confidence=float(np.mean(confidences)), avg_processing_time=float(np.mean(processing_times)), confidence_distribution=self._calculate_confidence_distribution(confidences), prediction_distribution=self._calculate_prediction_distribution(predictions), error_rate=self.get_current_error_rate(), response_time_percentiles=self._calculate_percentiles(processing_times), anomaly_score=self._calculate_anomaly_score(recent_predictions) ) def get_historical_metrics(self, hours: int = 24) -> List[MonitoringMetrics]: """Get historical metrics for specified time period""" cutoff_time = datetime.now() - timedelta(hours=hours) historical_metrics = [] for metrics in self.metrics_history: if datetime.fromisoformat(metrics.timestamp) > cutoff_time: historical_metrics.append(metrics) return historical_metrics def get_prediction_patterns(self, hours: int = 24) -> Dict[str, Any]: """Analyze prediction patterns for anomaly detection""" cutoff_time = datetime.now() - timedelta(hours=hours) recent_predictions = [ p for p in self.recent_predictions if datetime.fromisoformat(p.timestamp) > cutoff_time ] if not recent_predictions: return {'error': 'No recent predictions found'} # Analyze patterns hourly_distribution = defaultdict(int) confidence_trends = [] processing_time_trends = [] for prediction in recent_predictions: hour = datetime.fromisoformat(prediction.timestamp).hour hourly_distribution[hour] += 1 confidence_trends.append(prediction.confidence) processing_time_trends.append(prediction.processing_time) return { 'total_predictions': len(recent_predictions), 'hourly_distribution': dict(hourly_distribution), 'confidence_stats': { 'mean': float(np.mean(confidence_trends)), 'std': float(np.std(confidence_trends)), 'min': float(np.min(confidence_trends)), 'max': float(np.max(confidence_trends)) }, 'processing_time_stats': { 'mean': float(np.mean(processing_time_trends)), 'std': float(np.std(processing_time_trends)), 'min': float(np.min(processing_time_trends)), 'max': float(np.max(processing_time_trends)) }, 'anomaly_indicators': self._detect_anomaly_indicators(recent_predictions) } def get_current_error_rate(self) -> float: """Calculate current error rate""" if self.total_predictions == 0: return 0.0 return self.error_count / (self.total_predictions + self.error_count) def get_confidence_analysis(self) -> Dict[str, Any]: """Analyze confidence distribution and trends""" recent_predictions = self._get_recent_predictions(minutes=60) if not recent_predictions: return {'error': 'No recent predictions found'} confidences = [p.confidence for p in recent_predictions] # Confidence distribution distribution = self._calculate_confidence_distribution(confidences) # Confidence trends (last hour in 10-minute windows) trends = [] now = datetime.now() for i in range(6): # 6 ten-minute windows window_start = now - timedelta(minutes=(i+1)*10) window_end = now - timedelta(minutes=i*10) window_predictions = [ p for p in recent_predictions if window_start <= datetime.fromisoformat(p.timestamp) < window_end ] if window_predictions: avg_confidence = np.mean([p.confidence for p in window_predictions]) trends.append({ 'window_start': window_start.isoformat(), 'window_end': window_end.isoformat(), 'avg_confidence': float(avg_confidence), 'prediction_count': len(window_predictions) }) return { 'total_predictions': len(recent_predictions), 'overall_avg_confidence': float(np.mean(confidences)), 'confidence_std': float(np.std(confidences)), 'distribution': distribution, 'trends': trends[::-1], # Reverse to get chronological order 'low_confidence_alerts': len([c for c in confidences if c < self.confidence_thresholds['low']]) } def _monitoring_loop(self): """Background monitoring loop""" while self.monitoring_active: try: # Process any pending predictions if self.prediction_buffer: self._process_prediction_batch() # Generate and save metrics current_metrics = self.get_current_metrics() self.metrics_history.append(current_metrics) self._append_to_log(self.metrics_log_path, asdict(current_metrics)) # Check for alerts self._check_performance_alerts(current_metrics) # Sleep for 1 minute time.sleep(60) except Exception as e: logger.error(f"Error in monitoring loop: {e}") time.sleep(60) def _process_prediction_batch(self): """Process batch of predictions and save to log""" batch = list(self.prediction_buffer) self.prediction_buffer.clear() # Save batch to log file for prediction in batch: self._append_to_log(self.predictions_log_path, asdict(prediction)) def _get_recent_predictions(self, minutes: int) -> List[PredictionRecord]: """Get predictions from the last N minutes""" cutoff_time = datetime.now() - timedelta(minutes=minutes) return [ p for p in self.recent_predictions if datetime.fromisoformat(p.timestamp) > cutoff_time ] def _calculate_confidence_distribution(self, confidences: List[float]) -> Dict[str, int]: """Calculate confidence distribution buckets""" distribution = { 'very_low': 0, # < 0.5 'low': 0, # 0.5-0.7 'medium': 0, # 0.7-0.8 'high': 0, # 0.8-0.9 'very_high': 0 # > 0.9 } for confidence in confidences: if confidence < 0.5: distribution['very_low'] += 1 elif confidence < 0.7: distribution['low'] += 1 elif confidence < 0.8: distribution['medium'] += 1 elif confidence < 0.9: distribution['high'] += 1 else: distribution['very_high'] += 1 return distribution def _calculate_prediction_distribution(self, predictions: List[str]) -> Dict[str, int]: """Calculate prediction label distribution""" distribution = defaultdict(int) for prediction in predictions: distribution[prediction] += 1 return dict(distribution) def _calculate_percentiles(self, values: List[float]) -> Dict[str, float]: """Calculate response time percentiles""" if not values: return {} return { 'p50': float(np.percentile(values, 50)), 'p90': float(np.percentile(values, 90)), 'p95': float(np.percentile(values, 95)), 'p99': float(np.percentile(values, 99)) } def _calculate_anomaly_score(self, predictions: List[PredictionRecord]) -> float: """Calculate anomaly score based on various factors""" if not predictions: return 0.0 scores = [] # Confidence anomaly (low confidence spike) confidences = [p.confidence for p in predictions] low_confidence_ratio = len([c for c in confidences if c < 0.6]) / len(confidences) scores.append(low_confidence_ratio) # Processing time anomaly (slow responses) processing_times = [p.processing_time for p in predictions] slow_response_ratio = len([t for t in processing_times if t > 5.0]) / len(processing_times) scores.append(slow_response_ratio) # Prediction distribution anomaly (extreme skew) prediction_dist = self._calculate_prediction_distribution([p.prediction for p in predictions]) if prediction_dist: max_ratio = max(prediction_dist.values()) / len(predictions) if max_ratio > 0.9: # More than 90% same prediction scores.append(0.5) else: scores.append(0.0) return float(np.mean(scores)) def _detect_anomaly_indicators(self, predictions: List[PredictionRecord]) -> List[str]: """Detect specific anomaly indicators""" indicators = [] if not predictions: return indicators # Low confidence spike low_confidence_count = len([p for p in predictions if p.confidence < 0.6]) if low_confidence_count > len(predictions) * 0.3: indicators.append(f"High low-confidence predictions: {low_confidence_count}/{len(predictions)}") # Slow response spike slow_responses = len([p for p in predictions if p.processing_time > 5.0]) if slow_responses > len(predictions) * 0.1: indicators.append(f"Slow responses detected: {slow_responses}/{len(predictions)}") # Prediction skew prediction_dist = self._calculate_prediction_distribution([p.prediction for p in predictions]) if prediction_dist: max_count = max(prediction_dist.values()) if max_count > len(predictions) * 0.9: dominant_prediction = max(prediction_dist, key=prediction_dist.get) indicators.append(f"Extreme prediction skew: {max_count}/{len(predictions)} are '{dominant_prediction}'") return indicators def _check_performance_alerts(self, metrics: MonitoringMetrics): """Check for performance-based alerts""" alerts = [] # Response time alerts if metrics.avg_processing_time > self.performance_thresholds['response_time_critical']: alerts.append({ 'type': 'critical', 'category': 'response_time', 'message': f"Critical response time: {metrics.avg_processing_time:.2f}s", 'threshold': self.performance_thresholds['response_time_critical'] }) elif metrics.avg_processing_time > self.performance_thresholds['response_time_warning']: alerts.append({ 'type': 'warning', 'category': 'response_time', 'message': f"High response time: {metrics.avg_processing_time:.2f}s", 'threshold': self.performance_thresholds['response_time_warning'] }) # Confidence alerts if metrics.avg_confidence < self.performance_thresholds['confidence_warning']: alerts.append({ 'type': 'warning', 'category': 'confidence', 'message': f"Low average confidence: {metrics.avg_confidence:.2f}", 'threshold': self.performance_thresholds['confidence_warning'] }) # Error rate alerts if metrics.error_rate > self.performance_thresholds['error_rate_critical']: alerts.append({ 'type': 'critical', 'category': 'error_rate', 'message': f"Critical error rate: {metrics.error_rate:.2%}", 'threshold': self.performance_thresholds['error_rate_critical'] }) elif metrics.error_rate > self.performance_thresholds['error_rate_warning']: alerts.append({ 'type': 'warning', 'category': 'error_rate', 'message': f"High error rate: {metrics.error_rate:.2%}", 'threshold': self.performance_thresholds['error_rate_warning'] }) # Anomaly alerts if metrics.anomaly_score > 0.3: alerts.append({ 'type': 'warning', 'category': 'anomaly', 'message': f"Anomaly detected: score {metrics.anomaly_score:.2f}", 'threshold': 0.3 }) # Save alerts for alert in alerts: alert['timestamp'] = datetime.now().isoformat() alert['metrics_snapshot'] = asdict(metrics) self._append_to_log(self.alerts_log_path, alert) def _check_error_rate_alerts(self): """Check error rate and generate alerts if needed""" error_rate = self.get_current_error_rate() if error_rate > self.performance_thresholds['error_rate_critical']: alert = { 'timestamp': datetime.now().isoformat(), 'type': 'critical', 'category': 'error_rate', 'message': f"Critical error rate reached: {error_rate:.2%}", 'error_count': self.error_count, 'total_requests': self.total_predictions + self.error_count } self._append_to_log(self.alerts_log_path, alert) def _hash_text(self, text: str) -> str: """Generate hash for text content""" import hashlib return hashlib.md5(text.encode()).hexdigest()[:16] def _append_to_log(self, log_path: Path, data: Dict): """Append data to log file""" try: with open(log_path, 'a') as f: f.write(json.dumps(data) + '\n') except Exception as e: logger.error(f"Failed to write to log {log_path}: {e}") def load_historical_data(self): """Load historical data on startup""" try: # Load recent predictions if self.predictions_log_path.exists(): with open(self.predictions_log_path, 'r') as f: for line in f: try: data = json.loads(line.strip()) prediction = PredictionRecord(**data) # Only load recent predictions (last 24 hours) if datetime.fromisoformat(prediction.timestamp) > datetime.now() - timedelta(hours=24): self.recent_predictions.append(prediction) except Exception: continue # Load recent metrics if self.metrics_log_path.exists(): with open(self.metrics_log_path, 'r') as f: for line in f: try: data = json.loads(line.strip()) metrics = MonitoringMetrics(**data) # Only load recent metrics (last 24 hours) if datetime.fromisoformat(metrics.timestamp) > datetime.now() - timedelta(hours=24): self.metrics_history.append(metrics) except Exception: continue logger.info(f"Loaded {len(self.recent_predictions)} recent predictions and {len(self.metrics_history)} metrics records") except Exception as e: logger.error(f"Failed to load historical data: {e}")