"""
Anomaly injectors for the base functions created with `make_series`.
Provides labeled anomaly injectors for SyntheticSeries instances.
All injectors accept a SyntheticSeries, perform a non-destructive copy,
and return a new SyntheticSeries with modified values and updated label metadata.
"""
from __future__ import annotations
import copy
from dataclasses import dataclass
from typing import Any
import torch
import torch.nn.functional as F
from artificial_dataset._components import compose
from artificial_dataset.series import SyntheticSeries
# Trend shapes recognised by `add_trend_change`, mirroring the keys understood
# by `artificial_dataset._components.compose`.
_KNOWN_TREND_TYPES = (
"constant",
"linear",
"exponential",
"logarithmic",
"periodic_seasonal",
"polynomial",
"sinusoidal",
)
# ---------- Internal Helpers ----------
def _copy_series(series: SyntheticSeries) -> SyntheticSeries:
"""Create a deep copy of a SyntheticSeries to guarantee immutability."""
return SyntheticSeries(
x=series.x.clone(),
y=series.y.clone(),
is_anomaly=series.is_anomaly.clone(),
anomaly_type=list(series.anomaly_type),
anomalies=copy.deepcopy(series.anomalies),
meta=copy.deepcopy(series.meta),
)
def _mark(series: SyntheticSeries, idx: torch.Tensor, tag: str) -> None:
"""Flag indices as anomalous and append tag to labels."""
series.is_anomaly[idx] = True
idx_list = idx.tolist() if isinstance(idx, torch.Tensor) else list(idx)
for i in idx_list:
curr = series.anomaly_type[i]
if curr == "":
series.anomaly_type[i] = tag
elif tag not in curr.split("|"):
series.anomaly_type[i] = f"{curr}|{tag}"
def _log(series: SyntheticSeries, entry: dict[str, Any]) -> None:
"""Append entry to the series anomaly audit log."""
series.anomalies.append(entry)
def _resolve_indices(
n: int,
length: int,
existing_mask: torch.Tensor | None,
avoid_existing: bool,
gen: torch.Generator,
) -> torch.Tensor:
"""Pick `n` distinct indices, optionally avoiding existing anomalies."""
if avoid_existing and existing_mask is not None:
pool = torch.where(~existing_mask)[0]
else:
pool = torch.arange(length)
if n > len(pool):
raise ValueError(
f"Requested {n} anomaly points, but only {len(pool)} available candidate \
indices."
)
perm = torch.randperm(len(pool), generator=gen)
chosen = pool[perm[:n]]
sorted_idx, _ = torch.sort(chosen)
return sorted_idx
# ---------- o ----------
# SpikeParams Dataclass
@dataclass(frozen=True)
class SpikeParams:
"""Configuration parameters for generating synthetic spike anomalies."""
amplitude_range: tuple[float, float] = (4.0, 7.0)
width_range: tuple[int, int] = (3, 6)
margin: int = 20
# ---------- o ----------
# Injectors
[docs]
def add_point_anomalies(
series: SyntheticSeries,
n_anomalies: int = 5,
magnitude: float | tuple[float, float] = (3.0, 6.0),
direction: str = "both",
avoid_existing: bool = True,
random_state: int | None = None,
) -> SyntheticSeries:
"""Inject single-point spikes or dips."""
series = _copy_series(series)
series_length = len(series)
gen = torch.Generator()
if random_state is not None:
gen.manual_seed(random_state)
idx = _resolve_indices(
n_anomalies, series_length, series.is_anomaly, avoid_existing, gen
)
y_std = torch.std(series.y).item() or 1.0
for i in idx:
if isinstance(magnitude, tuple):
r = torch.rand(1, generator=gen).item()
mag = magnitude[0] + (magnitude[1] - magnitude[0]) * r
else:
mag = magnitude
if direction == "up":
sign = 1.0
elif direction == "down":
sign = -1.0
else:
sign = 1.0 if torch.rand(1, generator=gen).item() > 0.5 else -1.0
series.y[i] += sign * mag * y_std
_mark(series, idx, "point")
_log(series, {"type": "point", "indices": idx.tolist()})
return series
[docs]
def add_spike_anomalies(
series: SyntheticSeries,
n_anomalies: int = 5,
spike_params: SpikeParams | None = None,
random_state: int | None = None,
) -> SyntheticSeries:
"""Inject triangular positive spike events."""
series = _copy_series(series)
params = spike_params or SpikeParams()
t_len = len(series)
gen = torch.Generator()
if random_state is not None:
gen.manual_seed(random_state)
centres = torch.randint(
params.margin,
max(params.margin + 1, t_len - params.margin),
(n_anomalies,),
generator=gen,
)
for centre in centres:
c = centre.item()
w = torch.randint(
params.width_range[0], params.width_range[1] + 1, (1,), generator=gen
).item()
idx = torch.arange(max(0, c - w), min(t_len, c + w + 1))
profile = 1.0 - torch.abs(idx.float() - float(c)) / (w + 1.0)
amp_min, amp_max = params.amplitude_range
amp = amp_min + (amp_max - amp_min) * torch.rand(1, generator=gen).item()
series.y[idx] += amp * profile
_mark(series, idx, "spike")
_log(series, {"type": "spike", "indices": centres.tolist()})
return series
[docs]
def add_collective_anomaly(
series: SyntheticSeries,
start_idx: int,
length: int = 20,
pattern: str = "noise",
magnitude: float = 3.0,
random_state: int | None = None,
) -> SyntheticSeries:
"""Replace a subsequence with a collective anomaly pattern."""
series = _copy_series(series)
end_idx = min(start_idx + length, len(series))
idx = torch.arange(start_idx, end_idx)
gen = torch.Generator()
if random_state is not None:
gen.manual_seed(random_state)
seg = series.y[idx].clone()
seg_mean = torch.mean(seg)
seg_std = torch.std(seg).item() or 1.0
if pattern == "noise":
series.y[idx] = seg_mean + torch.randn(len(idx), generator=gen) * (
magnitude * seg_std
)
elif pattern == "flat":
series.y[idx] = seg_mean
elif pattern == "reverse":
series.y[idx] = torch.flip(seg, dims=[0])
elif pattern == "scale":
series.y[idx] = seg_mean + magnitude * (seg - seg_mean)
elif pattern == "constant":
series.y[idx] = magnitude
else:
raise ValueError(
f"Unknown pattern '{pattern}'. Choose from: noise, flat, reverse, scale, \
constant."
)
_mark(series, idx, "collective")
_log(
series,
{
"type": "collective",
"start_idx": start_idx,
"end_idx": end_idx,
"pattern": pattern,
},
)
return series
[docs]
def add_level_shift(
series: SyntheticSeries,
start_idx: int,
shift_magnitude: float | tuple[float, float] = (3.0, 5.0),
duration: int | None = None,
random_state: int | None = None,
) -> SyntheticSeries:
"""Apply a step shift in the mean value."""
series = _copy_series(series)
end_idx = (
len(series) if duration is None else min(start_idx + duration, len(series))
)
idx = torch.arange(start_idx, end_idx)
gen = torch.Generator()
if random_state is not None:
gen.manual_seed(random_state)
if isinstance(shift_magnitude, tuple):
mag = (
shift_magnitude[0]
+ (shift_magnitude[1] - shift_magnitude[0])
* torch.rand(1, generator=gen).item()
)
else:
mag = shift_magnitude
sign = 1.0 if torch.rand(1, generator=gen).item() > 0.5 else -1.0
y_std = torch.std(series.y).item() or 1.0
series.y[idx] += sign * mag * y_std
_mark(series, idx, "level_shift")
_log(series, {"type": "level_shift", "start_idx": start_idx, "end_idx": end_idx})
return series
[docs]
def add_trend_change(
series: SyntheticSeries,
start_idx: int,
new_function_type: str, # "constant" | "linear" | "exponential" |
# "logarithmic" | "periodic_seasonal" |
# "polynomial" | "sinusoidal"
new_function_params: dict[str, Any] | None = None,
duration: int | None = None,
continuity: bool = True,
) -> SyntheticSeries:
"""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 :func:`~artificial_dataset._components.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, optional
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, optional
Length of the affected segment. Defaults to the rest of the series.
continuity : bool, default True
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
-------
SyntheticSeries
A new series with the segment's trend replaced.
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
"""
if new_function_type not in _KNOWN_TREND_TYPES:
raise ValueError(
f"Unknown new_function_type '{new_function_type}'. Choose from: "
f"{', '.join(_KNOWN_TREND_TYPES)}."
)
series = _copy_series(series)
series_length = len(series)
end_idx = (
series_length if duration is None else min(start_idx + duration, series_length)
)
idx = torch.arange(start_idx, end_idx)
params = {new_function_type: new_function_params or {}}
new_values = compose(series.x[idx], params)
if continuity and start_idx > 0:
offset = series.y[start_idx - 1] - new_values[0]
new_values = new_values + offset
series.y[idx] = new_values
_mark(series, idx, "trend_change")
_log(
series,
{
"type": "trend_change",
"start_idx": start_idx,
"end_idx": end_idx,
"new_function_type": new_function_type,
"new_function_params": new_function_params or {},
"continuity": continuity,
},
)
return series
[docs]
def add_variance_change(
series: SyntheticSeries,
start_idx: int,
duration: int = 20,
scale_factor: float = 4.0,
random_state: int | None = None,
) -> SyntheticSeries:
"""Inject extra Gaussian noise variance into a segment."""
series = _copy_series(series)
end_idx = min(start_idx + duration, len(series))
idx = torch.arange(start_idx, end_idx)
gen = torch.Generator()
if random_state is not None:
gen.manual_seed(random_state)
y_std = torch.std(series.y).item() or 1.0
noise = torch.randn(len(idx), generator=gen) * (scale_factor * y_std)
series.y[idx] += noise
_mark(series, idx, "variance_change")
_log(
series,
{
"type": "variance_change",
"start_idx": start_idx,
"end_idx": end_idx,
"scale_factor": scale_factor,
},
)
return series
[docs]
def add_dropout(
series: SyntheticSeries,
start_idx: int,
duration: int = 10,
mode: str = "flatline",
) -> SyntheticSeries:
"""Simulate missing or frozen sensor signal."""
series = _copy_series(series)
end_idx = min(start_idx + duration, len(series))
idx = torch.arange(start_idx, end_idx)
if mode == "flatline":
fill_val = series.y[max(start_idx - 1, 0)].item()
series.y[idx] = fill_val
elif mode == "zero":
series.y[idx] = 0.0
elif mode == "nan":
series.y[idx] = float("nan")
else:
raise ValueError("mode must be one of: flatline, zero, nan.")
_mark(series, idx, "dropout")
_log(
series,
{"type": "dropout", "start_idx": start_idx, "end_idx": end_idx, "mode": mode},
)
return series
[docs]
def add_seasonal_distortion(
series: SyntheticSeries,
start_idx: int,
duration: int = 30,
mode: str = "stretch",
factor: float = 2.0,
) -> SyntheticSeries:
"""Distort periodic pattern in a time series segment.
Applies distortion via stretching, compressing, damping, or phase shifting.
"""
series = _copy_series(series)
end_idx = min(start_idx + duration, len(series))
idx = torch.arange(start_idx, end_idx)
seg = series.y[idx].clone()
src_len = len(seg)
if mode in ("stretch", "compress"):
warped_len = max(
2, round(src_len * factor if mode == "stretch" else src_len / factor)
)
seg_reshaped = seg.view(1, 1, -1)
# Resample onto new grid, then interpolate back to original segment length
warped = F.interpolate(
seg_reshaped, size=warped_len, mode="linear", align_corners=True
)
resampled = F.interpolate(
warped, size=src_len, mode="linear", align_corners=True
)
series.y[idx] = resampled.squeeze()
elif mode == "damp":
seg_mean = torch.mean(seg)
series.y[idx] = seg_mean + (seg - seg_mean) / factor
elif mode == "phase_shift":
shift_val = round(factor)
series.y[idx] = torch.roll(seg, shifts=shift_val)
else:
raise ValueError("mode must be one of: stretch, compress, damp, phase_shift.")
_mark(series, idx, "seasonal_distortion")
_log(
series,
{
"type": "seasonal_distortion",
"start_idx": start_idx,
"end_idx": end_idx,
"mode": mode,
"factor": factor,
},
)
return series
# ---------- o ----------
# Summary
[docs]
def anomaly_summary(series: SyntheticSeries) -> list[dict[str, Any]]:
"""Return the audit log list stored in series.anomalies."""
return series.anomalies