artificial_dataset package#

Submodules#

The public API is re-exported from the top-level artificial_dataset package and documented under Module contents below. The per-submodule sections that follow repeat the same objects for navigation and use :no-index: so each object has a single canonical cross-reference target.

artificial_dataset.anomaly module#

Anomaly detection dataset generator built from signal components.

The generator produces a single multivariate time series of shape (m, T): m channels, each of length T. Every channel shares a smooth baseline (a superposition of _components signals) plus Gaussian measurement noise. A configurable number of positive triangular spikes are added on top: each spike event shares its centre and width across all channels, while every channel responds with its own random amplitude. Spikes are always additive and positive, so an anomaly is always a peak that rises above the baseline.

class artificial_dataset.anomaly.AnomalyDataset(y, labels, t, peak_indices)[source]

Bases: object

A single multivariate anomaly-detection time series.

y

Observed signal values for m channels, each of length T. float32.

Type:

torch.Tensor, shape (m, T)

labels

Per-timestep anomaly mask: 1 where the timestep falls inside the support of a spike, 0 otherwise, dtype torch.long.

Type:

torch.Tensor, shape (T,)

t

Shared time grid on which every channel baseline is evaluated.

Type:

torch.Tensor, shape (T,)

peak_indices

Ground-truth spike-event centres as sample positions into the series, dtype torch.long and sorted ascending.

Type:

torch.Tensor, shape (k,)

split(fractions)[source]

Split the series along the time axis into train, val, and test.

The timeline is partitioned contiguously from the beginning: the first fractions[0] of the T timesteps go to train, the next fractions[1] to val, and the remainder to test. Each subset’s peak_indices are filtered to the spikes whose centre falls in that window and re-based to local sample positions.

Parameters:

fractions (tuple[float, float, float]) – (train, val, test) fractions; must sum to 1 (within a small tolerance).

Returns:

The three contiguous time segments, each an AnomalyDataset.

Return type:

AnomalySplits

Raises:

ValueError – If the fractions do not sum to 1.

class artificial_dataset.anomaly.AnomalySplits(train, val, test)[source]

Bases: object

Train/validation/test partition of an AnomalyDataset.

train, val, test

The three disjoint time segments, cut contiguously from the beginning of the timeline.

Type:

AnomalyDataset

class artificial_dataset.anomaly.SpikeParams(amplitude_range=(4.0, 7.0), width_range=(3, 6), count_range=(3, 8), margin=20)[source]

Bases: object

Configuration of the random positive anomaly spikes.

The series receives a random number of spike events. Every event has a shared centre and width across channels, while its amplitude is sampled independently per channel, so all channels react to the same event with different magnitudes. Every value is drawn uniformly from the inclusive range it controls, making the spikes random yet fully configurable.

amplitude_range

Inclusive (min, max) peak height of a spike, sampled per channel. Values are positive so spikes always rise above the baseline.

Type:

tuple[float, float]

width_range

Inclusive (min, max) half-width w of a spike in samples; the triangular bump spans [centre - w, centre + w].

Type:

tuple[int, int]

count_range

Inclusive (min, max) number of spike events in the series.

Type:

tuple[int, int]

margin

Minimum distance (in samples) kept between a spike centre and either end of the series, so spikes are not clipped at the boundaries.

Type:

int

artificial_dataset.anomaly.make_anomaly_dataset(series_length=1000, noise_std=0.4, x_range=(0.0, 18.84955592153876), channel_params=None, spike_params=None, split=None, random_state=None)[source]

Generate a single multivariate anomaly-detection time series.

Overloads:
  • series_length (int), noise_std (float), x_range (tuple[float, float]), channel_params (list[dict[str, Any]] | None), spike_params (SpikeParams | None), split (None), random_state (int | None) → AnomalyDataset

  • series_length (int), noise_std (float), x_range (tuple[float, float]), channel_params (list[dict[str, Any]] | None), spike_params (SpikeParams | None), split (tuple[float, float, float]), random_state (int | None) → AnomalySplits

Every channel shares a smooth baseline built from compose() plus Gaussian measurement noise. A random number of positive triangular spikes are then added: each spike event shares its centre and width across all channels, while every channel responds with its own random amplitude sampled from spike_params.

