03. PySATL CPD API: Change Point Detection

Section 1. Orientation

The data layer answers the question “what series are we working with?” The online core API answers the next question: “what happens when observations arrive one by one and we want a detector to react in streaming time?” This chapter is about that runtime story.

The goal here is not yet benchmarking or plotting. Instead, we want to understand the execution objects themselves: algorithms, detectors, and traces. Once those pieces are clear, the later notebooks on visualization and benchmarking become much easier to read because they stop feeling like magic wrappers around opaque runtime behavior.

What in this notebook

By the end of the notebook, you should understand three layers of the online execution stack.

  • The algorithm layer: objects that process one observation at a time and maintain internal state.

  • The detector layer: objects that decide how thresholds, resets, skip periods, and run-length logic are applied around the algorithm.

  • The trace layer: objects that store what happened during a complete run so the result can later be visualized or benchmarked.

We will use one small synthetic univariate scenario throughout the chapter so each execution detail has a concrete data source behind it.

from dataclasses import dataclass

import numpy as np
import pandas as pd
from IPython.display import display

from pysatl_cpd.algorithms.online import ShewhartControlChart
from pysatl_cpd.core.online import (
    BatchingOnlineAlgorithmWrapper,
    BatchReducer,
    OnlineAlgorithm,
    OnlineAlgorithmConfiguration,
    OnlineAlgorithmState,
    OnlineResetDetector,
    SkippingCondition,
    SkippingOnlineAlgorithmWrapper,
)
from pysatl_cpd.data.generator import preset_dataset

Core Concepts and Notation

Before building the running example, it helps to name the runtime objects precisely. The online core uses a small vocabulary that appears repeatedly in this notebook and in the benchmarking chapters.

  • An OnlineAlgorithm is the stateful numerical core. It receives one observation and emits one scalar detection-function value.

  • A detection function is the sequence of scalar statistics produced over time. Larger values usually mean stronger evidence for a change, but the algorithm does not declare alarms by itself.

  • A detector wraps an algorithm with runtime policy: thresholding, reset behavior, skip periods, forced run-length limits, state collection, and trace construction.

  • A detector description is the stable metadata form of that configured detector. It is used in traces, registries, benchmark tables, and plot labels.

  • A single run is one detector trace paired with the labeled provider that produced it.

  • An OnlineDetectionTrace is the durable record of one run: detection-function values, processing times, declared change points, learning periods, skip periods, and optional algorithm states.

The rest of the notebook works through these concepts in that order: first the algorithm, then the detector, then the trace, then optional wrappers that change algorithm cadence.

Section 2. Online Detection. The Running Example

Before discussing abstractions, we need one concrete online-detection problem that remains small enough to inspect line by line. A short univariate mean shift is ideal because the statistical behavior is easy to understand and the trace remains compact.

This section builds that source series and converts it into a labeled provider. The rest of the chapter will reuse the same provider so we can focus on execution behavior rather than on changing datasets.

2.1 A compact mean-shift scenario

The online API does not care where its provider came from, but the tutorial does. We want a signal with one obvious transition so we can interpret detector output without guessing where the ground truth lives.

The generator API gives us that source through a preset dataset. We ask for one univariate mean_shifts series, then reuse the resulting labeled provider throughout the notebook. The provider gives us both the observations and the known change points needed to interpret the trace.

online_demo_dataset = preset_dataset(
    "mean_shifts",
    n_series=1,
    seed=7,
    series_length=180,
    n_features=1,
)
online_demo_provider = online_demo_dataset[0]
online_demo_values = np.array(list(online_demo_provider), dtype=float)

print("Provider type:", type(online_demo_provider).__name__)
print("Provider length:", len(online_demo_provider))
print("Ground-truth change points:", list(online_demo_provider.change_points))
print("First 12 observations:", np.round(online_demo_values[:12], 3).tolist())
Provider type: PandasLabeledData
Provider length: 180
Ground-truth change points: [60, 120]
First 12 observations: [0.001, 0.299, -0.274, -0.891, -0.455, -0.992, 0.06, 1.34, -0.492, -0.62, 0.49, 0.357]

2.2 Why one provider is enough for this chapter

A large benchmark needs many providers, but an execution tutorial benefits from one stable example. The problem we are solving in this notebook is conceptual understanding of runtime objects, not statistical coverage across scenarios.

