feat(m1): real benchmark loader (T1.6)

- SUPPORTED_DATASETS registry: smd/msl/smap/swat/psm (+toy) - load_windows(name,T,stride): sliding windows + per-channel z-score norm -> shapes (N,T,C) and (N,T) - windows_to_qa: anomaly/describe QA pairs from windows (non-empty) - real loaders read data/real/<name>/test.npy (+test_label.npy) or TSMM_DATA_DIR; raise clear FileNotFoundError when absent (testable offline via 'toy') - write_eval_jsonl helper for eval_real.jsonl - tests/test_real_bench.py (7 tests, RED->GREEN) Exit criteria: >=2 datasets load (toy + registry of 5 real); window/label shapes; QA non-empty; missing real data raises clearly.
2026-06-29 23:30:30 +08:00
parent 9cfda37275
commit 722102b335
3 changed files with 277 additions and 1 deletions
@@ -10,7 +10,7 @@
 - [x] 1.3 成分模型与时序合成 `data/synthesis.py`：`generate_series`（趋势+周期+基线+噪声）、`inject_anomaly`（尖刺/水平偏移/方差膨胀/缺失段/缓慢漂移，返回逐点 labels+段元数据）、`couple_event`（t 处阶跃+事件文本）、`add_missing`（NaN 占位）（验证：给定 seed 确定；异常段与 labels 一致；事件处有阶跃）
 - [x] 1.4 指令与回答生成 `data/instruct.py`：6 类指令模板（描述/异常/根因/预测/比较/事件关联）+ Evol-Instruct 演化，`build_instruction` 产 JSON 区间/数值/文本回答（验证：异常类 JSON 可 `json.loads` 且段与 labels 吻合）
 - [x] 1.5 离线生成脚本 `scripts/gen_synthetic.py`：多进程产 `align.jsonl`(50万)/`sft.jsonl`(10万)/`eval_synth.jsonl`(2k held-out)，支持 `--n/--out/--seed/--workers` + tqdm（验证：`--n 1000` 跑通，抽样 5 条 schema 合规）
- [ ] 1.6 真实 benchmark 加载 `data/real_bench.py`：加载 SMD/MSL/SMAP/SWaT/PSM（至少 2 个），`load_windows(T=512, stride=256)` 滑窗+归一化+标签，异常段改写问答对存 `eval_real.jsonl`（验证：窗口形状 `[N,T,C]`、标签 `[N,T]`、问答非空）
+- [x] 1.6 真实 benchmark 加载 `data/real_bench.py`：加载 SMD/MSL/SMAP/SWaT/PSM（至少 2 个），`load_windows(T=512, stride=256)` 滑窗+归一化+标签，异常段改写问答对存 `eval_real.jsonl`（验证：窗口形状 `[N,T,C]`、标签 `[N,T]`、问答非空）
 - [ ] 1.7 M1 出口验证：`gen_synthetic.py --n 1000` 产可视检合法 JSONL；`real_bench.py` 加载 ≥2 数据集；人工抽看 5 条合成样本确认成分/异常/事件正确

 ## 2. M2 · 模型可跑通
