Drift API¶

Drift detectors consume one finite scalar with update. The drift_detected flag always refers to the most recent update.

BaseDriftDetector ¶

Bases: ABC

Abstract base class for scalar stream-change detectors.

update processes one finite scalar and returns the detector, while drift_detected describes the observation most recently processed. The monitored scalar can be an anomaly score, prediction error, residual, feature value, or another application-defined signal.

Subclasses implement update, drift_detected, and reset. A drift flag is evidence of change in the monitored signal; it does not diagnose the cause or prescribe a response for an anomaly model.

drift_detected `abstractmethod` `property` ¶

Python

drift_detected: bool

Return True if drift was detected on the last update.

Returns:

Type	Description
`bool`	True if drift was detected, False otherwise.

update `abstractmethod` ¶

Python

update(x: float) -> BaseDriftDetector

Update the detector with a single observation.

Parameters:

Name	Type	Description	Default
`x`	`float`	The observed value.	required

Returns:

Name	Type	Description
`self`	`BaseDriftDetector`	Returns self for method chaining.

reset `abstractmethod` ¶

Python

reset() -> None

Reset the detector to its initial state.

ADWIN ¶

Python

ADWIN(delta: float = 0.002, clock: int = 32, max_buckets: int = 5, min_window_length: int = 5, grace_period: int = 10)

Bases: BaseDriftDetector

ADWIN (ADaptive WINdowing) drift detector.

ADWIN maintains a variable-length window of recent data and detects concept drift by comparing the distributions of two subwindows. When drift is detected, it shrinks the window to remove old data.

The algorithm uses an exponential histogram (bucket structure) for memory-efficient storage and the Hoeffding bound for statistical significance testing.

Parameters:

Name	Type	Description	Default
`delta`	`float`	Significance level for drift detection. Lower values make the detector more conservative. Default is 0.002.	`0.002`
`clock`	`int`	How often to check for drift (every `clock` samples). Default is 32.	`32`
`max_buckets`	`int`	Maximum number of buckets per level. Default is 5.	`5`
`min_window_length`	`int`	Minimum subwindow size for comparison. Default is 5.	`5`
`grace_period`	`int`	Number of samples before drift detection starts. Default is 10.	`10`

Examples:

Python

from aberrant.drift import ADWIN

detector = ADWIN(delta=0.002)
drift_points = []
for index, value in enumerate([0.0] * 64 + [1.0] * 64):
    detector.update(value)
    if detector.drift_detected:
        drift_points.append(index)

References

Bifet, A., & Gavalda, R. (2007). Learning from time-changing data with adaptive windowing. In Proceedings of the 2007 SIAM International Conference on Data Mining (pp. 443-448). https://doi.org/10.1137/1.9781611972771.42 Reference implementation: https://github.com/Waikato/moa/blob/master/moa/src/main/java/moa/classifiers/core/driftdetection/ADWIN.java

drift_detected `property` ¶

Python

drift_detected: bool

Return True if drift was detected on the last update.

width `property` ¶

Python

width: int

Current window width (number of observations).

n_detections `property` ¶

Python

n_detections: int

Total number of drift detections.

estimation `property` ¶

Python

estimation: float

Current mean estimate of the window.

variance `property` ¶

Python

variance: float

Current variance estimate of the window.

reset ¶

Python

reset() -> None

Reset the detector to its initial state (clears all counters).

update ¶

Python

update(x: float) -> ADWIN

Update the detector with a new observation.

Parameters:

Name	Type	Description	Default
`x`	`float`	The observed value.	required

Returns:

Name	Type	Description
`self`	`ADWIN`	Returns self for method chaining.

KSWIN ¶

Python

KSWIN(alpha: float = 0.005, window_size: int = 100, stat_size: int = 30, seed: int | None = None)

Bases: BaseDriftDetector

KSWIN (Kolmogorov-Smirnov WINdowing) drift detector.

KSWIN detects concept drift by comparing recent observations with historical data using the Kolmogorov-Smirnov two-sample test. This is a distribution-free test that makes no assumptions about the underlying data distribution.

The detector maintains a sliding window and compares the most recent samples with a random sample from the earlier part of the window.

Parameters:

Name	Type	Description	Default
`alpha`	`float`	Significance level for the KS test. Lower values require stronger evidence to detect drift. Default is 0.005.	`0.005`
`window_size`	`int`	Size of the sliding window. Default is 100.	`100`
`stat_size`	`int`	Number of samples to use for comparison. Must be less than window_size / 2. Default is 30.	`30`
`seed`	`int \| None`	Random seed for reproducibility. Default is None.	`None`