Parameters:
  • series_length (int) – Length T of the time series.

  • noise_std (float) – Standard deviation of the additive Gaussian measurement noise applied to every channel.

  • x_range (tuple[float, float]) – Closed interval [min, max] spanned by the shared time grid on which the baselines are evaluated.

  • channel_params (list[dict[str, Any]] | None) – One signal-parameter dict per channel, each passed to compose(). The number of entries sets the channel count m. When None, a single default sinusoidal channel is used.

  • spike_params (SpikeParams | None) – Configuration of the random positive anomaly spikes. When None, the defaults of SpikeParams are used.

  • split (tuple[float, float, float] | None) – When given, the (train, val, test) fractions used to split the series along the time axis from the beginning; the function then returns an AnomalySplits. When None, a single AnomalyDataset is returned.

  • random_state (int | None) – Seed passed to torch.manual_seed() for reproducibility.

Returns:

An AnomalyDataset when split is None, otherwise an AnomalySplits holding the three time segments.

Return type:

AnomalyDataset or AnomalySplits

Examples

>>> data = make_anomaly_dataset(series_length=200, random_state=0)
>>> data.y.shape, data.labels.shape, data.t.shape
(torch.Size([1, 200]), torch.Size([200]), torch.Size([200]))
>>> bool((data.labels[data.peak_indices] == 1).all())
True
>>> splits = make_anomaly_dataset(
...     series_length=200, split=(0.5, 0.25, 0.25), random_state=0
... )
>>> splits.train.y.shape[1], splits.val.y.shape[1], splits.test.y.shape[1]
(100, 50, 50)

artificial_dataset.classification module#

Classification dataset generator built from signal components.

artificial_dataset.classification.make_classification(n_samples=1000, n_classes=2, noise_std=0.1, x_range=(0.0, 6.283185307179586), class_params=None, random_state=None)[source]

Generate a classification dataset from signal components.

Each class is characterised by a unique combination of linear, polynomial, and sinusoidal components. Samples are produced by evaluating the class signal at uniformly drawn x values and adding Gaussian noise.

Parameters:
  • n_samples (int) – Total number of samples across all classes.

  • n_classes (int) – Number of distinct classes.

  • noise_std (float) – Standard deviation of the additive Gaussian noise applied to every sample.

  • x_range (tuple[float, float]) – Closed interval [min, max] from which input values are drawn uniformly.

  • class_params (list[dict[str, Any]] | None) – Per-class component configuration. Each entry is a params dict understood by compose(). Must have exactly n_classes entries when provided. When None, a sensible default configuration is used for up to two classes; for more classes parameters are generated automatically.

  • random_state (int | None) – Seed passed to torch.manual_seed() for reproducibility.

Return type:

tuple[Tensor, Tensor]

Returns:

  • X (torch.Tensor, shape (n_samples, 2)) – Feature matrix. Column 0 contains the sampled x values; column 1 contains the corresponding signal value (components + noise).

  • y (torch.Tensor, shape (n_samples,)) – Integer class labels in [0, n_classes), dtype torch.long.

Raises:

ValueError – If len(class_params) != n_classes.

Examples

>>> X, y = make_classification(n_samples=200, n_classes=2, random_state=0)
>>> X.shape, y.shape
(torch.Size([200, 2]), torch.Size([200]))
>>> y.unique().tolist()
[0, 1]

artificial_dataset.metrics module#

Classifier evaluation metrics.

class artificial_dataset.metrics.ClassifierMetrics(y_true, y_pred)[source]

Bases: object

Evaluation metrics for a classifier.

Computes accuracy, macro-averaged precision, recall, F1 score, and a confusion matrix from predicted and ground-truth class labels.

Two construction modes are supported:

  • Constructor — supply full y_true / y_pred label tensors directly.

  • from_anomaly_indices() — supply the index positions of the positive (anomalous) samples as a torch.Tensor or list; binary label vectors are built internally.

All per-class metrics are macro-averaged over every class that appears in either y_true or y_pred.

Parameters:
  • y_true (Tensor) – Ground-truth class labels, dtype torch.long.

  • y_pred (Tensor) – Predicted class labels, dtype torch.long.

Raises:

ValueError – If y_true and y_pred differ in length, are not 1-D, or are empty.

Examples

>>> import torch
>>> y_true = torch.tensor([0, 1, 0, 1, 0])
>>> y_pred = torch.tensor([0, 1, 1, 1, 0])
>>> m = ClassifierMetrics(y_true, y_pred)
>>> round(m.accuracy, 2)
0.8
>>> m.confusion_matrix
tensor([[2, 1],
        [0, 2]])