By reusing one provider throughout the chapter, we can compare reset and no-reset behavior, inspect traces, and experiment with wrappers while keeping the data source fixed. That makes differences in output easier to attribute to the execution model rather than to the data.

print("Provider annotation:", online_demo_provider.annotation)
print("Provider states:", [dict(state) for state in online_demo_provider.states])
print("Provider transitions:", [repr(transition) for transition in online_demo_provider.transitions])
Provider annotation: TimeseriesAnnotation(name='mean_shifts_series_0000', source=None, metadata=frozendict({'preset': 'mean_shifts'}))
Provider states: [{'type': 'baseline'}, {'type': 'alternative_1'}, {'type': 'alternative_2'}]
Provider transitions: ["type='alternative_1'->type='alternative_2'", "type='baseline'->type='alternative_1'"]

Summary of Section 2: We now have one compact labeled provider with known change points. That provider will remain the fixed input for all algorithm, detector, and trace examples in the rest of the notebook.

Section 3. Online Algorithms

The first execution problem is algorithmic rather than detector-specific: how does one observation at a time get turned into a detection statistic? The online algorithm abstraction solves exactly that problem.

An online algorithm is not yet a complete detector. It has no threshold policy, no reset policy, and no benchmark wrapper. It only owns state and knows how to process observations sequentially.

The value returned by process(...) is the detection-function value for the current step. A complete run stores those values as a time-indexed detection function. Algorithms are free to define the statistic differently, but downstream detector code only assumes that it is a scalar sequence that can be thresholded or plotted.

3.1 OnlineAlgorithm as the stateful runtime core

The motivation for a dedicated algorithm abstraction is separation of concerns. The statistical logic of updating means, variances, windows, or scores should not be mixed together with detector policies such as thresholding or reset behavior.

The exported OnlineAlgorithm interface captures that division. A conforming algorithm exposes a configuration, a state snapshot, a process(...) method for one observation, a reset() method, and a recreate() method for safe independent copies.

print("OnlineAlgorithm abstract type:", OnlineAlgorithm.__name__)
OnlineAlgorithm abstract type: OnlineAlgorithm

3.2 A primitive custom algorithm

The quickest way to understand the OnlineAlgorithm contract is to implement a tiny algorithm directly. The example below learns a baseline mean for a fixed number of initial observations, then emits the absolute distance between each new observation and that learned baseline.

This is intentionally not a production detector. Its value is pedagogical: it shows that an algorithm owns configuration, exposes immutable state snapshots, supports reset and recreation, and returns one scalar statistic per processed observation.

@dataclass(frozen=True, kw_only=True)
class BaselineDistanceConfiguration(OnlineAlgorithmConfiguration):
    learning_period_size: int = 10


@dataclass(frozen=True, kw_only=True)
class BaselineDistanceState(OnlineAlgorithmState):
    step: int = 0
    baseline_values: tuple[float, ...] = ()
    baseline_mean: float = 0.0


class BaselineDistanceAlgorithm(OnlineAlgorithm[float, BaselineDistanceConfiguration, BaselineDistanceState]):
    def __init__(self, learning_period_size: int = 10) -> None:
        self._configuration = BaselineDistanceConfiguration(
            learning_period_size=learning_period_size,
        )
        self._state = BaselineDistanceState(is_in_learning_period=True)

    @property
    def configuration(self) -> BaselineDistanceConfiguration:
        return self._configuration

    @property
    def state(self) -> BaselineDistanceState:
        return self._state

    def process(self, observation: float) -> float:
        step = self._state.step + 1
        baseline_values = self._state.baseline_values
        if step <= self.configuration.learning_period_size:
            baseline_values = (*baseline_values, float(observation))
            baseline_mean = float(np.mean(baseline_values))
            self._state = BaselineDistanceState(
                is_in_learning_period=True,
                step=step,
                baseline_values=baseline_values,
                baseline_mean=baseline_mean,
            )
            return 0.0

        statistic = abs(float(observation) - self._state.baseline_mean)
        self._state = BaselineDistanceState(
            is_in_learning_period=False,
            step=step,
            baseline_values=baseline_values,
            baseline_mean=self._state.baseline_mean,
        )
        return statistic

    def reset(self) -> None:
        self._state = BaselineDistanceState(is_in_learning_period=True)

    def recreate(self) -> "BaselineDistanceAlgorithm":
        return type(self)(learning_period_size=self.configuration.learning_period_size)


