EvaluateDriftDetector#

class capymoa.drift.eval_detector.EvaluateDriftDetector[source]#

Bases: object

A class to evaluate the performance of concept drift detectors.

This class provides functionality to assess drift detection algorithms by comparing their predictions against ground truth drift points. Each drift point is represented as a tuple (start_location, end_location) to handle both abrupt and gradual drifts.

Max delay is the maximum number of instances to wait for a detection after a drift occurs. For gradual drifts, this window starts from the drift end_location. If a detector fails to signal within this window, it is considered to have missed the drift (false negative). The max delay can be thought of as the time after which a drift becomes apparent even without a detector.

Key Features:

Handles both abrupt and gradual drifts:
- Abrupt drifts: start_location = end_location, e.g., (100, 100)
- Gradual drifts: start_location < end_location, e.g., (100, 150)
Considers maximum acceptable detection delay
Calculates comprehensive performance metrics (precision, recall, F1)
Reports detection delay both in instances (DriftDetectionMetrics.mdt) and relative to max_delay (DriftDetectionMetrics.ndt)

Examples:

>>> import numpy as np
>>>
>>> from capymoa.drift.eval_detector import EvaluateDriftDetector
>>>
>>> preds = np.array([500, 1200, 1250, 2100])
>>> trues = np.array([1000, 2000])
>>>
>>> eval = EvaluateDriftDetector(max_delay=200)
>>> metrics = eval.calc_performance(preds=preds, trues=trues, tot_n_instances=200)
>>> print(metrics.f1)
0.6666666666666666

>>> # ``mdt`` counts instances, ``ndt`` states the same delay as a fraction of
>>> # ``max_delay``: here the detector used three quarters of its allowance.
>>> print(metrics.mdt, metrics.ndt)
150.0 0.75

>>> # Only ``ndt`` responds to a different tolerance, which is what makes it
>>> # comparable across streams evaluated with different ``max_delay``.
>>> lenient = EvaluateDriftDetector(max_delay=400)
>>> relaxed = lenient.calc_performance(preds=preds, trues=trues, tot_n_instances=200)
>>> print(relaxed.mdt, relaxed.ndt)
150.0 0.375

>>> # Example with actual detector
>>> import numpy as np
>>> from capymoa.drift.detectors import ADWIN
>>> from capymoa.drift.eval_detector import EvaluateDriftDetector
>>>
>>> detector = ADWIN(delta=0.001)
>>>
>>> data_stream = np.random.randint(2, size=2000)
>>> for i in range(999, 2000):
...     data_stream[i] = np.random.randint(4, high=8)
>>>
>>> for i in range(2000):
...     detector.add_element(data_stream[i])
>>>
>>> trues = np.array([1000])
>>> preds = detector.detection_index
>>> evaluator = EvaluateDriftDetector(max_delay=50)
>>> metrics = evaluator.calc_performance(trues=trues, preds=preds, tot_n_instances=2000)
>>> print(metrics.f1)
1.0

__init__( max_delay: int, rate_period: int = 1000, max_early_detection: int = 0, )[source]#

Initialize the drift detector evaluator.

Parameters:

max_delay (int) – Maximum number of instances to wait for a detection after a drift occurs. For gradual drifts, this window starts from the drift end_location. If a detector fails to signal within this window, it is considered to have missed the drift (false negative).
max_early_detection (int) – Maximum number of instances before a drift starts where a detection is still considered valid. For detections that occur earlier than this window, they are considered false positives.
rate_period (int) – The number of instances in a relevant time period. This number is used to calculate the false alarm rate and alarm rate. Default is 1000.

Raises:

ValueError – If max_delay is not a positive integer.

Note

The max_delay parameter is important for evaluating both the accuracy and timing of drift detection.
For gradual drifts (where start_location != end_location), the detection window extends from (start_location - max_delay) to (end_location + max_delay).
For abrupt drifts (where start_location == end_location), the detection window is (drift_point - max_delay) to (drift_point + max_delay).

Calculate performance metrics for drift detection.

Evaluates drift detection performance by comparing predicted drift points against true drift points, considering a maximum allowable delay. Calculates various metrics including precision, recall, F1-score, mean time to detect (MDT), normalized detection time (NDT), and false alarm rate.

Detection delay is defined only over the drifts that were actually detected, so both DriftDetectionMetrics.mdt and DriftDetectionMetrics.ndt are nan when the detector missed every drift.

This method supports two modes of operation:

Pass trues, preds, and tot_n_instances to evaluate from raw data.
Pass drift_episodes and tot_n_instances to evaluate from pre-computed episodes.

Parameters:

trues – Array-like of true drift points represented as (start, end) tuples indicating drift intervals. Required if drift_episodes is not provided.
preds – Array-like of predicted drift points (indices) where the detector signaled a drift. Required if drift_episodes is not provided.
tot_n_instances – Total number of instances in the data stream, used to calculate alarm rate and false alarm rate. Always required.
drift_episodes – Optional list of drift episodes. If provided, the trues and preds parameters will be ignored with a warning.

Returns:

DriftDetectionMetrics object containing the calculated metrics

Raises:

ValueError – If arrays are not ordered or contain invalid values, if tot_n_instances is not positive, or if neither drift_episodes nor (trues and preds) are provided.
AssertionError – If no drift points are given.

max_delay: int#: Maximum allowable delay for drift detection.

max_early_detection: int#: Maximum allowable earliness for drift detection.

metrics: DriftDetectionMetrics | None#: Latest calculated performance metrics.

rate_period: int#: The period used for calculating rates (e.g., per 1000 instances).