scan-py documentation

Documentation for the scan-py change-point detection framework. This consists of the complete documentation with usage examples.

SCAN

scan.detector.scan_cpd(x: Iterable[float], window_sizes: Sequence[int] | None = None, alpha: float = 0.05, n_boot: int = 400, vote_threshold: float = 0.5, min_window: int = 15, max_window: int | None = None, block_length: int | None = None, taper: str = 'tukey', ipm: str = 'wasserstein', tolerance_distance: int | None = None, random_state: int | None = None, n_jobs: int | None = -1, return_all: bool = True, *, change_type: str | None = None, batch_size: int = 32) → ScanResult

Run SCAN / Ensemble SCAN and return a structured result.

Parameters:

  • x – One-dimensional time series as input.

  • window_sizes – List of window sizes that we need to run the ensemble model. If omitted, set of windows are derived from min_window and max_window using builtin _default_window_sizes function.

  • alpha – Significance level for the hypothesis test.

  • n_boot – Number of tapered block bootstrap draws per local comparison.

  • vote_threshold – Minimum ensemble vote score required to retain a candidate change point.

  • min_window – Bounds to be used when window_sizes is omitted.

  • max_window

  • block_length – Optional bootstrap block length. If omitted, it is selected automatically using the n^(1/3) rule.

  • taper – Taper weights to be applied to tapered bootstrap blocks. Supported values are "tukey" for Tukey’s weights and "none" for classical block bootstrap.

  • ipm – IPM discrepancy measure used to detect changes. Currently, this supports "wasserstein" distance.

  • tolerance_distance – Optional maximum distance for grouping nearby candidates during ensemble voting. This defaults to the minimum window size in the window_sizes list.

  • random_state – Optional For reproducibility.

  • n_jobs – Number of worker threads used for parallel processing. The default -1 uses all available CPU threads.

  • return_all – When False, omit per-window diagnostics.

  • change_type – Explicit change type: "mean", "var", or "distribution".

  • batch_size – Number of local comparisons batched per worker.

Returns:

Structured result containing selected change points, vote scores, per-window diagnostics, parameters, and runtime metadata.

Return type:

ScanResult

ScanResult output

The scan_cpd function returns a ScanResult object. This object collects the main outputs of the SCAN procedure in one place, including the final detected change points, detection scores, voting information, per-window diagnostics, threshold values, input parameters, metadata, and the raw backend output.

Attribute

Type

Description

change_points

List[int]
Final set of estimated change-point locations returned by SCAN framework.
This is the main output users usually need for downstream analysis or plotting.

scores

Dict[int, float]
Detection scores associated with candidate change points.
These help assess the relative strength of detected changes.

votes

Dict[int, int]
Raw vote count by candidate change point.
This is useful for understanding how stable a detection is across multiple window sizes.

window_results

Dict[int, WindowResult]
Diagnostic information from each individual window size.
This helps inspect which window sizes contributed to each detection.

thresholds

Dict[int, Dict[str, List[float]]]
Bootstrap thresholds and local statistics keyed by window size.
These are important for reproducibility and for understanding the rejection rule.

parameters

Dict[str, Any]
Parameter values used in the call to scan_cpd.
This makes the result self-contained and easier to reproduce.

metadata

Dict[str, Any]
Additional run information, such as runtime and backend details.
This is useful for experiments, reporting, debugging, and reproducibility.

segments

Dict[str, Any]
Segment metadata returned by the backend, when available.
This is useful for advanced inspection of backend segmentation output.

raw

Dict[str, Any]
Raw output returned by the Rust/PyO3 backend before post-processing.
This is mainly useful for advanced users, debugging, or development.

The object returned by scan_cpd stores the main outputs of the detection procedure as attributes. Each attribute can be accessed using the dot operator (.), which makes it easy to inspect detected change points, detection scores, voting information, run parameters, metadata, and per-window results.

Example

# Simulate the time series

T = 200_000 # Length of the time series
K = 100 # Number of change-points
min_seg_len = 1000 # Minimum distance between two change-points
seed = 2000 # Seed for reproducibility

x, true_cps, _, _ = simulate_time_series(
    n=T,
    n_cps=K,
    min_seg_len=min_seg_len,
    change_type="mean",
    seed=seed,
)

# Detect change-points

from scan import choose_window_sizes

