HistoryConceptDrift

class frouros.callbacks.streaming.HistoryConceptDrift(name: str | None = None)

HistoryConceptDrift callback class that can be applied to concept_drift.streaming detectors.

Parameters:

name (Optional[str]) – name value, defaults to None. If None, the name will be set to HistoryConceptDrift.

Note:

By default the following variables are stored:

• value: list of values received by the detector

• drift: list of drift flags

• num_instances: list of number of instances received by the detector

Each detector may store additional variables if they are defined in an additional_vars dictionary in the detectors __init__ method. The user can add additional variables by calling the add_additional_vars() method.

Example:

>>> from frouros.callbacks import HistoryConceptDrift
>>> from frouros.detectors.concept_drift import ADWIN
>>> import numpy as np
>>> np.random.seed(seed=31)
>>> dist_a = np.random.normal(loc=0.2, scale=0.01, size=1000)
>>> dist_b = np.random.normal(loc=0.8, scale=0.04, size=1000)
>>> stream = np.concatenate((dist_a, dist_b))
>>> detector = ADWIN(callbacks=[HistoryConceptDrift(name="history")])
>>> for i, value in enumerate(stream):
...     callbacks_log = detector.update(value=value)
...     if detector.drift:
...         print(f"Change detected at step {i}")
...         break
Change detected at step 1055
>>> callbacks_log["history"]["drift"]
[False, False, ..., True]

add_additional_vars(vars_: list[str]) → None

Add additional variables to track.

Parameters:

vars (list[str]) – list of variables

on_update_end(value: int | float) → None

On update end method.

Parameters:

value (Union[int, float]) – value used to update the detector

reset() → None

Reset method.

property name: str

Name property.

Returns:

name value

Return type:

str

on_fit_end(X: ndarray) → None

On fit end method.

Parameters:

X (numpy.ndarray) – reference data

on_fit_start(X: ndarray) → None

On fit start method.

Parameters:

X (numpy.ndarray) – reference data

on_update_start(value: int | float) → None

On update start method.

Parameters:

value (Union[int, float]) – value used to update the detector

set_detector(detector) → None

Set detector method.