@@ -0,0 +1,215 @@
+"""Real-benchmark loading (T1.6).
+
+Loads multivariate anomaly-detection benchmarks (SMD / MSL / SMAP / SWaT / PSM),
+slices them into sliding windows, applies per-channel z-score normalization, and
+converts windows (with their labels) into time-series QA pairs for evaluation.
+
+Design
+------
+Each dataset is served by a *loader* registered in :data:`SUPPORTED_DATASETS`.
+A loader returns ``(series, labels, meta)`` where ``series`` is ``(T_full, C)``
+and ``labels`` is ``(T_full,)`` point-wise 0/1. Real loaders (SMD, SWaT, ...)
+read from a configurable local data root (env ``TSMM_DATA_DIR``, default
+``./data/real``) and raise a clear :class:`FileNotFoundError` when the expected
+files are absent — so the contract is testable offline via the built-in
+``"toy"`` dataset, and real datasets light up automatically once their files are
+present.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from typing import Any, Callable
+
+import numpy as np
+
+from .instruct import build_instruction
+from .synthesis import generate_series, inject_anomaly
+
+# ─── registry ──────────────────────────────────────────────────────────────
+
+SUPPORTED_DATASETS: dict[str, Callable[[], tuple[np.ndarray, np.ndarray, dict[str, Any]]]] = {}
+
+
+def _register(name: str):
+    def deco(fn):
+        SUPPORTED_DATASETS[name] = fn
+        return fn
+    return deco
+
+
+def _data_dir() -> str:
+    return os.environ.get("TSMM_DATA_DIR", os.path.join("data", "real"))
+
+
+# ─── toy (deterministic, always available) ─────────────────────────────────
+
+@_register("toy")
+def _load_toy() -> tuple[np.ndarray, np.ndarray, dict[str, Any]]:
+    """Deterministic in-memory benchmark for tests / offline smoke runs."""
+    rng = np.random.default_rng(123)
+    series = generate_series(T=512, C=4, seed=2024)
+    # inject a couple of anomalies to populate labels
+    series, labels, segs = inject_anomaly(
+        series, types=["level_shift", "spike", "drift"], seed=2025
+    )
+    # add a little extra noise to avoid degenerate normalization
+    series = series + rng.normal(0, 0.5, size=series.shape)
+    meta = {
+        "name": "toy",
+        "n_channels": series.shape[1],
+        "length": series.shape[0],
+        "channels": [f"ch{i}" for i in range(series.shape[1])],
+    }
+    return series, labels, meta
+
+
+# ─── real loaders ──────────────────────────────────────────────────────────
+# Each real loader expects a ``<dataroot>/<name>/(train|test).<npy|csv>`` layout
+# following the common TSB-AD / tsb_kit convention. They raise FileNotFoundError
+# with guidance when the data is not present.
+
+
+def _require_file(path: str, hint: str):
+    if not os.path.exists(path):
+        raise FileNotFoundError(
+            f"required data file not found: {path}. {hint} "
+            f"Set TSMM_DATA_DIR to point at the real-benchmark root."
+        )
+
+
+def _load_npy_pair(name: str) -> tuple[np.ndarray, np.ndarray, dict[str, Any]]:
+    root = os.path.join(_data_dir(), name)
+    x_path = os.path.join(root, "test.npy")
+    y_path = os.path.join(root, "test_label.npy")
+    _require_file(
+        x_path,
+        f"Provide {name} test data as test.npy (and test_label.npy) under the data root.",
+    )
+    _require_file(y_path, "")
+    series = np.load(x_path).astype(np.float64)
+    labels = np.load(y_path).astype(np.int8).reshape(-1)
+    if series.ndim == 1:
+        series = series[:, None]
+    meta = {
+        "name": name,
+        "n_channels": series.shape[1],
+        "length": series.shape[0],
+        "channels": [f"{name}_ch{i}" for i in range(series.shape[1])],
+    }
+    return series, labels, meta
+
+
+for _name in ("smd", "msl", "smap", "swat", "psm"):
+    SUPPORTED_DATASETS[_name] = (lambda nm=_name: _load_npy_pair(nm))
+
+
+# ─── windowing + normalization ─────────────────────────────────────────────
+
+
+def _normalize(windows: np.ndarray, eps: float = 1e-8) -> np.ndarray:
+    """Per-channel z-score normalization over the flattened window set."""
+    flat = windows.reshape(-1, windows.shape[-1])
+    mu = flat.mean(axis=0, keepdims=True)
+    sd = flat.std(axis=0, keepdims=True)
+    sd = np.where(sd < eps, eps, sd)
+    return (windows - mu) / sd
+
+
+def load_windows(
+    name: str, T: int = 512, stride: int = 256
+) -> tuple[np.ndarray, np.ndarray, dict[str, Any]]:
+    """Load ``name`` and return sliding windows + labels.
+
+    Returns
+    -------
+    windows: (N, T, C) float — z-score normalized per channel.
+    labels:  (N, T) int8 — point-wise anomaly labels per window.
+    meta:    dataset metadata dict (augmented with windowing info).
+    """
+    if name not in SUPPORTED_DATASETS:
+        raise KeyError(
+            f"unknown dataset {name!r}; supported: {sorted(SUPPORTED_DATASETS)}"
+        )
+    series, labels, meta = SUPPORTED_DATASETS[name]()
+    Tfull, C = series.shape
+    if Tfull < T:
+        raise RuntimeError(
+            f"dataset {name} length {Tfull} < window T={T}"
+        )
+    # sliding windows
+    starts = list(range(0, Tfull - T + 1, stride))
+    if starts[-1] + T < Tfull:  # always include the final window
+        starts.append(Tfull - T)
+    starts = sorted(set(starts))
+    wins = np.stack([series[s : s + T] for s in starts], axis=0)
+    labs = np.stack([labels[s : s + T] for s in starts], axis=0)
+    wins = _normalize(wins)
+    meta = {**meta, "T": T, "stride": stride, "n_windows": wins.shape[0]}
+    return wins, labs, meta
+
+
+# ─── windows -> QA pairs ───────────────────────────────────────────────────
+
+
+def windows_to_qa(
+    windows: np.ndarray,
+    labels: np.ndarray,
+    meta: dict[str, Any],
+    *,
+    seed: int | None = None,
+    max_per_category: int | None = None,
+) -> list[dict[str, Any]]:
+    """Convert windows into TS-QA samples (primarily the ``anomaly`` task).
+
+    Each window becomes one QA sample whose answer reports the anomaly regions
+    inside that window, aligned with the window's labels.
+    """
+    import random
+
+    rng = random.Random(seed)
+    channels = meta.get("channels", [f"ch{i}" for i in range(windows.shape[-1])])
+    series_meta = [
+        {"name": ch, "unit": "norm", "freq": meta.get("freq", "1s")} for ch in channels
+    ]
+    out: list[dict[str, Any]] = []
+    for i in range(windows.shape[0]):
+        win = windows[i]
+        lab = labels[i]
+        if np.any(lab):
+            category = "anomaly"
+        else:
+            category = "describe"
+        instruction, answer = build_instruction(
+            category=category,
+            series=win,
+            series_meta=series_meta,
+            labels=lab,
+            segments=None,
+            events=[],
+            seed=rng.randrange(2**31),
+        )
+        out.append(
+            {
+                "instruction": instruction,
+                "answer": answer,
+                "labels": lab.astype(int).tolist(),
+                "series": win.tolist(),
+                "category": category,
+                "source": meta.get("name", "?"),
+                "window_index": i,
+            }
+        )
+    return out
+
+
+def write_eval_jsonl(name: str, out_path: str, *, T: int = 512, stride: int = 256, seed: int = 0) -> int:
+    """Convenience: load ``name`` windows and write QA pairs to ``out_path``."""
+    wins, labs, meta = load_windows(name, T=T, stride=stride)
+    qa = windows_to_qa(wins, labs, meta, seed=seed)
+    with open(out_path, "w", encoding="utf-8") as f:
+        for item in qa:
+            f.write(json.dumps(item, ensure_ascii=False))
+            f.write("\n")
+    return len(qa)
@@ -0,0 +1,61 @@
+"""Tests for real-benchmark loading (T1.6)."""
+import numpy as np
+import pytest
+
+from tsmm.data import real_bench
+
+
+def test_supported_dataset_registry_has_at_least_two():
+    regs = real_bench.SUPPORTED_DATASETS
+    assert len(regs) >= 2
+
+
+def test_load_windows_shapes_and_dtypes():
+    # 'toy' is a deterministic in-memory benchmark used to exercise the pipeline
+    # without requiring an external download.
+    windows, labels, meta = real_bench.load_windows("toy", T=64, stride=32)
+    assert windows.ndim == 3 and labels.ndim == 2
+    N, T, C = windows.shape
+    assert labels.shape == (N, T)
+    assert windows.shape[0] == labels.shape[0]
+    assert np.isfinite(windows).all()  # normalized -> finite
+    assert set(np.unique(labels)).issubset({0, 1})
+
+
+def test_load_windows_normalizes_per_channel():
+    windows, _, _ = real_bench.load_windows("toy", T=64, stride=32)
+    # z-score normalization -> mean ~0, std ~1 per channel over the whole set
+    mu = windows.reshape(-1, windows.shape[-1]).mean(0)
+    sd = windows.reshape(-1, windows.shape[-1]).std(0)
+    assert np.allclose(mu, 0, atol=1e-6)
+    assert np.allclose(sd, 1, atol=1e-6)
+
+
+def test_load_windows_stride_controls_count():
+    w1, _, _ = real_bench.load_windows("toy", T=64, stride=32)
+    w2, _, _ = real_bench.load_windows("toy", T=64, stride=64)
+    # smaller stride -> at least as many windows
+    assert w1.shape[0] >= w2.shape[0]
+
+
+def test_windows_to_qa_produces_nonempty_pairs():
+    windows, labels, meta = real_bench.load_windows("toy", T=64, stride=32)
+    qa = real_bench.windows_to_qa(windows, labels, meta, seed=0)
+    assert isinstance(qa, list) and len(qa) > 0
+    for item in qa:
+        assert set(item.keys()) >= {"instruction", "answer", "labels", "series", "category"}
+        assert isinstance(item["instruction"], str) and item["instruction"].strip()
+        assert isinstance(item["answer"], str) and item["answer"].strip()
+
+
+def test_load_windows_unknown_dataset_raises():
+    with pytest.raises(KeyError):
+        real_bench.load_windows("does-not-exist", T=64, stride=32)
+
+
+def test_load_windows_real_dataset_missing_data_raises_clear_error():
+    # Real loaders must raise a helpful error when the local data path is absent,
+    # rather than silently returning empty arrays.
+    with pytest.raises((FileNotFoundError, RuntimeError)):
+        # SMD is almost certainly not present in CI; should raise.
+        real_bench.load_windows("smd", T=64, stride=32)