window_sizes = choose_window_sizes(
    series_length=200_000,
    n_windows=7,
    seed=500,
)
scan.detector.scan_single_window(x: Iterable[float], window_size: int, *, alpha: float = 0.05, n_boot: int = 400, block_length: int | None = None, taper: str = 'tukey', ipm: str = 'wasserstein', random_state: int | None = None, change_type: str | None = None, batch_size: int = 32) → WindowResult

Run SCAN for one window size and return per-location diagnostics.

Parameters:

  • x – One-dimensional time series to detect change-points.

  • window_size – Window size used for every local comparison.

  • alpha – Detection controls with the same meaning as in scan_cpd().

  • n_boot – Detection controls with the same meaning as in scan_cpd().

  • block_length – Detection controls with the same meaning as in scan_cpd().

  • taper – Detection controls with the same meaning as in scan_cpd().

  • ipm – Detection controls with the same meaning as in scan_cpd().

  • random_state – Detection controls with the same meaning as in scan_cpd().

  • change_type – Detection controls with the same meaning as in scan_cpd().

  • batch_size – Detection controls with the same meaning as in scan_cpd().

Returns:

Diagnostics for the supplied window size, including local statistics, thresholds, localized regions, and candidate change points.

Return type:

WindowResult

WindowResult output

The scan_single_window function returns a WindowResult object. This object contains the diagnostics for one scan window size.

Attribute

Type

Description

window_size

int
Window size used by the local scan.
This identifies which scan scale produced the diagnostics.

change_points

List[int]
Candidate change-points detected by the model.
This is useful for inspecting detections before ensemble voting.

starts

List[int]
Start indices of local windows evaluated by the backend.
These align each statistic and threshold with its local comparison.

statistics

List[float]
Local discrepancy statistics aligned with starts.
These help inspect where the local scan found strong evidence of change.

tapered_block_bootstrap_threshold

List[float]
Bootstrap thresholds aligned with starts.
These show the threshold each local statistic was compared against.

Example

import numpy as np
from scan import scan_single_window

x = np.r_[np.zeros(100), np.ones(100)]
window_result = scan_single_window(x, window_size=20, n_boot=100)
print(window_result.change_points)

Utils for SCAN

scan.metrics.covering_metric(true_cps: Iterable[int], estimated_cps: Iterable[int], n: int) → float

Compute a weighted segment-covering score in [0, 1].

Each true segment is compared with all estimated segments using intersection-over-union, then weighted by the true segment length uaing Jaccard distance between the segments.

Parameters:

  • true_cps – Ground-truth split locations.

  • estimated_cps – Estimated split locations.

  • n – Total series length.

Returns:

Weighted covering score, where 1 is perfect segment agreement.

Return type:

float

Example

from scan import covering_metric

score = covering_metric(true_cps=[100, 200], estimated_cps=[98, 205], n=300)
print(score)
scan.metrics.precision_recall_cpd(true_cps: Iterable[int], estimated_cps: Iterable[int], tolerance: int = 10) → Tuple[float, float]

Return tolerant precision and recall.

Empty sets are handled in the usual change-point evaluation convention: precision is 1 when there are no estimates and no true points, and recall is 1 when there are no true points and no estimates.

Example

from scan import precision_recall_cpd

precision, recall = precision_recall_cpd([100, 200], [98, 205], tolerance=10)
print(precision, recall)
scan.metrics.f1_score_cpd(true_cps: Iterable[int], estimated_cps: Iterable[int], tolerance: int = 10) → float

Return tolerant F1 score for change-point estimates.

Parameters:

  • true_cps – Ground-truth and estimated split locations.

  • estimated_cps – Ground-truth and estimated split locations.

  • tolerance – Maximum absolute distance allowed when matching points.

Returns:

Harmonic mean of tolerant precision and recall.

Return type:

float

Example

from scan import f1_score_cpd

f1 = f1_score_cpd([100, 200], [98, 205], tolerance=10)
print(f1)
scan.simulator.choose_window_sizes(series_length: int, n_windows: int = 7, seed: int = 500) → list[int]

Choose a reproducible set of safe scan window sizes.

Parameters:

  • series_length – Length of the series that will be scanned.

  • n_windows – Maximum number of window sizes to return.

  • seed – Random seed used when sampling from candidate sizes.

Returns:

Sorted scan window sizes.

Return type:

list[int]

Example

from scan import choose_window_sizes

windows = choose_window_sizes(series_length=1000, n_windows=5, seed=123)
print(windows)
scan.ensemble.ensemble_vote(window_results: Mapping[int, Sequence[int]], vote_threshold: float = 0.5, tolerance: int = 10) → Tuple[List[int], Dict[int, float], Dict[int, int]]

