Create predictive_models.py

predictive_models.py

@@ -0,0 +1,261 @@
+import numpy as np
+import pandas as pd
+from typing import Dict, List, Optional, Tuple
+from datetime import datetime, timedelta
+import json
+import warnings
+from dataclasses import dataclass
+
+@dataclass
+class ForecastResult:
+    metric: str
+    predicted_value: float
+    confidence: float
+    trend: str  # "increasing", "decreasing", "stable"
+    time_to_threshold: Optional[timedelta] = None
+    risk_level: str = "low"  # low, medium, high, critical
+
+class SimplePredictiveEngine:
+    """
+    Lightweight forecasting engine optimized for Hugging Face Spaces
+    Uses statistical methods instead of heavy ML models
+    """
+    
+    def __init__(self, history_window: int = 50):
+        self.history_window = history_window
+        self.service_history: Dict[str, List] = {}
+        self.prediction_cache: Dict[str, ForecastResult] = {}
+        
+    def add_telemetry(self, service: str, event_data: Dict):
+        """Add telemetry data to service history"""
+        if service not in self.service_history:
+            self.service_history[service] = []
+        
+        # Store key metrics with timestamp
+        telemetry_point = {
+            'timestamp': datetime.now(),
+            'latency': event_data.get('latency_p99', 0),
+            'error_rate': event_data.get('error_rate', 0),
+            'throughput': event_data.get('throughput', 0),
+            'cpu_util': event_data.get('cpu_util'),
+            'memory_util': event_data.get('memory_util')
+        }
+        
+        self.service_history[service].append(telemetry_point)
+        
+        # Keep only recent history
+        if len(self.service_history[service]) > self.history_window:
+            self.service_history[service].pop(0)
+    
+    def forecast_service_health(self, service: str, lookahead_minutes: int = 15) -> List[ForecastResult]:
+        """Forecast service health metrics"""
+        if service not in self.service_history or len(self.service_history[service]) < 10:
+            return []
+        
+        history = self.service_history[service]
+        forecasts = []
+        
+        # Forecast latency
+        latency_forecast = self._forecast_latency(history, lookahead_minutes)
+        if latency_forecast:
+            forecasts.append(latency_forecast)
+        
+        # Forecast error rate
+        error_forecast = self._forecast_error_rate(history, lookahead_minutes)
+        if error_forecast:
+            forecasts.append(error_forecast)
+        
+        # Forecast resource utilization
+        resource_forecasts = self._forecast_resources(history, lookahead_minutes)
+        forecasts.extend(resource_forecasts)
+        
+        # Cache results
+        for forecast in forecasts:
+            cache_key = f"{service}_{forecast.metric}"
+            self.prediction_cache[cache_key] = forecast
+        
+        return forecasts
+    
+    def _forecast_latency(self, history: List, lookahead_minutes: int) -> Optional[ForecastResult]:
+        """Forecast latency using linear regression and trend analysis"""
+        try:
+            latencies = [point['latency'] for point in history[-20:]]  # Last 20 points
+            
+            if len(latencies) < 5:
+                return None
+            
+            # Simple linear trend
+            x = np.arange(len(latencies))
+            slope, intercept = np.polyfit(x, latencies, 1)
+            
+            # Predict next value
+            next_x = len(latencies)
+            predicted_latency = slope * next_x + intercept
+            
+            # Calculate confidence based on data quality
+            residuals = latencies - (slope * x + intercept)
+            confidence = max(0, 1 - (np.std(residuals) / max(1, np.mean(latencies))))
+            
+            # Determine trend
+            if slope > 5:  # Increasing by more than 5ms per interval
+                trend = "increasing"
+                risk = "high" if predicted_latency > 300 else "medium"
+            elif slope < -2:  # Decreasing
+                trend = "decreasing" 
+                risk = "low"
+            else:
+                trend = "stable"
+                risk = "low"
+            
+            # Calculate time to reach critical threshold (500ms)
+            time_to_critical = None
+            if slope > 0 and predicted_latency < 500:
+                time_to_critical = timedelta(
+                    minutes=lookahead_minutes * (500 - predicted_latency) / (predicted_latency - latencies[-1])
+                )
+            
+            return ForecastResult(
+                metric="latency",
+                predicted_value=predicted_latency,
+                confidence=confidence,
+                trend=trend,
+                time_to_threshold=time_to_critical,
+                risk_level=risk
+            )
+            
+        except Exception as e:
+            print(f"Latency forecast error: {e}")
+            return None
+    
+    def _forecast_error_rate(self, history: List, lookahead_minutes: int) -> Optional[ForecastResult]:
+        """Forecast error rate using exponential smoothing"""
+        try:
+            error_rates = [point['error_rate'] for point in history[-15:]]
+            
+            if len(error_rates) < 5:
+                return None
+            
+            # Exponential smoothing
+            alpha = 0.3  # Smoothing factor
+            forecast = error_rates[0]
+            for rate in error_rates[1:]:
+                forecast = alpha * rate + (1 - alpha) * forecast
+            
+            predicted_rate = forecast
+            
+            # Trend analysis
+            recent_trend = np.mean(error_rates[-3:]) - np.mean(error_rates[-6:-3])
+            
+            if recent_trend > 0.02:  # Increasing trend
+                trend = "increasing"
+                risk = "high" if predicted_rate > 0.1 else "medium"
+            elif recent_trend < -0.01:  # Decreasing
+                trend = "decreasing"
+                risk = "low"
+            else:
+                trend = "stable"
+                risk = "low"
+            
+            # Confidence based on volatility
+            confidence = max(0, 1 - (np.std(error_rates) / max(0.01, np.mean(error_rates))))
+            
+            return ForecastResult(
+                metric="error_rate",
+                predicted_value=predicted_rate,
+                confidence=confidence,
+                trend=trend,
+                risk_level=risk
+            )
+            
+        except Exception as e:
+            print(f"Error rate forecast error: {e}")
+            return None
+    
+    def _forecast_resources(self, history: List, lookahead_minutes: int) -> List[ForecastResult]:
+        """Forecast CPU and memory utilization"""
+        forecasts = []
+        
+        # CPU forecast
+        cpu_values = [point['cpu_util'] for point in history if point.get('cpu_util') is not None]
+        if len(cpu_values) >= 5:
+            try:
+                predicted_cpu = np.mean(cpu_values[-5:])  # Simple moving average
+                trend = "increasing" if cpu_values[-1] > np.mean(cpu_values[-10:-5]) else "stable"
+                
+                risk = "low"
+                if predicted_cpu > 0.8:
+                    risk = "critical" if predicted_cpu > 0.9 else "high"
+                elif predicted_cpu > 0.7:
+                    risk = "medium"
+                
+                forecasts.append(ForecastResult(
+                    metric="cpu_util",
+                    predicted_value=predicted_cpu,
+                    confidence=0.7,  # Moderate confidence for resources
+                    trend=trend,
+                    risk_level=risk
+                ))
+            except Exception as e:
+                print(f"CPU forecast error: {e}")
+        
+        # Memory forecast (similar approach)
+        memory_values = [point['memory_util'] for point in history if point.get('memory_util') is not None]
+        if len(memory_values) >= 5:
+            try:
+                predicted_memory = np.mean(memory_values[-5:])
+                trend = "increasing" if memory_values[-1] > np.mean(memory_values[-10:-5]) else "stable"
+                
+                risk = "low"
+                if predicted_memory > 0.8:
+                    risk = "critical" if predicted_memory > 0.9 else "high"
+                elif predicted_memory > 0.7:
+                    risk = "medium"
+                
+                forecasts.append(ForecastResult(
+                    metric="memory_util",
+                    predicted_value=predicted_memory,
+                    confidence=0.7,
+                    trend=trend,
+                    risk_level=risk
+                ))
+            except Exception as e:
+                print(f"Memory forecast error: {e}")
+        
+        return forecasts
+    
+    def get_predictive_insights(self, service: str) -> Dict[str, any]:
+        """Generate actionable insights from forecasts"""
+        forecasts = self.forecast_service_health(service)
+        
+        critical_risks = [f for f in forecasts if f.risk_level in ["high", "critical"]]
+        warnings = []
+        recommendations = []
+        
+        for forecast in critical_risks:
+            if forecast.metric == "latency" and forecast.risk_level in ["high", "critical"]:
+                warnings.append(f"📈 Latency expected to reach {forecast.predicted_value:.0f}ms")
+                if forecast.time_to_threshold:
+                    minutes = int(forecast.time_to_threshold.total_seconds() / 60)
+                    recommendations.append(f"⏰ Critical latency (~500ms) in ~{minutes} minutes")
+                recommendations.append("🔧 Consider scaling or optimizing dependencies")
+            
+            elif forecast.metric == "error_rate" and forecast.risk_level in ["high", "critical"]:
+                warnings.append(f"🚨 Errors expected to reach {forecast.predicted_value*100:.1f}%")
+                recommendations.append("🐛 Investigate recent deployments or dependency issues")
+            
+            elif forecast.metric == "cpu_util" and forecast.risk_level in ["high", "critical"]:
+                warnings.append(f"🔥 CPU expected at {forecast.predicted_value*100:.1f}%")
+                recommendations.append("⚡ Consider scaling compute resources")
+            
+            elif forecast.metric == "memory_util" and forecast.risk_level in ["high", "critical"]:
+                warnings.append(f"💾 Memory expected at {forecast.predicted_value*100:.1f}%")
+                recommendations.append("🧹 Check for memory leaks or optimize usage")
+        
+        return {
+            'service': service,
+            'forecasts': [f.__dict__ for f in forecasts],
+            'warnings': warnings[:3],  # Top 3 warnings
+            'recommendations': list(dict.fromkeys(recommendations))[:3],  # Unique top 3
+            'critical_risk_count': len(critical_risks),
+            'forecast_timestamp': datetime.now().isoformat()
+        }