ts_data

Task-adapter layer over the TSData dataset library. TSData handles storage, catalog, and raw arrays; ts_data turns those into windowed, split, scaled torch datasets for forecasting, imputation, generation, classification, and anomaly detection.

Install

cd TSLib-tool/ts_data
pip install -e .

Pointing at TSData

ts_data reads the TSData library through a self-contained reader. Root resolution order:

1. root= argument to load(...) / TSDataSource(...)
2. TSDATA_ROOT environment variable
3. built-in default path

export TSDATA_ROOT=/data/songzy/workshop/data/timeseries/TSData

Quick start

import ts_data

# discovery
ts_data.list_datasets()                      # all dataset names
ts_data.list_datasets(task="forecasting")    # filter by task/domain/format
ts_data.info("etth1")                        # full meta.json

# one-liner: name + task -> Splits(train, val, test) of torch datasets
train, val, test = ts_data.load("appliances", task="forecasting",
                                window=96, horizon=96)

from torch.utils.data import DataLoader
from ts_data import collate
loader = DataLoader(train, batch_size=32, shuffle=True, collate_fn=collate)
sample = train[0]          # {"x", "y", "mask", "x_mark", "y_mark", "idx"} (unused -> None)

val/test may be None for formats without such a split (see below).

Datasets

87 datasets, indexed by the TSData catalog. Browse them with ts_data.list_datasets(...) and ts_data.info(name).

task	count	on-disk formats
forecasting	66	matrix (42), sequences (24)
classification	15	samples
anomaly_detection	6	anomaly

Forecasting datasets span domains such as weather, energy, traffic, finance, health, cloud, web, and forecasting-competition collections. Use the filters to narrow down:

ts_data.list_datasets(task="forecasting", domain="traffic")
ts_data.list_datasets(format="samples")

DataLoader note

Each sample is a dict with a fixed set of keys (x, y, mask, x_mark, y_mark, idx); fields a task does not use are None (e.g. a forecast sample has mask=None). Because the default torch collate cannot stack None, use the bundled ts_data.collate, which drops all-None keys and stacks the rest:

from torch.utils.data import DataLoader
from ts_data import collate

loader = DataLoader(train, batch_size=32, shuffle=True, collate_fn=collate)
batch = next(iter(loader))   # {"x": [B,C,L], "y": [B,C,H], "x_mark": ..., "idx": ...}

Custom sample format

To control what each sample looks like, pass a transform(sample) -> Any callable. It receives the fixed-key dict and can return anything — a renamed dict, a tuple, tensors on a device, etc. Attach it via load(..., transform=), create_dataset(..., transform=), or dataset.with_transform(fn). With a transform that returns pure tensors/tuples you can use the default collate.

# return (x, y) tuples instead of dicts
train, val, test = ts_data.load(
    "etth1", task="forecasting", window=96, horizon=96,
    transform=lambda s: (s["x"], s["y"]),
)
x, y = train[0]

# or rename / restructure
def to_tslib(s):
    return {"batch_x": s["x"], "batch_y": s["y"], "batch_x_mark": s["x_mark"]}

train, _, _ = ts_data.load("etth1", task="forecasting", transform=to_tslib)

Sample schema

Every dataset returns a dict with the same fixed keys; a field a task does not use is None.

key	meaning
x	input window, [C, L] (default) or [L, C] if channel_last=True
y	target (task-dependent; None for generation / anomaly-train)
mask	observation mask (imputation; None otherwise)
x_mark, y_mark	calendar time features from freq+start_time; None when there is no time axis
idx	window start index in the original series

load(...) options

ts_data.load(
    name, task=None, *,
    root=None,
    window=96,          # input window (also the window size for imputation/generation/classification)
    horizon=96,         # prediction length (forecast)
    label_len=0,        # teacher-forcing label length (forecast)
    stride=1,
    scale=True,
    scaler="standard",  # "standard" | "minmax" | "robust" | None (or an sklearn scaler)
    norm_each_channel=True,  # False -> one global statistic across channels
    channel_last=False, # False -> [C, L]; True -> [L, C]
    split_ratio=None,   # matrix ratio splitting, e.g. (0.7, 0.1, 0.2)
    split_mode=None,    # "ratio" | "standard" (ETT auto-detected)
    val_ratio=0.0,      # samples: carve a val split off train
    mmap=True,
)

If task is omitted, a sensible default is chosen from the dataset's format.

Formats and how they map to splits

TSData stores each dataset in one of four on-disk formats; load routes automatically:

TSData format	typical task	splits returned	notes
matrix (T, N)	forecasting / imputation / generation	train / val / test	ratio or ETT-standard split; x_mark derived from freq
samples (n, L, C) + labels	classification / generation	train / (val) / test	uses the archive's n_train; val_ratio carves val from train
anomaly	anomaly_detection	train / — / test	train is all-normal & unlabeled; test carries pointwise 0/1 labels
sequences (ragged)	forecasting / generation	single dataset / — / —	windows cut within each series; global standardization

Tasks

Task names match the TSData catalog vocabulary: forecasting, imputation, generation, classification, anomaly_detection.

# forecasting
train, val, test = ts_data.load("etth1", task="forecasting",
                                window=96, horizon=96, label_len=48)

# imputation
train, val, test = ts_data.load("appliances", task="imputation", window=96,
                                mask_ratio=0.25, mask_mode="random")

# generation
train, _, _ = ts_data.load("appliances", task="generation", window=64)

# classification (samples format)
train, val, test = ts_data.load("atrial_fibrillation", task="classification",
                                val_ratio=0.2)

