diff --git a/openspec/changes/ts-as-modality/tasks.md b/openspec/changes/ts-as-modality/tasks.md index 88b9db2..4a9edbc 100644 --- a/openspec/changes/ts-as-modality/tasks.md +++ b/openspec/changes/ts-as-modality/tasks.md @@ -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 · 模型可跑通 diff --git a/src/tsmm/data/real_bench.py b/src/tsmm/data/real_bench.py new file mode 100644 index 0000000..8ebdd2b --- /dev/null +++ b/src/tsmm/data/real_bench.py @@ -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 ``//(train|test).`` 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) diff --git a/tests/test_real_bench.py b/tests/test_real_bench.py new file mode 100644 index 0000000..43ea4e1 --- /dev/null +++ b/tests/test_real_bench.py @@ -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)