Detección de Anomalías de Rendimiento con IA: Más Allá de los Umbrales Estáticos

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

class DynamicBaseline:
    def __init__(self, lookback_days=30):
        self.lookback_days = lookback_days
        self.scaler = StandardScaler()
        self.hourly_patterns = {}
        self.daily_patterns = {}
        self.weekly_patterns = {}

    def learn_patterns(self, historical_data):
        """
        Aprende patrones temporales de datos históricos.
        historical_data: lista de (timestamp, value) tuplas
        """
        # Agrupa por hora del día
        for timestamp, value in historical_data:
            hour = timestamp.hour
            day_of_week = timestamp.weekday()

            if hour not in self.hourly_patterns:
                self.hourly_patterns[hour] = []
            self.hourly_patterns[hour].append(value)

            # Captura patrones de día de semana
            key = (day_of_week, hour)
            if key not in self.weekly_patterns:
                self.weekly_patterns[key] = []
            self.weekly_patterns[key].append(value)

        # Calcula estadísticas para cada patrón
        self.hourly_stats = {
            hour: {
                'mean': np.mean(values),
                'std': np.std(values),
                'p95': np.percentile(values, 95),
                'p99': np.percentile(values, 99)
            }
            for hour, values in self.hourly_patterns.items()
        }

    def get_expected_range(self, timestamp):
        """Retorna rango esperado para timestamp dado."""
        hour = timestamp.hour
        day_of_week = timestamp.weekday()

        # Primero intenta patrón semanal específico
        key = (day_of_week, hour)
        if key in self.weekly_patterns and len(self.weekly_patterns[key]) >= 10:
            values = self.weekly_patterns[key]
            return {
                'lower': np.percentile(values, 5),
                'expected': np.mean(values),
                'upper': np.percentile(values, 95)
            }

        # Cae a patrón horario
        if hour in self.hourly_stats:
            stats = self.hourly_stats[hour]
            return {
                'lower': stats['mean'] - 2 * stats['std'],
                'expected': stats['mean'],
                'upper': stats['mean'] + 2 * stats['std']
            }

        return None

    def is_anomaly(self, timestamp, value, sensitivity=2.0):
        """
        Verifica si valor es anómalo dado el tiempo.
        sensitivity: cuántas desviaciones estándar para umbral
        """
        expected = self.get_expected_range(timestamp)
        if expected is None:
            return False, 0.0

        # Calcula puntuación de anomalía
        deviation = abs(value - expected['expected'])
        range_size = expected['upper'] - expected['lower']

        if range_size > 0:
            anomaly_score = deviation / (range_size / 2)
        else:
            anomaly_score = 0.0

        is_anomalous = value < expected['lower'] or value > expected['upper']

        return is_anomalous, anomaly_score

class SeasonalBaseline(DynamicBaseline):
    def __init__(self, lookback_days=90):
        super().__init__(lookback_days)
        self.seasonal_factors = {}

    def learn_seasonal_patterns(self, historical_data):
        """Aprende patrones estacionales (mensuales, trimestrales)."""
        self.learn_patterns(historical_data)

        # Agrupa por mes para tendencias estacionales
        monthly_data = {}
        for timestamp, value in historical_data:
            month = timestamp.month
            if month not in monthly_data:
                monthly_data[month] = []
            monthly_data[month].append(value)

        # Calcula factores estacionales
        overall_mean = np.mean([v for _, v in historical_data])
        for month, values in monthly_data.items():
            self.seasonal_factors[month] = np.mean(values) / overall_mean

    def adjust_for_season(self, timestamp, value):
        """Ajusta valor por factores estacionales."""
        month = timestamp.month
        if month in self.seasonal_factors:
            return value / self.seasonal_factors[month]
        return value

from sklearn.ensemble import IsolationForest
import pandas as pd