property accuracy: float

Fraction of correctly classified samples.

Returns:

Accuracy in [0, 1].

Return type:

float

property confusion_matrix: Tensor

Confusion matrix of shape (n_classes, n_classes).

Entry [i, j] counts samples whose true class is classes[i] and predicted class is classes[j], where classes are the sorted unique labels seen across y_true and y_pred.

Returns:

Confusion matrix, dtype torch.long.

Return type:

torch.Tensor, shape (n_classes, n_classes)

property f1_score: float

Macro-averaged F1 score across all classes.

F1 for class c is the harmonic mean of its precision and recall. Classes where both are 0 contribute 0.

Returns:

Macro-averaged F1 score in [0, 1].

Return type:

float

classmethod from_anomaly_indices(n_samples, true_indices, pred_indices)[source]

Create metrics from anomaly index positions.

Both true_indices and pred_indices specify the positions of the positive (anomaly) class. Binary label vectors of length n_samples are constructed: 0 for normal and 1 for anomalous.

Parameters:
  • n_samples (int) – Total number of samples.

  • true_indices (Tensor | list[int]) – Index positions of the true anomalies.

  • pred_indices (Tensor | list[int]) – Index positions of the predicted anomalies.

Returns:

Backed by binary label vectors derived from the index positions.

Return type:

Self

Raises:

ValueError – If n_samples is not positive or any index is outside [0, n_samples).

Examples

>>> import torch
>>> m = ClassifierMetrics.from_anomaly_indices(
...     n_samples=5,
...     true_indices=[1, 3],
...     pred_indices=torch.tensor([1, 2]),
... )
>>> round(m.accuracy, 2)
0.6
property precision: float

Macro-averaged precision across all classes.

Precision for class c is TP_c / (TP_c + FP_c). Classes with no predicted samples contribute 0.

Returns:

Macro-averaged precision in [0, 1].

Return type:

float

property recall: float

Macro-averaged recall across all classes.

Recall for class c is TP_c / (TP_c + FN_c). Classes with no true samples contribute 0.

Returns:

Macro-averaged recall in [0, 1].

Return type:

float

Module contents#

Artificial dataset generation using signal component primitives.

The package exposes two high-level generators:

  • make_classification() - labelled multi-class data where each class follows a distinct signal shape.

  • make_anomaly_dataset() - a single multivariate time series with positive spike anomalies added to a smooth baseline.

  • make_series() plus the composable injectors in injectors (add_point_anomalies, add_level_shift, add_dropout, and others) - build a clean univariate series, then stack one or more labeled anomaly types onto it.

All return torch.Tensor-backed objects so the results integrate directly with PyTorch training loops.

class artificial_dataset.AnomalyDataset(y, labels, t, peak_indices)[source]#

Bases: object

A single multivariate anomaly-detection time series.

y#

Observed signal values for m channels, each of length T. float32.

Type:

torch.Tensor, shape (m, T)

labels#

Per-timestep anomaly mask: 1 where the timestep falls inside the support of a spike, 0 otherwise, dtype torch.long.

Type:

torch.Tensor, shape (T,)

t#

Shared time grid on which every channel baseline is evaluated.

Type:

torch.Tensor, shape (T,)

peak_indices#

Ground-truth spike-event centres as sample positions into the series, dtype torch.long and sorted ascending.

Type:

torch.Tensor, shape (k,)

split(fractions)[source]#

Split the series along the time axis into train, val, and test.

The timeline is partitioned contiguously from the beginning: the first fractions[0] of the T timesteps go to train, the next fractions[1] to val, and the remainder to test. Each subset’s peak_indices are filtered to the spikes whose centre falls in that window and re-based to local sample positions.

Parameters:

fractions (tuple[float, float, float]) – (train, val, test) fractions; must sum to 1 (within a small tolerance).

Returns:

The three contiguous time segments, each an AnomalyDataset.

Return type:

AnomalySplits

Raises:

ValueError – If the fractions do not sum to 1.

class artificial_dataset.AnomalySplits(train, val, test)[source]#

Bases: object

Train/validation/test partition of an AnomalyDataset.

train, val, test

The three disjoint time segments, cut contiguously from the beginning of the timeline.

Type:

AnomalyDataset

class artificial_dataset.ClassifierMetrics(y_true, y_pred)[source]#

