""" Machine Learning Models for Advanced Anomaly Detection Includes ensemble methods, causal inference, and adaptive thresholds """ import numpy as np from typing import Tuple, Optional, Dict, List import logging import datetime # Try importing optional ML libraries try: from sklearn.ensemble import IsolationForest from sklearn.preprocessing import StandardScaler SKLEARN_AVAILABLE = True except ImportError: SKLEARN_AVAILABLE = False logging.warning("scikit-learn not available. Using fallback detection only.") try: import torch import torch.nn as nn PYTORCH_AVAILABLE = True except ImportError: PYTORCH_AVAILABLE = False logging.warning("PyTorch not available. LSTM detector disabled.") logger = logging.getLogger(__name__) # === LSTM Model (Optional - Only if PyTorch available) === if PYTORCH_AVAILABLE: class LSTMAnomalyDetector(nn.Module): """ LSTM-based anomaly detector for time-series analysis. Uses sequence-to-sequence learning to predict next values and flag anomalies based on prediction error. """ def __init__(self, input_size: int = 5, hidden_size: int = 64, num_layers: int = 2): super(LSTMAnomalyDetector, self).__init__() self.hidden_size = hidden_size self.num_layers = num_layers # LSTM layers self.lstm = nn.LSTM( input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, batch_first=True, dropout=0.2 ) # Fully connected layers self.fc1 = nn.Linear(hidden_size, 32) self.fc2 = nn.Linear(32, input_size) self.relu = nn.ReLU() def forward(self, x): """Forward pass through the network""" # LSTM forward pass lstm_out, _ = self.lstm(x) # Take last time step last_output = lstm_out[:, -1, :] # Fully connected layers out = self.relu(self.fc1(last_output)) out = self.fc2(out) return out else: # Dummy class if PyTorch not available class LSTMAnomalyDetector: def __init__(self, *args, **kwargs): logger.warning("LSTM detector not available (PyTorch not installed)") # === Ensemble Anomaly Detector === class EnsembleAnomalyDetector: """ Ensemble of multiple anomaly detection algorithms for robust detection. Gracefully degrades if ML libraries aren't available. """ def __init__(self): self.isolation_forest = None self.lstm_model = None self.scaler = None self.is_trained = False self.training_data = [] # Initialize models if libraries are available if SKLEARN_AVAILABLE: try: self.isolation_forest = IsolationForest( contamination=0.1, random_state=42, n_estimators=100 ) self.scaler = StandardScaler() logger.info("Initialized Isolation Forest detector") except Exception as e: logger.error(f"Failed to initialize Isolation Forest: {e}") if PYTORCH_AVAILABLE: try: self.lstm_model = LSTMAnomalyDetector() logger.info("Initialized LSTM detector") except Exception as e: logger.error(f"Failed to initialize LSTM: {e}") logger.info(f"EnsembleAnomalyDetector initialized (sklearn={SKLEARN_AVAILABLE}, pytorch={PYTORCH_AVAILABLE})") def add_sample(self, features: np.ndarray) -> None: """ Add training sample Args: features: numpy array of [latency, error_rate, cpu, memory, throughput] """ if not isinstance(features, np.ndarray): features = np.array(features) self.training_data.append(features) # Auto-train when we have enough data if len(self.training_data) >= 100 and not self.is_trained: self.train() def train(self) -> None: """Train all available models in the ensemble""" if len(self.training_data) < 50: logger.warning(f"Insufficient data for training: {len(self.training_data)} samples (need 50+)") return try: X = np.array(self.training_data) # Train Isolation Forest if available if self.isolation_forest is not None and SKLEARN_AVAILABLE: self.isolation_forest.fit(X) logger.info(f"Trained Isolation Forest on {len(self.training_data)} samples") # Train LSTM if available (placeholder for now) if self.lstm_model is not None and PYTORCH_AVAILABLE: # TODO: Implement full LSTM training loop # For now, just scale the data if self.scaler is not None: X_scaled = self.scaler.fit_transform(X) logger.info("LSTM training not yet implemented (using fallback)") self.is_trained = True logger.info(f"✅ Ensemble trained on {len(self.training_data)} samples") except Exception as e: logger.error(f"Training failed: {e}", exc_info=True) self.is_trained = False def predict_anomaly(self, features: np.ndarray) -> Tuple[bool, float, Dict]: """ Predict if features represent an anomaly Args: features: numpy array of [latency, error_rate, cpu, memory, throughput] Returns: Tuple of (is_anomaly: bool, confidence: float, explanation: dict) """ if not isinstance(features, np.ndarray): features = np.array(features) # If not trained or no ML libraries, use fallback if not self.is_trained or not SKLEARN_AVAILABLE: return self._fallback_detection(features) try: # Isolation Forest prediction if_score = self.isolation_forest.score_samples(features.reshape(1, -1))[0] if_anomaly = self.isolation_forest.predict(features.reshape(1, -1))[0] == -1 # LSTM prediction (placeholder for now) lstm_score = 0.5 # TODO: Implement actual LSTM prediction # Statistical tests stat_score = self._statistical_tests(features) # Ensemble voting (weighted average) confidence = np.mean([ abs(if_score), lstm_score, stat_score ]) is_anomaly = if_anomaly or confidence > 0.7 explanation = { 'isolation_forest_score': float(if_score), 'isolation_forest_anomaly': bool(if_anomaly), 'lstm_reconstruction_error': float(lstm_score), 'statistical_score': float(stat_score), 'ensemble_confidence': float(confidence), 'primary_detector': 'isolation_forest' if if_anomaly else 'ensemble', 'models_used': ['isolation_forest', 'statistical'] } return is_anomaly, confidence, explanation except Exception as e: logger.error(f"Prediction failed, using fallback: {e}", exc_info=True) return self._fallback_detection(features) def _statistical_tests(self, features: np.ndarray) -> float: """ Perform statistical tests for anomaly detection using z-scores Args: features: Current feature values Returns: Anomaly probability (0-1) """ if len(self.training_data) < 10: return 0.5 try: # Calculate z-scores historical = np.array(self.training_data) mean = np.mean(historical, axis=0) std = np.std(historical, axis=0) # Avoid division by zero z_scores = np.abs((features - mean) / (std + 1e-8)) max_z_score = np.max(z_scores) # Convert z-score to probability (3-sigma rule) # z > 3 is very anomalous anomaly_prob = min(1.0, max_z_score / 3.0) return anomaly_prob except Exception as e: logger.error(f"Statistical test failed: {e}") return 0.5 def _fallback_detection(self, features: np.ndarray) -> Tuple[bool, float, Dict]: """ Fallback detection when ML models aren't trained or available Uses simple threshold-based detection Args: features: [latency, error_rate, cpu, memory, throughput] Returns: Tuple of (is_anomaly, confidence, explanation) """ latency_threshold = 150 error_rate_threshold = 0.05 cpu_threshold = 0.8 memory_threshold = 0.8 latency = features[0] if len(features) > 0 else 0 error_rate = features[1] if len(features) > 1 else 0 cpu = features[2] if len(features) > 2 else 0 memory = features[3] if len(features) > 3 else 0 is_anomaly = ( latency > latency_threshold or error_rate > error_rate_threshold or cpu > cpu_threshold or memory > memory_threshold ) confidence = 0.5 if is_anomaly else 0.1 explanation = { 'method': 'fallback_threshold', 'latency_exceeded': latency > latency_threshold, 'error_rate_exceeded': error_rate > error_rate_threshold, 'cpu_exceeded': cpu > cpu_threshold, 'memory_exceeded': memory > memory_threshold } return is_anomaly, confidence, explanation # === Causal Inference Engine === class CausalInferenceEngine: """ Bayesian causal inference for root cause analysis. Uses probabilistic graphical models to infer causality. """ def __init__(self): # Define causal relationships (cause -> effects) self.causal_graph = { 'database_latency': ['api_latency', 'error_rate'], 'network_issues': ['api_latency', 'timeout_errors'], 'memory_leak': ['memory_util', 'gc_time', 'response_time'], 'cpu_saturation': ['cpu_util', 'queue_length', 'latency'], 'traffic_spike': ['throughput', 'latency', 'error_rate'] } # Prior probabilities for each root cause self.prior_probabilities = { 'database_latency': 0.3, 'network_issues': 0.2, 'memory_leak': 0.15, 'cpu_saturation': 0.2, 'traffic_spike': 0.15 } logger.info("Initialized CausalInferenceEngine") def infer_root_cause(self, symptoms: Dict[str, float]) -> List[Tuple[str, float]]: """ Use Bayesian inference to determine likely root causes Args: symptoms: Dictionary of observed symptoms and their values e.g., {'api_latency': 500, 'error_rate': 0.15, 'cpu_util': 0.9} Returns: List of (root_cause, probability) tuples sorted by probability """ posterior_probs = {} for cause, effects in self.causal_graph.items(): # Calculate likelihood P(symptoms|cause) likelihood = self._calculate_likelihood(symptoms, effects) # Calculate posterior P(cause|symptoms) ∝ P(symptoms|cause) * P(cause) prior = self.prior_probabilities[cause] posterior = likelihood * prior posterior_probs[cause] = posterior # Normalize probabilities total = sum(posterior_probs.values()) if total > 0: posterior_probs = {k: v/total for k, v in posterior_probs.items()} else: # If all probabilities are 0, return uniform distribution posterior_probs = {k: 1.0/len(posterior_probs) for k in posterior_probs} # Sort by probability (descending) ranked_causes = sorted( posterior_probs.items(), key=lambda x: x[1], reverse=True ) logger.info(f"Inferred root causes: {ranked_causes[:3]}") return ranked_causes def _calculate_likelihood(self, symptoms: Dict[str, float], effects: List[str]) -> float: """ Calculate likelihood of symptoms given a cause Args: symptoms: Observed symptoms effects: Expected effects of the cause Returns: Likelihood score (0-1) """ matching_effects = sum(1 for effect in effects if effect in symptoms) if matching_effects == 0: return 0.1 # Low but non-zero probability # Higher likelihood if more effects are observed likelihood = matching_effects / len(effects) return likelihood # === Adaptive Threshold Learner === class AdaptiveThresholdLearner: """ Online learning system that adapts thresholds based on historical patterns. Uses exponential moving averages and seasonality detection. """ def __init__(self, window_size: int = 100): self.window_size = window_size self.historical_data: Dict[str, List[Dict]] = {} self.thresholds: Dict[str, Dict] = {} self.seasonality_patterns: Dict[str, Dict] = {} logger.info(f"Initialized AdaptiveThresholdLearner with window_size={window_size}") def update(self, metric: str, value: float, timestamp: datetime.datetime) -> None: """ Update historical data with new metric value Args: metric: Metric name (e.g., 'latency', 'error_rate') value: Metric value timestamp: Timestamp of the measurement """ if metric not in self.historical_data: self.historical_data[metric] = [] self.historical_data[metric].append({ 'value': value, 'timestamp': timestamp }) # Keep only recent data if len(self.historical_data[metric]) > self.window_size: self.historical_data[metric].pop(0) # Update threshold self._update_threshold(metric) def _update_threshold(self, metric: str) -> None: """ Calculate adaptive threshold using statistical methods Args: metric: Metric name """ data = self.historical_data[metric] if len(data) < 10: return try: values = [d['value'] for d in data] # Calculate statistics mean = np.mean(values) std = np.std(values) percentile_90 = np.percentile(values, 90) percentile_95 = np.percentile(values, 95) # Detect seasonality hour_of_day = data[-1]['timestamp'].hour day_of_week = data[-1]['timestamp'].weekday() # Adjust threshold based on time time_multiplier = self._get_time_multiplier(hour_of_day, day_of_week) # Set adaptive threshold (mean + 2*std, adjusted for time) threshold = (mean + 2 * std) * time_multiplier self.thresholds[metric] = { 'value': threshold, 'mean': mean, 'std': std, 'p90': percentile_90, 'p95': percentile_95, 'last_updated': datetime.datetime.now(), 'time_multiplier': time_multiplier } logger.debug(f"Updated threshold for {metric}: {threshold:.2f}") except Exception as e: logger.error(f"Failed to update threshold for {metric}: {e}") def _get_time_multiplier(self, hour: int, day_of_week: int) -> float: """ Adjust threshold based on time of day and day of week Args: hour: Hour of day (0-23) day_of_week: Day of week (0=Monday, 6=Sunday) Returns: Multiplier for threshold adjustment """ # Business hours (9 AM - 5 PM) on weekdays: higher threshold if 9 <= hour <= 17 and day_of_week < 5: return 1.2 # Off hours or weekends: lower threshold (more sensitive) return 0.8 def get_threshold(self, metric: str) -> Optional[float]: """ Get current adaptive threshold for metric Args: metric: Metric name Returns: Current threshold value or None if not available """ if metric in self.thresholds: return self.thresholds[metric]['value'] return None def get_statistics(self, metric: str) -> Optional[Dict]: """ Get full statistics for a metric Args: metric: Metric name Returns: Dictionary of statistics or None """ return self.thresholds.get(metric) # === Utility Functions === def create_feature_vector(event) -> np.ndarray: """ Convert ReliabilityEvent to feature vector for ML models Args: event: ReliabilityEvent object Returns: numpy array of [latency, error_rate, cpu, memory, throughput] """ return np.array([ event.latency_p99, event.error_rate, event.cpu_util if event.cpu_util is not None else 0.5, event.memory_util if event.memory_util is not None else 0.5, event.throughput ])