From 9cfda37275e21e74e17b2524e42dceda906333ef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=BC=A0=E5=AE=97=E5=B9=B3?= Date: Mon, 29 Jun 2026 23:29:04 +0800 Subject: [PATCH] feat(m1): offline generation pipeline + CLI (T1.5) - tsmm.data.gen_pipeline.gen_one_sample: composes attrs/series/anomaly/event/missing/instruct into one JSON-serializable sample (full schema) - write_jsonl helper - scripts/gen_synthetic.py: multiprocessing + tqdm CLI (--n/--out/--seed/--workers) - regression: build_instruction('event') with empty events no longer raises (TDD RED first) - tests/test_gen_pipeline.py (6 tests, RED->GREEN) Exit criteria: python scripts/gen_synthetic.py --n 1000 -> 1000 samples, schema-OK, all 6 categories balanced (~130-190 each). --- openspec/changes/ts-as-modality/tasks.md | 2 +- scripts/gen_synthetic.py | 102 +++++++++++++++ src/tsmm/data/gen_pipeline.py | 159 +++++++++++++++++++++++ src/tsmm/data/instruct.py | 3 +- tests/test_gen_pipeline.py | 73 +++++++++++ tests/test_instruct.py | 9 ++ 6 files changed, 346 insertions(+), 2 deletions(-) create mode 100644 scripts/gen_synthetic.py create mode 100644 src/tsmm/data/gen_pipeline.py create mode 100644 tests/test_gen_pipeline.py diff --git a/openspec/changes/ts-as-modality/tasks.md b/openspec/changes/ts-as-modality/tasks.md index 9a726bd..88b9db2 100644 --- a/openspec/changes/ts-as-modality/tasks.md +++ b/openspec/changes/ts-as-modality/tasks.md @@ -9,7 +9,7 @@ - [x] 1.2 属性词表与采样 `data/attributes.py`：定义变量名词表/单位/采样率档位，`sample_attributes(n_channels)` 返回结构化属性（验证：单测字段齐全、可复现） - [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 吻合） -- [ ] 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 合规） +- [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]`、问答非空） - [ ] 1.7 M1 出口验证：`gen_synthetic.py --n 1000` 产可视检合法 JSONL；`real_bench.py` 加载 ≥2 数据集；人工抽看 5 条合成样本确认成分/异常/事件正确 diff --git a/scripts/gen_synthetic.py b/scripts/gen_synthetic.py new file mode 100644 index 0000000..db8266d --- /dev/null +++ b/scripts/gen_synthetic.py @@ -0,0 +1,102 @@ +#!/usr/bin/env python3 +"""Offline synthetic data generation (T1.5). + +Produces JSONL datasets by composing :mod:`tsmm.data` primitives in parallel: + +* ``data/align.jsonl`` — alignment-style samples (default: many) +* ``data/sft.jsonl`` — Evol-Instruct rephrased (subset) +* ``data/eval_synth.jsonl`` — held-out eval set (fixed seed) + +Usage:: + + python scripts/gen_synthetic.py --n 1000 --out data/align.jsonl --seed 0 --workers 8 + +Each line is one JSON sample with keys: series, attributes, timestamps, +events, instruction, answer, labels, category, segments. +""" + +from __future__ import annotations + +import argparse +import json +import os +from concurrent.futures import ProcessPoolExecutor, as_completed +from pathlib import Path + +from tqdm import tqdm + +from tsmm.data.gen_pipeline import gen_one_sample, write_jsonl +from tsmm.data.instruct import evolve_instruction + + +def _one(args): + """Worker: generate a single sample (top-level for pickling).""" + T, C, seed, anomaly_types, missing_rate, event_prob, do_evolve = args + s = gen_one_sample( + T=T, C=C, seed=seed, + anomaly_types=anomaly_types, + missing_rate=missing_rate, + event_prob=event_prob, + ) + if do_evolve: + s["instruction"] = evolve_instruction(s["instruction"], seed=seed) + return s + + +def run( + n: int, + out: str, + *, + seed: int = 0, + workers: int = max(1, (os.cpu_count() or 2) - 1), + T: int = 512, + C: int = 5, + anomaly_types=("spike", "level_shift", "variance_change", "drift"), + missing_rate: float = 0.05, + event_prob: float = 0.5, + do_evolve: bool = False, +) -> int: + Path(out).parent.mkdir(parents=True, exist_ok=True) + work = [ + (T, C, seed + i, anomaly_types, missing_rate, event_prob, do_evolve) + for i in range(n) + ] + samples = [None] * n + with ProcessPoolExecutor(max_workers=workers) as ex: + futures = {ex.submit(_one, w): i for i, w in enumerate(work)} + for fut in tqdm(as_completed(futures), total=n, desc=os.path.basename(out)): + i = futures[fut] + samples[i] = fut.result() + return write_jsonl(samples, out) + + +def main(argv: list[str] | None = None) -> int: + p = argparse.ArgumentParser(description="Generate synthetic TS-QA JSONL data.") + p.add_argument("--n", type=int, default=1000, help="number of samples") + p.add_argument("--out", type=str, default="data/align.jsonl", help="output JSONL path") + p.add_argument("--seed", type=int, default=0, help="base RNG seed") + p.add_argument("--workers", type=int, default=max(1, (os.cpu_count() or 2) - 1)) + p.add_argument("--T", type=int, default=512, help="series length") + p.add_argument("--C", type=int, default=5, help="number of channels") + p.add_argument("--missing-rate", type=float, default=0.05) + p.add_argument("--event-prob", type=float, default=0.5) + p.add_argument("--evolve", action="store_true", help="apply Evol-Instruct rephrasing (for sft set)") + args = p.parse_args(argv) + + n = run( + n=args.n, + out=args.out, + seed=args.seed, + workers=args.workers, + T=args.T, + C=args.C, + missing_rate=args.missing_rate, + event_prob=args.event_prob, + do_evolve=args.evolve, + ) + print(f"wrote {n} samples -> {args.out}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/tsmm/data/gen_pipeline.py b/src/tsmm/data/gen_pipeline.py new file mode 100644 index 0000000..a5f38ed --- /dev/null +++ b/src/tsmm/data/gen_pipeline.py @@ -0,0 +1,159 @@ +"""Synthetic sample-composition pipeline (T1.5). + +Composes :mod:`tsmm.data` primitives into one fully-formed, JSON-serializable +sample:: + + generate_series ─┐ + sample_attributes ─┼─► gen_one_sample() ─► JSONL via write_jsonl() + inject_anomaly ─┤ + couple_event ─┤ + build_instruction ─┘ + +The heavy ``scripts/gen_synthetic.py`` entrypoint is a thin multiprocessing +wrapper around these two functions so the core logic stays unit-testable. +""" + +from __future__ import annotations + +import json +import random +from typing import Any, Iterable + +import numpy as np + +from .attributes import sample_attributes +from .instruct import INSTRUCTION_CATEGORIES, build_instruction +from .synthesis import add_missing, couple_event, generate_series, inject_anomaly + +# Default per-sample composition knobs; overridable by callers / the CLI. +_DEFAULTS = dict( + T=512, + C=5, + anomaly_types=("spike", "level_shift", "variance_change", "drift"), + missing_rate=0.05, + event_prob=0.5, +) + + +def _to_jsonable(obj: Any) -> Any: + """Recursively convert numpy types/arrays to plain JSON-serializable types.""" + if isinstance(obj, dict): + return {k: _to_jsonable(v) for k, v in obj.items()} + if isinstance(obj, (list, tuple)): + return [_to_jsonable(v) for v in obj] + if isinstance(obj, np.ndarray): + return obj.tolist() + if isinstance(obj, (np.integer,)): + return int(obj) + if isinstance(obj, (np.floating,)): + v = float(obj) + return None if np.isnan(v) else v + if isinstance(obj, float): + return None if np.isnan(obj) else obj + return obj + + +def gen_one_sample( + *, + T: int = 512, + C: int = 5, + seed: int | None = None, + anomaly_types: Iterable[str] = _DEFAULTS["anomaly_types"], + missing_rate: float = _DEFAULTS["missing_rate"], + event_prob: float = _DEFAULTS["event_prob"], + force_category: str | None = None, + horizon: int = 10, +) -> dict[str, Any]: + """Generate one fully-composed synthetic sample. + + Returns a JSON-serializable dict with keys: ``series``, ``attributes``, + ``timestamps``, ``events``, ``instruction``, ``answer``, ``labels``, + ``category`` (plus ``segments`` for downstream use). + """ + rng = random.Random(seed) + + # Derive independent sub-seeds so each stochastic primitive is reproducible + # and uncorrelated with the others. + s_attr, s_gen, s_anom, s_event, s_miss, s_cat, s_inst = ( + rng.randrange(2**31) for _ in range(7) + ) + + # 1) attributes + series + attrs = sample_attributes(C, seed=s_attr) + series = generate_series(T=T, C=C, seed=s_gen) + + # 2) anomalies + types = list(anomaly_types) + n_anom = rng.randint(1, max(1, len(types))) + chosen_types = rng.sample(types, n_anom) + series, labels, segments = inject_anomaly( + series, types=chosen_types, seed=s_anom + ) + + # 3) optional event coupling (may or may not happen) + events: list[dict[str, Any]] = [] + if rng.random() < event_prob: + t_event = int(rng.integers(0, T)) if hasattr(rng, "integers") else int(rng.randrange(0, T)) + kind = rng.choice(["deploy", "rollback", "incident", "config", "scale"]) + series, text = couple_event(series, t=t_event, kind=kind, seed=s_event) + events.append({"t": int(t_event), "text": text}) + + # 4) missing values + if missing_rate > 0: + series = add_missing(series, rate=missing_rate, seed=s_miss) + + # 5) timestamps (synthetic unix-ish index based on the first channel's freq) + freq_seconds = _freq_to_seconds(attrs[0]["freq"]) + t0 = 1_700_000_000 + timestamps = [int(t0 + i * freq_seconds) for i in range(T)] + + # 6) instruction category + answer + category = force_category if force_category is not None else random.Random(s_cat).choice( + INSTRUCTION_CATEGORIES + ) + instruction, answer = build_instruction( + category=category, + series=series, + series_meta=attrs, + labels=labels, + segments=segments, + events=events, + seed=s_inst, + horizon=horizon, + ) + + sample = { + "series": series, + "attributes": attrs, + "timestamps": timestamps, + "events": events, + "instruction": instruction, + "answer": answer, + "labels": labels, + "category": category, + "segments": segments, + } + return _to_jsonable(sample) + + +def _freq_to_seconds(freq: str) -> int: + """Convert a frequency band like ``'1s'``/``'5min'`` to seconds.""" + import re + + m = re.fullmatch(r"(\d+)\s*(s|min|ms|h)", freq.strip()) + if not m: + return 1 + n, unit = int(m.group(1)), m.group(2) + mult = {"ms": 0, "s": 1, "min": 60, "h": 3600}[unit] + return n * mult + + +def write_jsonl(samples: Iterable[dict[str, Any]], path: str) -> int: + """Write an iterable of samples to ``path`` as JSONL. Returns line count.""" + n = 0 + with open(path, "w", encoding="utf-8") as f: + for s in samples: + f.write(json.dumps(_to_jsonable(s), ensure_ascii=False)) + f.write("\n") + n += 1 + return n diff --git a/src/tsmm/data/instruct.py b/src/tsmm/data/instruct.py index 1af3a69..2a501e8 100644 --- a/src/tsmm/data/instruct.py +++ b/src/tsmm/data/instruct.py @@ -72,7 +72,8 @@ def _channel_names(series_meta: list[dict[str, Any]]) -> str: def _format_instruction(category: str, series_meta, *, horizon=10, event=None) -> str: tmpl = _INSTRUCTION_TEMPLATES[category] names = _channel_names(series_meta) - fields = {"names": names, "horizon": horizon} + # All templates only use a subset of these; missing keys are simply unused. + fields = {"names": names, "horizon": horizon, "et": "?"} if event is not None: fields["et"] = event["t"] return tmpl[0].format(**fields) diff --git a/tests/test_gen_pipeline.py b/tests/test_gen_pipeline.py new file mode 100644 index 0000000..566bb49 --- /dev/null +++ b/tests/test_gen_pipeline.py @@ -0,0 +1,73 @@ +"""Tests for offline synthetic generation (T1.5).""" +import json + +import numpy as np +import pytest + +from tsmm.data.gen_pipeline import gen_one_sample, write_jsonl + + +def test_gen_one_sample_has_required_schema(): + sample = gen_one_sample(T=128, C=2, seed=0) + required = { + "series", + "attributes", + "timestamps", + "events", + "instruction", + "answer", + "labels", + "category", + } + assert required.issubset(sample.keys()), f"missing keys: {required - sample.keys()}" + + +def test_gen_one_sample_shapes_and_types(): + T, C = 128, 3 + s = gen_one_sample(T=T, C=C, seed=1) + assert np.asarray(s["series"]).shape == (T, C) + assert np.asarray(s["labels"]).shape == (T,) + assert len(s["attributes"]) == C + assert len(s["timestamps"]) == T + # instruction + answer are non-empty strings + assert isinstance(s["instruction"], str) and s["instruction"].strip() + assert isinstance(s["answer"], str) and s["answer"].strip() + # category is one of the 6 + from tsmm.data.instruct import INSTRUCTION_CATEGORIES + assert s["category"] in INSTRUCTION_CATEGORIES + + +def test_gen_one_sample_reproducible(): + a = gen_one_sample(T=64, C=2, seed=42) + b = gen_one_sample(T=64, C=2, seed=42) + # JSON-serializable equality + assert json.dumps(a, sort_keys=True, default=str) == json.dumps(b, sort_keys=True, default=str) + + +def test_gen_one_sample_anomaly_answer_parses_and_matches_labels(): + s = gen_one_sample(T=128, C=2, seed=3, force_category="anomaly") + assert s["category"] == "anomaly" + parsed = json.loads(s["answer"]) + recon = np.zeros(128, dtype=int) + for r in parsed["anomaly_regions"]: + recon[r["start"]: r["end"]] = 1 + assert np.array_equal(recon, np.asarray(s["labels"]).astype(int)) + + +def test_gen_one_sample_json_serializable(): + s = gen_one_sample(T=64, C=2, seed=0) + # must round-trip through json with no custom types + blob = json.dumps(s) # raises if non-serializable + json.loads(blob) + + +def test_write_jsonl_creates_valid_file(tmp_path): + samples = [gen_one_sample(T=32, C=2, seed=i) for i in range(5)] + out = tmp_path / "out.jsonl" + write_jsonl(samples, str(out)) + assert out.exists() + lines = out.read_text().strip().splitlines() + assert len(lines) == 5 + for line in lines: + obj = json.loads(line) + assert "series" in obj and "instruction" in obj diff --git a/tests/test_instruct.py b/tests/test_instruct.py index 41e58a4..2043c3f 100644 --- a/tests/test_instruct.py +++ b/tests/test_instruct.py @@ -97,6 +97,15 @@ def test_event_answer_references_the_event(): assert "deploy" in (instruction + " " + answer).lower() +def test_event_category_handles_missing_events_gracefully(): + """Regression: selecting the 'event' category when events==[] must not raise.""" + ctx = _sample_context() + ctx["events"] = [] + instruction, answer = build_instruction(category="event", **ctx, seed=0) + assert isinstance(instruction, str) and instruction.strip() + assert isinstance(answer, str) and answer.strip() + + def test_root_cause_answer_is_text(): ctx = _sample_context() _, answer = build_instruction(category="root_cause", **ctx, seed=0)