"""Unit tests for streaming outlier detectors.""" import math import pytest from detector import OutlierDetector, OutlierDetectorMAD, OutlierDetectorFrugalMAD # --------------------------------------------------------------------------- # OutlierDetector (IQR) # --------------------------------------------------------------------------- class TestOutlierDetectorIQR: def _fill_buffer(self, det, values): """Feed values into the detector to fill its buffer.""" for v in values: det.process(v) def test_constant_buffer_no_false_outlier(self): det = OutlierDetector(buffer_size=11, k=1.5) self._fill_buffer(det, [5.0] * 11) # IQR is zero so the guard should prevent a false positive assert det.process(5.0) is False def test_clear_outlier_flagged(self): det = OutlierDetector(buffer_size=11, k=1.5) self._fill_buffer(det, list(range(11))) assert det.process(1000.0) is True def test_normal_value_not_flagged(self): det = OutlierDetector(buffer_size=11, k=1.5) self._fill_buffer(det, list(range(11))) assert det.process(5.0) is False def test_nan_returns_none(self): det = OutlierDetector(buffer_size=11) self._fill_buffer(det, list(range(11))) assert det.process(float("nan")) is None def test_none_returns_none(self): det = OutlierDetector(buffer_size=11) self._fill_buffer(det, list(range(11))) assert det.process(None) is None def test_buffer_not_full_returns_false(self): det = OutlierDetector(buffer_size=11) assert det.process(999.0) is False def test_invalid_k_raises(self): with pytest.raises(ValueError): OutlierDetector(k=0.0) with pytest.raises(ValueError): OutlierDetector(k=10.0) # --------------------------------------------------------------------------- # OutlierDetectorMAD # --------------------------------------------------------------------------- class TestOutlierDetectorMAD: def _fill_buffer(self, det, values): for v in values: det.process(v) def test_constant_buffer_no_false_outlier(self): det = OutlierDetectorMAD(buffer_size=11, k=1.5) self._fill_buffer(det, [5.0] * 11) assert det.process(5.0) is False def test_clear_outlier_flagged(self): det = OutlierDetectorMAD(buffer_size=11, k=1.5) self._fill_buffer(det, list(range(11))) assert det.process(1000.0) == True def test_nan_returns_none(self): det = OutlierDetectorMAD(buffer_size=11) self._fill_buffer(det, list(range(11))) assert det.process(float("nan")) is None def test_none_returns_none(self): det = OutlierDetectorMAD(buffer_size=11) self._fill_buffer(det, list(range(11))) assert det.process(None) is None def test_buffer_not_full_returns_false(self): det = OutlierDetectorMAD(buffer_size=11) assert det.process(999.0) is False def test_invalid_k_raises(self): with pytest.raises(ValueError): OutlierDetectorMAD(k=0.0) with pytest.raises(ValueError): OutlierDetectorMAD(k=10.0) # --------------------------------------------------------------------------- # OutlierDetectorFrugalMAD # --------------------------------------------------------------------------- class TestOutlierDetectorFrugalMAD: def test_nan_returns_none(self): det = OutlierDetectorFrugalMAD() assert det.process(float("nan")) is None def test_none_returns_none(self): det = OutlierDetectorFrugalMAD() assert det.process(None) is None def test_constant_values_non_outlier_not_flagged(self): det = OutlierDetectorFrugalMAD(k=4.0) for _ in range(200): det.process(5.0) # A value close to the constant should not be flagged assert det.process(5.0) is False def test_extreme_value_flagged_when_mad_positive(self): det = OutlierDetectorFrugalMAD(k=4.0) # Directly set converged internal state to test the outlier branch det.med = 5.0 det.mad = 1.0 # Fences: [5 - 4*1, 5 + 4*1] = [1, 9]; 100 is far outside result = det.process(100.0) assert result == True def test_zero_mad_guard_no_crash(self): det = OutlierDetectorFrugalMAD(k=4.0) # Feed identical values; MAD may reach zero for _ in range(100): result = det.process(3.0) # Should return False (not crash) assert result is False def test_mad_cannot_go_negative(self): """MAD should be clamped at zero, never negative.""" det = OutlierDetectorFrugalMAD(k=3.0) det.med = 5.0 det.mad = 0.001 # very small MAD # Feed values close to median to push MAD down for _ in range(100): det.process(5.0) assert det.mad >= 0.0, f"MAD went negative: {det.mad}" def test_converges_from_data_and_flags_spike(self): """Regression: driven only through process(), the detector must bootstrap its own scale estimate and flag an obvious spike. A previous version seeded the MAD from the first sample's deviation against the just-set median (always zero), which froze the step size at zero and made the detector permanently blind. The earlier tests missed this because they injected med/mad internal state directly. """ import numpy as np det = OutlierDetectorFrugalMAD(k=4.0) rng = np.random.default_rng(0) for v in rng.normal(0.0, 0.1, 500): det.process(v) assert det.mad > 0, "scale estimate never bootstrapped" assert det.process(5.0) == True, "obvious spike not flagged" assert det.process(0.05) == False, "in-range value wrongly flagged" def test_recovers_from_a_flat_stretch_without_flooding(self): """Regression: a quiet stretch must not turn the detector into a false-alarm generator when real signal returns. The MAD estimate correctly decays toward zero on constant input. An earlier version let it decay geometrically to a tiny nonzero value, leaving fences far narrower than the signal's real noise: every subsequent in-range sample was flagged, for thousands of samples. The floor plus gap-proportional recovery bounds that. """ import numpy as np det = OutlierDetectorFrugalMAD(k=4.0) rng = np.random.default_rng(0) for v in rng.normal(5.0, 0.05, 500): # converge on real noise det.process(v) for _ in range(3000): # a long flat stretch det.process(5.0) # Real, in-range signal returns. Allow a short recovery burst, # then require the false-positive rate to settle. first = sum(bool(det.process(v)) for v in rng.normal(5.0, 0.05, 200)) later = sum(bool(det.process(v)) for v in rng.normal(5.0, 0.05, 200)) assert first < 100, f"flooding on recovery: {first}/200 flagged" assert later < 20, f"still flooding after 200 samples: {later}/200" assert det.process(5.5) == True, "not sensitive again after recovery" def test_scale_floor_suppresses_flags_not_crashes(self): """At the floor the scale is unknown, so nothing is flagged.""" det = OutlierDetectorFrugalMAD(k=4.0, min_scale=1e-3) for _ in range(500): result = det.process(2.0) assert det.mad == pytest.approx(1e-3) assert result is False def test_rejects_nonpositive_min_scale(self): with pytest.raises(ValueError): OutlierDetectorFrugalMAD(min_scale=0.0)