┌──────────────────────────────┐
│ 1. DATA INGESTION (stream)   │  Kafka / API / DB
└──────────────┬───────────────┘
               ▼
┌──────────────────────────────┐
│ 2. PREPROCESSING             │  Cleaning + Scaling
└──────────────┬───────────────┘
               ▼
┌──────────────────────────────┐
│ 3. STATISTICAL LAYER         │  IQR + MAD (Filtri veloci e robusti)
└──────────────┬───────────────┘
               ▼
┌──────────────────────────────┐
│ 4. ML LAYER                  │  Isolation Forest (Pattern complessi)
└──────────────┬───────────────┘
               ▼
┌──────────────────────────────┐
│ 5. DECISION ENGINE           │  Ensemble pesato + Alerting calibrato
└──────────────┬───────────────┘
               ▼
        ALERT / DASHBOARD / API

Dipendenze

import numpy as np
import pandas as pd
from sklearn.ensemble import IsolationForest

🔹 1. Funzioni statistiche robuste

IQR detector

def iqr_detector(series):
    q1 = np.percentile(series, 25)
    q3 = np.percentile(series, 75)
    iqr = q3 - q1

    lower = q1 - 1.5 * iqr
    upper = q3 + 1.5 * iqr

    return (series < lower) | (series > upper)

MAD detector

def mad_detector(series, k=2.5):
    median = np.median(series)
    mad = np.median(np.abs(series - median))
    modified_z = (series - median) / (1.4826 * mad)

    return np.abs(modified_z) > k, modified_z

🔹 2. Isolation Forest layer

def isolation_forest_detector(series, contamination=0.05):
    model = IsolationForest(
        n_estimators=100,
        contamination=contamination,
        random_state=42
    )

    X = series.values.reshape(-1, 1)
    model.fit(X)

    scores = model.decision_function(X)
    preds = model.predict(X)

    # -1 anomaly, 1 normal
    return preds == -1, scores

🔹 3. Ensemble decision engine

def ensemble_anomaly(iqr_flag, mad_flag, if_flag):
    score = (
        0.2 * iqr_flag +
        0.4 * mad_flag +
        0.4 * if_flag
    )

    decision = np.where(
        score > 0.7, "ANOMALY",
        np.where(score > 0.4, "REVIEW", "OK")
    )

    return score, decision

🔹 4. Pipeline completa

def anomaly_pipeline(series):

    # Step 1: IQR
    iqr_flag = iqr_detector(series).astype(int)

    # Step 2: MAD
    mad_flag, mad_score = mad_detector(series)
    mad_flag = mad_flag.astype(int)

    # Step 3: Isolation Forest
    if_flag, if_score = isolation_forest_detector(series)
    if_flag = if_flag.astype(int)

    # Step 4: Ensemble
    score, decision = ensemble_anomaly(iqr_flag, mad_flag, if_flag)

    return pd.DataFrame({
        "value": series,
        "iqr_flag": iqr_flag,
        "mad_flag": mad_flag,
        "if_flag": if_flag,
        "score": score,
        "decision": decision
    })

Esempio reale (KPI e-commerce)

data = pd.Series([
    120, 130, 128, 135, 140, 138, 500, 145, 150, 148, 160, 1000
])

result = anomaly_pipeline(data)
print(result)

 Output tipico

    value  iqr_flag  mad_flag  if_flag  score decision
0     120         0         0        0    0.0       OK
1     130         0         0        0    0.0       OK
2     128         0         0        0    0.0       OK
3     135         0         0        0    0.0       OK
4     140         0         0        0    0.0       OK
5     138         0         0        0    0.0       OK
6     500         1         1        0    0.6   REVIEW
7     145         0         0        0    0.0       OK
8     150         0         0        0    0.0       OK
9     148         0         0        0    0.0       OK
10    160         0         0        0    0.0       OK
11   1000         1         1        1    1.0  ANOMALY

def mad_detector(series, k=2.5):
    # Conversione in array numpy per massima compatibilità
    arr = np.asarray(series)
    
    if len(arr) < 3:
        return np.zeros(len(arr), dtype=bool), np.zeros(len(arr))
        
    median = np.median(arr)
    mad = np.median(np.abs(arr - median))
    
    # Protezione anti-crash per flussi di dati piatti o sensori bloccati
    if mad == 0:
        mad = 1e-6
        
    modified_z = (arr - median) / (1.4826 * mad)
    return np.abs(modified_z) > k, modified_z

Perché questo esercizio è interessante 

Questo esercizio non è un semplice esempio accademico, ma rappresenta un bignami di architettura del software applicata alla data science. Le sue peculiarità principali sono:

 Dimostrazione visiva del “Voto di Maggioranza Qualificato” (Ensemble)

L’output sul valore 500 rispetto a 1000 mostra esattamente come l’architettura filtri il rumore. L’IQR e il MAD (che guardano la distribuzione monodimensionale statica) vedono 500 come un valore alieno. Tuttavia, l’Isolation Forest capisce che, sebbene sia alto, non ha ancora rotto la struttura geometrica globale dello spazio dei dati. L’assegnazione dello stato REVIEW impedisce l’invio di un finto alert critico, salvando i flussi di lavoro dei team Operations.

