RevIN（Reversible Instance Normalization）及其在时间序列中的应用

详细介绍 RevIN（Reversible Instance Normalization）及其在时间序列中的应用

1. RevIN 的定义与背景

RevIN（可逆实例归一化）是一种专门为时间序列预测设计的归一化方法，旨在处理非平稳数据（non-stationary）和分布漂移（distribution shift）。其核心思想是通过可逆的归一化-反归一化步骤，在保留实例（instance）特有统计特性的同时，提升模型对分布变化的鲁棒性。RevIN 在训练时对每个时间序列实例独立归一化，并在预测后通过存储的统计量还原数据，确保输出与原始尺度一致。

---

2. RevIN 的步骤详解

1. 归一化阶段（Normalization）

步骤 1.1：计算均值与标准差

对输入时间序列的每个样本（或通道）独立计算均值和标准差：

• 输入：时间序列 $X\in {\mathbb{R}}^{B\times T\times C}$（Batch × 时间步 × 通道）
• 计算：
  $${\mu }_c=\frac{1}{T}\sum _{t=1}^T{X}_{t,c},{\sigma }_c=\sqrt{\frac{1}{T}\sum _{t=1}^T(X_{t,c}-{\mu }_c{)}^2+ϵ}$$
  （每个通道 $c$ 独立计算，$ϵ$ 为数值稳定项）

步骤 1.2：标准化（Standardization）

对输入数据进行标准化：
$${\hat{X}}_{t,c}=\frac{X_{t,c}-{\mu }_c}{{\sigma }_c}$$

步骤 1.3：Affine 变换（可学习缩放与平移）

引入可学习的参数 $\gamma \in {\mathbb{R}}^C$ 和 $\beta \in {\mathbb{R}}^C$（每个通道独立），对标准化后的数据进行仿射变换：
$${\tilde{X}}_{t,c}={\gamma }_c\cdot {\hat{X}}_{t,c}+{\beta }_c$$

• 作用：允许模型自适应调整归一化后的分布，增强对不同时间序列模式的适应性。

2. 模型处理阶段

将归一化后的数据 $\tilde{X}$ 输入模型（如 Transformer、RNN 等），模型输出预测结果 $\tilde{Y}$。

3. 反归一化阶段（Denormalization）

步骤 3.1：逆 Affine 变换

利用学习到的 $\gamma$ 和 $\beta$，反向恢复标准化后的数据：
$${\hat{Y}}_{t,c}=\frac{{\tilde{Y}}_{t,c}-{\beta }_c}{{\gamma }_c}$$

步骤 3.2：恢复原始尺度

使用原始均值 ${\mu }_c$ 和标准差 ${\sigma }_c$ 恢复数据分布：
$$Y_{t,c}={\hat{Y}}_{t,c}\cdot {\sigma }_c+{\mu }_c$$

关键点说明

1. Affine 变换的作用：

   • 提供可学习的线性变换，使模型能动态调整归一化后的数据分布。
   • 增强对不同时间序列通道（特征）的适应性（如某些通道需要保留更大方差）。

2. 参数共享与学习：

   • $\gamma$ 和 $\beta$ 在训练中通过梯度下降学习，初始值通常为 $\gamma =1,\beta =0$。
   • 参数按通道（C）维度定义，支持多变量时间序列的独立调整。

3. 可逆性保障：

   • 在反归一化时严格逆序操作（先逆 Affine，再恢复均值和标准差），确保数值稳定。

---

3. 时序平稳性（Stationary）与 RevIN 的作用

• 非平稳性问题：传统时间序列模型（如 ARIMA）需通过差分或变换使数据平稳（均值、方差不随时间变化）。但深度模型直接处理非平稳数据时，可能因分布变化而性能下降。
• RevIN 的解决方案：
  • 在归一化阶段消除实例内的非平稳性（如局部趋势），使模型更容易学习残差模式。
  • 反归一化时恢复原始分布，确保预测结果与实际物理意义一致。

---

4. 分布漂移（Distribution Shift）的应对

分布漂移指训练集与测试集数据分布不一致的问题，常见于时间序列的跨时段预测。RevIN 通过以下机制缓解该问题：

• 实例级归一化：每个序列独立处理，避免跨实例的分布假设，适应测试阶段的未知分布。
• 局部统计量：仅依赖当前实例的均值和方差，而非全局统计量，减少对历史数据分布的依赖。