primitive_algorithm = BaselineDistanceAlgorithm(learning_period_size=8)
primitive_values = [primitive_algorithm.process(value) for value in online_demo_values[:16]]
display(
    pd.DataFrame(
        {
            "step": np.arange(len(primitive_values)),
            "observation": online_demo_values[:16],
            "detection_function": primitive_values,
        }
    ).round(3)
)
print("Primitive algorithm state:", primitive_algorithm.state)
step observation detection_function
0 0 0.001 0.000
1 1 0.299 0.000
2 2 -0.274 0.000
3 3 -0.891 0.000
4 4 -0.455 0.000
5 5 -0.992 0.000
6 6 0.060 0.000
7 7 1.340 0.000
8 8 -0.492 0.378
9 9 -0.620 0.507
10 10 0.490 0.604
11 11 0.357 0.471
12 12 0.105 0.219
13 13 -0.930 0.817
14 14 -0.029 0.085
15 15 0.695 0.809
Primitive algorithm state: BaselineDistanceState(is_in_learning_period=False, step=16, baseline_values=(0.0012301533574825742, 0.2987455375084699, -0.2741378553622176, -0.8905918387572742, -0.45467078517172255, -0.9916465549964624, 0.060143602597438485, 1.3402152455545335), baseline_mean=-0.11383906190871902)

3.3 ShewhartControlChart as the running algorithm

A tutorial needs one concrete algorithm that is simple enough to inspect and common enough to be useful later. ShewhartControlChart is a good fit because it emits a scalar detection statistic based on a sliding-window mean relative to a running baseline estimate.

The public API for the algorithm is compact: you choose a learning period and a window size, then call process(...) one observation at a time. The configuration and state remain accessible throughout the run.

algorithm = ShewhartControlChart(learning_period_size=30, window_size=10)
print("Algorithm description:", algorithm.description)
print("Algorithm configuration:", algorithm.configuration)
print("Initial state:", algorithm.state)
Algorithm description: OnlineAlgorithmDescription(name='ShewhartControlChart', configuration=w = 10)
Algorithm configuration: w = 10
Initial state: ShewhartControlChartState(is_in_learning_period=True, mean=0.0, variance=0.0, samples_count=0, window_contents=[])

3.4 Processing observations directly

Before adding detector behavior, it is worth solving the simplest runtime question directly: what values does the algorithm emit as it sees the series? This is the clearest way to understand the difference between an algorithm and a full detector.

The process(...) API takes one observation and returns one detection-function value. During the learning period, many algorithms emit zeros or otherwise special values because they are still estimating baseline behavior.

manual_algorithm = ShewhartControlChart(learning_period_size=5, window_size=5)
manual_detection_values = [manual_algorithm.process(value) for value in online_demo_provider]
manual_table = pd.DataFrame(
    {
        "step": np.arange(len(manual_detection_values)),
        "observation": online_demo_values,
        "detection_function": manual_detection_values,
    }
)
display(manual_table.head(12))
print("Peak detection statistic:", float(np.max(manual_detection_values)))
step observation detection_function
0 0 0.001230 0.000000
1 1 0.298746 0.000000
2 2 -0.274138 0.000000
3 3 -0.890592 0.000000
4 4 -0.454671 0.000000
5 5 -0.991647 0.000000
6 6 0.060144 0.377476
7 7 1.340215 0.933986
8 8 -0.492207 0.237011
9 9 -0.620475 0.162414
10 10 0.489842 0.213236
11 11 0.356887 1.019679
Peak detection statistic: 5.166162835553171

Summary of Section 3: An online algorithm is the stateful numerical core of online detection. It transforms one observation at a time into a detection-function value, but it does not yet decide when a change point should be declared or what should happen afterward.

Section 4. Wrapping all in detectors

Once an algorithm exists, the next problem is policy: when does a high statistic become a declared detection, and what happens after that declaration? Reset detectors solve one specific answer to that problem. After a change point is declared, the algorithm is reset and starts learning again.

This reset semantics is common when the goal is to detect one event, then restart the monitoring logic from the new regime. The detector wraps the algorithm and adds thresholding, optional skip periods, run-length control, and trace construction.

