Anomaly Injection

Anomalies are perturbations applied to metrics after the base mathematical trends have been computed. This decoupled design is critical: it keeps the clean baseline mathematical trend completely separate from the failure events.

This makes it exceptionally easy to benchmark anomaly detection models or test systems under failure conditions, as you can compare clean baselines with contaminated streams to get perfect ground-truth labels.

🛑 Available Anomaly Types

All anomaly classes are located in ts_data_generator.anomalies (and re-exported at the top level of ts_data_generator). They are applied stochastically based on your configuration.

1. PointAnomaly

Injects isolated spikes or drops in the metric value at individual timestamps.

  • probability (float): Probability in $[0, 1]$ of an anomaly occurring at any given timestamp (default 0.01).
  • mode ("additive" or "replacement"):
    • "additive": Adds the magnitude to the trend’s value ($y_{new} = y_{base} + \text{magnitude}$). Useful for simulating power spikes or network surges.
    • "replacement": Overwrites the trend’s value with the magnitude ($y_{new} = \text{magnitude}$). Useful for simulating dead sensors locking to a specific extreme value.
  • magnitude (float or tuple[float, float]): A fixed scalar magnitude, or a (min, max) range tuple. If a tuple is provided, values are sampled uniformly.
# API:
from ts_data_generator import PointAnomaly

# 1. Additive spikes of random height
random_spikes = PointAnomaly(probability=0.02, mode="additive", magnitude=(50.0, 100.0))

# 2. Replacement spikes locking to exactly 999.0
dead_locks = PointAnomaly(probability=0.01, mode="replacement", magnitude=999.0)

# CLI Shorthand:
PointAnomaly(probability=0.02,mode='additive',magnitude=(50,100))

2. MissingData

Simulates sensor failures, transmission drops, or periodic maintenance outages by injecting NaN values.

  • mode ("random", "burst", or "patterned"):
    • "random": Every individual timestamp has an independent probability of dropping out.
    • "burst": Simulates multi-interval network dropouts. Once a dropout starts, it holds NaN for a contiguous block of steps.
    • "patterned": Gaps are injected on a strict schedule. (API only - requires a callable).
  • probability (float): Probability for "random" mode (default 0.01).
  • burst_probability (float): Probability of a burst gap starting at any timestamp (default 0.02).
  • min_length (int): Minimum duration (in steps) of a burst gap (default 2).
  • max_length (int): Maximum duration (in steps) of a burst gap (default 5).
  • schedule (Callable[[pd.Timestamp], bool]): A function that takes a timestamp and returns True if that timestamp should be dropped. Required for "patterned" mode.
# API (Stochastic drops):
from ts_data_generator import MissingData

# Random 5% individual dropouts
random_drops = MissingData(mode="random", probability=0.05)

# Network burst drops lasting between 5 and 20 steps
network_outage = MissingData(mode="burst", burst_probability=0.01, min_length=5, max_length=20)

# API (Schedule-based patterned drops):
# Simulate weekly maintenance drops every Sunday
sunday_maintenance = MissingData(
    mode="patterned",
    schedule=lambda ts: ts.weekday() == 6
)

# CLI Shorthand (Random & Burst only):
MissingData(mode='burst',burst_probability=0.01,min_length=5,max_length=20)

3. ConceptDrift

Simulates gradual distribution-level shifts in metrics (e.g. baseline shifts due to device degradation, seasonal HVAC changes, or system recalibrations).

Drift is defined by passing an ordered list of DriftSegment objects to the ConceptDrift anomaly injector.

DriftSegment Parameters:

  • start_timestamp (str or pd.Timestamp): The absolute timestamp where the drift transition starts.
  • transition_window (float): The duration in seconds to transition gradually from the old baseline to the drifted baseline (default 1800 = 30 mins). A linear interpolation is used.
  • target_mean (float): The mean value of the new drifted distribution (default 0.0).
  • target_std (float): The standard deviation of the new drifted distribution (default 1.0).
  • hold_duration (float): The duration in seconds to hold/remain in the drifted regime (default 7200 = 2 hours).
  • restore (bool): If True, the pipeline will gradually transition back to the original baseline distribution over another transition_window immediately after the hold duration (default False).

The Drift Segment Timeline: 

  [Clean Baseline]
         |
  [start_timestamp]
         |  \   <-- transition_window (Linear interpolation to target distribution)
         |   \
         +----+ <-- hold_duration (Gaussian distribution: target_mean, target_std)
               \
                \ <-- transition_window (Optional restore back to baseline)
                 |
          [Clean Baseline]

