Bonus - Alertas de calidad de datos

1#data-team | Lunes 9:45 AM (semana 2)
2
3Marcos Delgado  9:47 AM
4Contexto para el nuevo: el mes pasado, antes de que llegaras,
5el evento board_viewed dejo de emitirse durante 3 días.
6Un deploy de backend rompio el tracking de boards sin querer.
7Nadie se entero hasta que Sandra noto que un dashboard de
8Looker mostraba ceros. Para entonces ya habiamos perdido
93 días de datos irrecuperables.
10
11Eso no puede volver a pasar. Tu tarea está semana:
12montar alertas de calidad de datos para el pipeline de
13eventos. Si algo falla, nos enteramos en horas, no en días.
14
15Sandra Vidal  9:50 AM
16+1 ?? Las métricas que construimos la semana pasada dependen
17de que los eventos lleguen. Si dejan de llegar y no nos
18enteramos, los números de retención son mentira.
19Y no puedes recalcular cohortes con datos que no existen.
20
21Pablo Herrera  9:52 PM
22?? Prioridad alta. Ticket DATA-048. Marcos te hace de
23sparring para el diseño.
24
25Lucia Fernandez  9:54 AM
26Ey, +1 ?? Si los datos de retención que mostró el nuevo
27la semana pasada hubieran estado basados en 3 días menos
28de eventos... uf. Me habria presentado al board con números
29incorrectos.
30
31    ?? 3   ? 2

1# data_quality_alerts.py - Sistema de alertas para eventos TaskFlow
2import pandas as pd
3from datetime import datetime, timedelta
4from typing import NamedTuple
5
6class AlertResult(NamedTuple):
7    check_name: str
8    status: str       # 'OK', 'WARNING', 'CRITICAL'
9    message: str
10    value: float
11    threshold: float
12
13def check_volume(df: pd.DataFrame, window_hours: int = 1) -> AlertResult:
14    """
15    Verifica que el volumen de eventos por hora está dentro de lo esperado.
16    Esperado: 80K-150K eventos/hora en horario laboral.
17    WARNING si < 50K. CRITICAL si < 10K o = 0.
18    """
19    now = datetime.utcnow()
20    recent = df[df['event_timestamp'] >= now - timedelta(hours=window_hours)]
21    volume = len(recent)
22    
23    if volume == 0:
24        return AlertResult('volume', 'CRITICAL', 
25                          f'0 eventos en las últimas {window_hours}h', 0, 10000)
26    elif volume < 10000:
27        return AlertResult('volume', 'CRITICAL',
28                          f'Solo {volume:,} eventos (esperado: 80K+)', volume, 10000)
29    elif volume < 50000:
30        return AlertResult('volume', 'WARNING',
31                          f'{volume:,} eventos (esperado: 80K+)', volume, 50000)
32    return AlertResult('volume', 'OK', f'{volume:,} eventos en {window_hours}h', volume, 80000)
33
34def check_freshness(df: pd.DataFrame, max_lag_minutes: int = 30) -> AlertResult:
35    """
36    Verifica que el evento más reciente no tiene más de X minutos.
37    WARNING si > 30 min. CRITICAL si > 120 min.
38    """
39    now = datetime.utcnow()
40    latest = df['event_timestamp'].max()
41    lag_minutes = (now - latest).total_seconds() / 60
42    
43    if lag_minutes > 120:
44        return AlertResult('freshness', 'CRITICAL',
45                          f'Ultimo evento hace {lag_minutes:.0f} min', lag_minutes, 120)
46    elif lag_minutes > max_lag_minutes:
47        return AlertResult('freshness', 'WARNING',
48                          f'Ultimo evento hace {lag_minutes:.0f} min', lag_minutes, max_lag_minutes)
49    return AlertResult('freshness', 'OK',
50                      f'Lag: {lag_minutes:.0f} min', lag_minutes, max_lag_minutes)
51
52def check_distribution(df: pd.DataFrame, window_hours: int = 24) -> AlertResult:
53    """
54    Verifica que la distribución de tipos de evento es estable.
55    Si un tipo que normalmente es >5% cae a 0%, es CRITICAL.
56    """
57    now = datetime.utcnow()
58    recent = df[df['event_timestamp'] >= now - timedelta(hours=window_hours)]
59    
60    expected_types = ['board_viewed', 'task_created', 'task_completed',
61                     'comment_added', 'user_login']
62    
63    missing = []
64    for et in expected_types:
65        if et not in recent['event_type'].values:
66            missing.append(et)
67    
68    if missing:
69        return AlertResult('distribution', 'CRITICAL',
70                          f'Tipos ausentes: {missing}', len(missing), 0)
71    return AlertResult('distribution', 'OK',
72                      'Todos los tipos presentes', 0, 0)
73
74def check_duplicates(df: pd.DataFrame, max_ratio: float = 0.02) -> AlertResult:
75    """
76    Verifica que el ratio de duplicados está en rango normal (<2%).
77    WARNING si > 2%. CRITICAL si > 5%.
78    """
79    total = len(df)
80    dupes = df['event_id'].duplicated().sum()
81    ratio = dupes / total if total > 0 else 0
82    
83    if ratio > 0.05:
84        return AlertResult('duplicates', 'CRITICAL',
85                          f'Ratio de duplicados: {ratio:.1%}', ratio, 0.05)
86    elif ratio > max_ratio:
87        return AlertResult('duplicates', 'WARNING',
88                          f'Ratio de duplicados: {ratio:.1%}', ratio, max_ratio)
89    return AlertResult('duplicates', 'OK',
90                      f'Duplicados: {ratio:.1%}', ratio, max_ratio)
91
92def run_all_checks(df: pd.DataFrame) -> list:
93    """Ejecuta todas las alertas y devuelve resultados."""
94    results = [
95        check_volume(df),
96        check_freshness(df),
97        check_distribution(df),
98        check_duplicates(df),
99    ]
100    
101    print("=== DATA QUALITY REPORT ===")
102    print(f"Timestamp: {datetime.utcnow().isoformat()}")
103    print(f"Eventos analizados: {len(df):,}")
104    print()
105    
106    for r in results:
107        icon = {'OK': '?', 'WARNING': '?', 'CRITICAL': '?'}[r.status]
108        print(f"  {icon} [{r.status:8}] {r.check_name}: {r.message}")
109    
110    critical = [r for r in results if r.status == 'CRITICAL']
111    if critical:
112        print(f"\n?? {len(critical)} ALERTAS CRITICAS -- requieren accion inmediata")
113    
114    return results

