scripts/tune_threshold.py

"""Find the threshold that maximises F1 on Phase C for the current detector config."""

from __future__ import annotations

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

import numpy as np
from sklearn.metrics import f1_score, precision_score, recall_score

from src.config import settings
from src.detector.anomaly import AnomalyDetector
from src.detector.drift import DriftDetector
from src.pipeline.runner import _drift_signal
from src.stream.generator import _generate_sync


def sweep(seed: int = 42) -> None:
    cfg   = settings.stream
    cfg_d = settings.detector
    cfg_dr = settings.drift

    observations = list(
        _generate_sync(
            phase_a_length=cfg.phase_a_length,
            phase_b_length=cfg.phase_b_length,
            phase_c_length=cfg.phase_c_length,
            drift_magnitude=cfg.drift_magnitude,
            anomaly_rate=cfg.anomaly_rate,
            seed=seed,
        )
    )

    pb_end = cfg.phase_a_length + cfg.phase_b_length
    phase_c = observations[pb_end:]
    y_c = np.array([0 if o.label == "normal" else 1 for o in phase_c])

    detector  = AnomalyDetector(
        threshold=0.5,
        n_trees=cfg_d.n_trees,
        height=cfg_d.height,
        window_size=cfg_d.window_size,
        seed=cfg_d.seed,
    )
    drift_det = DriftDetector(delta=cfg_dr.delta, grace_period=cfg_dr.grace_period)
    scores_c: list[float] = []

    _frozen = False
    for i, obs in enumerate(observations):
        if obs.phase == "A":
            detector.learn_scaler(obs)
        elif not _frozen:
            detector.freeze_baseline()
            _frozen = True
        score = detector.score(obs)
        drift_det.update(_drift_signal(obs))
        if i >= pb_end:
            scores_c.append(score)

    scores_arr = np.array(scores_c)

    print(f"\nPhase C score distribution (n={len(scores_arr)}, anomalies={y_c.sum()}):")
    print(f"  Normal  scores: mean={scores_arr[y_c==0].mean():.4f}  std={scores_arr[y_c==0].std():.4f}  max={scores_arr[y_c==0].max():.4f}")
    print(f"  Anomaly scores: mean={scores_arr[y_c==1].mean():.4f}  std={scores_arr[y_c==1].std():.4f}  min={scores_arr[y_c==1].min():.4f}")
    print()
    print(f"  {'Threshold':>10}  {'Prec':>6}  {'Recall':>6}  {'F1':>6}  {'TP':>4}  {'FP':>4}  {'FN':>4}")
    print("  " + "-" * 55)

    best_f1, best_t = 0.0, 0.5
    for t in np.arange(0.05, 0.95, 0.025):
        preds = (scores_arr >= t).astype(int)
        p  = precision_score(y_c, preds, zero_division=0)
        r  = recall_score(y_c, preds, zero_division=0)
        f1 = f1_score(y_c, preds, zero_division=0)
        tp = int(((preds == 1) & (y_c == 1)).sum())
        fp = int(((preds == 1) & (y_c == 0)).sum())
        fn = int(((preds == 0) & (y_c == 1)).sum())
        marker = " <-- best" if f1 > best_f1 else ""
        if f1 > best_f1:
            best_f1, best_t = f1, float(t)
        print(f"  {t:>10.3f}  {p:>6.3f}  {r:>6.3f}  {f1:>6.3f}  {tp:>4}  {fp:>4}  {fn:>4}{marker}")

    print()
    print(f"  Best threshold: {best_t:.3f}  ->  F1={best_f1:.3f}")
    print(f"  Current config: {cfg_d.threshold:.3f}")


if __name__ == "__main__":
    sweep()