AI for Performance Anomaly Detection in Testing

import numpy as np
from sklearn.preprocessing import StandardScaler
from datetime import datetime, timedelta

class PerformanceBaseline:
    def __init__(self, window_days=30):
        self.window_days = window_days
        self.scaler = StandardScaler()
        self.baseline_metrics = {}

    def train_baseline(self, metrics_data):
        """
        Train baseline model on historical performance data

        Args:
            metrics_data: DataFrame with columns ['timestamp', 'response_time',
                         'throughput', 'error_rate', 'cpu_usage', 'memory_usage']
        """
        # Filter data to training window
        cutoff_date = datetime.now() - timedelta(days=self.window_days)
        training_data = metrics_data[metrics_data['timestamp'] >= cutoff_date]

        # Calculate statistical baselines for each metric
        for metric in ['response_time', 'throughput', 'error_rate',
                       'cpu_usage', 'memory_usage']:
            self.baseline_metrics[metric] = {
                'mean': training_data[metric].mean(),
                'std': training_data[metric].std(),
                'percentile_95': training_data[metric].quantile(0.95),
                'percentile_99': training_data[metric].quantile(0.99),
                'min': training_data[metric].min(),
                'max': training_data[metric].max()
            }

        return self.baseline_metrics

    def is_anomaly(self, current_value, metric_name, threshold_std=3):
        """
        Detect if current value deviates from baseline
        """
        baseline = self.baseline_metrics[metric_name]
        z_score = abs((current_value - baseline['mean']) / baseline['std'])

        return z_score > threshold_std, z_score

import pandas as pd
from sklearn.ensemble import RandomForestRegressor

class TemporalBaselineModel:
    def __init__(self):
        self.model = RandomForestRegressor(n_estimators=100, random_state=42)

    def extract_temporal_features(self, timestamp):
        """Extract time-based features for pattern recognition"""
        return {
            'hour': timestamp.hour,
            'day_of_week': timestamp.dayofweek,
            'day_of_month': timestamp.day,
            'month': timestamp.month,
            'is_weekend': 1 if timestamp.dayofweek >= 5 else 0,
            'is_business_hours': 1 if 9 <= timestamp.hour <= 17 else 0
        }

    def train(self, historical_data):
        """Train model to predict expected performance based on time"""
        features = pd.DataFrame([
            self.extract_temporal_features(ts)
            for ts in historical_data['timestamp']
        ])

        self.model.fit(features, historical_data['response_time'])

    def predict_expected_performance(self, timestamp):
        """Predict expected response time for given timestamp"""
        features = pd.DataFrame([self.extract_temporal_features(timestamp)])
        return self.model.predict(features)[0]

from sklearn.ensemble import IsolationForest
import pandas as pd

class PerformanceAnomalyDetector:
    def __init__(self, contamination=0.1):
        self.model = IsolationForest(
            contamination=contamination,
            random_state=42,
            n_estimators=100
        )
        self.feature_columns = [
            'response_time', 'throughput', 'error_rate',
            'cpu_usage', 'memory_usage', 'db_query_time'
        ]

    def train(self, historical_metrics):
        """Train Isolation Forest on normal performance patterns"""
        X = historical_metrics[self.feature_columns]
        self.model.fit(X)

    def detect_anomalies(self, current_metrics):
        """
        Detect anomalies in current metrics

        Returns:
            predictions: -1 for anomalies, 1 for normal
            scores: anomaly scores (lower = more anomalous)
        """
        X = current_metrics[self.feature_columns]
        predictions = self.model.predict(X)
        scores = self.model.score_samples(X)

        anomalies = current_metrics[predictions == -1].copy()
        anomalies['anomaly_score'] = scores[predictions == -1]

        return anomalies

    def explain_anomaly(self, anomaly_record):
        """Identify which metrics contributed most to anomaly detection"""
        contributions = {}

        for feature in self.feature_columns:
            baseline_mean = self.baseline_metrics[feature]['mean']
            baseline_std = self.baseline_metrics[feature]['std']
            current_value = anomaly_record[feature]

            deviation = abs((current_value - baseline_mean) / baseline_std)
            contributions[feature] = deviation

        # Sort by contribution
        sorted_contributions = sorted(
            contributions.items(),
            key=lambda x: x[1],
            reverse=True
        )

        return sorted_contributions

