`MMD`#

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

Bases: BaseDataDriftDetector

Maximum Mean Discrepancy

MMD is a multivariate, kernel-based distance between two distributions. A permutation test is used to obtain a p-value.

By default, the RBF kernel with sigma=1.0 is used.

Example:#

>>> import numpy as np
>>> from capymoa.drift.detectors import MMD
>>> rng = np.random.default_rng(42)
>>> detector = MMD(window_size=50, n_permutations=100, sigma=1.0)
>>> detector.fit(rng.normal(0, 1, size=(100, 2)))
>>> for x in rng.normal(3, 1, size=(50, 2)):
...     detector.add_element(x)
>>> detector.detected_change()
True

Reference:#

Gretton, Arthur, et al. “A kernel two-sample test.” The Journal of Machine Learning Research 13.1 (2012): 723-773.

__init__( window_size: int, alpha: float = 0.05, sigma: float = 1.0, kernel: Callable | None = None, n_permutations: int = 100, auto_fit_samples: int | None = None, )[source]#

Create an MMD data drift detector.

Parameters:

window_size – Number of observations in the sliding window.
alpha – Significance level for drift decisions.
sigma – Bandwidth for the default RBF kernel. Ignored when a custom kernel is provided.
kernel – Callable (X, Y) -> K that returns an (n, m) kernel matrix. If None, an RBF kernel with the given sigma is used.
n_permutations – Number of permutations for the significance test. Higher values give more accurate p-values at the cost of computation. Must be at least 1.
auto_fit_samples – Number of initial samples for auto-fit.

Raises:

ValueError – If n_permutations < 1 or window_size < 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.

MMD#

Example:#

Reference:#

`MMD`#