当前位置：首页 > article >正文

N-Beats：一种用于时间序列预测的纯前馈神经网络模型

article 2025/2/11 14:05:37

介绍

N-Beats（Neural Basis Expansion Analysis for Interpretable Time Series Forecasting）是一种基于纯前馈神经网络的时间序列预测模型，由Boris Oreshkin等人在2019年提出。与传统的递归神经网络（如LSTM和GRU）不同，N-Beats通过堆叠多个简单的前馈块来生成预测，具有高度的可解释性和灵活性。

工作原理

模型架构

N-Beats的核心思想是使用一系列简单的前馈块（称为“基础块”）来逐步生成预测值。每个基础块包含一个全连接层（或多个全连接层），用于学习时间序列的模式，并输出两个部分：

后向分量（Backcast）：用于拟合输入的历史数据。
前向分量（Forecast）：用于生成未来的预测值。

基础块可以分为两种类型：

通用块（Generic Block）：直接学习时间序列的复杂模式。
趋势块（Trend Block）和季节性块（Seasonality Block）：分别用于捕捉时间序列的趋势成分和季节性成分。

训练过程

数据准备：
- 输入数据包括历史观测值。
- 每个时间序列被分成训练集和测试集。
模型训练：
- 对于每个基础块，输入为历史观测值的一部分（即“窗口”）。
- 基础块生成后向分量和前向分量。
- 后向分量用于拟合输入的历史数据，前向分量用于生成未来预测值。
- 损失函数通常采用均方误差（MSE），以最小化预测值与真实值之间的差异。
堆叠基础块：
- 多个基础块按顺序堆叠，每个块的后向分量用于校正前一个块的残差。
- 最终的预测值是所有基础块前向分量的累加结果。

优势

可解释性：通过堆叠不同类型的基础块（如趋势块和季节性块），N-Beats能够明确地分解时间序列的不同成分，提高模型的可解释性。
灵活性：N-Beats可以处理不同长度和频率的时间序列数据，并且支持多种类型的预测任务。
高效性：由于不依赖递归结构，N-Beats在训练和推理过程中具有较高的计算效率。
鲁棒性：N-Beats在处理噪声和异常值方面表现出色，能够在一定程度上抵抗这些因素的影响。

实现步骤

数据准备

收集数据：获取时间序列数据及其相关的协变量。
预处理：
- 处理缺失值。
- 标准化或归一化数据。
- 提取时间特征（如月份、星期几等）。

模型构建

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset

class GenericBlock(nn.Module):
    def __init__(self, input_size, theta_size, hidden_size, num_layers):
        super(GenericBlock, self).__init__()
        layers = [nn.Linear(input_size, hidden_size), nn.ReLU()]
        for _ in range(num_layers - 1):
            layers.append(nn.Linear(hidden_size, hidden_size))
            layers.append(nn.ReLU())
        layers.append(nn.Linear(hidden_size, theta_size))
        self.layers = nn.Sequential(*layers)

    def forward(self, x):
        return self.layers(x)

class NBeatsBlock(nn.Module):
    def __init__(self, input_size, theta_size, hidden_size, num_layers, basis_function):
        super(NBeatsBlock, self).__init__()
        self.input_size = input_size
        self.theta_size = theta_size
        self.basis_function = basis_function
        self.fc = GenericBlock(input_size, theta_size, hidden_size, num_layers)

    def forward(self, x):
        backcast, forecast = self.basis_function(self.fc(x))
        return backcast, forecast

class NBeatsNet(nn.Module):
    def __init__(self, input_size, output_size, stack_types, nb_blocks_per_stack, thetas_dim, hidden_layer_units, share_weights_in_stack=False):
        super(NBeatsNet, self).__init__()
        self.blocks = []
        for stack_id in range(len(stack_types)):
            for block_id in range(nb_blocks_per_stack[stack_id]):
                if share_weights_in_stack and block_id != 0:
                    self.blocks[-1].requires_grad_(True)
                else:
                    self.blocks.append(NBeatsBlock(input_size, thetas_dim[stack_id], hidden_layer_units[stack_id], 4, stack_types[stack_id]))
        self.blocks = nn.ModuleList(self.blocks)

    def forward(self, x):
        residuals = x
        forecast = 0
        for i, block in enumerate(self.blocks):
            backcast, block_forecast = block(residuals)
            residuals = (residuals - backcast)
            forecast = forecast + block_forecast
        return forecast

# 参数设置
input_size = 10  # 输入窗口大小
output_size = 1  # 输出窗口大小
stack_types = ['generic', 'trend', 'seasonality']
nb_blocks_per_stack = [2, 2, 2]
thetas_dim = [5, 3, 7]
hidden_layer_units = [64, 64, 64]
batch_size = 32
epochs = 100

# 初始化模型
model = NBeatsNet(input_size, output_size, stack_types, nb_blocks_per_stack, thetas_dim, hidden_layer_units)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

# 数据加载
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)

# 训练模型
for epoch in range(epochs):
    for i, (inputs, targets) in enumerate(train_loader):
        optimizer.zero_grad()
        outputs = model(inputs)
        loss = criterion(outputs, targets)
        loss.backward()
        optimizer.step()
    print(f'Epoch [{epoch+1}/{epochs}], Loss: {loss.item():.4f}')