class MultiMetricAnomalyDetector:
    def __init__(self, contamination=0.1):
        self.model = IsolationForest(
            contamination=contamination,
            random_state=42,
            n_estimators=100
        )
        self.scaler = StandardScaler()
        self.feature_names = []

    def prepare_features(self, metrics_df):
        """
        Prepara features de múltiples métricas.
        metrics_df: DataFrame con columnas de timestamp y métricas
        """
        # Agrega features derivadas
        features = metrics_df.copy()

        for col in metrics_df.columns:
            if col != 'timestamp':
                # Tasa de cambio
                features[f'{col}_rate'] = metrics_df[col].pct_change()
                # Desviación de media móvil
                rolling_mean = metrics_df[col].rolling(window=10).mean()
                features[f'{col}_deviation'] = metrics_df[col] - rolling_mean

        self.feature_names = [c for c in features.columns if c != 'timestamp']
        return features[self.feature_names].fillna(0)

    def fit(self, metrics_df):
        """Entrena detector con datos históricos."""
        features = self.prepare_features(metrics_df)
        scaled_features = self.scaler.fit_transform(features)
        self.model.fit(scaled_features)

    def detect(self, metrics_df):
        """Detecta anomalías en nuevos datos."""
        features = self.prepare_features(metrics_df)
        scaled_features = self.scaler.transform(features)

        # -1 para anomalías, 1 para normal
        predictions = self.model.predict(scaled_features)
        scores = self.model.decision_function(scaled_features)

        results = metrics_df.copy()
        results['is_anomaly'] = predictions == -1
        results['anomaly_score'] = -scores  # Mayor = más anómalo

        return results

    def explain_anomaly(self, metrics_row):
        """Explica por qué un punto se marcó como anomalía."""
        features = self.prepare_features(pd.DataFrame([metrics_row]))
        scaled = self.scaler.transform(features)

        # Encuentra features que más contribuyen
        explanations = []
        for i, (name, value) in enumerate(zip(self.feature_names, scaled[0])):
            if abs(value) > 2:  # Más de 2 std de la media
                explanations.append({
                    'feature': name,
                    'z_score': value,
                    'direction': 'high' if value > 0 else 'low'
                })

        return sorted(explanations, key=lambda x: abs(x['z_score']), reverse=True)

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

class LSTMAnomalyDetector:
    def __init__(self, sequence_length=60, threshold_percentile=95):
        self.sequence_length = sequence_length
        self.threshold_percentile = threshold_percentile
        self.model = None
        self.threshold = None

    def build_model(self, n_features):
        """Construye modelo LSTM para predicción de series temporales."""
        self.model = Sequential([
            LSTM(64, input_shape=(self.sequence_length, n_features),
                 return_sequences=True),
            Dropout(0.2),
            LSTM(32, return_sequences=False),
            Dropout(0.2),
            Dense(n_features)
        ])
        self.model.compile(optimizer='adam', loss='mse')

    def create_sequences(self, data):
        """Crea secuencias para entrenamiento LSTM."""
        sequences = []
        targets = []
        for i in range(len(data) - self.sequence_length):
            sequences.append(data[i:i + self.sequence_length])
            targets.append(data[i + self.sequence_length])
        return np.array(sequences), np.array(targets)

    def fit(self, data, epochs=50, batch_size=32):
        """Entrena LSTM con datos normales."""
        scaled_data = self.scaler.fit_transform(data)
        X, y = self.create_sequences(scaled_data)

        if self.model is None:
            self.build_model(data.shape[1])

        self.model.fit(
            X, y,
            epochs=epochs,
            batch_size=batch_size,
            validation_split=0.1,
            verbose=0
        )

        # Calcula umbral de errores de entrenamiento
        predictions = self.model.predict(X)
        errors = np.mean(np.abs(predictions - y), axis=1)
        self.threshold = np.percentile(errors, self.threshold_percentile)

    def detect(self, data):
        """Detecta anomalías comparando predicciones vs valores reales."""
        scaled_data = self.scaler.transform(data)
        X, y = self.create_sequences(scaled_data)

        predictions = self.model.predict(X)
        errors = np.mean(np.abs(predictions - y), axis=1)

        anomalies = errors > self.threshold

        return {
            'is_anomaly': anomalies,
            'error_scores': errors,
            'threshold': self.threshold,
            'predictions': self.scaler.inverse_transform(predictions)
        }