Bases: object

Evaluation metrics for a classifier.

Computes accuracy, macro-averaged precision, recall, F1 score, and a confusion matrix from predicted and ground-truth class labels.

Two construction modes are supported:

  • Constructor — supply full y_true / y_pred label tensors directly.

  • from_anomaly_indices() — supply the index positions of the positive (anomalous) samples as a torch.Tensor or list; binary label vectors are built internally.

All per-class metrics are macro-averaged over every class that appears in either y_true or y_pred.

Parameters:
  • y_true (Tensor) – Ground-truth class labels, dtype torch.long.

  • y_pred (Tensor) – Predicted class labels, dtype torch.long.

Raises:

ValueError – If y_true and y_pred differ in length, are not 1-D, or are empty.

Examples

>>> import torch
>>> y_true = torch.tensor([0, 1, 0, 1, 0])
>>> y_pred = torch.tensor([0, 1, 1, 1, 0])
>>> m = ClassifierMetrics(y_true, y_pred)
>>> round(m.accuracy, 2)
0.8
>>> m.confusion_matrix
tensor([[2, 1],
        [0, 2]])
property accuracy: float#

Fraction of correctly classified samples.

Returns:

Accuracy in [0, 1].

Return type:

float

property confusion_matrix: Tensor#

Confusion matrix of shape (n_classes, n_classes).

Entry [i, j] counts samples whose true class is classes[i] and predicted class is classes[j], where classes are the sorted unique labels seen across y_true and y_pred.

Returns:

Confusion matrix, dtype torch.long.

Return type:

torch.Tensor, shape (n_classes, n_classes)

property f1_score: float#

Macro-averaged F1 score across all classes.

F1 for class c is the harmonic mean of its precision and recall. Classes where both are 0 contribute 0.

Returns:

Macro-averaged F1 score in [0, 1].

Return type:

float

classmethod from_anomaly_indices(n_samples, true_indices, pred_indices)[source]#

Create metrics from anomaly index positions.

Both true_indices and pred_indices specify the positions of the positive (anomaly) class. Binary label vectors of length n_samples are constructed: 0 for normal and 1 for anomalous.

Parameters:
  • n_samples (int) – Total number of samples.

  • true_indices (Tensor | list[int]) – Index positions of the true anomalies.

  • pred_indices (Tensor | list[int]) – Index positions of the predicted anomalies.

Returns:

Backed by binary label vectors derived from the index positions.

Return type:

Self

Raises:

ValueError – If n_samples is not positive or any index is outside [0, n_samples).

Examples

>>> import torch
>>> m = ClassifierMetrics.from_anomaly_indices(
...     n_samples=5,
...     true_indices=[1, 3],
...     pred_indices=torch.tensor([1, 2]),
... )
>>> round(m.accuracy, 2)
0.6
property precision: float#

Macro-averaged precision across all classes.

Precision for class c is TP_c / (TP_c + FP_c). Classes with no predicted samples contribute 0.

Returns:

Macro-averaged precision in [0, 1].

Return type:

float

property recall: float#

Macro-averaged recall across all classes.

Recall for class c is TP_c / (TP_c + FN_c). Classes with no true samples contribute 0.

Returns:

Macro-averaged recall in [0, 1].

Return type:

float

class artificial_dataset.SpikeParams(amplitude_range=(4.0, 7.0), width_range=(3, 6), count_range=(3, 8), margin=20)[source]#

Bases: object

Configuration of the random positive anomaly spikes.

The series receives a random number of spike events. Every event has a shared centre and width across channels, while its amplitude is sampled independently per channel, so all channels react to the same event with different magnitudes. Every value is drawn uniformly from the inclusive range it controls, making the spikes random yet fully configurable.

amplitude_range#

Inclusive (min, max) peak height of a spike, sampled per channel. Values are positive so spikes always rise above the baseline.

Type:

tuple[float, float]

width_range#

Inclusive (min, max) half-width w of a spike in samples; the triangular bump spans [centre - w, centre + w].

Type:

tuple[int, int]

count_range#

Inclusive (min, max) number of spike events in the series.

Type:

tuple[int, int]

margin#

Minimum distance (in samples) kept between a spike centre and either end of the series, so spikes are not clipped at the boundaries.

Type:

int

class artificial_dataset.SyntheticSeries(x, y, is_anomaly, anomaly_type, anomalies=<factory>, meta=<factory>)[source]#