Examples:

Python

from aberrant.drift import KSWIN

detector = KSWIN(alpha=0.005)
drift_points = []
for index, value in enumerate([0.0] * 100 + [1.0] * 100):
    detector.update(value)
    if detector.drift_detected:
        drift_points.append(index)

References

Raab, C., Heusinger, M., & Schleif, F. M. (2020). Reactive Soft Prototype Computing for Concept Drift Streams. Neurocomputing, 416, 340-351. https://doi.org/10.1016/j.neucom.2019.11.111

drift_detected `property` ¶

Python

drift_detected: bool

Return True if drift was detected on the last update.

n_detections `property` ¶

Python

n_detections: int

Total number of drift detections.

p_value `property` ¶

Python

p_value: float | None

P-value from the last KS test, or None if not yet computed.

statistic `property` ¶

Python

statistic: float | None

KS statistic from the last test, or None if not yet computed.

reset ¶

Python

reset() -> None

Reset the detector to its initial state.

update ¶

Python

update(x: float) -> KSWIN

Update the detector with a new observation.

Parameters:

Name	Type	Description	Default
`x`	`float`	The observed value.	required

Returns:

Name	Type	Description
`self`	`KSWIN`	Returns self for method chaining.

PageHinkley ¶

Python

PageHinkley(min_instances: int = 30, delta: float = 0.005, threshold: float = 50.0, alpha: float = 0.9999, mode: Literal['up', 'down', 'both'] = 'both')

Bases: BaseDriftDetector

Page-Hinkley drift detector.

The Page-Hinkley test is a sequential analysis technique for detecting changes in the mean of a distribution. It is based on the cumulative sum (CUSUM) control chart method.

The detector monitors the cumulative deviation from the running mean and triggers drift when this deviation exceeds a threshold.

Parameters:

Name	Type	Description	Default
`min_instances`	`int`	Minimum number of observations before detection starts. Default is 30.	`30`
`delta`	`float`	Magnitude of changes to tolerate. Smaller values make the detector more sensitive. Default is 0.005.	`0.005`
`threshold`	`float`	Detection threshold (lambda). When the test statistic exceeds this value, drift is detected. Default is 50.0.	`50.0`
`alpha`	`float`	Forgetting factor for the cumulative sums. Values closer to 1 give more weight to historical data. Default is 0.9999.	`0.9999`
`mode`	`Literal['up', 'down', 'both']`	Direction of change to detect: - "up": Detect increases in the mean - "down": Detect decreases in the mean - "both": Detect both increases and decreases (default)	`'both'`

Examples:

Python

from aberrant.drift import PageHinkley

detector = PageHinkley(threshold=50.0)
drift_points = []
for index, value in enumerate([0.0] * 64 + [2.0] * 64):
    detector.update(value)
    if detector.drift_detected:
        drift_points.append(index)

References

Page, E. S. (1954). Continuous inspection schemes. Biometrika, 41(1/2), 100-115. https://doi.org/10.1093/biomet/41.1-2.100

drift_detected `property` ¶

Python

drift_detected: bool

Return True if drift was detected on the last update.

n_detections `property` ¶

Python

n_detections: int

Total number of drift detections.

mean `property` ¶

Python

mean: float

Current running mean.

reset ¶

Python

reset() -> None

Reset the detector to its initial state (clears all counters).

update ¶

Python

update(x: float) -> PageHinkley

Update the detector with a new observation.

Parameters:

Name	Type	Description	Default
`x`	`float`	The observed value.	required

Returns:

Name	Type	Description
`self`	`PageHinkley`	Returns self for method chaining.

Drift API¶

BaseDriftDetector ¶

drift_detected abstractmethod property ¶

update abstractmethod ¶

reset abstractmethod ¶

ADWIN ¶

drift_detected property ¶

width property ¶

n_detections property ¶

estimation property ¶

variance property ¶

reset ¶

update ¶

KSWIN ¶

drift_detected property ¶

n_detections property ¶

p_value property ¶

statistic property ¶

reset ¶

update ¶

PageHinkley ¶

drift_detected property ¶

n_detections property ¶

mean property ¶

reset ¶

update ¶

drift_detected `abstractmethod` `property` ¶

update `abstractmethod` ¶

reset `abstractmethod` ¶

drift_detected `property` ¶

width `property` ¶

n_detections `property` ¶

estimation `property` ¶

variance `property` ¶

drift_detected `property` ¶

n_detections `property` ¶

p_value `property` ¶

statistic `property` ¶

drift_detected `property` ¶

n_detections `property` ¶

mean `property` ¶