benchmark

Benchmark orchestrator for PySATL CPD.

class pysatl_cpd.benchmark.benchmark.Benchmark(dataset, registry, *, n_jobs=1)[source]

Bases: Generic

Orchestrates benchmark execution over a dataset through a detector registry.

Manages the full lifecycle of running multiple detector configurations against a dataset, persisting intermediate results in an associated registry, and collecting per-scenario analysis at the end.

Parameters:

dataset (Dataset[TypeVar(DataT), TimeseriesAnnotation]) – Labeled dataset whose providers serve as detector inputs.
registry (BenchmarkRegistry[TypeVar(DataT), TypeVar(TraceT, bound= DetectionTrace)]) – Registry that caches and retrieves per-detector execution results.
n_jobs (int) – Number of parallel worker processes (default 1). Must be non-zero.

__init__(dataset, registry, *, n_jobs=1)[source]

Parameters:

dataset (Dataset[DataT, TimeseriesAnnotation])
registry (BenchmarkRegistry[DataT, TraceT])
n_jobs (int)

Return type:

None

property dataset: Dataset[DataT, TimeseriesAnnotation]: Dataset bound to this benchmark instance.

property registry: BenchmarkRegistry[DataT, TraceT]: Registry of precomputed detector executions bound to this benchmark.

property n_jobs: int: Number of parallel worker processes used for benchmark execution.

upload_registry(upload_registry_path)[source]

Load a previously exported registry file from disk.

Parameters:: upload_registry_path (Path) – Filesystem path to a pickled registry file.
Return type:: None

export_registry(export_registry_path)[source]

Persist the current registry contents to disk as a pickle.

Parameters:: export_registry_path (Path) – Destination filesystem path for the pickled registry.
Return type:: None

run_scenario(scenario, *, force_recompute=False, n_jobs=None, backend='loky')[source]

Execute a benchmark scenario across all detectors in the registry.

Iterates over jobs prepared by the scenario, runs each detector against its assigned providers, and returns the aggregated scenario analysis. Entries already present in the registry are skipped unless force_recompute is set.

Parameters:

scenario (BenchmarkScenario[TypeVar(DataT), TypeVar(TraceT, bound= DetectionTrace), TypeVar(ResultT)]) – Scenario defining job preparation and result analysis logic.
force_recompute (bool) – If True, re-executes detectors even when a cached result exists.
n_jobs (int | None) – Worker count override; falls back to instance-level n_jobs when None.
backend (str) – Joblib parallel backend identifier (default "loky"; e.g. threading).

Returns:

Scenario analysis results keyed by detector description.

Return type:

dict[ChangePointDetectorDescription, TypeVar(ResultT)]