class AlertClassifier:
    def __init__(self):
        self.severity_rules = []
        self.suppression_rules = []
        self.correlation_window = timedelta(minutes=5)

    def add_severity_rule(self, condition, severity, description):
        """Agrega regla para clasificación de severidad."""
        self.severity_rules.append({
            'condition': condition,
            'severity': severity,
            'description': description
        })

    def classify_alert(self, anomaly):
        """Clasifica severidad de anomalía."""
        for rule in self.severity_rules:
            if rule['condition'](anomaly):
                return {
                    'severity': rule['severity'],
                    'reason': rule['description']
                }
        return {'severity': 'low', 'reason': 'No rules matched'}

    def should_suppress(self, alert, recent_alerts):
        """Determina si alerta debe suprimirse."""
        # Suprime duplicados dentro de ventana de correlación
        for recent in recent_alerts:
            if (alert['metric'] == recent['metric'] and
                alert['timestamp'] - recent['timestamp'] < self.correlation_window):
                return True, 'Duplicate within correlation window'

        # Suprime durante mantenimiento conocido
        if self.is_maintenance_window(alert['timestamp']):
            return True, 'Maintenance window'

        return False, None

    def correlate_alerts(self, alerts):
        """Agrupa alertas relacionadas."""
        groups = []
        used = set()

        for i, alert in enumerate(alerts):
            if i in used:
                continue

            group = [alert]
            used.add(i)

            for j, other in enumerate(alerts[i+1:], i+1):
                if j in used:
                    continue
                if self.are_related(alert, other):
                    group.append(other)
                    used.add(j)

            groups.append({
                'root_cause': self.identify_root_cause(group),
                'alerts': group,
                'severity': max(a.get('severity', 'low') for a in group)
            })

        return groups

    def are_related(self, alert1, alert2):
        """Verifica si dos alertas están relacionadas."""
        time_diff = abs(alert1['timestamp'] - alert2['timestamp'])
        return time_diff < self.correlation_window

    def identify_root_cause(self, alert_group):
        """Intenta identificar causa raíz de grupo de alertas."""
        # Ordena por timestamp - la más temprana es probable causa raíz
        sorted_alerts = sorted(alert_group, key=lambda x: x['timestamp'])
        return sorted_alerts[0]

from prometheus_client import Counter, Gauge, Histogram
import requests

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

        # Métricas para exponer
        self.anomalies_detected = Counter(
            'ml_anomalies_detected_total',
            'Total number of anomalies detected',
            ['metric_name', 'severity']
        )
        self.anomaly_score = Gauge(
            'ml_anomaly_score',
            'Current anomaly score',
            ['metric_name']
        )
        self.detection_latency = Histogram(
            'ml_detection_latency_seconds',
            'Time to detect anomalies'
        )

    def query_prometheus(self, query, start_time, end_time, step='1m'):
        """Obtiene datos de Prometheus."""
        response = requests.get(
            f'{self.prometheus_url}/api/v1/query_range',
            params={
                'query': query,
                'start': start_time.timestamp(),
                'end': end_time.timestamp(),
                'step': step
            }
        )
        return response.json()['data']['result']

    def record_anomaly(self, metric_name, severity, score):
        """Registra detección de anomalía."""
        self.anomalies_detected.labels(
            metric_name=metric_name,
            severity=severity
        ).inc()
        self.anomaly_score.labels(metric_name=metric_name).set(score)

{
  "panels": [
    {
      "title": "Anomaly Score Over Time",
      "type": "timeseries",
      "targets": [
        {
          "expr": "ml_anomaly_score",
          "legendFormat": "{{metric_name}}"
        }
      ],
      "fieldConfig": {
        "defaults": {
          "thresholds": {
            "steps": [
              {"color": "green", "value": null},
              {"color": "yellow", "value": 0.5},
              {"color": "red", "value": 0.8}
            ]
          }
        }
      }
    },
    {
      "title": "Anomalies by Severity",
      "type": "piechart",
      "targets": [
        {
          "expr": "sum by (severity) (increase(ml_anomalies_detected_total[24h]))"
        }
      ]
    }
  ]
}