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).
This commit is contained in:
@@ -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 条合成样本确认成分/异常/事件正确
|
||||
|
||||
|
||||
@@ -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())
|
||||
@@ -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
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user