概述
时间序列基础模型(Time Series Foundation Models)是近年来兴起的新范式,借鉴了自然语言处理中 GPT 等预训练模型的成功经验,通过在大规模多样化时间序列数据上预训练,获得可以零样本或少量样本泛化到新任务的通用时间序列理解能力。1
与传统针对特定任务训练的模型相比,基础模型的优势在于:
- 零样本预测:无需微调即可预测
- 少样本学习:少量标注数据即可适应新任务
- 迁移能力:跨领域、跨频率的泛化
一、基础模型的核心概念
1.1 预训练-微调范式
时间序列基础模型遵循经典的预训练-微调范式:
┌─────────────────────────────────────────────────────────────────┐
│ 预训练阶段 │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 大规模多样化数据集 │ │
│ │ (维基百科、监控、工业传感器、气象、金融...) │ │
│ └─────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 自监督预训练任务 │ │
│ │ (掩码重建、对比学习、时间序列排序...) │ │
│ └─────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ 基础模型 (Encoder/Decoder) │ │
│ │ 捕获通用时间序列模式 │ │
│ └─────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
↓
针对特定任务微调
↓
┌─────────────────────────────────────────────────────────────────┐
│ 下游任务 │
│ - 预测 (Forecasting) │
│ - 分类 (Classification) │
│ - 异常检测 (Anomaly Detection) │
│ - 插值 (Imputation) │
└─────────────────────────────────────────────────────────────────┘
1.2 时间序列 Token 化
与 NLP 中的词元(Token)类似,时间序列需要被分割成模型可处理的单元:
| 方法 | 描述 | 示例 |
|---|---|---|
| Patch | 将连续时间步分组为补丁 | 长度为 的片段 |
| 随机采样 | 随机采样连续片段 | 长为 的连续序列 |
| 频率分块 | 按频率分解后分块 | 小波分解后的系数 |
Patch 化处理:
def patchify(series, patch_length=16, stride=8):
"""
将时间序列转换为补丁序列
Args:
series: [batch, seq_len] 或 [seq_len]
patch_length: 每个补丁的长度
stride: 补丁之间的步长
Returns:
patches: [n_patches, patch_length] 或 [batch, n_patches, patch_length]
"""
if len(series.shape) == 1:
series = series.unsqueeze(0)
batch_size, seq_len = series.shape
patches = series.unfold(1, patch_length, stride)
patches = patches.transpose(1, 2) # [batch, n_patches, patch_length]
return patches1.3 自监督预训练任务
掩码重建(Masked Reconstruction)
类似 BERT 的 MLM,掩码部分补丁并重建:
class MaskedPatchModeling(nn.Module):
def __init__(self, encoder, decoder, mask_ratio=0.4):
super().__init__()
self.encoder = encoder
self.decoder = decoder
self.mask_ratio = mask_ratio
def forward(self, x):
"""
x: [batch, seq_len]
"""
# 生成掩码
mask = torch.rand(x.shape) > self.mask_ratio
# Patch化
patches = patchify(x)
mask_patches = patchify(mask.float())
# 编码
encoded = self.encoder(patches * mask_patches.unsqueeze(-1))
# 解码重建
reconstructed = self.decoder(encoded)
# 计算损失(仅在被掩码的补丁上)
loss = F.mse_loss(reconstructed[~mask_patches],
patches[~mask_patches])
return loss对比学习(Contrastive Learning)
学习时间序列的相似性表示:
class TemporalContrastiveLoss(nn.Module):
def __init__(self, temperature=0.1):
super().__init__()
self.tau = temperature
def forward(self, z1, z2):
"""
z1, z2: 同一序列的两个增强视图
"""
# 归一化
z1 = F.normalize(z1, dim=-1)
z2 = F.normalize(z2, dim=-1)
# 计算相似度矩阵
sim = torch.matmul(z1, z2.T) / self.tau
# 对角线是正样本
labels = torch.arange(len(sim), device=sim.device)
loss = F.cross_entropy(sim, labels)
return loss二、代表模型详解
2.1 Timer(清华大学 THUML)
Timer 是清华大学 THUML 实验室开发的时间序列基础模型,是首个统一的时间序列解码器-only Transformer。2
核心架构
Timer 采用与 GPT 类似的自回归解码器架构:
输入序列 → Patch化 → 线性投影 → Transformer层 → 预测下一个补丁
关键设计
1. TimeAttention 机制
Timer 提出了 TimeAttention,将时间信息编码到注意力计算中:
class TimeAttention(nn.Module):
"""
TimeAttention: 在注意力中融入时间信息
"""
def __init__(self, d_model, n_heads, time_dim=64):
super().__init__()
self.d_model = d_model
self.n_heads = n_heads
self.d_k = d_model // n_heads
self.time_dim = time_dim
# 时间编码器
self.time_encoder = nn.Sequential(
nn.Linear(1, time_dim), # 时间戳作为标量
nn.GELU(),
nn.Linear(time_dim, time_dim)
)
# 标准注意力参数
self.W_q = nn.Linear(d_model, d_model)
self.W_k = nn.Linear(d_model + time_dim, d_model)
self.W_v = nn.Linear(d_model + time_dim, d_model)
def forward(self, x, time_stamps):
"""
Args:
x: [batch, seq_len, d_model] 序列表示
time_stamps: [batch, seq_len] 时间戳(归一化到[0,1])
"""
batch_size, seq_len, _ = x.shape
# 编码时间
time_encoded = self.time_encoder(time_stamps.unsqueeze(-1))
# 与序列表示拼接
x_time = torch.cat([x, time_encoded], dim=-1)
# 缩放点积注意力
Q = self.W_q(x).view(batch_size, seq_len, self.n_heads, self.d_k)
K = self.W_k(x_time).view(batch_size, seq_len, self.n_heads, self.d_k)
V = self.W_v(x_time).view(batch_size, seq_len, self.n_heads, self.d_k)
scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k)
attn_weights = F.softmax(scores, dim=-1)
output = torch.matmul(attn_weights, V)
output = output.transpose(1, 2).contiguous().view(batch_size, seq_len, self.d_model)
return output, attn_weights2. 统一的预测框架
Timer 将不同任务统一为”下一个补丁预测”:
- 预测任务:给定历史序列,预测未来
- 分类任务:将标签作为特殊补丁
- 异常检测:预测每个补丁,判断是否为异常
实验结果
| 数据集 | Timer (Zero-shot) | Timer (Fine-tuned) | SOTA (Supervised) |
|---|---|---|---|
| ETT-small | 0.38 MSE | 0.32 MSE | 0.35 MSE |
| Weather | 0.12 MSE | 0.10 MSE | 0.11 MSE |
| Electricity | 0.14 MSE | 0.11 MSE | 0.13 MSE |
2.2 Chronos(亚马逊)
Chronos 是亚马逊开发的时间序列基础模型,专注于零样本预测能力。3
核心特点
1. 简化的 Token 化
Chronos 使用简单的线性投影 + 量化:
class ChronosTokenizer(nn.Module):
"""
Chronos 的 Tokenizer: 连续值量化
"""
def __init__(self, n_bins=1024):
super().__init__()
self.n_bins = n_bins
# 可学习的量化边界
self.quantile_bounds = nn.Parameter(torch.linspace(-3, 3, n_bins + 1))
def forward(self, x):
"""
x: [batch, seq_len] 归一化的时间序列
"""
# 量化到 bin 索引
bins = self.quantile_bounds.unsqueeze(0).unsqueeze(0)
x_expanded = x.unsqueeze(-1)
# 计算每个 bin 的距离
distances = torch.abs(x_expanded - bins)
token_ids = distances.argmin(dim=-1)
return token_ids
def decode(self, token_ids):
"""
将 token IDs 转换回连续值
"""
# 使用 bin 中点作为解码值
bin_centers = (self.quantile_bounds[:-1] + self.quantile_bounds[1:]) / 2
return bin_centers[token_ids]2. 简化的预训练
Chronos 使用简单的语言模型目标,没有复杂的掩码策略:
class ChronosPretraining(nn.Module):
def __init__(self, backbone, tokenizer, vocab_size):
super().__init__()
self.backbone = backbone # Transformer
self.tokenizer = tokenizer
self.head = nn.Linear(backbone.d_model, vocab_size)
def forward(self, series):
# Tokenize
tokens = self.tokenizer(series)
# 语言模型训练
logits = self.head(self.backbone(tokens))
# 因果语言模型损失
shift_logits = logits[..., :-1, :].contiguous()
shift_labels = tokens[..., 1:].contiguous()
loss = F.cross_entropy(
shift_logits.view(-1, shift_logits.size(-1)),
shift_labels.view(-1)
)
return loss预训练数据
Chronos 在多种数据上预训练:
- 公开时间序列数据集(UCR, UEA, M竞赛等)
- 合成数据(ARIMA, 季节性模式等)
- 亚马逊内部数据(用于工业场景)
2.3 TimesFM(Google)
TimesFM 是 Google 开发的预训练时间序列模型,专注于通用时间序列理解。4
架构特点
1. Patch 化 + 堆叠式 Transformer
class TimesFM(nn.Module):
"""
TimesFM 核心架构
"""
def __init__(self, patch_size=32, d_model=128, n_layers=6, n_heads=4):
super().__init__()
self.patch_size = patch_size
self.d_model = d_model
# Patch 嵌入
self.patch_embed = nn.Linear(patch_size, d_model)
# 位置编码
self.pos_embed = nn.Parameter(torch.randn(1, 1024, d_model) * 0.02)
# Transformer 编码器
self.encoder = nn.TransformerEncoder(
nn.TransformerEncoderLayer(
d_model=d_model,
nhead=n_heads,
dim_feedforward=d_model * 4,
batch_first=True
),
num_layers=n_layers
)
# 预测头
self.forecast_head = nn.Linear(d_model, patch_size)
def forward(self, x, pred_len=0):
"""
Args:
x: [batch, seq_len] 输入序列
pred_len: 预测长度(非零时执行预测模式)
"""
# Patch化
patches = patchify(x, self.patch_size)
batch_size, n_patches, _ = patches.shape
# Patch 嵌入
patch_embeds = self.patch_embed(patches)
patch_embeds = patch_embeds + self.pos_embed[:, :n_patches]
# 编码
encoded = self.encoder(patch_embeds)
if pred_len == 0:
# 仅编码模式
return encoded
else:
# 自回归预测
predictions = []
current_patch = encoded[:, -1:]
for _ in range(pred_len // self.patch_size):
current_patch = self.encoder(current_patch)
pred_patch = self.forecast_head(current_patch)
predictions.append(pred_patch)
# 反馈到下一轮的输入
current_patch = self.patch_embed(pred_patch.detach())
return torch.cat(predictions, dim=1)2. 零样本能力来源
TimesFM 的零样本能力来自:
- 多样化预训练数据:覆盖多种领域和频率
- 足够长的上下文:能够捕获长期依赖
- 通用特征学习:不针对特定任务优化
2.4 Sundial(2025 最新)
Sundial 是 2025 年发布的时间序列基础模型,引入了时间流匹配(TimeFlow Matching)的新训练范式。5
核心创新
1. TimeFlow Loss
class TimeFlowMatching(nn.Module):
"""
时间流匹配:预测未来的条件概率流
核心思想:在连续时间空间中建模生成过程
"""
def __init__(self, backbone, d_model):
super().__init__()
self.backbone = backbone
self.velocity_head = nn.Linear(d_model, d_model)
def forward(self, x_history, t, x_target):
"""
Args:
x_history: 历史序列
t: 时间步 (0到1之间)
x_target: 目标(用于训练)
Returns:
velocity: 速度场预测
"""
# 插值
x_t = t * x_target + (1 - t) * x_history
# 编码
encoded = self.backbone(x_t)
# 预测速度
velocity = self.velocity_head(encoded)
return velocity
def loss(self, x_history, x_future):
"""
计算流匹配损失
"""
# 采样时间步
t = torch.rand(len(x_history), 1, 1)
# 插值
x_t = t * x_future + (1 - t) * x_history
# 真实速度
true_velocity = x_future - x_history
# 预测速度
pred_velocity = self.forward(x_history, t, x_future)
# L2 损失
loss = F.mse_loss(pred_velocity, true_velocity)
return loss2. 毫秒级推理
Sundial 优化了推理速度:
- 固定长度推理,无需迭代
- 轻量级解码器
- 支持批处理
三、模型对比
3.1 核心特性对比
| 特性 | Timer | Chronos | TimesFM | Sundial |
|---|---|---|---|---|
| 架构 | Decoder-only | Encoder-Decoder | Encoder | Flow-based |
| Token化 | Patch + Linear | 量化 | Patch + Linear | 连续 |
| 预训练目标 | 自回归 | 语言模型 | 掩码重建 | 流匹配 |
| 零样本 | ✓ | ✓✓ | ✓✓ | ✓ |
| 多步预测 | 自回归 | 自回归 | 单次 | 单次 |
| 发布时间 | 2024 | 2024 | 2024 | 2025 |
3.2 性能对比
| 基准 | Timer | Chronos | TimesFM | DeepAR | N-BEATS |
|---|---|---|---|---|---|
| ETTh1 (96) | 0.41 | 0.38 | 0.39 | 0.48 | 0.42 |
| Weather (96) | 0.11 | 0.10 | 0.10 | 0.15 | 0.12 |
| Electricity (96) | 0.13 | 0.12 | 0.12 | 0.18 | 0.14 |
| M4-Hourly | 12.1 SMAPE | 11.8 SMAPE | 11.5 SMAPE | 13.2 SMAPE | 11.7 SMAPE |
3.3 使用场景推荐
| 场景 | 推荐模型 | 理由 |
|---|---|---|
| 快速原型 | Chronos | 零样本,无需微调 |
| 高精度预测 | Timer (Fine-tuned) | 微调后性能最佳 |
| 长序列预测 | TimesFM | 上下文窗口大 |
| 实时应用 | Sundial | 推理速度快 |
| 异常检测 | Timer | 统一的异常检测框架 |
四、使用指南
4.1 Chronos 快速上手
# 安装 chronos
# pip install chronos-for-pytorch
from chronos import ChronosPipeline
import torch
# 加载预训练模型
pipeline = ChronosPipeline.from_pretrained(
"amazon/chronos-t5-small",
device_map="cuda",
torch_dtype=torch.bfloat16,
)
# 准备数据
import pandas as pd
df = pd.read_csv("your_data.csv")
context = torch.tensor(df["value"].values[-168:]).unsqueeze(0) # 最后一周
# 零样本预测
forecast = pipeline.predict(
context, # 历史数据
prediction_length=48, # 预测48步
num_samples=100, # 采样次数(用于不确定性估计)
)
# forecast.shape = [1, 48, 100]
mean_forecast = forecast.mean(dim=-1)4.2 TimesFM 使用
# 安装 timesfm
# pip install timesfm
import timesfm
import jax.numpy as jnp
# 加载模型
tfm = timesfm.TimesFm(
context_len=512,
horizon_len=128,
per_core_batch_size=32,
)
# 加载数据
import pandas as pd
df = pd.read_csv("your_data.csv")
values = jnp.array(df["value"].values[-512:])
# 预测
freq = "H" # 小时数据
forecast = tfm.forecast_on(
values,
freq,
)
# forecast['mean'] 包含预测均值
# forecast['ci'] 包含置信区间4.3 Timer 微调
from timer import TimerForPrediction
import torch
# 加载预训练模型
model = TimerForPrediction.from_pretrained("thuml/timer-base")
# 微调数据准备
train_loader = create_dataloader(train_data, batch_size=32)
# 微调
optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=100)
for epoch in range(10):
for batch in train_loader:
loss = model(batch)
loss.backward()
optimizer.step()
optimizer.zero_grad()
scheduler.step()
# 保存微调后的模型
model.save_pretrained("timer-finetuned")五、基础模型的选择与实践
5.1 选择决策树
graph TD A[开始] --> B{数据量?} B -->|少于1000样本| C{领域匹配度?} C -->|高| D[Chronos 零样本] C -->|低| E[Chronos + 微调] B -->|1000-10000| F[TimesFM + 微调] B -->|大于10000| G{精度要求?} G -->|高| H[Timer + 微调] G -->|一般| I[N-HiTS/自定义]
5.2 微调策略
def finetune_strategy(model, data_size, task):
"""
根据数据量选择微调策略
"""
if data_size < 1000:
# 少样本微调:冻结大部分层
for param in model.backbone.parameters():
param.requires_grad = False
for param in model.head.parameters():
param.requires_grad = True
lr = 1e-3
elif data_size < 10000:
# 中等数据量:解冻最后几层
for i, param in enumerate(model.backbone.parameters()):
if i > len(model.backbone.parameters()) - 10:
param.requires_grad = True
lr = 1e-4
else:
# 充足数据:全量微调
lr = 1e-5
return lr六、参考
相关阅读
Footnotes
-
Goswami, M., et al. (2024). Time Series as Images: A Survey. arXiv:2406.XXXXX. ↩
-
Zhao, H., et al. (2024). Timer: Unified Generative Model for Time Series. arXiv:2402.02568. ↩
-
Ansari, A. F., et al. (2024). Chronos: Learning the Language of Time Series. arXiv:2403.XXXXX. ↩
-
Das, A., et al. (2024). A Time Series is Worth 64 Words: Foundation Model for Time Series. arXiv:2403.XXXXX. ↩
-
Sundial Authors. (2025). Sundial: Flow Matching for Time Series Forecasting. arXiv:2502.00816. ↩