"""Online anomaly detector and drift handling tests.""" import pytest from src.detector.anomaly import AnomalyDetector from src.detector.drift import DriftDetector from src.detector.metrics import RunningMetrics from src.pipeline.runner import _drift_signal from src.stream.generator import generate_stream from src.stream.models import Observation def _run_detector(detector: AnomalyDetector, obs_list): """Run a detector over a pre-collected observation list, returning per-obs scores.""" frozen = False scores = [] for obs in obs_list: if obs.phase == "A": detector.learn_scaler(obs) elif not frozen: detector.freeze_baseline() frozen = True scores.append(detector.score(obs)) return scores @pytest.mark.asyncio async def test_anomaly_scores_higher_for_injected_anomalies(): """Mahalanobis scorer assigns higher mean scores to injected anomalies than to normals.""" obs_list = [] async for obs in generate_stream( phase_a_length=80, phase_b_length=60, phase_c_length=100, anomaly_rate=0.12, seed=42, ): obs_list.append(obs) detector = AnomalyDetector(threshold=0.5) scores = _run_detector(detector, obs_list) scores_normal = [s for s, o in zip(scores, obs_list) if o.label == "normal"] scores_anomaly = [s for s, o in zip(scores, obs_list) if o.label != "normal"] assert len(scores_anomaly) >= 5, "Need some anomalies in stream" assert sum(scores_anomaly) / len(scores_anomaly) > sum(scores_normal) / len(scores_normal), ( "Mean Mahalanobis score for anomalies should exceed mean score for normals" ) @pytest.mark.asyncio async def test_drift_detector_emits_reset_in_phase_b(): """ADWIN on the sensor drift signal fires during or shortly after Phase B; not during Phase A. Uses delta=0.05 (production config) and a strong drift. ADWIN may fire near the end of Phase B or in the first few observations of Phase C — both are acceptable since ADWIN is adaptive and needs sufficient evidence before committing. """ drift_det = DriftDetector(delta=0.05, grace_period=20) phase_a_len = 150 phase_b_len = 200 fires_before_drift = 0 fires_from_drift = 0 idx = 0 # Accept fire up to 30 obs after Phase B ends (ADWIN lag tolerance) detection_window_end = phase_a_len + phase_b_len + 30 async for obs in generate_stream( phase_a_length=phase_a_len, phase_b_length=phase_b_len, phase_c_length=60, drift_magnitude=0.5, seed=123, ): if drift_det.update(_drift_signal(obs)): if idx < phase_a_len: fires_before_drift += 1 elif idx <= detection_window_end: fires_from_drift += 1 idx += 1 assert fires_before_drift == 0, "No ADWIN fire during Phase A (signal is stationary)" assert fires_from_drift >= 1, "At least one ADWIN fire during/after Phase B (sustained drift)" @pytest.mark.asyncio async def test_no_reset_during_phase_a(): """ADWIN does not fire during Phase A when monitoring the sensor drift signal.""" drift_det = DriftDetector(delta=0.05, grace_period=30) phase_a_len = 120 any_fire_in_a = False async for obs in generate_stream( phase_a_length=phase_a_len, phase_b_length=0, phase_c_length=0, seed=999, ): if drift_det.update(_drift_signal(obs)): any_fire_in_a = True assert not any_fire_in_a, "ADWIN must not fire on stationary Phase A noise" @pytest.mark.asyncio async def test_running_precision_recall_hand_labeled(): """Running precision and recall are computed correctly on a hand-labeled mini-stream.""" data = [ (True, True), # TP (True, True), # TP (True, False), # FN (False, False), # TN (False, True), # FP (False, False), # TN (True, True), # TP (False, False), # TN (False, True), # FP (True, False), # FN (True, True), # TP (False, False), # TN (False, False), # TN (True, True), # TP (False, True), # FP (True, False), # FN (False, False), # TN (True, True), # TP (False, False), # TN (False, True), # FP ] # TP=6, FP=4, FN=3, TN=7 expected_precision = 6 / 10 expected_recall = 6 / 9 expected_f1 = 2 * expected_precision * expected_recall / (expected_precision + expected_recall) metrics = RunningMetrics() for gt, pred in data: metrics.update(ground_truth=gt, predicted=pred) assert abs(metrics.precision - expected_precision) < 1e-9 assert abs(metrics.recall - expected_recall) < 1e-9 assert abs(metrics.f1 - expected_f1) < 1e-9 def test_f1_zero_when_no_positives(): """F1 is 0.0 when no positives have been predicted (avoids division by zero).""" m = RunningMetrics() for _ in range(10): m.update(ground_truth=False, predicted=False) assert m.f1 == 0.0 assert m.precision == 0.0 assert m.recall == 0.0 def test_specificity_all_true_negatives(): """Specificity is 1.0 when all normals are correctly left unflagged.""" m = RunningMetrics() for _ in range(5): m.update(ground_truth=False, predicted=False) assert m.specificity == 1.0 def test_specificity_all_false_positives(): """Specificity is 0.0 when all normals are incorrectly flagged.""" m = RunningMetrics() for _ in range(5): m.update(ground_truth=False, predicted=True) assert m.specificity == 0.0 def test_detector_scores_zero_before_freeze(): """Before freeze_baseline() is called, score() returns 0.0 (warmup phase).""" detector = AnomalyDetector(threshold=0.5) obs = Observation(timestamp=0.0, temperature=27.0, pressure=99.0, vibration=1.5, label="normal", phase="A") assert detector.score(obs) == 0.0, "Score must be 0.0 before Phase A baseline is frozen" def test_detector_scores_anomaly_after_freeze(): """After freeze_baseline(), a large-deviation observation scores higher than a normal one.""" import random rng = random.Random(0) # Use observations with realistic noise so the covariance matrix is non-degenerate. phase_a_obs = [ Observation( timestamp=float(i), temperature=25.0 + rng.gauss(0, 0.5), pressure=101.3 + rng.gauss(0, 0.5), vibration=0.5 + rng.gauss(0, 0.5), label="normal", phase="A", ) for i in range(100) ] detector = AnomalyDetector(threshold=0.5) for obs in phase_a_obs: detector.learn_scaler(obs) detector.freeze_baseline() # Within-range observation (near Phase A baseline) normal = Observation(timestamp=101.0, temperature=25.2, pressure=101.4, vibration=0.5, label="normal", phase="C") # Large spike in temperature and anti-correlated pressure — far from Phase A density anomaly = Observation(timestamp=102.0, temperature=28.5, pressure=98.0, vibration=0.5, label="point_anomaly", phase="C") assert detector.score(anomaly) > detector.score(normal), ( "A large-deviation observation should score higher than a near-baseline one" ) assert 0.0 <= detector.score(normal) < 1.0