Implementation:Online ml River Anomaly Base
| Knowledge Sources | |
|---|---|
| Domains | Online_Learning, Anomaly_Detection, Outlier_Detection |
| Last Updated | 2026-02-08 16:00 GMT |
Overview
Base classes defining interfaces for anomaly detection in River, including unsupervised, supervised, and filter-based detectors.
Description
The module defines three abstract base classes. AnomalyDetector is for unsupervised anomaly detection with learn_one(x) and score_one(x) methods. SupervisedAnomalyDetector extends this for supervised scenarios where both features x and target y are available. AnomalyFilter wraps an anomaly detector and adds a classify method to convert scores into binary decisions, with optional protection to prevent the detector from learning anomalies. This architecture enables flexible composition of anomaly detection components.
Usage
Subclass AnomalyDetector for unsupervised methods or SupervisedAnomalyDetector for supervised methods. Use AnomalyFilter when you need to convert anomaly scores into binary classifications or want to protect the detector from learning anomalous examples.
Code Reference
Source Location
- Repository: Online_ml_River
- File: river/anomaly/base.py
Signature
class AnomalyDetector(base.Estimator):
@abc.abstractmethod
def learn_one(self, x: dict) -> None:
...
@abc.abstractmethod
def score_one(self, x: dict) -> float:
...
class SupervisedAnomalyDetector(base.Estimator):
@abc.abstractmethod
def learn_one(self, x: dict, y: base.typing.Target) -> None:
...
@abc.abstractmethod
def score_one(self, x: dict, y: base.typing.Target) -> float:
...
class AnomalyFilter(base.Wrapper, base.Estimator):
def __init__(
self,
anomaly_detector: AnomalyDetector,
protect_anomaly_detector=True
):
...
@abc.abstractmethod
def classify(self, score: float) -> bool:
...
Import
from river import anomaly
I/O Contract
| Class | Key Methods | Returns |
|---|---|---|
| AnomalyDetector | score_one(x) | float (anomaly score) |
| SupervisedAnomalyDetector | score_one(x, y) | float (anomaly score) |
| AnomalyFilter | classify(score) | bool (is_anomaly) |
Usage Examples
from river import anomaly
from river import stats
# Implement custom unsupervised detector
class SimpleThresholdDetector(anomaly.base.AnomalyDetector):
def __init__(self, threshold=3.0):
self.mean = stats.Mean()
self.std = stats.Var()
self.threshold = threshold
def learn_one(self, x):
value = x['value']
self.mean.update(value)
self.std.update(value)
def score_one(self, x):
if self.std.get() == 0:
return 0.0
z_score = abs(
(x['value'] - self.mean.get()) /
(self.std.get() ** 0.5)
)
return z_score / self.threshold
# Use the detector
detector = SimpleThresholdDetector(threshold=3.0)
for value in [1, 2, 3, 100, 2, 3]:
x = {'value': value}
score = detector.score_one(x)
print(f"Value: {value}, Score: {score:.3f}")
detector.learn_one(x)