# anomaly detection (anomaly format)
train, _, test = ts_data.load("msl", task="anomaly_detection", window=100)

Normalization

The scaler is fit on the training split and reused for val/test:

scaler	transform	typical use
"standard" (default)	z-score (x-μ)/σ	forecasting
"minmax"	scale to [0, 1]	generation (GAN/diffusion)
"robust"	median / IQR	outlier-heavy data, anomaly detection
None	no scaling	pre-normalized data

You can also pass any sklearn-like scaler instance. Set scale=False to disable entirely. Use norm_each_channel=False to fit a single global statistic across all channels instead of per-channel.

train, val, test = ts_data.load("appliances", task="generation",
                                window=64, scaler="minmax")

Inverting predictions

Every returned split exposes the fitted scaler via .scaler and an .inverse_transform(pred, mask=None) helper to map predictions back to the original scale for metrics. It expects features on the last axis ([..., C]), so transpose channel-first outputs first. mask (imputation) keeps the original value where False.

train, val, test = ts_data.load("etth1", task="forecasting", window=96, horizon=96)
pred = model(...)                 # normalized space, e.g. [B, C, L]
pred_real = test.inverse_transform(pred.transpose(-1, -2))  # -> [..., C] in original units

Lower-level access

from ts_data import DataModule, TSDataSource

# raw arrays + metadata straight from TSData
src = TSDataSource()
data, meta = src.load_matrix("etth1")
X, y, meta = src.load_samples("atrial_fibrillation")
train, test, test_label, meta = src.load_anomaly("msl")
values, offsets, meta = src.load_sequences("air")

# manual matrix splitting / windowing
dm = DataModule.from_tsdata("etth1", split_ratio=(0.7, 0.1, 0.2))
ds = dm.create_dataset("train", "forecasting", input_len=96, pred_len=96)

# your own CSV/NPY files still work
dm = DataModule(data="my.csv", date_col="date", freq="h")

使用新数据集（不注册进 TSData 库）

新数据不一定要写进全局 TSData 库（catalog + 默认 root）。两条路：

方式一：数组直接传 DataModule（推荐，零落盘）

DataModule 直接接受 (T, F) numpy 数组，完全绕过 catalog：

import numpy as np
from torch.utils.data import DataLoader
from ts_data import DataModule, collate
from ts_data.utils import marks_from_freq

data = np.load("my_data.npy")                      # (T, F) 连续多变量序列
marks = marks_from_freq(len(data), "h", "2020-01-01")  # 可选时间特征

dm = DataModule(
    data,
    time_marks=marks,                  # 可为 None：样本里 x_mark/y_mark 为 None
    split_ratio=(0.7, 0.1, 0.2),       # ratio 切分；名字为 etth1 等时自动走 ETT 标准切分
    scale=True, scaler="standard",     # scaler 只在 train 段 fit
)

train_ds = dm.create_dataset("train", "forecasting",
                             input_len=96, pred_len=24, label_len=0)
test_ds = dm.create_dataset("test", "forecasting",
                            input_len=96, pred_len=24, label_len=0, stride=5)

loader = DataLoader(train_ds, batch_size=32, shuffle=True, collate_fn=collate)
# batch: {"x": (B,F,96), "y": (B,F,24), "x_mark", "y_mark", "idx"} —— 直接喂 ts_trainer

create_dataset(flag, task, **kwargs)：

• flag："train" / "val" / "test"。
• task：forecasting / imputation / generation / classification / anomaly_detection。
• 其余参数透传给对应 Dataset：预测用 input_len/pred_len/label_len； 其余任务用 window_size，插补另有 mask_ratio/mask_mode/seed， 分类需构造时给 labels，异常检测 test 段需点级 labels。
• 正 stride 只对 test 段降采样，train/val 恒为 1；负数对三段都生效。

还原尺度用 dm.inverse_transform（sklearn 风格，2D、特征在最后一轴）； 样本结构、scaler 语义与 ts_data.load 完全一致。

方式二：私有 root，继续用 ts_data.load

想要 load() 的全套参数（test_stride、transform、mmap 等），可以搭一个 最小私有库，不动全局默认库：

my_root/
├── catalog.json                 # 空 {} 即可，索引实际在 task 文件里
└── forecasting/
    ├── metadata.json            # 数据集索引（见下）
    └── mydata/
        ├── data.npy             # (T, F)
        └── meta.json            # "freq": "h", "start_time": "2020-01-01" 等

forecasting/metadata.json：

{"mydata": {"path": "forecasting/mydata", "format": "matrix", "domain": "custom"}}

splits = ts_data.load("mydata", task="forecasting", root="my_root",
                      window=96, horizon=24, test_stride=5)

也可以 export TSDATA_ROOT=my_root 省掉每次传 root=。 samples / anomaly / sequences 格式同理，按 source.py 的文件约定落盘即可。

Layout

ts_data/
├── __init__.py        # public API re-exports
├── api.py             # high-level load() + discovery
├── source.py          # self-contained TSData reader (catalog + raw arrays)
├── datamodule.py      # matrix splitting + create_dataset routing
├── utils.py           # CSV/NPY loading + time-feature extraction
└── datasets/          # torch Dataset implementations
    ├── base.py            # sliding-window base dataset
    ├── forecast.py        # forecast (matrix / continuous)
    ├── imputation.py      # imputation
    ├── generation.py      # generation
    ├── classification.py  # classification (continuous)
    ├── anomaly.py         # anomaly detection (train/test/test_label)
    ├── samples.py         # pre-windowed (n, L, C) datasets
    └── sequences.py       # variable-length (values + offsets) windowing

License

MIT