ja4-platform/services/bot-detector/bot_detector/metrics.py

"""Module de métriques de performance du pipeline ML.

Enregistre un résumé de chaque cycle dans ml_performance_metrics :
  - Taux d'anomalie par niveau (CRITICAL/HIGH/MEDIUM/LOW)
  - Taux de corrélation (correlated=1 vs 0)
  - Drift rate, latence, taille baseline, seuil adaptatif
  - Alertes automatiques sur calibration, drift, corrélation, latence

Utilisation dans cycle.py :
    from .metrics import record_cycle_metrics
    record_cycle_metrics(client, cycle_id, model_name, ...)
"""
import time
import logging
from datetime import datetime
from typing import Optional

import pandas as pd

logger = logging.getLogger(__name__)

# Seuils d'alerte (configurables via env vars)
import os
ALERT_ANOMALY_RATE_HIGH  = float(os.getenv('ALERT_ANOMALY_RATE_HIGH', '0.10'))
ALERT_ANOMALY_RATE_LOW   = float(os.getenv('ALERT_ANOMALY_RATE_LOW', '0.005'))
ALERT_DRIFT_RATE         = float(os.getenv('ALERT_DRIFT_RATE', '0.30'))
ALERT_CORRELATION_RATE   = float(os.getenv('ALERT_CORRELATION_RATE', '0.50'))
ALERT_LATENCY_MS         = int(os.getenv('ALERT_LATENCY_MS', '300000'))


def record_cycle_metrics(
    client,
    db: str,
    cycle_id: str,
    model_name: str,
    df_all: pd.DataFrame,
    anomalies: pd.DataFrame,
    all_scored: pd.DataFrame,
    drift_rate: float,
    cycle_start_time: float,
    baseline_size: int,
    threshold: float,
    valid_features: int,
    total_features: int,
    meta_learner_active: bool = False,
) -> None:
    """Enregistre les métriques d'un cycle dans ml_performance_metrics.

    Émet également des alertes dans les logs si les seuils sont dépassés.

    Args:
        client           : ClickHouseClient actif
        db               : nom de la base ja4_processing
        cycle_id         : identifiant du cycle (timestamp)
        model_name       : 'Complet' ou 'Applicatif' (ou variante 24h)
        df_all           : DataFrame complet du cycle (avant scoring)
        anomalies        : DataFrame des anomalies détectées
        all_scored       : DataFrame de tous les scores
        drift_rate       : taux de features en dérive [0, 1]
        cycle_start_time : timestamp (time.time()) du début du cycle
        baseline_size    : nombre de sessions dans la baseline humaine
        threshold        : seuil adaptatif utilisé
        valid_features   : nombre de features valides
        total_features   : nombre total de features demandées
        meta_learner_active : True si le meta-learner a été utilisé ce cycle
    """
    now = datetime.utcnow()
    latency_ms = int((time.time() - cycle_start_time) * 1000)

    n_total = max(len(df_all), 1)
    n_scored = max(len(all_scored), 1)

    # Taux de corrélation
    if 'correlated' in df_all.columns:
        n_correlated = int((df_all['correlated'] == 1).sum())
        correlated_rate = n_correlated / n_total
    else:
        correlated_rate = 0.0

    # Comptage par niveau de menace
    n_critical = n_high = n_medium = n_low = 0
    if not anomalies.empty and 'threat_level' in anomalies.columns:
        levels = anomalies['threat_level'].value_counts()
        n_critical = int(levels.get('CRITICAL', 0))
        n_high     = int(levels.get('HIGH', 0))
        n_medium   = int(levels.get('MEDIUM', 0))
        n_low      = int(levels.get('LOW', 0))

    # Bots connus et navigateurs légitimes
    n_known_bot = 0
    n_anubis_deny = 0
    n_legit_browser = 0
    if not df_all.empty:
        if 'bot_name' in df_all.columns:
            n_known_bot = int((df_all['bot_name'].fillna('') != '').sum())
        if 'anubis_bot_action' in df_all.columns:
            n_anubis_deny = int((df_all['anubis_bot_action'] == 'DENY').sum())
        if 'browser_confidence' in df_all.columns:
            from .config import BROWSER_CONFIDENCE_THRESHOLD
            n_legit_browser = int((df_all['browser_confidence'] >= BROWSER_CONFIDENCE_THRESHOLD).sum())

    anomaly_rate = (n_critical + n_high + n_medium + n_low) / n_scored
    drift_alert = 1 if drift_rate > ALERT_DRIFT_RATE else 0

    # Alertes
    _emit_alerts(model_name, anomaly_rate, drift_rate, correlated_rate, latency_ms, drift_alert)

    try:
        # Vérifier que la table existe avant d'insérer (optionnelle)
        table_check = client.query(
            f"SELECT name FROM system.tables WHERE database = '{db}' AND name = 'ml_performance_metrics'"
        )
        if not table_check.result_rows:
            logger.debug("[Metrics] Table ml_performance_metrics absente — métriques non enregistrées")
            return

        client.insert(
            f"{db}.ml_performance_metrics",
            [[
                now,
                model_name,
                n_total,
                round(float(correlated_rate), 4),
                round(float(anomaly_rate), 4),
                n_critical,
                n_high,
                n_medium,
                n_low,
                n_known_bot,
                n_anubis_deny,
                n_legit_browser,
                round(float(drift_rate), 4),
                drift_alert,
                latency_ms,
                valid_features,
                total_features,
                baseline_size,
                round(float(threshold), 6),
                1 if meta_learner_active else 0,
            ]]
        )
        logger.debug(f"[Metrics] Cycle {cycle_id} enregistré ({latency_ms}ms)")
    except Exception as e:
        logger.warning(f"[Metrics] Erreur d'enregistrement des métriques : {e}")


def _emit_alerts(model_name: str, anomaly_rate: float, drift_rate: float,
                 correlated_rate: float, latency_ms: int, drift_alert: int) -> None:
    """Émet des alertes dans les logs si les seuils sont dépassés."""
    if anomaly_rate > ALERT_ANOMALY_RATE_HIGH:
        logger.warning(
            f"[{model_name}] ⚠ ALERTE CALIBRATION : taux d'anomalie élevé "
            f"({anomaly_rate:.1%} > {ALERT_ANOMALY_RATE_HIGH:.1%})"
        )
    elif anomaly_rate < ALERT_ANOMALY_RATE_LOW and anomaly_rate > 0:
        logger.warning(
            f"[{model_name}] ⚠ ALERTE CALIBRATION : taux d'anomalie très bas "
            f"({anomaly_rate:.3%} < {ALERT_ANOMALY_RATE_LOW:.1%})"
        )
    if drift_alert:
        logger.warning(
            f"[{model_name}] ⚠ ALERTE DRIFT : {drift_rate:.1%} des features en dérive "
            f"(seuil {ALERT_DRIFT_RATE:.1%})"
        )
    if correlated_rate < ALERT_CORRELATION_RATE:
        logger.warning(
            f"[{model_name}] ⚠ ALERTE CORRÉLATION : taux de corrélation bas "
            f"({correlated_rate:.1%} < {ALERT_CORRELATION_RATE:.1%}) — "
            "vérifier ja4sentinel/logcorrelator"
        )
    if latency_ms > ALERT_LATENCY_MS:
        logger.warning(
            f"[{model_name}] ⚠ ALERTE PERFORMANCE : latence cycle {latency_ms}ms "
            f"> {ALERT_LATENCY_MS}ms"
        )