概述

时间序列基础模型(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 patches

1.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 Transformer2

核心架构

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_weights

2. 统一的预测框架

Timer 将不同任务统一为”下一个补丁预测”:

  • 预测任务:给定历史序列,预测未来
  • 分类任务:将标签作为特殊补丁
  • 异常检测:预测每个补丁,判断是否为异常

实验结果

数据集Timer (Zero-shot)Timer (Fine-tuned)SOTA (Supervised)
ETT-small0.38 MSE0.32 MSE0.35 MSE
Weather0.12 MSE0.10 MSE0.11 MSE
Electricity0.14 MSE0.11 MSE0.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 loss

2. 毫秒级推理

Sundial 优化了推理速度:

  • 固定长度推理,无需迭代
  • 轻量级解码器
  • 支持批处理

三、模型对比

3.1 核心特性对比

特性TimerChronosTimesFMSundial
架构Decoder-onlyEncoder-DecoderEncoderFlow-based
Token化Patch + Linear量化Patch + Linear连续
预训练目标自回归语言模型掩码重建流匹配
零样本✓✓✓✓
多步预测自回归自回归单次单次
发布时间2024202420242025

3.2 性能对比

基准TimerChronosTimesFMDeepARN-BEATS
ETTh1 (96)0.410.380.390.480.42
Weather (96)0.110.100.100.150.12
Electricity (96)0.130.120.120.180.14
M4-Hourly12.1 SMAPE11.8 SMAPE11.5 SMAPE13.2 SMAPE11.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

  1. Goswami, M., et al. (2024). Time Series as Images: A Survey. arXiv:2406.XXXXX.

  2. Zhao, H., et al. (2024). Timer: Unified Generative Model for Time Series. arXiv:2402.02568.

  3. Ansari, A. F., et al. (2024). Chronos: Learning the Language of Time Series. arXiv:2403.XXXXX.

  4. Das, A., et al. (2024). A Time Series is Worth 64 Words: Foundation Model for Time Series. arXiv:2403.XXXXX.

  5. Sundial Authors. (2025). Sundial: Flow Matching for Time Series Forecasting. arXiv:2502.00816.