`Wasserstein`#

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

Bases: BaseDataDriftDetector

Wasserstein Distance (Earth Mover’s Distance)

The first Wasserstein distance (EMD) measures the minimum “work” needed to transform one distribution into another. Intuitively, if the distributions are piles of earth, it is the minimum cost of moving earth to reshape one pile into the other.

Drift is declared when the distance exceeds threshold. Applied independently to each feature using the configured threshold. Overall drift is reported when any feature flags drift. This detector does not return p-values, and the correction parameter has no effect on its drift decisions. No Bonferroni correction is applied.

Note

The distance is scale-dependent. Choose threshold based on the expected range of your features.

Example:#

>>> import numpy as np
>>> from capymoa.drift.detectors import Wasserstein
>>> rng = np.random.default_rng(42)
>>> detector = Wasserstein(window_size=50, threshold=0.5)
>>> 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:#

Rubner, Yossi, Carlo Tomasi, and Leonidas J. Guibas. “The earth mover’s distance as a metric for image retrieval.” International Journal of Computer Vision 40.2 (2000): 99-121.

__init__( window_size: int, threshold: float = 0.5, correction: Literal['bonferroni', 'none'] = 'bonferroni', auto_fit_samples: int | None = None, )[source]#

Create a Wasserstein data drift detector.

Parameters:

window_size – Number of observations in the sliding window.
threshold – Distance above which drift is declared (per feature).
correction – Accepted for interface consistency with other detectors, but has no effect: this detector does not produce p-values, so no multiple-testing correction is applied to its drift decisions.
auto_fit_samples – Number of initial samples for auto-fit.

Raises:

ValueError – If threshold <= 0.

_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 = True#: 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.

Wasserstein#

Example:#

Reference:#

`Wasserstein`#