Apply ensemble voting across window-specific candidate lists.

Parameters:

  • window_results – Mapping from window size to candidate split locations.

  • vote_threshold – Minimum fraction of windows required to retain a cluster leader.

  • tolerance – Maximum gap used to cluster nearby candidates before voting.

Returns:

Selected change points, normalized vote scores, and raw vote counts.

Return type:

tuple[list[int], dict[int, float], dict[int, int]]

Example

from scan import ensemble_vote

selected, scores, votes = ensemble_vote({20: [100], 30: [102]}, tolerance=5)
print(selected, scores, votes)
scan.ensemble.merge_change_points(change_points: Iterable[int], tolerance: int = 10) → List[List[int]]

Cluster nearby candidate change points.

Parameters:

  • change_points – Candidate split locations to group.

  • tolerance – Maximum gap between consecutive candidates in the same cluster.

Returns:

Sorted clusters of nearby candidate locations.

Return type:

list[list[int]]

Example

from scan import merge_change_points

clusters = merge_change_points([98, 100, 205], tolerance=5)
print(clusters)
scan.bootstrap.adaptive_threshold(left: Iterable[float], right: Iterable[float], alpha: float = 0.05, n_boot: int = 400, block_length: int | None = None, taper: str = 'tukey', random_state: int | None = None) → float

Compute a local bootstrap threshold for a two-window comparison.

The two windows are mean-centered and pooled before bootstrapping so the null distribution represents no local distributional change.

Parameters:

  • left – Adjacent samples to compare.

  • right – Adjacent samples to compare.

  • alpha – Upper-tail probability for the returned threshold.

  • n_boot – Number of bootstrap samples used to approximate the null distribution.

  • block_length – Optional bootstrap block length.

  • taper – Taper shape passed to tapered_block_bootstrap().

  • random_state – Optional NumPy random seed.

Returns:

Empirical 1 - alpha quantile of bootstrapped Wasserstein statistics.

Return type:

float

Example

import numpy as np
from scan import adaptive_threshold

rng = np.random.default_rng(123)
threshold = adaptive_threshold(rng.normal(size=50), rng.normal(size=50), n_boot=100)
print(threshold)
scan.bootstrap.tapered_block_bootstrap(x: Iterable[float], sample_length: int, block_length: int | None = None, n_boot: int = 1, taper: str = 'tukey', random_state: int | None = None) → ndarray

Generate tapered block bootstrap samples.

Parameters:

  • x – One-dimensional source series used to draw overlapping blocks.

  • sample_length – Number of observations in each bootstrap replicate.

  • block_length – Length of each sampled block. When omitted, uses n ** (1 / 3) with a minimum of 3 observations.

  • n_boot – Number of bootstrap replicates to generate.

  • taper – Taper shape. Use "tukey" for Hann-style tapering or "none" for rectangular blocks.

  • random_state – Optional NumPy random seed for reproducible sampling.

Returns:

Array with shape (n_boot, sample_length) containing bootstrap replicates.

Return type:

numpy.ndarray

Example

import numpy as np
from scan import tapered_block_bootstrap

samples = tapered_block_bootstrap(np.arange(100), sample_length=50, n_boot=3)
print(samples.shape)

Plotting functions

scan.plotting.plot_time_series(x: Iterable[float], change_points: Iterable[int] | None = None, true_change_points: Iterable[int] | None = None, index: Iterable[float] | None = None, x_label: str = 'Time', y_label: str = 'Value', title: str = 'Time series')

Plot a univariate time series with optional change-point markers.

Parameters:

  • x – One-dimensional time series.

  • change_points – Optional detected split locations.

  • true_change_points – Optional ground-truth split locations.

  • index – Optional x-axis values. If omitted and x has an index attribute, that index is used; otherwise integer positions are used.

  • x_label – Plot labels.

  • y_label – Plot labels.

  • title – Plot labels.

Returns:

Time-series plot with detected and true change-point markers when provided.

Return type:

plotnine.ggplot

Example

from scan import plot_time_series

plot = plot_time_series([0.1, 0.2, 1.4, 1.6], change_points=[2])
print(plot)
scan.plotting.plot_change_points(x: Iterable[float], result: ScanResult, true_change_points: Iterable[int] | None = None, index: Iterable[float] | None = None, x_label: str = 'Time', y_label: str = 'Series', title: str | None = None)

Plot a time series with detected and optional true change points.