Bases: object

Data container for a generated synthetic 1D time series.

x#

Timeline array of shape (T,).

Type:

torch.Tensor

y#

Series values array of shape (T,).

Type:

torch.Tensor

is_anomaly#

Boolean mask of shape (T,), True at any anomalous timestep.

Type:

torch.Tensor

anomaly_type#

List of strings of length T, containing “” for normal timesteps or “|”-joined anomaly tags (e.g., “point|level_shift”).

Type:

list[str]

anomalies#

Audit trail logging every injected anomaly and its metadata.

Type:

list[dict[str, Any]]

meta#

Generation metadata (e.g., function_type, parameters, noise_std).

Type:

dict[str, Any]

property label: Tensor#

Binary classification target derived from is_anomaly.

Returns:

Integer tensor of shape (T,), dtype torch.long. 1 at every timestep flagged anomalous (is_anomaly[i] == True), 0 otherwise. For continuous anomalies (e.g. level shifts, collective anomalies), every point in the affected span is marked 1, since injectors already flag each index in the span via is_anomaly.

Return type:

torch.Tensor

pipe(func, *args, **kwargs)[source]#

Pass self to func(self, *args, **kwargs) and return the result.

Return type:

Any

split(fractions)[source]#

Split the series along the time axis into train, val, and test.

The timeline is partitioned contiguously from the beginning: the first fractions[0] of the T timesteps go to train, the next fractions[1] to val, and the remainder to test. Each subset’s anomalies audit-log entries are clipped to that window and re-based to local sample positions; entries that fall entirely outside the window are dropped.

Parameters:

fractions (tuple[float, float, float]) – (train, val, test) fractions; must sum to 1 (within a small tolerance).

Returns:

The three contiguous time segments, each a SyntheticSeries.

Return type:

SyntheticSeriesSplits

Raises:

ValueError – If the fractions do not sum to 1.

class artificial_dataset.SyntheticSeriesSplits(train, val, test)[source]#

Bases: object

Train/validation/test partition of a SyntheticSeries.

train, val, test

The three disjoint time segments, cut contiguously from the beginning of the timeline.

Type:

SyntheticSeries

artificial_dataset.add_collective_anomaly(series, start_idx, length=20, pattern='noise', magnitude=3.0, random_state=None)[source]#

Replace a subsequence with a collective anomaly pattern.

Return type:

SyntheticSeries

artificial_dataset.add_dropout(series, start_idx, duration=10, mode='flatline')[source]#

Simulate missing or frozen sensor signal.

Return type:

SyntheticSeries

artificial_dataset.add_level_shift(series, start_idx, shift_magnitude=(3.0, 5.0), duration=None, random_state=None)[source]#

Apply a step shift in the mean value.

Return type:

SyntheticSeries

artificial_dataset.add_point_anomalies(series, n_anomalies=5, magnitude=(3.0, 6.0), direction='both', avoid_existing=True, random_state=None)[source]#

Inject single-point spikes or dips.

Return type:

SyntheticSeries

artificial_dataset.add_seasonal_distortion(series, start_idx, duration=30, mode='stretch', factor=2.0)[source]#

Distort periodic pattern in a time series segment.

Applies distortion via stretching, compressing, damping, or phase shifting.

Return type:

SyntheticSeries

artificial_dataset.add_spike_anomalies(series, n_anomalies=5, spike_params=None, random_state=None)[source]#

Inject triangular positive spike events.

Return type:

SyntheticSeries

artificial_dataset.add_trend_change(series, start_idx, new_function_type, new_function_params=None, duration=None, continuity=True)[source]#

Replace a segment’s trend with a different known trend shape.

Simulates a concept-drift-style anomaly: over [start_idx, end_idx) the series stops following its original generative shape (e.g. sinusoidal) and instead follows new_function_type, evaluated with new_function_params, on the segment’s own time values. The new shape is computed by compose(), so any single component recognised there (with all of its own parameters) can be used as the anomalous trend.

