change_point

change_point ¤

Change-point detection: where did the process level (or spread) shift?

Offline segmentation of one signal into stretches with constant mean (and optionally constant variance) using PELT, the exact pruned dynamic program of Killick, Fearnhead & Eckley (2012). Unlike a CUSUM or a rolling-window test it finds all change points at once, places them at the exact sample, and needs no window length. Typical uses: a tool change, a new raw-material lot, a recalibration, a drifting setpoint that was corrected.

The implementation is plain numpy (no extra dependency). Costs are Gaussian negative log-likelihoods computed from cumulative sums, so each candidate segment costs O(1).

ChangePointEvents ¤

ChangePointEvents(
    dataframe: DataFrame,
    signal_uuid: str,
    *,
    value_column: str = "value_double",
    event_uuid: str = "quality:change_point",
    time_column: str = "systime"
)

Bases: Base

Quality: change-point detection (PELT).

detect_change_points: One point event per detected shift, with the level and spread before and after.
segment_statistics: The stable stretches between change points as intervals with their mean, std and sample count.

Example

import ts_shape df = ts_shape.make_timeseries(["pressure"], n_points=100) df.loc[50:, "value_double"] += 5 # level shift half-way out = ts_shape.ChangePointEvents(df, "pressure").detect_change_points() numbers = ["mean_before", "mean_after", "shift_sigma"] out[["systime", *numbers]].round(dict.fromkeys(numbers, 1)) systime mean_before mean_after shift_sigma 0 2025-01-01 00:25:00 100.1 104.8 6.1

detect_change_points ¤

detect_change_points(
    cost: Cost = "normal_mean",
    penalty: float | None = None,
    min_size: int = 5,
    method: Method = "auto",
) -> pd.DataFrame

Detect level (and optionally spread) shifts.

Parameters:

Name	Type	Description	Default
`cost`	`Cost`	`"normal_mean"` finds mean shifts; `"normal_meanvar"` also finds changes in variance.	`'normal_mean'`
`penalty`	`float \| None`	Cost of one change point; `None` = BIC (see :func:`pelt`). Larger values find fewer, larger changes.	`None`
`min_size`	`int`	Minimum samples per segment.	`5`
`method`	`Method`	`"auto"` (default) runs `"pelt"` on up to :data:`PELT_WARN_SAMPLES` (20,000) samples and `"binseg"` above, where PELT may turn quadratic. `"pelt"` is exact at any length (it warns above 20,000 samples); `"binseg"` is fast and approximate (for long series with few changes).	`'auto'`

Returns:

Type	Description
`DataFrame`	Point events at the first sample of each new segment: `systime`,
`DataFrame`	`uuid`, `source_uuid`, `value` (the new level),
`DataFrame`	`mean_before`, `mean_after`, `std_before`, `std_after`,
`DataFrame`	`shift` (`mean_after - mean_before`) and `shift_sigma`
`DataFrame`	(the shift in units of the pooled segment std).

segment_statistics ¤

segment_statistics(
    cost: Cost = "normal_mean",
    penalty: float | None = None,
    min_size: int = 5,
    method: Method = "auto",
) -> pd.DataFrame

The stable segments between change points.

A segment holds until the first sample of the next one; the last segment ends at the last sample.

Parameters:

Name	Type	Description	Default
`cost`	`Cost`	As in :meth:`detect_change_points`.	`'normal_mean'`
`penalty`	`float \| None`	As in :meth:`detect_change_points`.	`None`
`min_size`	`int`	As in :meth:`detect_change_points`.	`5`
`method`	`Method`	As in :meth:`detect_change_points`.	`'auto'`

Returns:

Name	Type	Description
`Intervals`	`DataFrame`	`start`, `end`, `duration_seconds`, `uuid`,
	`DataFrame`	`source_uuid`, `segment`, `mean`, `std`, `sample_count`.

pelt ¤

pelt(
    x: ndarray,
    *,
    cost: Cost = "normal_mean",
    penalty: float | None = None,
    min_size: int = 2
) -> list[int]

Optimal change points of x under a penalized Gaussian cost (PELT).

Exact: returns the segmentation minimizing total cost plus penalty per change. Runtime is near-linear when changes are frequent and quadratic when they are rare; for long series with few changes use :func:binseg or resample first.

Parameters:

Name	Type	Description	Default
`x`	`ndarray`	1-D finite values.	required
`cost`	`Cost`	`"normal_mean"` (mean changes, common noise level estimated robustly) or `"normal_meanvar"` (mean and variance change). `normal_mean` assumes the noise level is constant: on a signal whose spread changes it splits the noisy stretch into many segments, so use `normal_meanvar` there.	`'normal_mean'`
`penalty`	`float \| None`	Cost of one extra change point. `None` uses the BIC: `2 log n` for `normal_mean` and `3 log n` for `normal_meanvar` (parameters added per change: location + mean, or location + mean + variance).	`None`
`min_size`	`int`	Minimum samples per segment.	`2`

Returns:

Type	Description
`list[int]`	Sorted indices `i` where a new segment starts at `x[i]`.

binseg ¤

binseg(
    x: ndarray,
    *,
    cost: Cost = "normal_mean",
    penalty: float | None = None,
    min_size: int = 2
) -> list[int]

Change points by binary segmentation: fast (O(n log n)), approximate.

Splits a segment at its best point while the cost reduction exceeds penalty, then recurses into both halves. Same arguments and return value as :func:pelt; close changes may be merged or misplaced.