Detected change points are taken from result.change_points. True change points should be provided separately using true_change_points.

Parameters:

  • x – One-dimensional time series to visualize.

  • result – SCAN result object from the scan_cpd and scan_single_window containing detected change points.

  • true_change_points – Optional True change-points (or annotated).

  • index – Optional x-axis values.

  • x_label – Plot labels.

  • y_label – Plot labels.

  • title – Plot labels.

Returns:

Time-series plot with detected and optional true change-point markers.

Return type:

plotnine.ggplot

Example

import numpy as np
from scan import plot_change_points, scan_cpd

x = np.r_[np.zeros(100), np.ones(100)]
result = scan_cpd(x, window_sizes=[20], n_boot=50)
plot = plot_change_points(x, result)
print(plot)
scan.plotting.plot_swal_curve(x: Iterable[float], x_label: str = 'Time series', y_label: str = 'Scaled Wasserstein statistic', title: str | None = None)

Plot the SWAL/Wasserstein localization curve for one change point.

Parameters:

  • x – One-dimensional time series containing a single change point.

  • x_label – Plot labels.

  • y_label – Plot labels.

  • title – Plot labels.

Returns:

Local refinement curve with the selected split marked.

Return type:

plotnine.ggplot

Example

import numpy as np
from scan import plot_swal_curve

x = np.r_[np.zeros(50), np.ones(50)]
plot = plot_swal_curve(x)
print(plot)
scan.plotting.plot_vote_scree(result: ScanResult, x_label: str = 'Voting threshold $(\\nu)$', y_label: str = 'Number of retained change points', title: str | None = None)

Plot number of retained change points versus voting threshold.

Parameters:

  • result – SCAN result containing ensemble vote scores.

  • x_label – Plot labels.

  • y_label – Plot labels.

  • title – Plot labels.

Returns:

Scree plot showing how many candidates remain as the vote threshold changes.

Return type:

plotnine.ggplot

Example

import numpy as np
from scan import plot_vote_scree, scan_cpd

x = np.r_[np.zeros(100), np.ones(100)]
result = scan_cpd(x, window_sizes=[20, 30], n_boot=50)
plot = plot_vote_scree(result)
print(plot)
scan.plotting.plot_window_votes(result: ScanResult, max_x_labels: int = 12, x_label_angle: int = 45, x_label: str = 'Candidate change point', y_label: str = 'Window votes', title: str | None = None)

Plot ensemble vote counts for candidate change points.

Parameters:

  • result – SCAN result containing raw vote counts and scores.

  • max_x_labels – Maximum number of x-axis labels to display.

  • x_label_angle – Rotation angle for x-axis labels.

  • x_label – Plot labels.

  • y_label – Plot labels.

  • title – Plot labels.

Returns:

Bar chart of candidate vote counts with the selected vote threshold.

Return type:

plotnine.ggplot

Example

import numpy as np
from scan import plot_window_votes, scan_cpd

x = np.r_[np.zeros(100), np.ones(100)]
result = scan_cpd(x, window_sizes=[20, 30], n_boot=50)
plot = plot_window_votes(result)
print(plot)
scan.plotting.plot_thresholds()

Placeholder for a future threshold diagnostics plot.

This function is exported for API continuity but is not implemented yet. It currently returns None.

Example

from scan import plot_thresholds

plot = plot_thresholds()
print(plot)

Simulator

Functions for simulating univariate time series. This simulator generates AR, ARMA, and ARFIMA series with change points for use in change-point detection simulation studies.

UnivariateSeriesSimulator.simulate_ar_series(rho: float, error_type: str, error_mean: float = 0.0, error_variance: float = 1.0, error_location: float = 0.0, error_scale: float = 0.03) → ndarray

Simulate an AR(1) series with normal or Cauchy innovations.

Parameters:

  • rho – Autoregressive coefficient.

  • error_type – Innovation family: "normal" or "cauchy".

  • error_mean – Mean and variance used for normal innovations.

  • error_variance – Mean and variance used for normal innovations.

  • error_location – Location and scale used for Cauchy innovations.

  • error_scale – Location and scale used for Cauchy innovations.

Returns:

Simulated series of length len_series.

Return type:

numpy.ndarray

Example

from scan import UnivariateSeriesSimulator

simulator = UnivariateSeriesSimulator(len_series=200, seed=123)
x = simulator.simulate_ar_series(rho=0.4, error_type="normal")
UnivariateSeriesSimulator.simulate_ar_unif(error_variance: float = 1.0) → ndarray