Parameters:
  • series (SyntheticSeries) – The base series to modify.

  • start_idx (int) – Index (inclusive) where the trend change begins.

  • new_function_type (str) – Name of the replacement trend shape. One of: constant, linear, exponential, logarithmic, periodic_seasonal, polynomial, sinusoidal.

  • new_function_params (dict[str, Any] | None) – Keyword arguments forwarded to the chosen trend function (e.g. {"slope": 0.5, "intercept": 0.0} for "linear"). Defaults to that function’s own defaults when omitted.

  • duration (int | None) – Length of the affected segment. Defaults to the rest of the series.

  • continuity (bool) – If True, the new trend is vertically shifted so its first value matches the series value immediately before start_idx, avoiding an artificial level jump at the boundary while still exposing the change in shape/slope. If False, the new trend is used exactly as computed, which may introduce a visible jump.

Returns:

A new series with the segment’s trend replaced.

Return type:

SyntheticSeries

Raises:

ValueError – If new_function_type is not a recognised trend shape.

Examples

>>> from artificial_dataset.series import make_series
>>> base = make_series(
...     200, "sinusoidal", {"amplitude": 2.0, "frequency": 0.05}
... )
>>> anomalous = add_trend_change(
...     base, start_idx=100, new_function_type="linear",
...     new_function_params={"slope": 0.05}, duration=50,
... )
>>> bool(anomalous.is_anomaly[100:150].all())
True
artificial_dataset.add_variance_change(series, start_idx, duration=20, scale_factor=4.0, random_state=None)[source]#

Inject extra Gaussian noise variance into a segment.

Return type:

SyntheticSeries

artificial_dataset.anomaly_summary(series)[source]#

Return the audit log list stored in series.anomalies.

Return type:

list[dict[str, Any]]

artificial_dataset.compose(x, params)[source]#

Evaluate a superposition of signal components at x.

Recognised keys in params and their expected value types:

  • "constant" - dict of keyword arguments for constant()

  • "linear" - dict of keyword arguments for linear()

  • "exponential" - dict of keyword arguments for exponential()

  • "logarithmic" - dict of keyword arguments for logarithmic()

  • "periodic_seasonal" - dict of keyword arguments for :func:

    periodic_seasonal

  • "polynomial" - dict of keyword arguments for polynomial()

  • "sinusoidal" - dict of keyword arguments for sinusoidal()

Unknown keys are silently ignored so callers can attach metadata to the same dict without interfering with signal generation.

Parameters:
  • x (Tensor) – Input values, shape (n,).

  • params (dict[str, Any]) – Component specifications as described above.

Returns:

Superposed signal of the same shape as x.

Return type:

Tensor

artificial_dataset.compose_weighted(x, components)[source]#

Evaluate a weighted sum of several composed signals at x.

Unlike compose(), which sums at most one instance of each component type (dict keys must be unique), compose_weighted accepts a list, so the same component type can appear more than once — e.g. two sinusoids at different frequencies — each contributing with its own weight.

Parameters:
  • x (Tensor) – Input values, shape (n,).

  • components (list[dict[str, Any]]) – One entry per term in the sum. Each entry is a params dict as understood by compose() (e.g. {"sinusoidal": {...}}), plus an optional "weight" key (default 1.0) scaling that term.

Returns:

Superposed signal of the same shape as x.

Return type:

Tensor

Raises:

ValueError – If components is empty.

Examples

>>> import torch
>>> x = torch.linspace(0, 10, steps=50)
>>> y = compose_weighted(x, [
...     {"linear": {"slope": 0.2}},
...     {"sinusoidal": {"amplitude": 1.0, "frequency": 0.5}, "weight": 0.5},
... ])
>>> y.shape
torch.Size([50])
artificial_dataset.constant(x, value=1.0)[source]#

Compute a constant signal: y = value.

Parameters:
  • x (Tensor) – Input values, used only to determine the output shape.

  • value (float) – Constant value.

Returns:

Output tensor of the same shape as x, filled with value.

Return type:

Tensor

artificial_dataset.exponential(x, initial_value=1.0, growth_rate=0.05)[source]#

Compute an exponential signal: y = initial_value * exp(growth_rate * x).

Parameters:
  • x (Tensor) – Input values.

  • initial_value (float) – Value at x = 0.

  • growth_rate (float) – Exponential growth (positive) or decay (negative) rate.

Returns:

Output tensor of the same shape as x.

Return type:

Tensor

artificial_dataset.gaussian_noise(size, mean=0.0, std=1.0)[source]#

Generate a tensor of Gaussian noise.

Parameters:
  • size (tuple[int, ...]) – Shape of the output tensor.

  • mean (float) – Mean of the Gaussian distribution.

  • std (float) – Standard deviation of the Gaussian distribution.

