Chart CUSUM Detector

class source.detector.cusum.ChartCUSUM_Detector(warmup_period: int = 10, level: int = 3, deviation_type: str = 'sqr-dev', target_mean: float = None)

Bases: Cusum

Change Point Detector using CUSUM Control Chart. A class to detect change points in sequential data using the CUSUM Control Chart algorithm.

Parameters:

  • warmup_period (int) – The warmup period for the detector. Must be equal or greater than 10.

  • level (int) – The control limit level. Must be between 1 and 3.

  • deviation_type (str) – The type of deviation to use. Must be ‘sqr-dev’ or ‘dev’.

  • target_mean (float, optional) – The target mean for the CUSUM chart. If None, the mean of the warmup period is used.

Initializes the Change Point Detector with the specified parameters.

Parameters:

  • warmup_period (int) – The warmup period for the detector. Must be equal or greater than 10.

  • level (int) – The control limit level. Must be between 1 and 3.

  • deviation_type (str) – The type of deviation to use. Must be ‘sqr-dev’ or ‘dev’.

  • target_mean (float, optional) – The target mean for the CUSUM chart. If None, the mean of the warmup period is used.

__init__(warmup_period: int = 10, level: int = 3, deviation_type: str = 'sqr-dev', target_mean: float = None)

Initializes the Change Point Detector with the specified parameters.

Parameters:

  • warmup_period (int) – The warmup period for the detector. Must be equal or greater than 10.

  • level (int) – The control limit level. Must be between 1 and 3.

  • deviation_type (str) – The type of deviation to use. Must be ‘sqr-dev’ or ‘dev’.

  • target_mean (float, optional) – The target mean for the CUSUM chart. If None, the mean of the warmup period is used.

detection(observation: float)

Predicts the next data point and detects change points.

Parameters:

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

Returns:

  • upper (float) – The upper control limit.

  • lower (float) – The lower control limit.

  • cusum (float) – The cumulative sum of deviations.

  • 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: - ‘upper_limits’: numpy.ndarray of upper limits for each observation. - ‘lower_limits’: numpy.ndarray of lower limits for each observation. - ‘cusums’: numpy.ndarray of cumulative sums 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, cusums: list, upper_limits: list, lower_limits: list)

Plots data with detected change points and CUSUM statistics.

Parameters:

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

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

  • cusums (list) – List of cumulative sums for each data point.

  • upper_limits (list) – List of upper control limits for each data point.

  • lower_limits (list) – List of lower control limits for each data point.

Examples

Instance-based Detection

from source.detector.cusum import ChartCUSUM_Detector

detector = ChartCUSUM_Detector(warmup_period=20, level=3, deviation_type='sqr-dev')
data_stream = np.concatenate([np.random.normal(0, 1, 100),
                    np.random.normal(5, 1, 100)])
for data in data_stream:
    cusum, upper_limit, lower_limit, is_change = detector.detection(data)
    print(f"Change Detected: {is_change} \n -CUSUM: {cusum[0]}, \n -Upper Limit: {upper_limit[0]}, \n -Lower Limit: {lower_limit[0]}")

Batch-based Detection

from source.detector.cusum import ChartCUSUM_Detector

detector = ChartCUSUM_Detector(warmup_period=20, level=3, deviation_type='sqr-dev')
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["cusums"],
                            results["upper_limits"],
                            results["lower_limits"])

Plotting

Chart CUSUM Example