KS-CUM Detector Class

class source.detector.cusum.KS_CUM_Detector(window_pre: int = 30, window_post: int = 30, alpha: float = 0.05)

Bases: Cusum

A class to detect change points in sequential data using the Kolmogorov-Smirnov Test, loosley named Kolmogorov-Smirnov Cumulative Sum (KS-CUM) algorithm.

Parameters:

  • window_pre (int) – The size of the pre-change window. Must be greater than 30 and greater than or equal to window_post.

  • window_post (int) – The size of the post-change window. Must be greater than 30.

  • alpha (float) – The significance level for the KS test. Must be between 0 and 0.1

Initializes the KS-CUM Detector with the specified parameters.

Parameters:

  • window_pre (int) – The size of the pre-change window. Must be greater than 30 and greater than or equal to window_post.

  • window_post (int) – The size of the post-change window. Must be greater than 30.

  • alpha (float) – The significance level for the KS test. Must be between 0 and 0.1.

Raises:

ValueError – If window_pre < 30, window_post < 30, window_pre < window_post, or alpha is not between 0 and 0.1.

__init__(window_pre: int = 30, window_post: int = 30, alpha: float = 0.05)

Initializes the KS-CUM Detector with the specified parameters.

Parameters:

  • window_pre (int) – The size of the pre-change window. Must be greater than 30 and greater than or equal to window_post.

  • window_post (int) – The size of the post-change window. Must be greater than 30.

  • alpha (float) – The significance level for the KS test. Must be between 0 and 0.1.

Raises:

ValueError – If window_pre < 30, window_post < 30, window_pre < window_post, or alpha is not between 0 and 0.1.

detection(observation: float)

Predicts the next data point and detects change points.

Parameters:

observation (float) – The next data point to predict.

Returns:

  • ks_statistic (numpy.ndarray) – The KS statistic.

  • p_value (numpy.ndarray) – The p-value of the KS test.

  • is_changepoint (bool) – Indicates if a change point is detected.

offline_detection(data: ndarray)

Detects change points in the given data in an offline manner.

Parameters:

data (numpy.ndarray) – Data points to be analyzed.

Returns:

results – A dictionary containing: - ‘ks_statistics’: numpy.ndarray of KS statistics for each observation. - ‘p_values’: numpy.ndarray of p-values for each observation. - ‘is_drift’: list of booleans indicating detected change points. - ‘change_points’: numpy.ndarray of detected change point indices.

Return type:

dict

plot_change_points(data: ndarray, change_points: list, p_values: list)

Plots data with detected change points and KS statistics.

Parameters:

  • data (numpy.ndarray) – Original data points.

  • change_points (list) – List of detected change points.

  • p_values (list) – List of p-values for each data point.

Examples

Instance-based Detection

from source.detector.cusum import KS_CUM_Detector

detector = KS_CUM_Detector(window_pre=30, window_post=30, alpha=0.05)
data_stream = np.concatenate([np.random.normal(0, 1, 100),
                    np.random.normal(5, 1, 100)])
for data in data_stream:
    p_value, is_change = detector.detection(data)
    print(f"Change Detected: {is_change} \n -P-Value: {p_value[0]}")

Batch-based Detection

from source.detector.cusum import KS_CUM_Detector

detector = KS_CUM_Detector(window_pre=30, window_post=30, alpha=0.05)
data = np.concatenate([np.random.normal(0, 1, 100),
                    np.random.normal(5, 1, 100)])
results = detector.offline_detection(data)
detector.plot_change_points(data,
                            results["change_points"],
                            results["p_values"])

Plotting

KS-CUM Example