Returns:

Noise tensor with the requested shape.

Return type:

Tensor

artificial_dataset.linear(x, slope=1.0, intercept=0.0)[source]#

Compute a linear signal: y = slope * x + intercept.

Parameters:
  • x (Tensor) – Input values.

  • slope (float) – Slope coefficient.

  • intercept (float) – Intercept (bias) term.

Returns:

Output tensor of the same shape as x.

Return type:

Tensor

artificial_dataset.logarithmic(x, scale=1.0, shift=1.0)[source]#

Compute a logarithmic signal: y = scale * log(x + shift).

Parameters:
  • x (Tensor) – Input values.

  • scale (float) – Multiplicative scale applied to the logarithm.

  • shift (float) – Additive shift applied to x before taking the log, keeping the argument positive when x starts at 0.

Returns:

Output tensor of the same shape as x.

Return type:

Tensor

Raises:

ValueError – If shift is not strictly positive.

artificial_dataset.make_anomaly_dataset(series_length=1000, noise_std=0.4, x_range=(0.0, 18.84955592153876), channel_params=None, spike_params=None, split=None, random_state=None)[source]#

Generate a single multivariate anomaly-detection time series.

Overloads:
  • series_length (int), noise_std (float), x_range (tuple[float, float]), channel_params (list[dict[str, Any]] | None), spike_params (SpikeParams | None), split (None), random_state (int | None) → AnomalyDataset

  • series_length (int), noise_std (float), x_range (tuple[float, float]), channel_params (list[dict[str, Any]] | None), spike_params (SpikeParams | None), split (tuple[float, float, float]), random_state (int | None) → AnomalySplits

Every channel shares a smooth baseline built from compose() plus Gaussian measurement noise. A random number of positive triangular spikes are then added: each spike event shares its centre and width across all channels, while every channel responds with its own random amplitude sampled from spike_params.

Parameters:
  • series_length (int) – Length T of the time series.

  • noise_std (float) – Standard deviation of the additive Gaussian measurement noise applied to every channel.

  • x_range (tuple[float, float]) – Closed interval [min, max] spanned by the shared time grid on which the baselines are evaluated.

  • channel_params (list[dict[str, Any]] | None) – One signal-parameter dict per channel, each passed to compose(). The number of entries sets the channel count m. When None, a single default sinusoidal channel is used.

  • spike_params (SpikeParams | None) – Configuration of the random positive anomaly spikes. When None, the defaults of SpikeParams are used.

  • split (tuple[float, float, float] | None) – When given, the (train, val, test) fractions used to split the series along the time axis from the beginning; the function then returns an AnomalySplits. When None, a single AnomalyDataset is returned.

  • random_state (int | None) – Seed passed to torch.manual_seed() for reproducibility.

Returns:

An AnomalyDataset when split is None, otherwise an AnomalySplits holding the three time segments.

Return type:

AnomalyDataset or AnomalySplits

Examples

>>> data = make_anomaly_dataset(series_length=200, random_state=0)
>>> data.y.shape, data.labels.shape, data.t.shape
(torch.Size([1, 200]), torch.Size([200]), torch.Size([200]))
>>> bool((data.labels[data.peak_indices] == 1).all())
True
>>> splits = make_anomaly_dataset(
...     series_length=200, split=(0.5, 0.25, 0.25), random_state=0
... )
>>> splits.train.y.shape[1], splits.val.y.shape[1], splits.test.y.shape[1]
(100, 50, 50)
artificial_dataset.make_classification(n_samples=1000, n_classes=2, noise_std=0.1, x_range=(0.0, 6.283185307179586), class_params=None, random_state=None)[source]#

Generate a classification dataset from signal components.

Each class is characterised by a unique combination of linear, polynomial, and sinusoidal components. Samples are produced by evaluating the class signal at uniformly drawn x values and adding Gaussian noise.

Parameters:
  • n_samples (int) – Total number of samples across all classes.

  • n_classes (int) – Number of distinct classes.

  • noise_std (float) – Standard deviation of the additive Gaussian noise applied to every sample.

  • x_range (tuple[float, float]) – Closed interval [min, max] from which input values are drawn uniformly.

  • class_params (list[dict[str, Any]] | None) – Per-class component configuration. Each entry is a params dict understood by compose(). Must have exactly n_classes entries when provided. When None, a sensible default configuration is used for up to two classes; for more classes parameters are generated automatically.

  • random_state (int | None) – Seed passed to torch.manual_seed() for reproducibility.