Sistema de alertas con 4 checks: volumen, frescura, distribución, unicidad

1# Configuracion del alerting (simplificado)
2ALERT_CONFIG = {
3    "schedule": "0 * * * *",   # Cada hora en punto
4    "slack_channel": "#data-quality",
5    "critical_notify": ["marcos.delgado", "tu.nombre"],
6    "thresholds": {
7        "volume_warning": 50000,
8        "volume_critical": 10000,
9        "freshness_warning_min": 30,
10        "freshness_critical_min": 120,
11        "duplicate_warning": 0.02,
12        "duplicate_critical": 0.05,
13    },
14    "expected_event_types": [
15        "board_viewed", "task_created", "task_completed",
16        "comment_added", "user_login", "task_assigned",
17    ],
18}
19
20# Ejemplo de mensaje Slack para un CRITICAL:
21SLACK_CRITICAL_TEMPLATE = """
22?? *DATA QUALITY ALERT - CRITICAL*
23Check: {check_name}
24Status: {status}
25Message: {message}
26Value: {value}
27Threshold: {threshold}
28Timestamp: {timestamp}
29Action required: investigar inmediatamente
30cc: @marcos.delgado
31"""

Configuración de alertas: umbrales, canal de Slack y template

1Marcos Delgado  (review del PR, viernes)
2
3"Bien. Solo un cambio: anade un check de 'schema drift'.
4Si de repente aparece un campo nuevo en properties que no
5existía antes, quiero saberlo. No es crítico pero es útil
6para detectar deploys de backend que cambian el tracking."
7
8"Y anade un cooldown: si la alerta ya se disparo hace menos
9de 1 hora, no la repitas. Si no, el canal se llena de spam
10cuando algo está roto."
11
12"Aprobado con esos dos cambios. Buen trabajó."