# API (Overheating Event with Restore):
from ts_data_generator import ConceptDrift, DriftSegment

overheat = ConceptDrift(segments=[
    DriftSegment(
        start_timestamp="2024-01-05T12:00:00",
        transition_window=3600,  # 1 hour to ramp up
        target_mean=95.0,         # Shifts average to 95 degrees
        target_std=2.0,           # With tight volatility
        hold_duration=14400,     # Overheats for 4 hours
        restore=True             # Gradually cools back down to normal
    )
])

# CLI Shorthand:
ConceptDrift(segments=[DriftSegment(start_timestamp='2024-01-05T12:00:00',transition_window=3600,target_mean=95,target_std=2,hold_duration=14400,restore=True)])

4. PeriodicBurstAnomaly (with EventSchedule)

Simulates recurring, duty-cycled failures — RF jammers, scheduled batch jobs, intermittent interference — where the effect spikes at onset and decays back toward baseline rather than switching cleanly on/off, and where several metrics must react on the exact same timestamps.

Unlike PointAnomaly (memoryless, one timestamp at a time) or ConceptDrift (a single one-shot regime shift), PeriodicBurstAnomaly models a repeating renewal process: bursts fire roughly every period samples, each lasting duration samples, with an occasional skipped cycle to keep it from looking too mechanical.

  • schedule (EventSchedule): A shared, precomputed burst timetable. Build one EventSchedule and pass it to several PeriodicBurstAnomaly instances (one per metric) so every metric spikes/recovers on identical windows. Mutually exclusive with period.
  • period (int): Shorthand for standalone (single-metric) use — builds an internal EventSchedule from period/duration/skip_probability/max_skip_periods/jitter/phase. Mutually exclusive with schedule.
  • peak_magnitude (float or tuple): Value injected at burst onset; a tuple samples uniformly per burst.
  • decay (float): Exponential decay rate per sample from onset (0 = constant for the whole burst, then a hard drop).
  • tail_length (int): Extra samples appended after the flagged burst window over which the effect keeps decaying — this is what produces the “spike then fades out” shape seen in real BLER/SNR recovery curves.
  • mode ("additive" or "replacement"): Same semantics as PointAnomaly.

EventSchedule parameters

  • period (int): Nominal samples between burst starts.
  • duration (int or tuple): Burst length in samples, fixed or sampled uniformly per burst.
  • skip_probability (float): Chance that a scheduled burst’s next gap stretches to 2..max_skip_periods periods instead of exactly one — models a device that occasionally misses a cycle.
  • max_skip_periods (int): Upper bound on how many periods a skipped gap can span.
  • jitter (int): Max random +/- sample offset applied to each burst’s start.
  • phase (int): Sample index of the first burst.
# API: correlated jammer-style bursts across multiple metrics
from ts_data_generator import (
    ARNoiseTrend,
    DataGenerator,
    EventSchedule,
    LinearTrend,
    PeriodicBurstAnomaly,
)

dg = DataGenerator(seed=42)
dg.start_datetime = "2024-01-01"
dg.end_datetime = "2024-01-01 00:30:00"
dg.to_granularity("100ms")

# One schedule shared by every affected metric — same on/off windows everywhere.
jammer = EventSchedule(period=100, duration=(15, 20), phase=250, skip_probability=0.03)

dg.add_metric(
    "DL_BLER",
    trends=[LinearTrend(offset=0.01, slope=0.0), ARNoiseTrend(decay=0.3, noise_std=0.002)],
    anomalies=[PeriodicBurstAnomaly(schedule=jammer, peak_magnitude=(0.03, 0.08), decay=0.3, tail_length=12)],
)
dg.add_metric(
    "UL_SNR",
    trends=LinearTrend(offset=21.0, slope=0.0),
    anomalies=[PeriodicBurstAnomaly(schedule=jammer, peak_magnitude=(-4.0, -1.0), decay=0.3, tail_length=12)],
)
# decay=0, tail_length=0, replacement -> a clean 0/1 event flag column
dg.add_metric(
    "Jamming",
    trends=LinearTrend(offset=0.0, slope=0.0),
    anomalies=[PeriodicBurstAnomaly(schedule=jammer, peak_magnitude=1.0, decay=0.0, mode="replacement")],
)

# CLI Shorthand (standalone, single metric — use the Python API to share a schedule across metrics):
PeriodicBurstAnomaly(period=100,duration=(15,20),skip_probability=0.03,peak_magnitude=(0.03,0.08),decay=0.3,tail_length=12)

🥞 Stacking Multiple Anomalies

