张宗平
e2c0756a78
- generate_series: baseline + trend + seasonality + noise (opt heavy-tail)
- inject_anomaly: spike / level_shift / variance_change / missing_segment / drift
* non-overlapping segments (TDD caught an overlap bug) -> labels exactly
match contiguous [start,end) runs
* returns (series, point labels, segment metadata)
- couple_event: persistent per-channel step at t + event text
- add_missing: NaN drop at given rate
- tests/test_synthesis.py (14 tests, RED->GREEN; fixed overlap regression)
Exit criteria: seed-deterministic; labels cover segments; event step persistent.
130 lines
5.2 KiB
Python
130 lines
5.2 KiB
Python
"""Tests for time-series synthesis (T1.3)."""
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from tsmm.data.synthesis import (
|
|
generate_series,
|
|
inject_anomaly,
|
|
couple_event,
|
|
add_missing,
|
|
)
|
|
|
|
|
|
# ─── generate_series ───────────────────────────────────────────────────────
|
|
def test_generate_series_shape():
|
|
s = generate_series(T=256, C=3, seed=0)
|
|
assert s.shape == (256, 3)
|
|
assert np.isfinite(s).all(), "baseline series must be finite (no NaN before inject)"
|
|
|
|
|
|
def test_generate_series_reproducible():
|
|
a = generate_series(T=128, C=2, seed=7)
|
|
b = generate_series(T=128, C=2, seed=7)
|
|
assert np.array_equal(a, b)
|
|
|
|
|
|
def test_generate_series_different_seed_differs():
|
|
a = generate_series(T=128, C=2, seed=1)
|
|
b = generate_series(T=128, C=2, seed=2)
|
|
assert not np.array_equal(a, b)
|
|
|
|
|
|
def test_generate_series_has_structure():
|
|
# baseline + trend + seasonality + noise: variance should be > 0 and the
|
|
# series should not be constant.
|
|
s = generate_series(T=512, C=2, seed=3)
|
|
assert np.std(s) > 0
|
|
assert np.ptp(s) > 0 # peak-to-peak
|
|
|
|
|
|
# ─── inject_anomaly ────────────────────────────────────────────────────────
|
|
def test_inject_anomaly_returns_series_and_labels_and_segments():
|
|
s = generate_series(T=256, C=2, seed=0)
|
|
out, labels, segments = inject_anomaly(s, types=["spike"], seed=0)
|
|
assert out.shape == s.shape
|
|
assert labels.shape == (256,)
|
|
assert set(np.unique(labels)).issubset({0, 1})
|
|
assert isinstance(segments, list)
|
|
|
|
|
|
def test_inject_anomaly_labels_cover_segments():
|
|
s = generate_series(T=256, C=2, seed=0)
|
|
out, labels, segments = inject_anomaly(s, types=["level_shift", "variance_change"], seed=1)
|
|
# Each segment must mark a contiguous run of label==1 exactly matching [start,end)
|
|
for seg in segments:
|
|
st, en = seg["start"], seg["end"]
|
|
assert 0 <= st < en <= len(labels)
|
|
assert labels[st:en].all(), f"segment {seg} not fully labeled"
|
|
# immediately outside (if exists) must be 0
|
|
if st > 0:
|
|
assert labels[st - 1] == 0
|
|
if en < len(labels):
|
|
assert labels[en] == 0
|
|
|
|
|
|
def test_inject_anomaly_changes_the_series_at_segments():
|
|
s = generate_series(T=256, C=2, seed=0)
|
|
out, labels, segments = inject_anomaly(s, types=["spike"], seed=0)
|
|
assert np.any(labels == 1), "expected at least one anomaly segment"
|
|
# at least somewhere the injected series differs from baseline
|
|
assert not np.allclose(out, s)
|
|
|
|
|
|
def test_inject_anomaly_each_supported_type():
|
|
s = generate_series(T=256, C=2, seed=0)
|
|
for t in ["spike", "level_shift", "variance_change", "missing_segment", "drift"]:
|
|
out, labels, segments = inject_anomaly(s, types=[t], seed=0)
|
|
# missing_segment may place NaN, so use labels to assert effect
|
|
assert np.any(labels == 1), f"type {t} produced no anomaly segment"
|
|
|
|
|
|
def test_inject_anomaly_missing_segment_inserts_nan():
|
|
s = generate_series(T=256, C=2, seed=0)
|
|
out, labels, segments = inject_anomaly(s, types=["missing_segment"], seed=0)
|
|
assert np.isnan(out).any(), "missing_segment must inject NaN values"
|
|
|
|
|
|
def test_inject_anomaly_reproducible():
|
|
s = generate_series(T=256, C=2, seed=0)
|
|
o1, l1, _ = inject_anomaly(s, types=["spike", "level_shift"], seed=5)
|
|
o2, l2, _ = inject_anomaly(s, types=["spike", "level_shift"], seed=5)
|
|
assert np.array_equal(o1, o2) and np.array_equal(l1, l2)
|
|
|
|
|
|
# ─── couple_event ──────────────────────────────────────────────────────────
|
|
def test_couple_event_introduces_step_and_returns_text():
|
|
s = generate_series(T=256, C=2, seed=0)
|
|
before = s.copy()
|
|
out, text = couple_event(s, t=128, kind="deploy", seed=0)
|
|
assert isinstance(text, str) and len(text) > 0
|
|
# a step change near t should alter the series magnitude after t
|
|
assert not np.allclose(out[130:], before[130:])
|
|
# immediately before t the series is unchanged
|
|
assert np.allclose(out[:128], before[:128])
|
|
|
|
|
|
def test_couple_event_step_has_persistent_offset():
|
|
s = generate_series(T=256, C=1, seed=0)
|
|
out, _ = couple_event(s, t=100, kind="incident", seed=1)
|
|
# mean of post-event window differs from pre-event baseline
|
|
pre = out[:100].mean()
|
|
post = out[120:].mean()
|
|
assert abs(post - pre) > 1e-6
|
|
|
|
|
|
# ─── add_missing ───────────────────────────────────────────────────────────
|
|
def test_add_missing_introduces_nan_within_rate():
|
|
s = generate_series(T=512, C=2, seed=0)
|
|
out = add_missing(s, rate=0.1, seed=0)
|
|
assert out.shape == s.shape
|
|
frac = np.isnan(out).mean()
|
|
# within reasonable tolerance of requested rate
|
|
assert 0.0 < frac <= 0.15
|
|
|
|
|
|
def test_add_missing_reproducible():
|
|
s = generate_series(T=256, C=2, seed=0)
|
|
a = add_missing(s, rate=0.1, seed=3)
|
|
b = add_missing(s, rate=0.1, seed=3)
|
|
assert np.array_equal(a, b, equal_nan=True)
|