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)

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

>>> # 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 (MTD), and false alarm rate.

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).

EvaluateDriftDetector#

This Page