zeetay's picture
initial deploy
b0add2b verified
Raw
History Blame Contribute Delete
3.05 kB
"""Online anomaly scorer using Mahalanobis distance from Phase A baseline."""
from __future__ import annotations
import math
import numpy as np
from src.stream.models import Observation
def _to_array(obs: Observation) -> np.ndarray:
return np.array([obs.temperature, obs.pressure, obs.vibration], dtype=float)
class AnomalyDetector:
"""Scores observations using Mahalanobis distance from a Phase A baseline.
Observations are buffered during Phase A to compute mean and covariance.
After freeze_baseline() is called, score() returns tanh(d2/SCALE) where
d2 is the squared Mahalanobis distance from the frozen baseline.
"""
# tanh scale: chi2(3) 97.5th percentile (d2=9.348) maps to ~0.71
_TANH_SCALE: float = 15.0
def __init__(
self,
threshold: float = 0.60,
seed: int = 42,
n_trees: int = 25,
height: int = 8,
window_size: int = 150,
) -> None:
# n_trees / height / window_size kept for API compatibility (used by ADWIN tests)
self.threshold = threshold
self._phase_a_buffer: list[list[float]] = []
self._phase_a_mean: np.ndarray | None = None
self._inv_cov: np.ndarray | None = None
self._frozen: bool = False
# ------------------------------------------------------------------
# Phase A fitting
# ------------------------------------------------------------------
def learn_scaler(self, obs: Observation) -> None:
"""Buffer one Phase A observation. Call for every Phase A observation."""
self._phase_a_buffer.append(_to_array(obs).tolist())
def freeze_baseline(self) -> None:
"""Fit mean and inverse covariance from the Phase A buffer.
Call once when the first Phase B observation arrives.
"""
X = np.array(self._phase_a_buffer, dtype=float)
self._phase_a_mean = X.mean(axis=0)
cov = np.cov(X.T) + 1e-6 * np.eye(3) # regularize for numerical stability
self._inv_cov = np.linalg.inv(cov)
self._frozen = True
# ------------------------------------------------------------------
# Scoring
# ------------------------------------------------------------------
def score(self, obs: Observation) -> float:
"""Return anomaly score in [0, 1). Returns 0.0 before freeze_baseline() is called."""
if not self._frozen:
return 0.0
diff = _to_array(obs) - self._phase_a_mean # type: ignore[operator]
d_sq = float(diff @ self._inv_cov @ diff) # type: ignore[operator]
return math.tanh(d_sq / self._TANH_SCALE)
# ------------------------------------------------------------------
# No-ops kept for pipeline compatibility
# ------------------------------------------------------------------
def learn(self, obs: Observation) -> None:
"""No-op: Mahalanobis uses the frozen Phase A baseline only."""
def reset(self) -> None:
"""No-op: Phase A baseline is intentionally preserved across drift events."""