L’Edge Case del “Mondo Reale” (mad == 0)

Nei libri di testo si assume che la varianza o la deviazione dei dati sia sempre presente. In produzione, un sensore industriale che perde la connessione o si blocca su un valore fisso (es. telemetria congelata) azzera completamente la deviazione. Questo esercizio evidenzia come un sistema puramente matematico possa fallire non per la complessità del dato, ma per la sua staticità anomala, introducendo il concetto di programmazione difensiva nei sistemi di Machine Learning.

Separazione dei Layer Computazionali

L’esercizio mostra l’efficienza computazionale. In uno scenario di streaming reale di grandi volumi, i layer 1 e 2 (IQR, MAD) richiedono pochissime risorse CPU ([math]\mathcal{O}(N)[/math] o [math]\mathcal{O}(N \log N)[/math] per l’ordinamento dei quantili), agendo da barriera. L’Isolation Forest, più pesante, interviene o viene calibrata solo su dati che superano determinati controlli o viene usata in parallelo sapendo che le metriche veloci hanno già rimosso le anomalie grossolane (gli spike impossibili).

Dal voto alla priorità: Il Severity Scoring

L’etichetta discreta (OK, REVIEW, ANOMALY) è comoda per i report mattutini, ma in produzione la vera domanda a cui il sistema deve rispondere alle tre di notte è sempre e solo una: “Quanto devo preoccuparmi adesso?”.

Avere un sistema di incident management (come PagerDuty, Opsgenie o i workflow di Slack) basato su stringhe di testo rigide è limitante. Questi sistemi vivono di priorità continue. Ecco come possiamo implementare un Severity Score a partire dal punteggio grezzo dell’ensemble (da 0.0 a 1.0):

def severity_from_score(score):
    # Converte lo score [0, 1] in una severità [0, 100]
    severity = int(score * 100)
    
    if severity <= 30:
        level = "NOISE"
    elif severity <= 60:
        level = "REVIEW"
    elif severity <= 85:
        level = "WARNING"
    else:
        level = "CRITICAL"
        
    return severity, level

 Perché questo approccio domina nel mondo reale

• Gradualità operativa: Un valore che passa da 88 a 92 non cambia la macro-azione (è sempre un CRITICAL), ma fa salire automaticamente il ticket in cima alla coda di priorità del team.
• Soglie asimmetriche: Le zone 0-30 e 86-100 sono le più importanti. Servono, rispettivamente, per filtrare spietatamente il rumore di fondo (evitando l’alert fatigue) e per garantire che lo stato critico scatti solo quando il sistema è davvero compromesso.
• Integrazione nativa: I sistemi di escalation aziendali accettano nativamente un intero 0-100 e applicano le loro policy di routing senza bisogno di parsing complessi.

Un errore da non fare: La trappola della linearità

C’è un’insidia nascosta nella funzione qui sopra. Non linearizzare mai lo score se la distribuzione delle anomalie è fortemente asimmetrica (skewed).

Se il 90% degli score storici del tuo sistema è inferiore a 0.2, mappare linearmente lo 0.2 al livello 20 di severità potrebbe generare una valanga di falsi WARNING non appena il valore tocca lo 0.4. In questi casi, la severità non deve riflettere il valore assoluto, ma la sua rarità. Si usa quindi una calibrazione sui quantili empirici:

import numpy as np
from scipy.stats import percentileofscore

def calibrated_severity(historical_scores, current_score):
    # Soglie dinamiche basate sulla distribuzione storica reale
    noise_th = np.percentile(historical_scores, 70)   # Il 70% dei dati è rumore
    review_th = np.percentile(historical_scores, 85)
    warning_th = np.percentile(historical_scores, 95)
    
    if current_score <= noise_th:
        # Mappa dinamicamente nella fascia 0-30
        return int((current_score / noise_th) * 30), "NOISE"
    elif current_score <= warning_th:
        return 75, "WARNING" # Esempio di mapping semplificato
    else:
        return 95, "CRITICAL"

Pattern Avanzato: Il decadimento della severità

Cosa succede se un sensore IoT subisce un danno fisico e inizia a inviare un valore anomalo ma costante nel tempo?

La pipeline rileverà un’anomalia continua. Generare un alert critico al minuto 1 è corretto.

Continuare a trattarlo come “critico ed emergente” dopo 3 ore non ha senso e intasa la dashboard operativa.

Un’architettura matura implementa il decadimento della severità (Severity Decay). Se un’anomalia persiste senza aggravarsi, la sua priorità deve calare fisiologicamente nel tempo:

def decay_severity(original_severity, minutes_since_first_alert):
    # Dimezzamento della severità spalmato su una finestra di 3 ore (180 min)
    decay_factor = max(0.2, 1 - (minutes_since_first_alert / 180))
    
    current_severity = int(original_severity * decay_factor)
    return current_severity

Applicando questa logica, un allarme che parte a livello 90 (CRITICAL) ma rimane identico per tre ore, scenderà gradualmente verso quota 50-45, declassandosi automaticamente a REVIEW. Questo trasforma il nostro semplice ensemble statistico in un motore di priorità auto-regolante.