Quadro branco com sticky notes organizados em categorias de produtividade e workflow

Análise Temporal de Frequência de Erros: Detector de Picos e Tendências com Python Puro

PorOlivetto

Seu log está gritando, mas você só percebe quando já virou incidente.

Eu estive aí. 3 da manhã, pager disparando, e a pergunta inevitável: “quando isso começou a piorar?” A resposta sempre estava nos logs das últimas 6 horas, mas ninguém tinha parado pra olhar a frequência dos erros, só o conteúdo.

Hoje vou te mostrar como construir um detector de picos e tendências que analisa logs temporalmente, identifica anomalias antes que virem incidentes, e te avisa quando algo está “esquentando”. Sem Elasticsearch, sem Datadog, sem mensalidade. Só Python puro, estatística básica e aquela velha filosofia: se dá pra fazer local, não precisa de nuvem.

Por que frequência importa mais que conteúdo

Quando um erro aparece pela primeira vez, você investiga. Quando aparece dez vezes em uma hora, você entra em pânico. O problema é que o pânico geralmente vem tarde demais.

A maioria dos sistemas de monitoramento foca no que aconteceu (a mensagem de erro, o stack trace), mas ignora o quando e o com que frequência. Isso é como olhar pra temperatura de um paciente sem ver se ela está subindo, caindo ou estável.

O que vamos construir faz três coisas:

  • Detector de picos: identifica quando a frequência de um tipo de erro aumenta subitamente
  • Analisador de tendências: calcula se um erro está ficando mais comum ao longo do tempo
  • Agregador inteligente: agrupa erros similares por padrão, não só por string exata

A arquitetura: SQLite + Python + estatística básica

Vou usar SQLite porque é local, rápido o suficiente pra logs de aplicações médias, e não precisa de servidor. O esquema é simples:

CREATE TABLE error_events (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    error_signature TEXT NOT NULL,
    timestamp REAL NOT NULL,
    raw_message TEXT,
    source TEXT
);

CREATE INDEX idx_signature_time ON error_events(error_signature, timestamp);

A error_signature é o pulo do gato. Em vez de salvar “ConnectionError: timeout after 30s” e “ConnectionError: timeout after 45s” como coisas diferentes, normalizamos pra “ConnectionError: timeout after *s”. Isso agrupa variações do mesmo problema.

Normalização: o segredo está no regex

Aqui está a função que transforma mensagens de erro em assinaturas agrupáveis:

import re
import hashlib

def normalize_error(message: str) -> str:
    """Remove números, UUIDs, timestamps e caminhos específicos."""
    # Remove UUIDs
    normalized = re.sub(
        r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}',
        '<UUID>', message
    )
    # Remove números (mas preserva nomes de erro tipo HTTP 500)
    normalized = re.sub(r'\b\d+\b', '<N>', normalized)
    # Remove caminhos de arquivo específicos
    normalized = re.sub(r'/[\w/.-]+', '<PATH>', normalized)
    # Remove timestamps ISO
    normalized = re.sub(
        r'\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}',
        '<TIMESTAMP>', normalized
    )
    return normalized

def error_signature(message: str) -> str:
    """Gera hash curto da mensagem normalizada."""
    normalized = normalize_error(message)
    return hashlib.sha256(normalized.encode()).hexdigest()[:12]

Isso transforma “Timeout connecting to 192.168.1.42:5432 after 30s” e “Timeout connecting to 10.0.0.15:5432 after 45s” na mesma assinatura. Perfeito.

Detectando picos com desvio padrão móvel

Um pico acontece quando a frequência atual de um erro está significativamente acima da média histórica. Vou usar uma janela móvel de 1 hora e calcular quantos desvios padrão acima da média estamos.

import sqlite3
from datetime import datetime, timedelta
import statistics

