generator
Synthetic data-generation API for change-point detection.
This package provides a declarative framework for generating synthetic time series with known regime structure, change points, and state labels. It is designed for benchmarking, testing, and studying the behavior of change-point detection algorithms under controlled conditions.
Scenarios are defined as immutable specifications (ScenarioSpec) that
separate the ordered sequence of segment occurrences from reusable segment
plans. Each plan carries a distribution specification, a state descriptor,
and optional metadata. Distribution specs support univariate distributions
(normal, uniform, exponential, Student-t), multivariate normal with
correlated features, and independent per-column distributions.
Generated series can be converted into labeled-provider types used by the
data layer via the build_* helper functions, or assembled into full
datasets via ScenarioDatasetGenerator. Scenarios can also be loaded
from YAML files or plain Python mappings for config-driven workflows.
Public API
- Distribution specifications
NormalSpec– Normal (Gaussian) distribution parameters.UniformSpec– Uniform distribution over a fixed interval.ExponentialSpec– Exponential distribution with scale parameter.StudentTSpec– Student’s t-distribution parameters.UnivariateDistributionSpec– Type alias for the four univariate specs above.MultivariateNormalSpec– Multivariate normal with named means and covariance structure.IndependentColumnsSpec– Per-column independent univariate distributions.DistributionSpec– Type alias covering all distribution specs.
- Scenario specifications
SegmentSpec– Ordered segment occurrence (plan name and length).SegmentPlan– Reusable plan defining distribution, state, and metadata for a segment type.ScenarioSpec– Top-level scenario blueprint combining segments, plans, and metadata.
- Generators
GenericSeriesGenerator– Core engine that samples series fromScenarioSpecobjects or segment generator sequences.SegmentGenerator– Protocol for custom segment generators. See thesegmentssubpackage for details.LabeledDataGenerator– Protocol for objects that produce labeled data instances.ScenarioDatasetGenerator– BuildsDatasetcollections from named scenario specifications.
- Provider builders
build_pandas_labeled_data– Convert aGeneratedSeriesto a multivariatePandasLabeledDataprovider.build_pandas_univariate_labeled_data– Convert a single feature from aGeneratedSeriesto a univariatePandasLabeledData.build_plain_multivariate_labeled_data– Convert to a NumPy-backedPlainMultivariateLabeledDataprovider.build_plain_univariate_labeled_data– Convert a single feature to a NumPy-backedPlainUnivariateLabeledDataprovider.
- Configuration loaders
scenario_from_mapping– Build aScenarioSpecfrom a plain Python mapping.scenario_from_yaml– Load a single scenario from a YAML file.scenarios_from_yaml– Load one or more scenarios from a YAML file, returning a name-to-spec dictionary.
- Presets
PRESET_SCENARIOS– Frozen mapping of built-in preset scenario specifications (currently empty; presets are constructed on demand).preset_dataset– One-call dataset generation from a named preset (e.g.,"mean_shifts","variance_shifts","covariance_shifts").
Subpackages
segments– Core segment-generation building blocks, sampling utilities, and theSegmentGeneratorprotocol. See its module docstring for details.providers– Provider-builder functions that convertGeneratedSeriesinto labeled-provider types. See its module docstring for details.
Examples
Examples
Generate a univariate mean-shift series from a scenario specification:
>>> from pysatl_cpd.data.generator import (
... GenericSeriesGenerator,
... NormalSpec,
... ScenarioSpec,
... SegmentPlan,
... SegmentSpec,
... )
>>> from pysatl_cpd.data.typedefs import StateDescriptor, frozendict
>>> scenario = ScenarioSpec(
... name="mean_shift",
... segments=(
... SegmentSpec(plan_name="baseline", length=100),
... SegmentSpec(plan_name="shifted", length=60),
... SegmentSpec(plan_name="baseline", length=40),
... ),
... plans=frozendict(
... baseline=SegmentPlan(
... distribution=NormalSpec(mean=0.0, std=1.0),
... state=StateDescriptor(type="baseline"),
... ),
... shifted=SegmentPlan(
... distribution=NormalSpec(mean=3.0, std=1.0),
... state=StateDescriptor(type="shifted"),
... ),
... ),
... )
>>> series = GenericSeriesGenerator(seed=42).generate_from_scenario(
... scenario, name="example_series",
... )
>>> series.data.shape
(200, 1)
>>> series.change_points
(99, 159)
Generate a multivariate series with independent columns:
>>> from pysatl_cpd.data.generator import (
... GenericSeriesGenerator,
... IndependentColumnsSpec,
... NormalSpec,
... ScenarioSpec,
... SegmentPlan,
... SegmentSpec,
... )
>>> from pysatl_cpd.data.typedefs import StateDescriptor, frozendict
>>> mv_scenario = ScenarioSpec(
... name="independent_mv",
... segments=(
... SegmentSpec(plan_name="a", length=50),
... SegmentSpec(plan_name="b", length=30),
... ),
... plans=frozendict(
... a=SegmentPlan(
... distribution=IndependentColumnsSpec(
... columns=frozendict(
... x=NormalSpec(mean=0.0, std=1.0),
... y=NormalSpec(mean=10.0, std=2.0),
... ),
... ),
... state=StateDescriptor(type="regime_a"),
... ),
... b=SegmentPlan(
... distribution=IndependentColumnsSpec(
... columns=frozendict(
... x=NormalSpec(mean=5.0, std=1.0),
... y=NormalSpec(mean=15.0, std=2.0),
... ),
... ),
... state=StateDescriptor(type="regime_b"),
... ),
... ),
... )
>>> mv_series = GenericSeriesGenerator(seed=0).generate_from_scenario(
... mv_scenario,
... )
>>> mv_series.feature_names
('x', 'y')
>>> mv_series.data.shape
(80, 2)
Build a labeled provider from a generated series:
>>> from pysatl_cpd.data.generator import (
... GenericSeriesGenerator,
... NormalSpec,
... ScenarioSpec,
... SegmentPlan,
... SegmentSpec,
... build_pandas_labeled_data,
... )
>>> from pysatl_cpd.data.typedefs import StateDescriptor, frozendict
>>> scenario = ScenarioSpec(
... name="provider_example",
... segments=(
... SegmentSpec(plan_name="a", length=50),
... SegmentSpec(plan_name="b", length=30),
... ),
... plans=frozendict(
... a=SegmentPlan(
... distribution=NormalSpec(mean=0.0, std=1.0),
... state=StateDescriptor(type="baseline"),
... ),
... b=SegmentPlan(
... distribution=NormalSpec(mean=3.0, std=1.0),
... state=StateDescriptor(type="shifted"),
... ),
... ),
... )
>>> series = GenericSeriesGenerator(seed=0).generate_from_scenario(scenario)
>>> provider = build_pandas_labeled_data(series, name="provider_example")
>>> list(provider.feature_columns)
['value']
>>> provider.change_points
(50,)
Generate a dataset from multiple named scenarios:
>>> from pysatl_cpd.data.generator import (
... GenericSeriesGenerator,
... NormalSpec,
... ScenarioDatasetGenerator,
... ScenarioSpec,
... SegmentPlan,
... SegmentSpec,
... )
>>> from pysatl_cpd.data.typedefs import StateDescriptor, frozendict
>>> s1 = ScenarioSpec(
... name="small_shift",
... segments=(SegmentSpec(plan_name="a", length=50), SegmentSpec(plan_name="b", length=30)),
... plans=frozendict(
... a=SegmentPlan(distribution=NormalSpec(mean=0.0, std=1.0), state=StateDescriptor(type="a")),
... b=SegmentPlan(distribution=NormalSpec(mean=1.0, std=1.0), state=StateDescriptor(type="b")),
... ),
... )
>>> s2 = ScenarioSpec(
... name="large_shift",
... segments=(SegmentSpec(plan_name="a", length=50), SegmentSpec(plan_name="b", length=30)),
... plans=frozendict(
... a=SegmentPlan(distribution=NormalSpec(mean=0.0, std=1.0), state=StateDescriptor(type="a")),
... b=SegmentPlan(distribution=NormalSpec(mean=5.0, std=1.0), state=StateDescriptor(type="b")),
... ),
... )
>>> gen = ScenarioDatasetGenerator({"small": s1, "large": s2}, seed=0)
>>> dataset = gen.generate("small", size=2)
>>> len(dataset)
2
Load a scenario from a Python mapping:
>>> from pysatl_cpd.data.generator import scenario_from_mapping
>>> mapping = {
... "name": "from_mapping",
... "segments": [
... {"plan_name": "a", "length": 40},
... {"plan_name": "b", "length": 20},
... ],
... "plans": {
... "a": {"distribution": {"kind": "normal", "mean": 0.0, "std": 1.0}},
... "b": {"distribution": {"kind": "normal", "mean": 2.0, "std": 1.0}},
... },
... }
>>> spec = scenario_from_mapping(mapping)
>>> spec.name
'from_mapping'
Generate a dataset from a built-in preset:
>>> from pysatl_cpd.data.generator import preset_dataset
>>> ds = preset_dataset("mean_shifts", n_series=2, seed=0, series_length=120)
>>> len(ds)
2
Notes
Notes
All change-point indices are zero-based throughout the package.
Univariate distribution specs (
NormalSpec,UniformSpec,ExponentialSpec,StudentTSpec) produce single-column arrays with the default feature name"value".MultivariateNormalSpecrequires non-emptymeanswith feature names as keys. The covariance can be a scalar, a 1-D sequence (diagonal), or a nested sequence (full matrix).IndependentColumnsSpecrequires each column to reference a univariate distribution spec.GenericSeriesGeneratoruses NumPy’sdefault_rnginternally. Passseedfor reproducible results.The
preset_datasetfunction supports presets such as"mean_shifts","variance_shifts","covariance_shifts","no_shifts","extreme_mean_shifts","3d_mean_shifts", and"mixed_shifts".YAML loading requires the
PyYAMLdependency (included in the project’s dev installation).The
segmentsandproviderssubpackages contain additional utilities and protocols; consult their module docstrings for segment-level sampling and provider-conversion details.