4.1 OnlineResetDetector

The motivation for OnlineResetDetector is to package the whole reset-style runtime contract around an online algorithm. You should not have to re-implement threshold logic, skip-period handling, and trace packaging every time you want to run one algorithm on one provider.

Its public constructor accepts the algorithm plus runtime options such as threshold, skip_period, max_runlength, and collect_states. The detector’s detect(...) method then returns a full OnlineDetectionTrace.

reset_detector = OnlineResetDetector(
    ShewhartControlChart(learning_period_size=30, window_size=10),
    threshold=2.0,
    skip_period=5,
    collect_states=True,
)
print("Reset detector description:", reset_detector.description)
Reset detector description: ChangePointDetectorDescription(name='online_reset_detector', parameters=frozendict({'algorithm': OnlineAlgorithmDescription(name='ShewhartControlChart', configuration=w = 10), 'threshold': 2.0, 'skip_period': 5, 'max_runlength': None, 'collect_states': True, 'data_transformer': None}))

4.2 Threshold, skip period, and run-length controls

A reset detector does more than simply compare a statistic to a threshold. The runtime problem is richer than that. After a declared change point, we may want to suppress immediate repeated detections for a few steps, and in some workflows we may want to force a reset if a run lasts too long.

The API exposes those concerns explicitly. threshold controls signal-based detections, skip_period suppresses detection decisions for a fixed number of later steps, and max_runlength can force a declaration if a run grows too long without a signal.

print("Threshold:", reset_detector.threshold)
print("Skip period:", reset_detector.skip_period)
print("Max runlength:", reset_detector.max_runlength)
print("Collect states:", reset_detector.collect_states)
Threshold: 2.0
Skip period: 5
Max runlength: None
Collect states: True

4.3 Running one reset detector end to end

The main practical question is what the detector produces when all those policies are active together. We want to see the declared change points, the signal-driven change points, the skip periods, and the size of the resulting per-step arrays.

The detect(...) call solves that full execution problem. It consumes the provider, runs the algorithm under the detector policy, and packages the whole run into one trace object.

reset_trace = reset_detector.detect(online_demo_provider)

print("Detected change points:", reset_trace.detected_change_points)
print("Signal change points:", reset_trace.signal_change_points)
print("Forced change points:", reset_trace.forced_change_points)
print("Skip periods:", reset_trace.skip_periods)
print("Detection-function length:", len(reset_trace.detection_function))
print("Processing-time length:", len(reset_trace.processing_time))
Detected change points: [56, 122]
Signal change points: [56, 122]
Forced change points: []
Skip periods: [(57, 61), (123, 127)]
Detection-function length: 180
Processing-time length: 180

Summary of Section 4: OnlineResetDetector turns an online algorithm into a full reset-style detector with explicit runtime policy. Thresholding, skip periods, and optional run-length forcing are detector concerns, not algorithm concerns, and the result of a complete run is an OnlineDetectionTrace.

Section 5. OnlineDetectionTrace

So far we have treated traces as outputs, but not yet as first-class objects worthy of study on their own. The next problem is understanding what a trace stores and why later notebooks rely on it so heavily.

A trace is the execution record of one detector run. It stores detection-function values, processing times, optional state snapshots, declared change points, and interval-style metadata such as skip periods or learning periods.

5.1 Trace anatomy

The motivation for studying trace anatomy is practical. Visualization and benchmarking operate on traces, not directly on detector internals. If the trace fields are familiar, later notebooks become much easier to navigate.

The most important fields are detection_function, processing_time, detected_change_points, signal_change_points, forced_change_points, skip_periods, and algorithm_states. Together they describe both numerical behavior and detector decisions.

trace_field_summary = {
    "threshold": reset_trace.threshold,
    "detected_change_points": reset_trace.detected_change_points,
    "signal_change_points": reset_trace.signal_change_points,
    "forced_change_points": reset_trace.forced_change_points,
    "skip_periods": reset_trace.skip_periods,
    "algorithm_states_count": len(reset_trace.algorithm_states),
}
print(trace_field_summary)
{'threshold': 2.0, 'detected_change_points': [56, 122], 'signal_change_points': [56, 122], 'forced_change_points': [], 'skip_periods': [(57, 61), (123, 127)], 'algorithm_states_count': 180}