def detect_spike(db_path: str, signature: str, 
                 window_hours: int = 1, threshold_sigma: float = 2.0) -> dict:
    """Detecta se há pico na frequência de um erro específico."""
    conn = sqlite3.connect(db_path)
    now = datetime.now().timestamp()
    window_start = now - (window_hours * 3600)
    baseline_start = now - (24 * 3600)  # últimas 24h como baseline
    
    # Conta ocorrências na janela atual
    cursor = conn.execute(
        "SELECT COUNT(*) FROM error_events "
        "WHERE error_signature = ? AND timestamp >= ?",
        (signature, window_start)
    )
    current_count = cursor.fetchone()[0]
    
    # Calcula média e desvio padrão das últimas 24h (em janelas de 1h)
    hourly_counts = []
    for hour_offset in range(24):
        hour_start = now - ((hour_offset + 1) * 3600)
        hour_end = now - (hour_offset * 3600)
        cursor = conn.execute(
            "SELECT COUNT(*) FROM error_events "
            "WHERE error_signature = ? AND timestamp >= ? AND timestamp < ?",
            (signature, hour_start, hour_end)
        )
        hourly_counts.append(cursor.fetchone()[0])
    
    conn.close()
    
    if len(hourly_counts) < 3:
        return {"spike": False, "reason": "dados insuficientes"}
    
    mean = statistics.mean(hourly_counts)
    stdev = statistics.stdev(hourly_counts) if len(hourly_counts) > 1 else 0
    
    if stdev == 0:
        # Sem variação histórica, qualquer aumento é significativo
        is_spike = current_count > mean * 2
        z_score = 999 if current_count > mean else 0
    else:
        z_score = (current_count - mean) / stdev
        is_spike = z_score >= threshold_sigma
    
    return {
        "spike": is_spike,
        "current_count": current_count,
        "mean_hourly": mean,
        "z_score": z_score,
        "threshold": threshold_sigma
    }

O z-score é a mágica aqui. Um z-score de 2.0 significa que a frequência atual está 2 desvios padrão acima da média. Na distribuição normal, isso acontece em ~5% dos casos. Ou seja, é raro o suficiente pra merecer atenção.

Análise de tendência: regressão linear simples

Às vezes não é um pico súbito, mas um aumento gradual. Um erro que acontecia 2x por hora e agora acontece 10x pode passar despercebido se você só olha picos.

Para tendências, uso regressão linear nos últimos 7 dias:

def analyze_trend(db_path: str, signature: str, days: int = 7) -> dict:
    """Analisa se um erro está aumentando ou diminuindo ao longo do tempo."""
    conn = sqlite3.connect(db_path)
    now = datetime.now().timestamp()
    start_time = now - (days * 24 * 3600)
    
    # Agrupa por dia
    daily_counts = []
    for day_offset in range(days):
        day_start = now - ((day_offset + 1) * 24 * 3600)
        day_end = now - (day_offset * 24 * 3600)
        cursor = conn.execute(
            "SELECT COUNT(*) FROM error_events "
            "WHERE error_signature = ? AND timestamp >= ? AND timestamp < ?",
            (signature, day_start, day_end)
        )
        daily_counts.append(cursor.fetchone()[0])
    
    conn.close()
    
    # Inverte pra ordem cronológica (mais antigo primeiro)
    daily_counts.reverse()
    
    # Regressão linear simples
    n = len(daily_counts)
    x_values = list(range(n))
    sum_x = sum(x_values)
    sum_y = sum(daily_counts)
    sum_xy = sum(x * y for x, y in zip(x_values, daily_counts))
    sum_x2 = sum(x ** 2 for x in x_values)
    
    denominator = (n * sum_x2 - sum_x ** 2)
    if denominator == 0:
        return {"trend": "stable", "slope": 0, "daily_avg": daily_counts[0]}
    
    slope = (n * sum_xy - sum_x * sum_y) / denominator
    intercept = (sum_y - slope * sum_x) / n
    
    # Classifica tendência
    daily_avg = sum_y / n if n > 0 else 0
    if daily_avg == 0:
        trend = "stable"
    elif slope > daily_avg * 0.1:  # aumento de 10% ao dia
        trend = "increasing"
    elif slope < -daily_avg * 0.1:
        trend = "decreasing"
    else:
        trend = "stable"
    
    return {
        "trend": trend,
        "slope": slope,  # mudança por dia
        "daily_avg": daily_avg,
        "predicted_tomorrow": intercept + slope * n
    }

