Spaces:
Sleeping
Sleeping
File size: 3,197 Bytes
b0add2b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | """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()
|