MMD

MMD#

class frouros.detectors.data_drift.streaming.distance_based.MMD(window_size: int, kernel: ~typing.Callable = <function rbf_kernel>, chunk_size: int | None = None, callbacks: ~frouros.callbacks.streaming.base.BaseCallbackStreaming | list[~frouros.callbacks.streaming.base.BaseCallbackStreaming] | None = None)#

MMD (Maximum Mean Discrepancy) [gretton2012kernel] detector.

Parameters:

  • window_size (int) – window size value

  • kernel (Callable) – kernel function, defaults to rbf_kernel()

  • chunk_size (Optional[int]) – chunk size value, defaults to None

  • callbacks (Optional[Union[BaseCallbackStreaming,) – callbacks, defaults to None

list[BaseCallbackStreaming]]]

References:

[gretton2012kernel]

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

Example:

>>> from functools import partial
>>> from frouros.detectors.data_drift import MMDStreaming
>>> from frouros.utils.kernels import rbf_kernel
>>> import numpy as np
>>> np.random.seed(seed=31)
>>> X = np.random.multivariate_normal(mean=[1, 1], cov=[[2, 0], [0, 2]], size=100)
>>> Y = np.random.multivariate_normal(mean=[0, 0], cov=[[2, 1], [1, 2]], size=100)
>>> detector = MMDStreaming(window_size=10, kernel=partial(rbf_kernel, sigma=0.5))
>>> _ = detector.fit(X=X)
>>> for sample in Y:
...     distance, _ = detector.update(value=sample)
...     if distance is not None:
...         print(distance)
property window_size: int#

Window size property.

Returns:

window size

Return type:

int

property X_ref: ndarray | None#

Reference data property.

Returns:

reference data

Return type:

Optional[numpy.ndarray]

property callbacks: list[BaseCallback] | None#

Callbacks property.

Returns:

callbacks

Return type:

Optional[list[BaseCallback]]

compare(X: ndarray) → Tuple[DistanceResult | None, dict[str, Any]]#

Compare detector.

Parameters:

X (np.ndarray) – data to use to compare the detector

Returns:

update result

Return type:

Tuple[Optional[DistanceResult], dict[str, Any]]

property data_type: BaseDataType#

Data type property.

Returns:

data type

Return type:

BaseDataType

fit(X: ndarray, **kwargs: Any) → dict[str, Any]#

Fit detector.

Parameters:

  • X (numpy.ndarray) – feature data

  • kwargs (Any) – additional fit parameters

Returns:

callbacks logs

Return type:

dict[str, Any]

reset() → None#

Reset method.

property statistical_type: BaseStatisticalType#

Statistical type property.

Returns:

statistical type

Return type:

BaseStatisticalType

update(value: int | float) → Tuple[BaseResult | None, dict[str, Any]]#

Update detector.

Parameters:

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

Returns:

update result and callbacks logs

Return type:

Tuple[Optional[BaseResult], dict[str, Any]]

Contents