Return type:

tuple[Tensor, Tensor]

Returns:

  • X (torch.Tensor, shape (n_samples, 2)) – Feature matrix. Column 0 contains the sampled x values; column 1 contains the corresponding signal value (components + noise).

  • y (torch.Tensor, shape (n_samples,)) – Integer class labels in [0, n_classes), dtype torch.long.

Raises:

ValueError – If len(class_params) != n_classes.

Examples

>>> X, y = make_classification(n_samples=200, n_classes=2, random_state=0)
>>> X.shape, y.shape
(torch.Size([200, 2]), torch.Size([200]))
>>> y.unique().tolist()
[0, 1]
artificial_dataset.make_series(series_length, function_type, function_params=None, noise_std=0.0, random_state=None)[source]#

Generate a synthetic 1D time series using a single base function.

Parameters:
  • series_length (int) – Number of timesteps.

  • function_type (str) – One of: constant, linear_trend, sinusoidal, exponential, logarithmic, periodic_seasonal.

  • function_params (dict[str, Any] | None) – Function-specific parameters.

  • noise_std (float) – Standard deviation of additive Gaussian noise.

  • random_state (int | None) – Seed for reproducible noise.

Return type:

SyntheticSeries

artificial_dataset.periodic_seasonal(x, period=10.0, amplitude=1.0, offset=0.0, waveform='sine')[source]#

Compute a repeating pattern with an explicit period, in samples.

Unlike sinusoidal(), which is parameterised by frequency, this is parameterised by period and supports a few common waveform shapes.

Parameters:
  • x (Tensor) – Input values.

  • period (float) – Period of the pattern, in the same units as x.

  • amplitude (float) – Peak amplitude.

  • offset (float) – Constant vertical offset.

  • waveform (str) – One of "sine", "square", "triangle", "sawtooth".

Returns:

Output tensor of the same shape as x.

Return type:

Tensor

Raises:

ValueError – If period is not strictly positive, or waveform is unrecognised.

artificial_dataset.plot_series(series, title=None, ax=None, save_path=None)[source]#

Plot a SyntheticSeries: the 1D signal against time (a.u.), with anomalies marked.

Single anomalous points are drawn as scatter markers; contiguous anomalous runs (e.g. a level shift or dropout span) are additionally shaded to make their extent visible.

Builds and returns the figure without displaying it. In a notebook, the figure is shown automatically (either as the cell’s returned value, or, with %matplotlib inline, because it is still open at the end of cell execution); in a script, call plt.show() on the result if you want to display it.

Parameters:
  • series (SyntheticSeries) – The series to visualize (e.g. as returned by make_series or after one or more add_* injectors have been applied).

  • title (str | None) – Plot title. Defaults to the function_type recorded in series.meta, if present.

  • ax (Axes | None) – Axes to draw into. A new figure/axes is created when omitted.

Returns:

The figure containing the plot.

Return type:

Figure

artificial_dataset.polynomial(x, coefficients)[source]#

Compute a polynomial signal: y = sum(c_i * x^i).

Parameters:
  • x (Tensor) – Input values.

  • coefficients (list[float]) – Coefficients [c_0, c_1, ..., c_n] where c_i multiplies x**i.

Returns:

Output tensor of the same shape as x.

Return type:

Tensor

artificial_dataset.save_series(series, path)[source]#

Save a SyntheticSeries to a CSV file.

Writes one row per timestep with columns: x, y, label, anomaly_type. label is the binary classification target (1 = anomalous, matching series.label); anomaly_type is the pipe-delimited tag string for that timestep (empty string when not anomalous).

Parameters:
  • series (SyntheticSeries) – The series to export.

  • path (str | PathLike[str]) – Destination file path. Parent directories are not created automatically.

Return type:

None

artificial_dataset.sinusoidal(x, amplitude=1.0, frequency=1.0, phase=0.0)[source]#

Compute a sinusoidal signal: y = amplitude * sin(frequency * x + phase).

Parameters:
  • x (Tensor) – Input values (in radians when using default frequency).

  • amplitude (float) – Peak amplitude.

  • frequency (float) – Angular frequency in rad/unit.

  • phase (float) – Phase offset in radians.

Returns:

Output tensor of the same shape as x.

Return type:

Tensor