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.

References:

[gretton2012kernel]

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

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

Return type:

Optional[BaseResult]