Simulate an AR process with random time-varying coefficients.

Parameters:

error_variance – Variance of Gaussian innovations.

Returns:

Simulated series with rho_t drawn uniformly from [0, 1].

Return type:

numpy.ndarray

Example

from scan import UnivariateSeriesSimulator

simulator = UnivariateSeriesSimulator(len_series=200, seed=123)
x = simulator.simulate_ar_unif(error_variance=1.0)
print(x[:5])
UnivariateSeriesSimulator.simulate_arma(phi: Sequence[float] | float, theta: Sequence[float] | float, error_mean: float = 0.0, error_variance: float = 1.0) → ndarray

Simulate an ARMA process using statsmodels when it is installed.

Parameters:

  • phi – Autoregressive coefficients.

  • theta – Moving-average coefficients.

  • error_mean – Mean and variance of Gaussian innovations.

  • error_variance – Mean and variance of Gaussian innovations.

Returns:

Simulated ARMA series.

Return type:

numpy.ndarray

Example

from scan import UnivariateSeriesSimulator

simulator = UnivariateSeriesSimulator(len_series=200, seed=123)
x = simulator.simulate_arma(phi=0.5, theta=0.2)
print(x[:5])
UnivariateSeriesSimulator.arfima_sim(d: float = 0.3, ar: Sequence[float] | None = None, ma: Sequence[float] | None = None, sigma: float = 1.0) → ndarray

Simulate an ARFIMA(p,d,q) series.

Parameters:

  • d – Fractional integration parameter.

  • ar – Optional autoregressive coefficients.

  • ma – Optional moving-average coefficients.

  • sigma – Innovation standard deviation.

Returns:

Simulated ARFIMA series of length len_series.

Return type:

numpy.ndarray

Example

from scan import UnivariateSeriesSimulator

simulator = UnivariateSeriesSimulator(len_series=500, seed=123)
x = simulator.arfima_sim(d=0.3, sigma=1.0)
print(x.shape)
UnivariateSeriesSimulator.apply_random_shifts(series: Sequence[float], change_point_locations: Sequence[int], min_shift: float = 3.0, max_shift: float = 10.0, shifts: Sequence[float] | float | None = None, seed: int | None = None, change_type: str = 'mean', variance_multipliers: Sequence[float] | float | None = None, lognormal_mean: float = 0.0, lognormal_sigma: float | None = None, variance_center: str | float = 'pre_change', variance_reference: str = 'first_segment') → dict[str, ndarray]

Apply mean and/or variance shifts at supplied split locations.

Parameters:

  • series – Base one-dimensional series to transform.

  • change_point_locations – Split locations where segment parameters change.

  • min_shift – Range for randomly generated mean-shift magnitudes.

  • max_shift – Range for randomly generated mean-shift magnitudes.

  • shifts – Optional scalar or per-change mean shifts.

  • seed – Optional seed overriding the simulator seed for shift generation.

  • change_type – Shift type: "mean", "variance", or "distribution".

  • variance_multipliers – Optional scalar or per-change variance multipliers.

  • lognormal_mean – Parameters used for random variance multipliers.

  • lognormal_sigma – Parameters used for random variance multipliers.

  • variance_center – Center used when rescaling segment variance.

  • variance_reference – Reference variance source for multipliers.

Returns:

Original series, shifted series, change points, mean shifts, and variance multipliers.

Return type:

dict[str, numpy.ndarray]

Example

from scan import UnivariateSeriesSimulator

simulator = UnivariateSeriesSimulator(len_series=300, seed=123)
base = simulator.simulate_ar_series(rho=0.0, error_type="normal")
shifted = simulator.apply_random_shifts(base, [100, 200], change_type="mean")
print(shifted["change_points"])
UnivariateSeriesSimulator.select_change_point_locations(min_points: int, ratio: int = 100, min_first_cp: int = 30, n_cps: int | None = None) → ndarray

Select sorted split locations with a minimum spacing constraint.

Parameters:

  • min_points – Minimum distance between consecutive selected split locations.

  • ratio – Used to infer the number of change points when n_cps is not provided.

  • min_first_cp – Earliest allowed split location.

  • n_cps – Optional explicit number of change points to select.

Returns:

Sorted integer split locations.

Return type:

numpy.ndarray

Example

from scan import UnivariateSeriesSimulator

simulator = UnivariateSeriesSimulator(len_series=500, seed=123)
cps = simulator.select_change_point_locations(min_points=80, n_cps=3)
print(cps)