基于CEEMDAN和Transformer的时序信号分解与预测实战

1次阅读

没有评论

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

时序预测在金融、气象等领域有着广泛的应用，但传统的时序预测方法在处理非平稳信号时存在诸多局限性。以 ARIMA 和单一 LSTM 为例，这些方法在面对非平稳信号时往往表现不佳，主要原因包括：

模态混叠问题 ：传统经验模态分解（EMD）在处理复杂信号时容易产生模态混叠，导致不同时间尺度的信号成分混杂在一起，影响预测精度。
长期依赖捕捉困难 ：RNN 和 LSTM 虽然在时序建模上有一定优势，但对于超长序列的长期依赖关系捕捉能力有限，容易出现梯度消失或爆炸问题。
噪声敏感 ：单一模型对噪声的鲁棒性较差，尤其在金融或气象数据中，噪声干扰会显著降低预测性能。

CEEMDAN（自适应噪声完备集合经验模态分解）是 EMD 的改进版本，相较于 EMD 和 EEMD，它具有以下优势：

自适应噪声添加 ：CEEMDAN 通过自适应添加噪声的方式，有效解决了 EMD 的模态混叠问题，同时避免了 EEMD 中噪声残留的缺陷。
完备性 ：CEEMDAN 的分解过程更加完备，能够更准确地提取信号的多尺度成分（IMF）。

Transformer 相较于 RNN 在时序预测中的优势主要体现在：

并行计算 ：Transformer 的自注意力机制允许并行处理序列数据，显著提升了训练效率。
长期依赖建模 ：通过自注意力机制，Transformer 能够直接捕捉序列中任意位置的关系，避免了 RNN 的梯度消失问题。

CEEMDAN 的分解过程可以描述为以下步骤：

添加噪声 ：对原始信号添加自适应白噪声，生成多个噪声扰动信号。
EMD 分解 ：对每个噪声扰动信号进行 EMD 分解，得到一组 IMF 分量。
平均处理 ：对所有噪声扰动信号对应的 IMF 分量进行平均，得到最终的 IMF 分量。

数学公式描述如下：

[IMF_k = \frac{1}{N} \sum_{i=1}^N EMD_k(x + \epsilon w_i) ]

其中，(x) 为原始信号，(w_i) 为白噪声，(\epsilon) 为噪声强度，(N) 为噪声扰动次数，(EMD_k) 表示第 k 个 IMF 分量。

针对时序数据，我们对传统 Transformer 进行了以下改进：

位置编码优化 ：使用可学习的位置编码替代固定的正弦 / 余弦编码，以适应不同时间尺度的信号。
注意力机制调整 ：在自注意力层中引入局部注意力窗口，减少计算复杂度的同时保留局部时序特征。

from PyEMD import CEEMDAN
import numpy as np

# 生成示例信号
t = np.linspace(0, 1, 200)
signal = np.sin(2 * np.pi * 5 * t) + np.sin(2 * np.pi * 20 * t)

# 初始化 CEEMDAN
ceemdan = CEEMDAN()
# 设置噪声标准差（根据信号幅度调整）ceemdan.noise_std = 0.1
# 执行分解
IMFs = ceemdan(signal)

# 查看分解结果
for i, imf in enumerate(IMFs):
    print(f'IMF {i+1} shape:', imf.shape)

import torch
import torch.nn as nn

class TimeSeriesTransformer(nn.Module):
    def __init__(self, input_dim, output_dim, d_model=64, nhead=4, num_layers=3):
        super().__init__()
        self.encoder = nn.Linear(input_dim, d_model)
        self.pos_encoder = nn.Parameter(torch.randn(1, 1000, d_model))  # 可学习位置编码
        encoder_layer = nn.TransformerEncoderLayer(d_model, nhead)
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers)
        self.decoder = nn.Linear(d_model, output_dim)

    def forward(self, x):
        # x shape: (batch, seq_len, input_dim)
        x = self.encoder(x)
        x = x + self.pos_encoder[:, :x.size(1)]  # 添加位置编码
        x = self.transformer(x)
        x = self.decoder(x)
        return x