Page-Hinkley Test Detector Class

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

Bases: Cusum

Page-Hinkley Test Change Point Detector. A class to detect change points in sequential data using the Page-Hinkley CUSUM algorithm.

Parameters:

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

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

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

Initializes the Page-Hinkley Test Detector with the specified parameters.

Parameters:

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

  • 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)

Initializes the Page-Hinkley Test Detector with the specified parameters.

Parameters:

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

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

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

Raises:

ValueError If warmup_period < 10. –

detection(observation: float)

Predicts the next data point and detects change points.

Parameters:

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

Returns:

  • PH_stat_pos (numpy.ndarray) – The positive Page-Hinkley statistic.

  • PH_stat_neg (numpy.ndarray) – The negative Page-Hinkley statistic.

  • 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: - ‘pos_changes’: numpy.ndarray of positive Page-Hinkley statistics. - ‘neg_changes’: numpy.ndarray of negative Page-Hinkley statistics. - ‘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 Page-Hinkley statistics.

Parameters:

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

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

  • pos_changes (list) – List of positive Page-Hinkley statistics.

  • neg_changes (list) – List of negative Page-Hinkley statistics.

Examples

Instance-based Detection

from source.detector.cusum import PHTest_Detector

ph_test_detector = PHTest_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 = ph_test_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 PHTest_Detector

ph_test_detector = PHTest_Detector(warmup_period=10, delta=10, threshold=20)
data = np.concatenate([np.random.normal(0, 1, 100),
                    np.random.normal(5, 1, 100)])
results = ph_test_detector.offline_detection(data)
ph_test_detector.plot_change_points(data,
                            results["change_points"],
                            results["pos_changes"],
                            results["neg_changes"])