tests/test_detector.py

"""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