Isso te dá uma previsão simples: se a tendência continuar, quantos erros vamos ter amanhã? Se a resposta é "muitos", talvez seja hora de investigar antes que vire incidente.

O monitor completo: rodando em background

Agora juntamos tudo num script que roda a cada 15 minutos e alerta quando algo está errado:

import time
import json
from pathlib import Path

def run_monitor(db_path: str, alert_threshold_spike: float = 2.5, 
                alert_threshold_trend: str = "increasing"):
    """Monitor principal: verifica picos e tendências."""
    conn = sqlite3.connect(db_path)
    
    # Pega todas as assinaturas únicas das últimas 24h
    now = datetime.now().timestamp()
    yesterday = now - (24 * 3600)
    cursor = conn.execute(
        "SELECT DISTINCT error_signature FROM error_events WHERE timestamp >= ?",
        (yesterday,)
    )
    signatures = [row[0] for row in cursor.fetchall()]
    conn.close()
    
    alerts = []
    
    for sig in signatures:
        # Verifica pico
        spike_result = detect_spike(db_path, sig, threshold_sigma=alert_threshold_spike)
        if spike_result["spike"]:
            alerts.append({
                "type": "spike",
                "signature": sig,
                "current_count": spike_result["current_count"],
                "z_score": spike_result["z_score"],
                "message": f"Pico detectado: {spike_result['current_count']} ocorrências (z={spike_result['z_score']:.2f})"
            })
        
        # Verifica tendência
        trend_result = analyze_trend(db_path, sig)
        if trend_result["trend"] == alert_threshold_trend:
            alerts.append({
                "type": "trend",
                "signature": sig,
                "slope": trend_result["slope"],
                "predicted_tomorrow": trend_result["predicted_tomorrow"],
                "message": f"Tendência de alta: +{trend_result['slope']:.1f}/dia, previsão amanhã: {trend_result['predicted_tomorrow']:.0f}"
            })
    
    # Salva alertas
    if alerts:
        alert_file = Path("/tmp/error_monitor_alerts.json")
        with open(alert_file, "w") as f:
            json.dump({"timestamp": now, "alerts": alerts}, f, indent=2)
        
        # Aqui você integra com seu sistema de notificação
        # (webhook, email, Telegram, etc)
        for alert in alerts:
            print(f"[ALERT] {alert['message']}")

if __name__ == "__main__":
    # Roda a cada 15 minutos
    while True:
        run_monitor("errors.db")
        time.sleep(900)

Integrando com logs existentes

Você provavelmente já tem logs. Aqui está um parser simples que lê um arquivo de log e popula o banco:

def ingest_log_file(db_path: str, log_file: str):
    """Lê um arquivo de log e insere eventos no banco."""
    conn = sqlite3.connect(db_path)
    
    with open(log_file, "r") as f:
        for line in f:
            # Ajuste esse regex pro formato do seu log
            match = re.match(
                r'\[(.*?)\] ERROR: (.*)',
                line.strip()
            )
            if match:
                timestamp_str, message = match.groups()
                # Converte timestamp (ajuste o formato)
                timestamp = datetime.fromisoformat(timestamp_str).timestamp()
                signature = error_signature(message)
                
                conn.execute(
                    "INSERT INTO error_events (error_signature, timestamp, raw_message, source) "
                    "VALUES (?, ?, ?, ?)",
                    (signature, timestamp, message, log_file)
                )
    
    conn.commit()
    conn.close()

Quando isso não é suficiente

Vou ser honesto: essa abordagem funciona bem pra aplicações com até ~1000 erros/hora. Se você tem milhões de eventos, vai precisar de algo como ClickHouse ou Elasticsearch. Mas a lógica de detecção (z-score, regressão linear) é a mesma.

O ponto é: não espere ter "escala de verdade" pra começar a monitorar frequência. A maioria dos incidentes que vi na carreira poderia ter sido detectada 2-4 horas antes se alguém tivesse olhado pra tendência.

E aí, qual automação de monitoramento você quer ver aqui no AutoMente? Detector de padrões em stack traces? Agregador de logs multi-serviço? Manda nos comentários que eu construo.

Posts Similares