Vanilla CUSUM Detector Class

class source.detector.cusum.CUSUM_Detector(warmup_period: int = 10, delta: int = 10, threshold: int = 20)

Bases: Cusum

CUSUM Change Point Detector.

This class implements a sequential CUSUM algorithm for detecting positive and negative change points in a data series.

Parameters:

  • warmup_period (int) – Number of initial observations before detecting change points.

  • delta (float) – Sensitivity parameter for detecting changes.

  • threshold (float) – Threshold for detecting a change point.

Initialize the CUSUM detector with the given parameters.

Parameters:

  • warmup_period (int) – Number of initial observations before detecting change points.

  • delta (float) – Sensitivity parameter for detecting changes.

  • threshold (float) – Threshold for detecting a change point.

Raises:

ValueError – If warmup_period < 10.

__init__(warmup_period: int = 10, delta: int = 10, threshold: int = 20)

Initialize the CUSUM detector with the given parameters.

Parameters:

  • warmup_period (int) – Number of initial observations before detecting change points.

  • delta (float) – Sensitivity parameter for detecting changes.

  • threshold (float) – Threshold for detecting a change point.

Raises:

ValueError – If warmup_period < 10.

detection(observation: float)

Process a single observation and detect change points.

Parameters:

observation (float) – New data point.

Returns:

  • S_pos (numpy.ndarray) – Positive cumulative sum values.

  • S_neg (numpy.ndarray) – Negative cumulative sum values.

  • is_changepoint (bool) – True if a change point is detected, False otherwise.

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: - ‘pos_changes’: numpy.ndarray of positive cumulative sums. - ‘neg_changes’: numpy.ndarray of negative cumulative sums. - ‘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, pos_changes: list, neg_changes: list)

Plots data with detected change points and cumulative sums.

Parameters:

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

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

  • pos_changes (list) – List of positive cumulative sums.

  • neg_changes (list) – List of negative cumulative sums.

Examples

Instance-based Detection

from source.detector.cusum import CUSUM_Detector

detector = CUSUM_Detector(warmup_period=10, delta=10, threshold=20)
data_stream = np.concatenate([np.random.normal(0, 1, 100),
                    np.random.normal(5, 1, 100)])
for data in data_stream:
    pos, neg, is_change = detector.detection(data)
    print(f"Change Detected: {is_change} \n -Positives: {pos[0]}, \n -Negatives: {neg[0]}")

Batch-based Detection

from source.detector.cusum import CUSUM_Detector

detector = CUSUM_Detector(warmup_period=10, delta=10, threshold=20)
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["pos_changes"],
                            results["neg_changes"])

Plotting

Vanilla CUSUM Example