Alertas y monitorización: enterarte ANTES de que el CEO pregunte

1import pandas as pd
2import numpy as np
3from datetime import datetime, timedelta
4
5class MonitorDatos:
6    """Monitor de calidad con umbrales estáticos y dinámicos."""
7    
8    def __init__(self, nombre_dataset: str):
9        self.nombre = nombre_dataset
10        self.historial: list[dict] = []
11    
12    def registrar_metricas(self, metricas: dict):
13        """Registra las métricas del día (simula historial)."""
14        metricas['timestamp'] = datetime.now()
15        self.historial.append(metricas)
16    
17    def check_estatico(self, metrica: str, valor: float, umbral: float) -> dict:
18        """Check simple: ¿el valor supera el umbral fijo?"""
19        dispara = valor > umbral
20        return {
21            'tipo': 'estático',
22            'metrica': metrica,
23            'valor': valor,
24            'umbral': umbral,
25            'alerta': dispara,
26            'mensaje': f"{metrica} = {valor:.2f} ({'🔴 > ' if dispara else '✅ <= '}{umbral})",
27        }
28    
29    def check_dinamico(self, metrica: str, valor_actual: float, 
30                        historial_valores: list[float], sigma: float = 3.0) -> dict:
31        """Check dinámico: ¿el valor se desvía del comportamiento histórico?"""
32        if len(historial_valores) < 7:
33            return {'tipo': 'dinámico', 'metrica': metrica, 'alerta': False,
34                    'mensaje': 'Historial insuficiente (< 7 días)'}
35        
36        media = np.mean(historial_valores)
37        std = np.std(historial_valores)
38        
39        if std == 0:
40            limite_superior = media * 1.5
41            limite_inferior = media * 0.5
42        else:
43            limite_superior = media + sigma * std
44            limite_inferior = media - sigma * std
45        
46        fuera = valor_actual > limite_superior or valor_actual < limite_inferior
47        
48        return {
49            'tipo': 'dinámico',
50            'metrica': metrica,
51            'valor': valor_actual,
52            'media_historica': round(media, 2),
53            'rango_normal': f"[{limite_inferior:.1f}, {limite_superior:.1f}]",
54            'alerta': fuera,
55            'mensaje': f"{metrica} = {valor_actual:.1f} vs media {media:.1f} "
56                       f"{'🔴 ANOMALÍA' if fuera else '✅ Normal'}",
57        }
58
59# Ejemplo: monitorizar volumen diario
60monitor = MonitorDatos("pedidos_diarios")
61
62# Historial de los últimos 14 días (volumen normal: ~10000 ± 1000)
63historial_volumen = [9800, 10200, 10100, 9900, 10500, 10300, 9700, 
64                     10100, 10400, 9600, 10200, 10000, 10300, 9800]
65
66# Hoy: solo llegaron 3200 pedidos (¡anomalía!)
67volumen_hoy = 3200
68
69# Check estático (umbral fijo: mínimo 5000)
70resultado_estatico = monitor.check_estatico("volumen", volumen_hoy, 5000)
71print(f"Estático: {resultado_estatico['mensaje']}")
72
73# Check dinámico (basado en historial)
74resultado_dinamico = monitor.check_dinamico("volumen", volumen_hoy, historial_volumen)
75print(f"Dinámico: {resultado_dinamico['mensaje']}")
76print(f"  Rango normal: {resultado_dinamico.get('rango_normal', 'N/A')}")

El umbral dinámico detecta la anomalía aunque el estático no se dispare — se adapta al comportamiento real

1from dataclasses import dataclass
2from datetime import datetime
3from enum import Enum
4
5class Severidad(Enum):
6    INFO = "info"
7    WARNING = "warning" 
8    CRITICAL = "critical"
9
10@dataclass
11class Alerta:
12    """Una alerta de calidad de datos."""
13    dataset: str
14    metrica: str
15    mensaje: str
16    severidad: Severidad
17    valor_actual: float
18    valor_esperado: float
19    timestamp: datetime = None
20    
21    def __post_init__(self):
22        if self.timestamp is None:
23            self.timestamp = datetime.now()
24
25class SistemaAlertas:
26    """Sistema de alertas multi-canal."""
27    
28    def __init__(self):
29        self.alertas_enviadas: list[Alerta] = []
30    
31    def enviar(self, alerta: Alerta):
32        """Enruta la alerta al canal apropiado según severidad."""
33        self.alertas_enviadas.append(alerta)
34        
35        if alerta.severidad == Severidad.CRITICAL:
36            self._enviar_slack(alerta)
37            self._enviar_email(alerta)
38            self._enviar_pagerduty(alerta)
39        elif alerta.severidad == Severidad.WARNING:
40            self._enviar_slack(alerta)
41        else:
42            self._log(alerta)
43    
44    def _enviar_slack(self, alerta: Alerta):
45        """Simula envío a Slack (en producción: webhook o SDK)."""
46        emoji = "🔴" if alerta.severidad == Severidad.CRITICAL else "⚠️"
47        print(f"[SLACK #data-alerts] {emoji} {alerta.dataset}: {alerta.mensaje}")
48        print(f"  Valor: {alerta.valor_actual} (esperado: {alerta.valor_esperado})")
49        # En producción:
50        # requests.post(SLACK_WEBHOOK_URL, json={"text": mensaje})
51    
52    def _enviar_email(self, alerta: Alerta):
53        """Simula envío de email (en producción: SMTP o SES)."""
54        print(f"[EMAIL → data-oncall@empresa.com] CRÍTICO: {alerta.dataset}")
55        print(f"  {alerta.mensaje}")
56        # En producción:
57        # import smtplib / boto3.client('ses').send_email(...)
58    
59    def _enviar_pagerduty(self, alerta: Alerta):
60        """Simula envío a PagerDuty (en producción: API)."""
61        print(f"[PAGERDUTY] 📟 Incidente creado: {alerta.dataset} — {alerta.metrica}")
62        # En producción:
63        # requests.post("https://events.pagerduty.com/v2/enqueue", json=payload)
64    
65    def _log(self, alerta: Alerta):
66        """Solo log para alertas informativas."""
67        print(f"[LOG] ℹ️ {alerta.dataset}: {alerta.mensaje}")
68
69# Uso
70alertas = SistemaAlertas()
71
72# Alerta crítica: pipeline no se ejecutó
73alertas.enviar(Alerta(
74    dataset="pedidos_diarios",
75    metrica="frescura",
76    mensaje="Datos no actualizados en 6 horas (SLA: 1h)",
77    severidad=Severidad.CRITICAL,
78    valor_actual=6.0,
79    valor_esperado=1.0,
80))
81
82print()
83
84# Alerta warning: volumen bajo
85alertas.enviar(Alerta(
86    dataset="pedidos_diarios",
87    metrica="volumen",
88    mensaje="Volumen 40% menor que la media histórica",
89    severidad=Severidad.WARNING,
90    valor_actual=6000,
91    valor_esperado=10000,
92))

La severidad determina el canal: INFO → log, WARNING → Slack, CRITICAL → Slack + email + PagerDuty