import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
import numpy as np

class LSTMAnomalyDetector:
    def __init__(self, sequence_length=50):
        self.sequence_length = sequence_length
        self.model = None
        self.threshold = None

    def build_model(self, n_features):
        """Build LSTM autoencoder for anomaly detection"""
        model = Sequential([
            LSTM(64, activation='relu', input_shape=(self.sequence_length, n_features),
                 return_sequences=True),
            Dropout(0.2),
            LSTM(32, activation='relu', return_sequences=False),
            Dropout(0.2),
            Dense(32, activation='relu'),
            Dense(n_features)
        ])

        model.compile(optimizer='adam', loss='mse')
        self.model = model
        return model

    def create_sequences(self, data):
        """Convert time series data into sequences"""
        sequences = []
        for i in range(len(data) - self.sequence_length):
            sequences.append(data[i:i + self.sequence_length])
        return np.array(sequences)

    def train(self, normal_data, epochs=50, batch_size=32):
        """Train LSTM on normal performance data"""
        n_features = normal_data.shape[1]

        if self.model is None:
            self.build_model(n_features)

        X_train = self.create_sequences(normal_data)

        # Train autoencoder to reconstruct normal patterns
        self.model.fit(
            X_train,
            normal_data[self.sequence_length:],
            epochs=epochs,
            batch_size=batch_size,
            validation_split=0.1,
            verbose=0
        )

        # Calculate reconstruction error threshold
        predictions = self.model.predict(X_train)
        reconstruction_errors = np.mean(np.abs(predictions - normal_data[self.sequence_length:]), axis=1)
        self.threshold = np.percentile(reconstruction_errors, 95)

    def detect_anomalies(self, test_data):
        """Detect anomalies based on reconstruction error"""
        X_test = self.create_sequences(test_data)
        predictions = self.model.predict(X_test)

        reconstruction_errors = np.mean(np.abs(predictions - test_data[self.sequence_length:]), axis=1)

        anomalies = reconstruction_errors > self.threshold
        return anomalies, reconstruction_errors

from statsmodels.tsa.holtwinters import ExponentialSmoothing
from sklearn.metrics import mean_absolute_error
import warnings
warnings.filterwarnings('ignore')

class PerformanceTrendPredictor:
    def __init__(self):
        self.models = {}

    def train_predictor(self, metric_data, metric_name, seasonal_periods=24):
        """
        Train exponential smoothing model for trend prediction

        Args:
            metric_data: Time series data for specific metric
            metric_name: Name of the metric (e.g., 'response_time')
            seasonal_periods: Number of periods in seasonal cycle (24 for hourly data)
        """
        model = ExponentialSmoothing(
            metric_data,
            seasonal_periods=seasonal_periods,
            trend='add',
            seasonal='add'
        ).fit()

        self.models[metric_name] = model
        return model

    def predict_future(self, metric_name, steps_ahead=24):
        """Predict future values for specified metric"""
        if metric_name not in self.models:
            raise ValueError(f"No trained model for {metric_name}")

        forecast = self.models[metric_name].forecast(steps=steps_ahead)
        return forecast

    def detect_degradation_trend(self, metric_name, threshold_slope=0.05):
        """
        Detect if metric shows degradation trend

        Returns:
            is_degrading: Boolean indicating degradation trend
            slope: Rate of degradation
            forecast: Predicted values
        """
        forecast = self.predict_future(metric_name, steps_ahead=24)

        # Calculate trend slope
        time_steps = np.arange(len(forecast))
        slope = np.polyfit(time_steps, forecast, 1)[0]

        is_degrading = slope > threshold_slope

        return is_degrading, slope, forecast

from enum import Enum

class AlertSeverity(Enum):
    INFO = 1
    WARNING = 2
    CRITICAL = 3
    EMERGENCY = 4

