online_detection_trace

Module contains online detection trace container for streaming changepoint detection.

This module provides containers for storing step-by-step results and aggregated traces from online changepoint detection algorithms.

pysatl_cpd.core.online.detectors.online_detection_trace.extract_periods(in_periods)[source]

Extract continuous periods where condition is True.

Parameters:

in_periods (Sequence[bool | None]) – Sequence of boolean values indicating whether each position is in a period. None values are ignored.

Returns:

List of (start, end) indices for each continuous period.

Return type:

list[tuple[int, int]]

class pysatl_cpd.core.online.detectors.online_detection_trace.OnlineDetectionStepResult(*, step_num=0, is_forced_change_point=False, is_signal_change_point=False, is_in_skip_period=False, detection_function=nan, processing_time=nan, algorithm_state=None)[source]

Bases: Generic

Result of processing a single observation in online changepoint detection.

This class captures the complete output for one step of an online detection algorithm, including detection flags, computed statistics, and timing information.

Variables:

  • step_num – Zero-based index of the processed observation.

  • is_forced_change_point – Whether a changepoint was forced due to maximum runlength constraint.

  • is_signal_change_point – Whether a changepoint was detected by the algorithm’s detection function exceeding the threshold.

  • is_in_skip_period – Whether this step occurred during a post-detection skip period.

  • detection_function – The value of the detection statistic computed for this observation.

  • processing_time – Wall-clock time in seconds spent processing this step.

  • algorithm_state – Snapshot of algorithm internal state after processing this step.

Parameters:

  • step_num (int)

  • is_forced_change_point (bool)

  • is_signal_change_point (bool)

  • is_in_skip_period (bool)

  • detection_function (Number)

  • processing_time (Number)

  • algorithm_state (StateT | None)

step_num: int = 0
is_forced_change_point: bool = False
is_signal_change_point: bool = False
is_in_skip_period: bool = False
detection_function: Number = nan
processing_time: Number = nan
algorithm_state: StateT | None = None
property is_change_point: bool

Whether a changepoint was detected at this step.

A changepoint is considered detected if it was either forced (due to maximum runlength) or signaled (detection function exceeded threshold).

Returns:

True if a forced or signal changepoint occurred, False otherwise.

Return type:

bool

__init__(*, step_num=0, is_forced_change_point=False, is_signal_change_point=False, is_in_skip_period=False, detection_function=nan, processing_time=nan, algorithm_state=None)
Parameters:

  • step_num (int)

  • is_forced_change_point (bool)

  • is_signal_change_point (bool)

  • is_in_skip_period (bool)

  • detection_function (Number)

  • processing_time (Number)

  • algorithm_state (StateT | None)

Return type:

None

class pysatl_cpd.core.online.detectors.online_detection_trace.OnlineDetectionTrace(*, detected_change_points=<factory>, detector_description, threshold=None, processing_time, detection_function, forced_change_points=<factory>, signal_change_points=<factory>, skip_periods=<factory>, learning_periods=<factory>, algorithm_states)[source]

Bases: DetectionTrace, Generic

Complete trace of online changepoint detection execution.

This class aggregates the results of running an online detection algorithm over a complete data sequence. It extends DetectionTrace with additional metadata specific to online detection, including per-step statistics, processing times, and forced detection markers.

Variables:

  • threshold – Detection threshold used during the run.

  • processing_time – Processing time for each observation step as a 1-D NumPy array.

  • detection_function – Detection function values for each observation as a 1-D NumPy array.

  • forced_change_points – Indices where changepoints were forced due to maximum runlength constraint.

  • signal_change_points – Indices where changepoints were detected due to the algorithm’s detection function exceeding the threshold.

  • skip_periods – Indices of beginning and ending of segments where observations were skipped during post-detection periods.

  • learning_periods – Indices of beginning and ending of segments where algorithm was learning data distribution before change point.

  • algorithm_states – Algorithm state snapshots after processing each observation.

Parameters:
threshold: Number | None = None
processing_time: UnivariateNumericArray
detection_function: UnivariateNumericArray
forced_change_points: list[int]
signal_change_points: list[int]
skip_periods: list[tuple[int, int]]
learning_periods: list[tuple[int, int]]
algorithm_states: list[StateT | None]
cut(start, end)[source]

Create a new trace representing a cut of the current trace [start, end] (inclusive).

Automatically recalculates all relative indices (change points, periods).

Parameters:

  • start (int) – Start index of the cut (inclusive).

  • end (int) – End index of the cut (inclusive).

Returns:

New trace containing the cut portion with shifted indices.

Return type:

OnlineDetectionTrace[TypeVar(StateT, bound= OnlineAlgorithmState)]

classmethod from_run(steps, detector_description, threshold=None)[source]

Construct an OnlineDetectionTrace from a data provider and step results.

This factory method aggregates per-step results into a complete trace, extracting detection indices, processing times, and state snapshots.

Parameters:
Returns:

Aggregated trace containing all detection results and metadata.

Return type:

OnlineDetectionTrace[TypeVar(StateT, bound= OnlineAlgorithmState)]

Examples

>>> from pysatl_cpd.data import NDArrayUnivariateProvider
>>> import numpy as np
>>> steps = [
...     OnlineDetectionStepResult(step_num=0, detection_function=0.1),
...     OnlineDetectionStepResult(step_num=1, detection_function=0.8, is_change_point=True)
... ]
>>> trace = OnlineDetectionTrace.from_run(steps=steps, threshold=0.5)
>>> trace.detected_changes
[1]
__init__(*, detected_change_points=<factory>, detector_description, threshold=None, processing_time, detection_function, forced_change_points=<factory>, signal_change_points=<factory>, skip_periods=<factory>, learning_periods=<factory>, algorithm_states)
Parameters:
Return type:

None