5.2 Programmatic inspection of stepwise results

A trace is easiest to understand when turned back into a step table. That solves the same interpretation problem that many debugging sessions run into: “what exactly was happening near the true transition or near a declared detection?”

The code below builds a compact inspection table from the trace arrays and highlights the first region around the true shift. This is often enough to explain why a detector fired early, late, or repeatedly.

inspection_window = pd.DataFrame(
    {
        "step": np.arange(len(reset_trace.detection_function)),
        "observation": online_demo_values,
        "detection_function": reset_trace.detection_function,
        "processing_time": reset_trace.processing_time,
        "is_detected_change_point": [
            step in set(reset_trace.detected_change_points) for step in range(len(reset_trace.detection_function))
        ],
    }
)
display(inspection_window.iloc[70:95].reset_index(drop=True))
step observation detection_function processing_time is_detected_change_point
0 70 2.646903 0.000000 0.000001 False
1 71 0.007580 0.000000 0.000001 False
2 72 1.536830 0.000000 0.000002 False
3 73 1.902713 0.000000 0.000002 False
4 74 3.257015 0.000000 0.000001 False
5 75 2.689404 0.000000 0.000001 False
6 76 1.672787 0.000000 0.000001 False
7 77 1.631424 0.000000 0.000001 False
8 78 1.749805 0.000000 0.000001 False
9 79 3.523529 0.000000 0.000001 False
10 80 1.571975 0.000000 0.000001 False
11 81 1.696320 0.000000 0.000001 False
12 82 2.352589 0.000000 0.000001 False
13 83 1.879230 0.000000 0.000001 False
14 84 1.802716 0.000000 0.000001 False
15 85 0.885933 0.000000 0.000001 False
16 86 1.988479 0.000000 0.000001 False
17 87 1.556419 0.000000 0.000001 False
18 88 3.166128 0.000000 0.000001 False
19 89 2.653089 0.000000 0.000001 False
20 90 1.975856 0.000000 0.000003 False
21 91 2.668381 0.000000 0.000001 False
22 92 1.660130 0.505068 0.000002 False
23 93 3.052126 0.276813 0.000002 False
24 94 1.994600 0.597611 0.000002 False

5.3 Cutting traces for local analysis

The final trace-level problem is locality. Full traces are useful for complete runs, but debugging often focuses on one region around one event. The trace object therefore supports cut(...), just as providers do.

The cut API re-bases all relative indices inside the trace. That means detected change points and skip periods are shifted to the local coordinate system of the window, which makes local analysis much easier than manual index arithmetic.

local_trace = reset_trace.cut(70, 100)
print("Local trace length:", len(local_trace.detection_function))
print("Local detected change points:", local_trace.detected_change_points)
print("Local skip periods:", local_trace.skip_periods)
print("Local peak statistic:", float(np.max(local_trace.detection_function)))
Local trace length: 31
Local detected change points: []
Local skip periods: []
Local peak statistic: 1.2953987438650323

Summary of Section 5: OnlineDetectionTrace is the common record of online execution. It keeps numerical signals, timing data, event locations, and optional state snapshots together in one object, and its local cut operation makes detailed inspection tractable.

Section 6. Optional Runtime Wrappers

The final runtime problem in this chapter is execution cadence. Sometimes you want an algorithm to process every observation, but sometimes you want to skip observations or batch them before calling the underlying algorithm. Those concerns should not require rewriting the algorithm itself.

The wrapper classes solve this by adapting one online algorithm into another while preserving the same runtime interface.

6.1 SkippingOnlineAlgorithmWrapper

The motivation for skipping is computational efficiency or deliberate coarsening. If an algorithm is expensive, or if you only want it evaluated under certain conditions, a wrapper can enforce that cadence without modifying the algorithm’s own implementation.

SkippingOnlineAlgorithmWrapper is part of the current public pysatl_cpd.core.online surface. When the condition is true, the wrapper does not pass that observation to the wrapped algorithm; it returns the previous detection statistic instead. The table below includes samples_seen_by_algorithm so skipped observations visibly do not advance the underlying algorithm state.