class SmartAlertSystem:
    def __init__(self):
        self.alert_history = []
        self.suppression_rules = {}

    def classify_alert(self, anomaly_score, metric_name, impact_score):
        """
        Classify alert severity based on multiple factors

        Args:
            anomaly_score: How anomalous the metric is (0-100)
            metric_name: Name of affected metric
            impact_score: Business impact score (0-100)
        """
        # Weighted severity calculation
        severity_score = (anomaly_score * 0.6) + (impact_score * 0.4)

        if severity_score >= 90:
            return AlertSeverity.EMERGENCY
        elif severity_score >= 70:
            return AlertSeverity.CRITICAL
        elif severity_score >= 40:
            return AlertSeverity.WARNING
        else:
            return AlertSeverity.INFO

    def should_suppress_alert(self, metric_name, current_time):
        """
        Determine if alert should be suppressed based on recent history
        """
        # Check for alert fatigue (same metric, multiple alerts in short time)
        recent_alerts = [
            a for a in self.alert_history
            if a['metric'] == metric_name
            and (current_time - a['timestamp']).seconds < 600  # 10 minutes
        ]

        if len(recent_alerts) >= 3:
            return True  # Suppress to avoid alert fatigue

        return False

    def generate_alert(self, anomaly_data, root_cause_analysis):
        """
        Generate actionable alert with context
        """
        alert = {
            'timestamp': anomaly_data['timestamp'],
            'severity': self.classify_alert(
                anomaly_data['score'],
                anomaly_data['metric'],
                anomaly_data['impact']
            ),
            'metric': anomaly_data['metric'],
            'current_value': anomaly_data['value'],
            'expected_value': anomaly_data['baseline'],
            'deviation_percent': anomaly_data['deviation'],
            'root_cause': root_cause_analysis,
            'recommended_actions': self.get_remediation_steps(anomaly_data['metric'])
        }

        return alert

    def get_remediation_steps(self, metric_name):
        """Provide context-specific remediation guidance"""
        remediation_map = {
            'response_time': [
                'Check database query performance',
                'Review recent code deployments',
                'Verify external API dependencies',
                'Check server resource utilization'
            ],
            'error_rate': [
                'Review application logs for errors',
                'Check database connectivity',
                'Verify third-party service status',
                'Review recent configuration changes'
            ],
            'throughput': [
                'Check load balancer configuration',
                'Verify auto-scaling policies',
                'Review rate limiting settings',
                'Check network bandwidth'
            ]
        }

        return remediation_map.get(metric_name, ['Investigate metric anomaly'])

from prometheus_client import Gauge, Counter
import requests

class PrometheusAnomalyIntegration:
    def __init__(self, prometheus_url, grafana_url):
        self.prometheus_url = prometheus_url
        self.grafana_url = grafana_url

        # Define custom metrics
        self.anomaly_score_gauge = Gauge(
            'performance_anomaly_score',
            'Current anomaly score for performance metrics',
            ['metric_name', 'service']
        )

        self.anomaly_counter = Counter(
            'performance_anomalies_total',
            'Total number of performance anomalies detected',
            ['severity', 'metric_name']
        )

    def query_metrics(self, query, start_time, end_time):
        """Query historical metrics from Prometheus"""
        params = {
            'query': query,
            'start': start_time,
            'end': end_time,
            'step': '1m'
        }

        response = requests.get(
            f"{self.prometheus_url}/api/v1/query_range",
            params=params
        )

        return response.json()['data']['result']

    def publish_anomaly_metrics(self, anomalies):
        """Publish detected anomalies back to Prometheus"""
        for anomaly in anomalies:
            self.anomaly_score_gauge.labels(
                metric_name=anomaly['metric'],
                service=anomaly['service']
            ).set(anomaly['score'])

            self.anomaly_counter.labels(
                severity=anomaly['severity'].name,
                metric_name=anomaly['metric']
            ).inc()

    def create_grafana_annotation(self, anomaly):
        """Create annotation in Grafana for detected anomaly"""
        annotation = {
            'time': int(anomaly['timestamp'].timestamp() * 1000),
            'tags': ['anomaly', anomaly['severity'].name, anomaly['metric']],
            'text': f"Anomaly detected: {anomaly['metric']} - {anomaly['description']}"
        }

        requests.post(
            f"{self.grafana_url}/api/annotations",
            json=annotation,
            headers={'Authorization': f'Bearer {self.grafana_token}'}
        )