Spaces:
Sleeping
Sleeping
| """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.""" | |