Implementation:Online ml River Anomaly SAD
| Knowledge Sources | |
|---|---|
| Domains | Online_Learning, Anomaly_Detection, Statistical_Methods, Univariate_Analysis |
| Last Updated | 2026-02-08 16:00 GMT |
Overview
Standard Absolute Deviation computes anomaly scores by measuring how many standard deviations a value is from a central statistic (mean or median).
Description
SAD (StandardAbsoluteDeviation) implements the 3-sigma rule for anomaly detection. It maintains running statistics of the target variable's variance and a central tendency measure (mean by default, but median can be used). The anomaly score is the absolute z-score: |y - central| / sqrt(variance). This gives a measure of how many standard deviations the value is from the center. Values more than 3 standard deviations away typically score above 3.0, indicating anomalies. The algorithm is adapted from PySAD's implementation.
Usage
Use SAD for simple, interpretable univariate anomaly detection based on statistical deviation. It works well when anomalies are values that deviate significantly from the mean. Use stats.Median() as sub_stat for robustness to outliers in the central tendency estimate.
Code Reference
Source Location
- Repository: Online_ml_River
- File: river/anomaly/sad.py
Signature
class StandardAbsoluteDeviation(anomaly.base.SupervisedAnomalyDetector):
def __init__(self, sub_stat: stats.base.Univariate | None = None):
...
def learn_one(self, x, y):
...
def score_one(self, x, y):
...
Import
from river import anomaly
I/O Contract
| Parameter | Type | Description |
|---|---|---|
| sub_stat | stats.Univariate (optional) | Central tendency statistic (defaults to stats.Mean()) |
| Method | Returns | Interpretation |
|---|---|---|
| score_one(x, y) | float | Absolute number of standard deviations from center |
Usage Examples
import random
from river import anomaly
from river import stats
rng = random.Random(42)
# Use with default mean
model = anomaly.StandardAbsoluteDeviation(sub_stat=stats.Mean())
# Learn normal behavior
for _ in range(150):
y = rng.gauss(0, 1)
model.learn_one(None, y)
# Score values
print(model.score_one(None, 2)) # 2.057 (about 2 std devs)
print(model.score_one(None, 0)) # 0.084 (close to mean)
print(model.score_one(None, 1)) # 0.986 (about 1 std dev)
# Detect anomalies with threshold
threshold = 3.0
for _ in range(100):
y = rng.gauss(0, 1)
score = model.score_one(None, y)
if score > threshold:
print(f"Anomaly detected: y={y:.2f}, score={score:.2f}")
model.learn_one(None, y)
# Use median for robustness
robust_model = anomaly.StandardAbsoluteDeviation(
sub_stat=stats.Median()
)
# Process data with outliers
data = [1, 2, 3, 100, 2, 3, 2, 3]
for y in data:
score = robust_model.score_one(None, y)
print(f"y={y}, score={score:.3f}")
robust_model.learn_one(None, y)