The CUSUM repository contains multiple change point detectors for sequential analysis, enabling the detection of changes in the statistical properties of time-ordered or streaming data.
A change point is a point in time at which the statistical properties of a target variable—or its underlying data-generating process—undergo a significant shift. Detecting such changes is critical in domains such as finance, energy markets, healthcare, environmental monitoring, industrial processes, and online advertising. In these settings, predictive models and decision-making systems must adapt continuously to non-stationary and evolving conditions.
This project implements multiple variants of the CUSUM algorithm for change point detection, enabling robust identification of distributional shifts in sequential data.
CUSUM-based detectors are sequential algorithms designed to identify shifts in time-ordered or streaming data. They operate by incrementally processing observations—either instance by instance or in batches—and signaling a change when accumulated evidence indicates a statistically significant deviation from the expected behavior.
The implemented detectors support both:
-
Batch-based detection, where change points are identified over fixed-size windows or data batches
-
Instance-based detection, where each incoming observation is evaluated upon arrival
These detectors are, therefore, well suitable for both offline analysis and real-time monitoring in streaming environmnets.
Clone the repository:
git clone https://github.com/giobbu/CUSUM.git
cd CUSUMand install dependencies:
uv syncDocumentation is available at CUSUM Docs
View details on docs - Here
from source.generator.change_point_generator import ChangePointGenerator
from source.detector.cusum import ProbCUSUM_Detector
# data generator
generator = ChangePointGenerator(num_segments=2, segment_length=300, change_point_type='sudden_shift', seed=1)
generator.generate_data()
data_stream = generator.get_data()
# cusum detector
threshold_probability = 0.01
warmup_period = 150
detector = ProbCUSUM_Detector(warmup_period=warmup_period, threshold_probability=threshold_probability )
for data in data_stream:
prob, is_change = detector.detection(data)
print(f"Change Detected: {is_change} \n -Probability: {prob[0]}")
Performance Monitoring of an instance-based linear learning model applying the CUSUM algorithm.
At each time step:
- Generate a prediction with recursive least squares (RLS) model;
- Acquire the true observed value;
- Compute residual;
- Apply the CUSUM detector on the residuals to identify potential change points;
- Update the model parameters with the new data instance.
View details on notebook - Here
This project is under the GPL-3.0 license.