张宗平
0723b257fd
Root cause (found during 50k sample generation at seed 6975, surfacing as 'ValueError: Number of samples, -187, must be non-negative' from linspace in _inject_drift): _random_window's best-effort fallback after 50 failed non-overlap attempts drew start and end from TWO INDEPENDENT random integers, so end could be < start. The primary loop was consistent (start, start+length). Fix: reuse the same start in the fallback so end >= start always holds (overlap is still allowed in best-effort, just no inversion). - tests/test_synthesis.py: regression sweeping 300 seeds × 4 type combos, asserting every segment's end >= start. 146 tests passing.
149 lines
6.0 KiB
Python
149 lines
6.0 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)
|
|
|
|
|
|
# ─── regression: _random_window end >= start (found during 50k generation) ─
|
|
def test_inject_anomaly_end_ge_start_across_many_seeds():
|
|
"""Regression: _random_window fallback used to draw end independently,
|
|
producing end < start (ValueError in linspace). Sweep many seeds + types."""
|
|
import numpy as np
|
|
s0 = generate_series(T=512, C=5, seed=0)
|
|
for seed in range(300):
|
|
for types in (
|
|
["drift"], ["level_shift"], ["variance_change"],
|
|
["drift", "level_shift", "variance_change", "spike"],
|
|
):
|
|
s = s0.copy()
|
|
out, labels, segments = inject_anomaly(s, types=types, seed=seed)
|
|
for seg in segments:
|
|
assert seg["end"] >= seg["start"], (seed, types, seg)
|
|
assert out.shape == s.shape
|
|
assert labels.shape == (512,)
|