CEEMDAN Transformer 入门指南：从信号分解到时间序列预测实战

1次阅读

共计 2540 个字符，预计需要花费 7 分钟才能阅读完成。

传统时间序列分析方法在处理现实世界的复杂信号时常常力不从心。ARIMA 模型虽然经典，但它假设数据是平稳的，而且线性关系限制了其对非线性模式的捕捉能力。而基础的 EMD（经验模态分解）方法虽然能处理非平稳信号，却存在模态混叠问题——不同频率的成分会混杂在同一个 IMF（本征模态函数）中，严重影响分解效果。

以一个实际的电商销售预测场景为例：销售数据往往包含长期趋势（如增长）、季节性波动（如节假日高峰）、促销脉冲（短期尖峰）和随机噪声。传统方法要么无法同时捕捉这些特征，要么会因为模态混叠导致预测失真。

方法	模态混叠抑制	计算复杂度	噪声敏感性
EMD	弱	O(nlogn)	高
EEMD	中	O(kn^2)	中
CEEMDAN	强	O(kn^2)	低

注：k 为噪声添加次数，n 为数据长度

CEEMDAN（Complete Ensemble EMD with Adaptive Noise）的核心创新在于：
1. 自适应地添加特定比例的白噪声
2. 通过完备集合平均有效消除残留噪声
3. 每个 IMF 分量的计算都基于前一个分量的残差

数学表达如下：

IMF_1 = \frac{1}{N} \sum_{i=1}^N (x + \beta_0 w_i) - r_0

其中：
– $\beta_0$ 是初始噪声系数
– $w_i$ 是第 i 次添加的白噪声
– $r_0$ 是初始残差（通常为 0）

具体步骤：

对原始信号添加 N 组自适应白噪声
对每组噪声信号进行 EMD 分解得到第一组 IMF
计算第一组 IMF 的均值作为最终 IMF1
用原始信号减去 IMF1 得到残差 r1
对 r1 重复上述过程直至满足停止条件

关键改进点：

位置编码 ：将标准 Transformer 的 sin/cos 位置编码替换为可学习的时序嵌入

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=5000):
        super().__init__()
        self.pos_embed = nn.Parameter(torch.zeros(max_len, d_model))

    def forward(self, x):
        return x + self.pos_embed[:x.size(1)]

注意力机制 ：对不同的 IMF 分量采用独立的注意力头处理

from PyEMD import CEEMDAN
import numpy as np

# 关键参数说明
ceemdan = CEEMDAN(
    trials=100,       # 噪声添加次数
    noise_width=0.05, # 初始噪声系数
    ext_EMD=None      # 使用默认 EMD 实现
)

# 模拟电商销售数据（7 天周期 + 趋势 + 噪声）t = np.linspace(0, 30, 300)
sales = 50 + 2*t + 20*np.sin(2*np.pi*t/7) + np.random.normal(0, 5, 300)

# 执行分解
IMFs = ceemdan(sales)

# 可视化各 IMF 分量
import matplotlib.pyplot as plt
for i, imf in enumerate(IMFs):
    plt.subplot(len(IMFs), 1, i+1)
    plt.plot(imf)
plt.show()

import torch
import torch.nn as nn

class IMFTransformer(nn.Module):
    def __init__(self, n_imfs, d_model=64, nhead=4):
        super().__init__()
        self.imf_embed = nn.ModuleList([nn.Linear(1, d_model) for _ in range(n_imfs)]
        )
        self.pos_encoder = PositionalEncoding(d_model)
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=d_model, 
            nhead=nhead,
            batch_first=True
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=2)
        self.forecaster = nn.Linear(d_model, 1)

    def forward(self, x):  # x: [batch, seq_len, n_imfs]
        # 对各 IMF 分别嵌入
        imf_features = []
        for i in range(x.shape[-1]):
            imf_i = self.imf_embed[i](x[..., i:i+1])
            imf_features.append(imf_i)
        x = torch.stack(imf_features, dim=-1).mean(dim=-1)

        # Transformer 处理
        x = self.pos_encoder(x)
        x = self.transformer(x)
        return self.forecaster(x[:, -1:])