HDDMWeighted#

class capymoa.drift.detectors.HDDMWeighted[source]#

Bases: MOADriftDetector

Weighted Hoeffding’s bounds Drift Detector

Online and non-parametric drift detection methods based on Hoeffding’s bounds [1].

>>> import numpy as np
>>> from capymoa.drift.detectors import HDDMWeighted
>>> np.random.seed(0)
>>>
>>> detector = HDDMWeighted(lambda_=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])
...     if detector.detected_change():
...         print("Change detected in data: " + str(data_stream[i]) + " - at index: " + str(i))
Change detected in data: 6 - at index: 1234
__init__(
drift_confidence: float = 0.001,
warning_confidence: float = 0.005,
lambda_: float = 0.05,
test_type: Literal['Two-sided', 'One-sided'] = 'Two-sided',
CLI: str | None = None,
)[source]#
Parameters:

  • drift_confidence – Significance level for drift detection (p-value threshold).

  • warning_confidence – Significance level for warning detection (p-value threshold).

  • lambda – Forgetting factor (decay rate) for the weighted mean. A smaller value gives more weight to recent observations.

  • test_type –

    The type of statistical hypothesis test to use.

    • "Two-sided" detects drift in both directions, i.e. both increases and decreases in the monitored weighted average. Use this when the direction of change is unknown.

    • "One-sided" only detects increases in the monitored weighted average (i.e. performance degradation). Use this when only positive drift is relevant.

  • CLI – (Advanced) Override the CLI arguments passed directly to the underlying MOA detector, bypassing all other parameters.

add_element(element: float) → None[source]#

Update the drift detector with a new input value.

Parameters:

element – A value to update the drift detector with. Usually, this is the prediction error of a model.

cli_help()[source]#
detected_change() → bool[source]#

Is the detector currently detecting a concept drift?

detected_warning() → bool[source]#

Is the detector currently warning of an upcoming concept drift?

get_params() → Dict[str, Any][source]#

Get the hyper-parameters of the drift detector.

reset(clean_history: bool = False) → None[source]#

Reset the drift detector. :param clean_history: Whether to reset detection history, defaults to False

TEST_TYPES = ['Two-sided', 'One-sided']#