IncrementalKSTest

IncrementalKSTest#

class frouros.detectors.data_drift.streaming.statistical_test.IncrementalKSTest(window_size: int, callbacks: BaseCallbackStreaming | list[BaseCallbackStreaming] | None = None)#

IncrementalKSTest (Incremental Kolmogorov-Smirnov test) [dos2016fast] detector.

Parameters:

  • window_size (int) – window size value

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

References:

[dos2016fast]

dos Reis, Denis Moreira, et al. “Fast unsupervised online drift detection using incremental kolmogorov-smirnov test.” Proceedings of the 22nd ACM SIGKDD international conference on knowledge discovery and data mining. 2016.

Example:

>>> from frouros.detectors.data_drift import IncrementalKSTest
>>> import numpy as np
>>> np.random.seed(seed=31)
>>> X = np.random.normal(loc=0, scale=1, size=100)
>>> Y = np.random.normal(loc=1, scale=1, size=100)
>>> detector = IncrementalKSTest(window_size=10)
>>> _ = detector.fit(X=X)
>>> for sample in Y:
...     test, _ = detector.update(value=sample)
...     if test is not None:
...         print(test.statistic, test.p_value)
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]]

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