Implementation:Online ml River Anomaly GaussianScorer
| Knowledge Sources | |
|---|---|
| Domains | Online_Learning, Anomaly_Detection, Statistical_Methods, Univariate_Analysis |
| Last Updated | 2026-02-08 16:00 GMT |
Overview
A supervised univariate anomaly detector that fits a Gaussian distribution to target values and scores based on CDF distance from the median.
Description
GaussianScorer maintains a Gaussian distribution of observed target values (optionally over a sliding window). The anomaly score is computed as 2 * |CDF(y) - 0.5|, which measures how far the target value is from the median in terms of the cumulative distribution. Values at the median score 0, while extreme values approach 1.0. A grace period prevents unreliable scores during initial distribution stabilization. This approach assumes the normal behavior follows a Gaussian distribution.
Usage
Use GaussianScorer for univariate target anomaly detection when you expect normal values to be Gaussian-distributed. Set window_size to adapt to non-stationary distributions. The grace_period parameter prevents false alarms during the warm-up phase when the distribution estimate is unstable.
Code Reference
Source Location
- Repository: Online_ml_River
- File: river/anomaly/gaussian.py
Signature
class GaussianScorer(anomaly.base.SupervisedAnomalyDetector):
def __init__(self, window_size=None, grace_period=100):
...
def learn_one(self, x, y):
...
def score_one(self, x, y):
...
Import
from river import anomaly
I/O Contract
| Parameter | Type | Description |
|---|---|---|
| window_size | int (optional) | Sliding window size for adaptive Gaussian |
| grace_period | int (default: 100) | Samples before scoring (returns 0 during grace) |
Usage Examples
import random
from river import anomaly
rng = random.Random(42)
detector = anomaly.GaussianScorer(grace_period=100)
# Learn normal behavior
for _ in range(100):
y = rng.gauss(0, 1)
detector.learn_one(None, y)
# Score potential anomalies
print(detector.score_one(None, -3)) # 0.999477 (extreme)
print(detector.score_one(None, 3)) # 0.999153 (extreme)
print(detector.score_one(None, 0)) # 0.052665 (normal)
print(detector.score_one(None, 0.5)) # 0.383717 (slightly unusual)
# Use with windowed adaptation
adaptive_detector = anomaly.GaussianScorer(
window_size=1000,
grace_period=50
)
# Process stream with potential drift
for i in range(2000):
y = rng.gauss(0 if i < 1000 else 5, 1)
score = adaptive_detector.score_one(None, y)
adaptive_detector.learn_one(None, y)
if score > 0.95:
print(f"Anomaly at {i}: y={y:.2f}, score={score:.3f}")