You can apply multiple anomalies to the same metric. The generation engine applies them sequentially in the order they are listed in your Python array or terminal command.

Here is a fully runnable Python API script demonstrating trend composition and stacked anomalies (spikes + dropouts):

from ts_data_generator import (
    DataGenerator,
    LinearTrend,
    MissingData,
    PointAnomaly,
    SinusoidalTrend,
)

# 1. Setup Generator
dg = DataGenerator(seed=999)
dg.start_datetime = "2024-01-01"
dg.end_datetime = "2024-01-02"
dg.to_granularity("5min")

# 2. Setup Base Trends (Baseline CPU)
cpu_trends = [
    LinearTrend(offset=30.0, slope=15.0), # Creeping baseline load
    SinusoidalTrend(amplitude=10.0, freq=1.0, noise_level=2.0) # Daily cycles
]

# 3. Setup Anomalies to stack
spikes = PointAnomaly(probability=0.05, mode="additive", magnitude=(10.0, 15.0))
outages = MissingData(mode="burst", burst_probability=0.01, min_length=2, max_length=4)


# 4. Add Metric (providing both trends and the ordered anomalies list)
dg.add_metric(
    name="cpu_usage",
    trends=cpu_trends,
    anomalies=[spikes, outages]
)

# 5. Plot
dg.plot()

Output: anomaly_composition

Stacking via the CLI

In the CLI, use the + operator to chain anomaly classes:

tsdata generate \
  --start 2024-01-01 --end 2024-01-05 --granularity h \
  --mets "cpu_usage:LinearTrend(offset=30,slope=25)+SinusoidalTrend(amplitude=10,freq=1)" \
  --anomalies "cpu_usage:PointAnomaly(probability=0.015,magnitude=(30,50))+MissingData(mode=burst,burst_probability=0.01)" \
  --output cpu_telemetry.csv

🏷️ Auto-Labeled Output for ML

Any metric that has anomalies configured automatically emits a boolean ground-truth label column alongside the metric. This turns your generated data into a ready-to-train labeled-anomaly dataset: you can compare the clean baseline against the contaminated stream to get perfect, point-level labels for benchmarking anomaly-detection models — no manual labeling required.

For a metric named cpu_usage, the generator adds a cpu_usage_anomaly boolean column where a True marks a data point that an anomaly actually changed. A point is labeled True when the contaminated signal differs from the clean baseline. The comparison is NaN-aware, so MissingData gaps (injected NaNs) are flagged True just like PointAnomaly spikes.

Key behaviors:

  • Naming: The label column is always <metric_name>_anomaly (e.g. cpu_usage → cpu_usage_anomaly).
  • All anomaly types covered: PointAnomaly (additive and replacement), MissingData (random/burst/patterned), and ConceptDrift. For drift, the entire drift window — transition in, hold, and restore — is labeled True, because every point in that regime deviates from the clean baseline.
  • Appears only when needed: The label column is created only for metrics that actually have anomalies configured. Metrics with no anomalies get no label column.
  • Safe with normalize & plot: The column is bool dtype, so normalize() and plot() — which only operate on numeric columns — skip it automatically. The labels are never rescaled or plotted.
  • Survives aggregation: When you call dg.aggregate(...), the label column is aggregated as a boolean OR (max): a coarser resample window is labeled True if any point within it was anomalous.
  • Automatic in CSV/CLI/presets: No extra flag or config is needed. Every tsdata generate command (and every preset) that declares --anomalies automatically gains the matching _anomaly columns in its output CSV.
from ts_data_generator import DataGenerator, LinearTrend, MissingData, PointAnomaly

dg = DataGenerator(seed=42)
dg.start_datetime = "2024-01-01"
dg.end_datetime = "2024-01-02"
dg.to_granularity("5min")

dg.add_metric(
    name="cpu_usage",
    trends=LinearTrend(offset=30.0, slope=0.1),
    anomalies=[
        PointAnomaly(probability=0.05, mode="additive", magnitude=(10.0, 15.0)),
        MissingData(mode="burst", burst_probability=0.01, min_length=2, max_length=4),
    ],
)

# dg.data now includes a boolean `cpu_usage_anomaly` column — your ground-truth labels
print(dg.data[["cpu_usage", "cpu_usage_anomaly"]].head())
print(f"{dg.data['cpu_usage_anomaly'].sum()} labeled anomalies")

# CLI: the cpu_usage_anomaly column is added to the CSV automatically — no extra flag needed
tsdata generate --anomalies "cpu_usage:PointAnomaly(probability=0.05,magnitude=(10,15))" ...
This site uses Just the Docs, a documentation theme for Jekyll.