MMD#

class frouros.detectors.data_drift.streaming.distance_based.MMD(window_size: int, kernel: ~typing.Callable = <function rbf_kernel>, chunk_size: ~typing.Optional[int] = None, callbacks: ~typing.Optional[~typing.Union[~frouros.callbacks.streaming.base.BaseCallbackStreaming, ~typing.List[~frouros.callbacks.streaming.base.BaseCallbackStreaming]]] = 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: Optional[ndarray]#

Reference data property.

Returns:: reference data
Return type:: Optional[numpy.ndarray]

property callbacks: Optional[List[BaseCallback]]#

Callbacks property.

Returns:: callbacks
Return type:: Optional[List[BaseCallback]]

compare(X: ndarray) → Tuple[Optional[DistanceResult], 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) → Dict[str, Any]#

Fit detector.

Parameters:: X (numpy.ndarray) – feature data
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: Union[int, float]) → Tuple[Optional[BaseResult], 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]]

MMD

Contents

MMD#