`D3`#

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

Bases: BaseDataDriftDetector

Discriminative Drift Detector (D3)

Detects drift by training a classifier to distinguish reference samples from test samples. If the classifier achieves a high ROC-AUC, the two distributions must differ, indicating drift.

This is a fundamentally different approach from statistical tests or distance measures: it frames drift detection as a binary classification problem.

Stratified k-fold cross-validation is used so that every sample receives a prediction. Drift is declared when the ROC-AUC exceeds threshold.

Example:#

>>> import numpy as np
>>> from capymoa.drift.detectors import D3
>>> rng = np.random.default_rng(42)
>>> detector = D3(window_size=50, threshold=0.7, seed=42)
>>> detector.fit(rng.normal(0, 1, size=(200, 2)))
>>> for x in rng.normal(3, 1, size=(50, 2)):
...     detector.add_element(x)
>>> detector.detected_change()
True

Reference:#

Gözüaçık, Ö., Büyükçakir, A., Bonab, H., and Can, F. “Unsupervised concept drift detection with a discriminative classifier.” Proceedings of the 28th ACM International Conference on Information and Knowledge Management (2019). ACM.

__init__( window_size: int, threshold: float = 0.7, n_splits: int = 2, seed: int | None = None, auto_fit_samples: int | None = None, )[source]#

Create a D3 data drift detector.

Parameters:

window_size – Number of observations in the sliding window.
threshold – ROC-AUC above which drift is declared. Values near 0.5 mean the classifier cannot distinguish the windows (no drift); values near 1.0 mean clear separation (drift).
n_splits – Number of cross-validation folds.
seed – Random seed for the classifier and CV splits.
auto_fit_samples – Number of initial samples for auto-fit.

Raises:

ValueError – If threshold not in (0.5, 1] or n_splits < 2.

_fit(X: ndarray) → None[source]#

Store or preprocess the reference data.

Called by fit() after validation. X is guaranteed to be a 2-D np.ndarray with shape (n_samples, n_features).

At minimum, implementations should set self._X_ref = X.

_test( X_ref: ndarray, X_test: ndarray, ) → DataDriftResult[source]#

Run the underlying statistical test or distance measure.

When IS_UNIVARIATE = True this receives one feature at a time: both arrays are 1-D with shapes (n_ref,) and (n_test,). The subclass should set statistic and p_value (if available) on the returned result.

If p_value is set, the base class applies the configured correction and ignores the per-feature is_drift flag. Example:

def _test(self, x_ref, x_test):
    stat, p = scipy.stats.ks_2samp(x_ref, x_test)
    return DataDriftResult(
        is_drift=False, statistic=stat, p_value=p
    )

If p_value is left None (distance- or score-based tests), the base class uses the per-feature is_drift flag directly and correction has no effect – the subclass is responsible for its own per-feature decision (e.g. comparing a distance to a threshold).

When IS_UNIVARIATE = False this receives all features: both arrays are 2-D with shapes (n_ref, n_features) and (n_test, n_features). The subclass is responsible for the full DataDriftResult including is_drift.

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

Add one observation and check for drift.

The observation is appended to a sliding window of size window_size. Once full, the detector compares the window against the reference on every call.

If the detector is not yet fitted and auto_fit_samples is set, those initial observations build the reference. No comparison happens until then. If it is not fitted and auto-fit is not enabled, this raises RuntimeError.

Parameters:

element – A single observation – a scalar for univariate data, or a 1-D array for multivariate data.

Raises:

RuntimeError – If the detector is not fitted and auto-fit mode is not enabled. Call fit() first, or construct with auto_fit_samples.
ValueError – If the observation has a different number of features than the reference.

compare( X_test: ndarray, ) → DataDriftResult[source]#

One-shot batch comparison against the reference.

Unlike add_element(), this does not update the internal window or detection history. It is useful for offline evaluation. auto_fit_samples does not apply here; call fit() first.

Parameters:

X_test – Test data with the same number of features as the reference.

Raises:

RuntimeError – If fit() has not been called.
ValueError – If X_test has a different number of features than the reference.

Returns:

Comparison result.

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?

fit( X: ndarray | Any, feature_names: Sequence[str] | None = None, ) → None[source]#

Set the reference distribution.

This detector has REQUIRES_FIT set to True. Call fit() before add_element() or compare(), unless auto_fit_samples was set so add_element() can collect the reference. Calling fit() again replaces the reference (a sliding reference window).

Parameters:

X – Reference data, shape (n_samples,) or (n_samples, n_features). May also be a pandas DataFrame, in which case column names are extracted automatically.
feature_names – Optional names for each feature. If provided, must have length equal to the number of features. Overrides column names extracted from a DataFrame. When set, per-feature dicts in DataDriftResult use these names as keys instead of integer indices.

Raises:

ValueError – If X is empty or feature_names length does not match the number of features.

classmethod from_params( schema: Any = None, params: dict[str, Any] | None = None, random_seed: int = 1, ) → Any[source]#: Construct an instance from parameters produced by get_params.

get_params() → dict[str, Any][source]#: Return the hyper-parameters of this detector.

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

Reset the detector state.

Parameters:: clean_history – If True, also clear the reference data and detection history. If False (default), only the sliding window and current result are cleared; the reference and detection indices are preserved.

IS_UNIVARIATE: bool = False#: If True the test is applied to each feature separately and results are combined. If False the test runs on the joint distribution.

REQUIRES_FIT: bool = True#: Data drift detectors need a reference distribution. Call fit() or set auto_fit_samples so add_element() can collect it.

property X_ref: ndarray | None#: The reference data set with fit.

property alpha: float#: Significance level for drift decisions.

property auto_fit_samples: int | None#: Number of samples for auto-fit, or None if explicit fit.

property correction: str#: Multiple-testing correction across features.

property feature_names: list[str] | None#: Feature names passed to fit(), or None.

property is_fitted: bool#

Whether the reference distribution has been set.

True after fit(), or after add_element() has collected auto_fit_samples observations.

property n_features: int | None#: Number of features in the reference data, or None before fit.

property result: DataDriftResult | None#: Most recent comparison result, or None during warm-up.

property window_size: int#: Size of the sliding test window.

D3#

Example:#

Reference:#

`D3`#