---

5. Instance Normalization vs. 其他归一化方法

方法	归一化维度	适用场景	时间序列中的局限性
Batch Norm (BN)	跨批次的样本同一特征	图像、固定分布数据	破坏时间依赖关系，对分布漂移敏感
Layer Norm (LN)	单个样本的所有神经元	NLP、RNN/Transformer	忽略特征间差异，对多变量时序效果有限
Group Norm (GN)	将特征分组后归一化	小批量或动态网络	组划分需人工设定，不灵活
Instance Norm (IN)	单个样本的每个特征独立归一化	风格迁移、图像生成	直接用于时序会丢失全局时序模式
RevIN	实例级归一化 + 可逆性	非平稳时间序列预测	需存储统计量，对短序列敏感

RevIN 的独特优势：

• 可逆性：唯一支持无损反归一化的方法，确保预测结果可解释。
• 实例自适应：克服全局归一化（如 BN）对分布漂移的脆弱性。

---

6. 应用场景与实验结果

• 典型任务：多变量时间序列预测（如电力需求预测、股票价格预测）。
• 效果验证：
  • 在公开数据集（如ETTh1、Traffic）上，RevIN 显著提升 Transformer、N-BEATS 等模型的预测精度。
  • 对分布漂移场景（如COVID-19期间的突变数据）鲁棒性更强。

---

7. 总结与局限性

• 优势：
  • 显式处理非平稳性和分布漂移。
  • 兼容任意模型结构（如 LSTM、Transformer）。
• 局限性：
  • 对极短时间序列（统计量估计不准）效果有限。
  • 假设序列内部统计量稳定，对突变或异常值敏感。

未来方向：结合自适应归一化（如动态调整统计量）或异常检测机制，进一步提升鲁棒性。

8. 代码

class RevIN(nn.Module):def __init__(self, num_features: int, eps=1e-5):"""Args:num_features (int): 时间序列的特征维度（C）。eps (float): 用于数值稳定性的小值。"""super(RevIN, self).__init__()self.num_features = num_featuresself.eps = eps# 可学习的仿射参数self.affine_weight = nn.Parameter(torch.ones(num_features))self.affine_bias = nn.Parameter(torch.zeros(num_features))def forward(self, x, mode, mask = None):"""Args:x (torch.Tensor): 输入时间序列，形状为 (B, T, C)。mode (str): 操作模式，'norm' 或 'denorm'。Returns:torch.Tensor: 归一化或反归一化的时间序列。"""if mode == 'norm':self._get_statistics(x, mask)  # 计算均值和标准差x = self._normalize(x)  # 归一化elif mode == 'denorm':x = self._denormalize(x)  # 反归一化else:raise ValueError("mode 必须是 'norm' 或 'denorm'")return xdef _get_statistics(self, x, mask):"""计算均值和标准差"""if torch.isnan(x).any():if mask is None:mask = torch.isnan(x)else:raise ValueError("mask is given but x still contains nan values.")if mask is None:self.mean = x.mean(1, keepdim=True).detach()x_enc = x - self.meanvariance = torch.var(x_enc, dim=1, keepdim=True, unbiased=False) + self.epsself.stdev = torch.sqrt(variance).detach()else:missing_sum = torch.sum(mask, dim=1, keepdim=True) + self.epsmean = torch.sum(x * mask, dim=1, keepdim=True) / missing_sumx_enc = x - meanx_enc = x_enc.masked_fill(mask == 0, 0)variance = torch.sum(x_enc * x_enc, dim=1, keepdim=True) + self.epsstdev = torch.sqrt(variance / missing_sum)self.mean = meanself.stdev = stdevdef _normalize(self, x):"""归一化"""x = x - self.meanx = x / self.stdev# 应用仿射变换x = x * self.affine_weight + self.affine_biasreturn xdef _denormalize(self, x):"""反归一化"""# 撤销仿射变换x = (x - self.affine_bias) / self.affine_weightx = x * self.stdev + self.meanreturn x

8.1 代码关键点

假设时序数据集的输入$X\in R^{B\ast T\ast F}$，$B,T,F$分别代表batch_size, time steps和每个时间步的特征数量。在销量预测场景中，由于新品的存在，可能某些商品的历史销量并不足$T$，这时候需要padding技术。所以需要在原始RevIN的实现上增加mask的处理。