Spaces:
Sleeping
Sleeping
File size: 7,397 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 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 | """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
|