skipping_condition = SkippingCondition(
    name="large-absolute-observation",
    condition=lambda value: abs(float(value)) > 1.0,
)
skipping_algorithm = SkippingOnlineAlgorithmWrapper(
    ShewhartControlChart(learning_period_size=5, window_size=5),
    skipping_condition=skipping_condition,
)

skipping_rows = []
for step, value in enumerate(online_demo_values[:24]):
    statistic = skipping_algorithm.process(value)
    skipping_rows.append(
        {
            "step": step,
            "observation": value,
            "skipped": skipping_condition.condition(value),
            "samples_seen_by_algorithm": skipping_algorithm.state.samples_count,
            "detection_function": statistic,
        }
    )

display(pd.DataFrame(skipping_rows).round(3))
print("Skipping wrapper description:", skipping_algorithm.description)
step observation skipped samples_seen_by_algorithm detection_function
0 0 0.001 False 1 0.000
1 1 0.299 False 2 0.000
2 2 -0.274 False 3 0.000
3 3 -0.891 False 4 0.000
4 4 -0.455 False 5 0.000
5 5 -0.992 False 6 0.000
6 6 0.060 False 7 0.377
7 7 1.340 True 7 0.377
8 8 -0.492 False 8 0.934
9 9 -0.620 False 9 1.107
10 10 0.490 False 10 0.685
11 11 0.357 False 11 0.112
12 12 0.105 False 12 0.868
13 13 -0.930 False 13 0.800
14 14 -0.029 False 14 0.622
15 15 0.695 False 15 1.117
16 16 -1.344 True 15 1.117
17 17 -0.458 False 16 0.940
18 18 -1.901 True 16 0.940
19 19 -1.290 True 16 0.940
20 20 -1.842 True 16 0.940
21 21 -0.235 False 17 0.320
22 22 -1.267 True 17 0.320
23 23 0.271 False 18 0.031
Skipping wrapper description: OnlineAlgorithmDescription(name='ShewhartControlChart{skip[on=large-absolute-observation]}', configuration=w = 5)
batch_reducer = BatchReducer(name="mean", reducer=lambda batch: float(np.mean(batch)))
batching_algorithm = BatchingOnlineAlgorithmWrapper(
    ShewhartControlChart(learning_period_size=3, window_size=3),
    batch_size=4,
    reducer=batch_reducer,
)
batching_values = [batching_algorithm.process(value) for value in online_demo_values[:24]]

display(
    pd.DataFrame(
        {
            "step": np.arange(len(batching_values)),
            "observation": online_demo_values[:24],
            "batch_completed": [(step + 1) % 4 == 0 for step in range(24)],
            "detection_function": batching_values,
        }
    ).round(3)
)
print("Batching wrapper description:", batching_algorithm.description)
step observation batch_completed detection_function
0 0 0.001 False NaN
1 1 0.299 False NaN
2 2 -0.274 False NaN
3 3 -0.891 True 0.000
4 4 -0.455 False 0.000
5 5 -0.992 False 0.000
6 6 0.060 False 0.000
7 7 1.340 True 0.000
8 8 -0.492 False 0.000
9 9 -0.620 False 0.000
10 10 0.490 False 0.000
11 11 0.357 True 0.000
12 12 0.105 False 0.000
13 13 -0.930 False 0.000
14 14 -0.029 False 0.000
15 15 0.695 True 0.000
16 16 -1.344 False 0.000
17 17 -0.458 False 0.000
18 18 -1.901 False 0.000
19 19 -1.290 True 0.969
20 20 -1.842 False 0.969
21 21 -0.235 False 0.969
22 22 -1.267 False 0.969
23 23 0.271 True 0.496
Batching wrapper description: OnlineAlgorithmDescription(name='ShewhartControlChart{batch[size=4, reduce=mean]}', configuration=w = 3)

Summary of Section 6: Runtime wrappers are the cadence-control layer of the online API. Even when they are not used in a simple tutorial run, they show that execution behavior can be composed around algorithms without rewriting the underlying detection logic.

Section 7. Final Recap

This chapter solved the runtime side of the tutorial series. We started from one labeled provider, examined the stateful algorithm interface, wrapped the algorithm into a reset detector, studied the trace object it produces, and located the optional wrapper layer that controls cadence.

That is the exact context needed for the next notebook. Visualization will not introduce a new execution model; it will render the same providers and traces we have already created here.