feat(m1): synthesis — components + anomaly/event/missing (T1.3)

- 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.
2026-06-29 23:25:08 +08:00
parent 6d10454484
commit e2c0756a78
4 changed files with 1288 additions and 1 deletions
@@ -0,0 +1,328 @@
+"""Component model + time-series synthesis (T1.3).
+
+Synthesizes multivariate time series as a superposition of interpretable
+**components** (baseline + trend + seasonality + noise) and supports controlled
+anomaly / event injection for supervised data generation.
+
+All functions are deterministic given the same seed.
+
+Conventions
+-----------
+* Series shape: ``(T, C)`` — ``T`` timesteps, ``C`` channels.
+* ``labels`` are point-wise ``0/1`` of shape ``(T,)`` (an anomaly covers all
+  channels within a segment's time window).
+* Segments are dicts with at least ``{"type", "start", "end"}`` describing the
+  half-open time interval ``[start, end)`` that is labeled anomalous.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import numpy as np
+
+# ─── baseline series generation ────────────────────────────────────────────
+
+
+def generate_series(
+    T: int,
+    C: int,
+    seed: int | None = None,
+    *,
+    baseline_range: tuple[float, float] = (20.0, 80.0),
+    trend_strength: float = 0.02,
+    seasonal_periods: tuple[int, ...] = (16, 64),
+    noise_std: float = 1.0,
+    student_t: bool = False,
+) -> np.ndarray:
+    """Generate a multivariate series as baseline + trend + seasonality + noise.
+
+    Parameters
+    ----------
+    T, C:
+        Length and number of channels.
+    seed:
+        RNG seed.
+    baseline_range:
+        Per-channel baseline level sampled uniformly from this range.
+    trend_strength:
+        Max slope magnitude of a linear trend per channel.
+    seasonal_periods:
+        Periods (in timesteps) of additive sinusoidal components.
+    noise_std:
+        Std of additive Gaussian noise. If ``student_t`` additionally draws
+        heavy-tailed noise (Student-t via Gaussian mixture) to mimic outliers.
+    student_t:
+        Use heavy-tailed noise (heavier tails than Gaussian).
+    """
+    rng = np.random.default_rng(seed)
+    t = np.arange(T, dtype=np.float64)
+
+    # per-channel baseline level
+    levels = rng.uniform(*baseline_range, size=C)
+    series = np.broadcast_to(levels, (T, C)).astype(np.float64).copy()
+
+    # per-channel linear trend
+    slopes = rng.uniform(-trend_strength, trend_strength, size=C)
+    series += slopes[None, :] * t[:, None]
+
+    # additive seasonality: each channel picks amplitude + phase per period
+    for period in seasonal_periods:
+        amps = rng.uniform(0.5, 3.0, size=C)
+        phases = rng.uniform(0, 2 * np.pi, size=C)
+        series += amps[None, :] * np.sin(2 * np.pi * t[:, None] / period + phases[None, :])
+
+    # noise
+    noise = rng.normal(0.0, noise_std, size=(T, C))
+    if student_t:
+        # heavy tail: occasionally inflate a few points
+        mask = rng.random((T, C)) < 0.02
+        noise = np.where(mask, noise * rng.uniform(5, 10, size=(T, C)), noise)
+    series += noise
+    return series
+
+
+# ─── anomaly injection ─────────────────────────────────────────────────────
+
+# Each injector mutates a series copy in-place over [start, end) and returns
+# the segment metadata dict. They share a label array filled by the caller.
+
+_MIN_SEG_LEN = 8  # minimum segment length to keep labels meaningful
+
+
+def _random_window(rng: np.random.Generator, T: int, length: int, occupied=None) -> tuple[int, int]:
+    """Draw a half-open window ``[start, end)`` of ~``length`` that does not
+    overlap any interval in ``occupied`` (list of ``(start, end)``).
+
+    Falls back to allowing overlap if no free window exists after several tries.
+    """
+    occupied = occupied or []
+    length = min(max(length, _MIN_SEG_LEN), T)
+    for _ in range(50):
+        start = int(rng.integers(0, T - length + 1))
+        cand = (start, start + length)
+        if all(cand[1] <= s or cand[0] >= e for (s, e) in occupied):
+            return cand
+    # could not find a non-overlapping window; return best effort
+    return int(rng.integers(0, T - length + 1)), int(rng.integers(0, T - length + 1)) + length
+
+
+def _inject_spike(rng, series, T, C, occupied):
+    for _ in range(50):
+        start = int(rng.integers(0, T))
+        end = min(start + int(rng.integers(1, 4)), T)  # 1-3 points
+        if all(end <= s or start >= e for (s, e) in occupied):
+            break
+    mag = rng.uniform(6, 12, size=C)
+    series[start:end] += mag[None, :]
+    return {"type": "spike", "start": start, "end": end, "magnitude": mag.tolist()}
+
+
+def _inject_level_shift(rng, series, T, C, occupied):
+    start, end = _random_window(rng, T, length=int(rng.integers(40, 90)), occupied=occupied)
+    offset = rng.uniform(10, 30, size=C) * rng.choice([-1, 1], size=C)
+    series[start:end] += offset[None, :]
+    return {
+        "type": "level_shift",
+        "start": start,
+        "end": end,
+        "offset": offset.tolist(),
+    }
+
+
+def _inject_variance_change(rng, series, T, C, occupied):
+    start, end = _random_window(rng, T, length=int(rng.integers(40, 90)), occupied=occupied)
+    factor = rng.uniform(3, 6)
+    local_std = np.std(series[start:end], axis=0, keepdims=True)
+    # re-draw noise scaled up within the window, centered on the local mean
+    mean = np.mean(series[start:end], axis=0, keepdims=True)
+    series[start:end] = mean + rng.normal(0, 1, size=(end - start, C)) * (local_std * factor)
+    return {
+        "type": "variance_change",
+        "start": start,
+        "end": end,
+        "factor": float(factor),
+    }
+
+
+def _inject_missing_segment(rng, series, T, C, occupied):
+    start, end = _random_window(rng, T, length=int(rng.integers(20, 50)), occupied=occupied)
+    series[start:end] = np.nan
+    return {"type": "missing_segment", "start": start, "end": end}
+
+
+def _inject_drift(rng, series, T, C, occupied):
+    start, end = _random_window(rng, T, length=int(rng.integers(60, 120)), occupied=occupied)
+    # gradual ramp added across the segment
+    ramp = np.linspace(0, 1, end - start)[:, None]
+    slope = rng.uniform(8, 20, size=C) * rng.choice([-1, 1], size=C)
+    series[start:end] += slope[None, :] * ramp
+    return {
+        "type": "drift",
+        "start": start,
+        "end": end,
+        "slope": slope.tolist(),
+    }
+
+
+_INJECTORS = {
+    "spike": _inject_spike,
+    "level_shift": _inject_level_shift,
+    "variance_change": _inject_variance_change,
+    "missing_segment": _inject_missing_segment,
+    "drift": _inject_drift,
+}
+
+
+def inject_anomaly(
+    series: np.ndarray,
+    types: list[str] | tuple[str, ...] | str = "spike",
+    *,
+    seed: int | None = None,
+) -> tuple[np.ndarray, np.ndarray, list[dict[str, Any]]]:
+    """Inject one or more anomaly segments into a *copy* of ``series``.
+
+    Parameters
+    ----------
+    series:
+        Array of shape ``(T, C)``.
+    types:
+        One anomaly type or a list of types. Each entry produces one segment.
+        Supported: ``spike``, ``level_shift``, ``variance_change``,
+        ``missing_segment``, ``drift``.
+    seed:
+        RNG seed.
+
+    Returns
+    -------
+    out:
+        Mutated copy of ``series`` (NaN where ``missing_segment`` applied).
+    labels:
+        Point-wise ``0/1`` array of shape ``(T,)`` — ``1`` inside any segment.
+    segments:
+        List of metadata dicts; each has at least ``type``, ``start``, ``end``.
+    """
+    if isinstance(types, str):
+        types = [types]
+    unknown = [t for t in types if t not in _INJECTORS]
+    if unknown:
+        raise ValueError(f"unknown anomaly type(s): {unknown}")
+
+    rng = np.random.default_rng(seed)
+    out = np.array(series, dtype=np.float64, copy=True)
+    T = out.shape[0]
+    C = out.shape[1] if out.ndim == 2 else 1
+    if out.ndim == 1:  # tolerate 1-D input gracefully
+        out = out[:, None]
+
+    labels = np.zeros(T, dtype=np.int8)
+    segments: list[dict[str, Any]] = []
+    occupied: list[tuple[int, int]] = []
+    for t in types:
+        seg = _INJECTORS[t](rng, out, T, C, occupied)
+        labels[seg["start"]: seg["end"]] = 1
+        occupied.append((seg["start"], seg["end"]))
+        segments.append(seg)
+
+    if series.ndim == 1:
+        out = out[:, 0]
+    return out, labels, segments
+
+
+# ─── event coupling ────────────────────────────────────────────────────────
+
+_EVENT_TEMPLATES = {
+    "deploy": "deployed new version v{ver} of service {svc}",
+    "rollback": "rolled back service {svc} to v{ver}",
+    "incident": "incident declared on {svc} (severity {sev})",
+    "config": "configuration changed for {svc}",
+    "scale": "scaled {svc} from {n0} to {n1} replicas",
+}
+
+
+def couple_event(
+    series: np.ndarray,
+    t: int,
+    kind: str = "deploy",
+    *,
+    seed: int | None = None,
+) -> tuple[np.ndarray, str]:
+    """Inject a persistent step change at time ``t`` and return event text.
+
+    The series before ``t`` is left untouched; from ``t`` onward a per-channel
+    offset is added (modeling the persistent effect of the event).
+
+    Parameters
+    ----------
+    series:
+        Array of shape ``(T, C)``.
+    t:
+        Event time index (clamped to ``[0, T-1]``).
+    kind:
+        Event key; see :data:`_EVENT_TEMPLATES`.
+    seed:
+        RNG seed.
+
+    Returns
+    -------
+    out:
+        Mutated copy with a step change at ``t``.
+    text:
+        Human-readable description of the event.
+    """
+    rng = np.random.default_rng(seed)
+    if kind not in _EVENT_TEMPLATES:
+        raise ValueError(f"unknown event kind: {kind!r}")
+    out = np.array(series, dtype=np.float64, copy=True)
+    if out.ndim == 1:
+        out = out[:, None]
+        squeeze = True
+    else:
+        squeeze = False
+
+    T, C = out.shape
+    t = int(np.clip(t, 0, T - 1))
+    offset = rng.uniform(8, 25, size=C) * rng.choice([-1, 1], size=C)
+    out[t:] += offset[None, :]
+
+    # render event text
+    ctx = {
+        "ver": f"{rng.integers(1, 9)}.{rng.integers(0, 20)}.{rng.integers(0, 10)}",
+        "svc": f"svc-{chr(ord('a') + int(rng.integers(0, 8)))}",
+        "sev": f"SEV{int(rng.integers(1, 4))}",
+        "n0": int(rng.integers(1, 10)),
+    }
+    ctx["n1"] = max(1, ctx["n0"] + int(rng.integers(-3, 6)))
+    text = _EVENT_TEMPLATES[kind].format(**ctx)
+
+    if squeeze:
+        out = out[:, 0]
+    return out, text
+
+
+# ─── missing values ────────────────────────────────────────────────────────
+
+
+def add_missing(
+    series: np.ndarray,
+    rate: float = 0.05,
+    *,
+    seed: int | None = None,
+) -> np.ndarray:
+    """Randomly drop a fraction ``rate`` of points, replacing them with NaN.
+
+    Parameters
+    ----------
+    series:
+        Array of shape ``(T, C)``.
+    rate:
+        Expected fraction of points to drop (clamped to ``[0, 0.5]``).
+    seed:
+        RNG seed.
+    """
+    rate = float(np.clip(rate, 0.0, 0.5))
+    rng = np.random.default_rng(seed)
+    out = np.array(series, dtype=np.float64, copy=True)
+    mask = rng.random(out.shape) < rate
+    out[mask] = np.nan
+    return out