时间序列预测实战指南：从ARIMA到LSTM与Transformer的完整建模方案与Python代码详解

时间序列预测是机器学习中最常见也最具挑战性的任务之一。无论是销售预测、流量预估、股票分析还是能源负荷预测，时间序列数据无处不在。与常规表格数据不同，时间序列数据具有时序依赖性、季节性和趋势性等独特特征，需要专门的建模方法。本文将从经典统计方法到现代深度学习模型，系统讲解时间序列预测的核心技术，并给出完整的 Python 实战代码。

一、时间序列数据的核心特征

在动手建模之前，必须先理解时间序列数据的几个核心构成要素。任何时间序列都可以分解为以下四个组成部分：

• 趋势（Trend）：数据在长期内呈现的上升或下降走势，例如电商年交易额持续增长。
• 季节性（Seasonality）：数据中重复出现的周期性波动，例如冰淇淋销量每年夏天达到高峰。
• 周期性（Cyclic）：非固定周期的波动，通常与经济周期相关。
• 残差/噪声（Residual）：剔除趋势和季节性后的随机波动。

使用 Python 的

库可以轻松进行时间序列分解：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import pandas as pd

import matplotlib.pyplot as plt

from statsmodels.tsa.seasonal import seasonal_decompose



# 加载数据

df = pd.read_csv('sales.csv', parse_dates=['date'], index_col='date')



# 加法模型分解（当季节性波动幅度不随趋势变化时使用）

decomposition = seasonal_decompose(df['value'], model='additive', period=12)



# 乘法模型分解（当季节性波动幅度随趋势按比例变化时使用）

# decomposition = seasonal_decompose(df['value'], model='multiplicative', period=12)



fig, axes = plt.subplots(4, 1, figsize=(14, 10))

decomposition.observed.plot(ax=axes[0], title='原始数据')

decomposition.trend.plot(ax=axes[1], title='趋势')

decomposition.seasonal.plot(ax=axes[2], title='季节性')

decomposition.resid.plot(ax=axes[3], title='残差')

plt.tight_layout()

plt.savefig('decomposition.png', dpi=150)

平稳性检验：ADF检验

许多经典模型（如 ARIMA）要求数据是平稳的，即均值和方差不随时间变化。Augmented Dickey-Fuller（ADF）检验是最常用的平稳性检验方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
from statsmodels.tsa.stattools import adfuller



def adf_test(series, title=''):

    result = adfuller(series.dropna(), autolag='AIC')

    labels = ['ADF统计量', 'p值', '使用滞后阶数', '观测数', '临界值']

    print(f'--- {title} ---')

    for i, label in enumerate(labels[:4]):

        print(f'{label}: {result[i]:.6f}')

    for key, val in result[4].items():

        print(f'  临界值({key}): {val:.4f}')

    if result[1] <= 0.05:

        print('=> 序列是平稳的（拒绝原假设）')

    else:

        print('=> 序列非平稳（不能拒绝原假设）')



adf_test(df['value'], '原始序列')



# 差分使其平稳

df['value_diff1'] = df['value'].diff()

adf_test(df['value_diff1'], '一阶差分后')

如果原始序列非平稳，通常通过差分（

）来消除趋势，必要时还需进行季节性差分。

二、经典统计方法：ARIMA 与 SARIMA

ARIMA（AutoRegressive Integrated Moving Average）是时间序列预测的经典模型，包含三个核心参数

：

• p（自回归阶数）：使用过去多少期的值来预测当前值。
• d（差分阶数）：对数据做几次差分使其平稳。
• q（移动平均阶数）：使用过去多少期的预测误差来修正当前预测。

SARIMA 在 ARIMA 基础上增加了季节性参数

，其中

是季节周期长度。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from statsmodels.tsa.statespace.sarimax import SARIMAX

from sklearn.metrics import mean_absolute_error, mean_squared_error

import numpy as np



# 划分训练集和测试集

train_size = int(len(df) * 0.8)

train, test = df['value'][:train_size], df['value'][train_size:]



# 构建 SARIMA 模型

# order=(p,d,q), seasonal_order=(P,D,Q,m)

model = SARIMAX(

    train,

    order=(1, 1, 1),

    seasonal_order=(1, 1, 1, 12),

    enforce_stationarity=False,

    enforce_invertibility=False

)

results = model.fit(disp=False)

print(results.summary())



# 预测

forecast = results.forecast(steps=len(test))



# 评估

mae = mean_absolute_error(test, forecast)

rmse = np.sqrt(mean_squared_error(test, forecast))

mape = np.mean(np.abs((test - forecast) / test)) * 100

print(f'MAE: {mae:.2f}, RMSE: {rmse:.2f}, MAPE: {mape:.2f}%')

自动参数选择：pmdarima

手动选择 ARIMA 参数既繁琐又依赖经验。

库提供了自动搜索最优参数的功能：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import pmdarima as pm



# 自动搜索最优 SARIMA 参数

auto_model = pm.auto_arima(

    train,

    seasonal=True,

    m=12,               # 季节周期

    d=None,             # 自动确定差分阶数

    D=None,             # 自动确定季节性差分阶数

    start_p=0, start_q=0,

    max_p=3, max_q=3,

    start_P=0, start_Q=0,

    max_P=2, max_Q=2,

    trace=True,         # 打印搜索过程

    error_action='ignore',

    suppress_warnings=True,

    stepwise=True       # 使用步进搜索加速

)



print(f'最优参数: order={auto_model.order}, seasonal_order={auto_model.seasonal_order}')



# 预测

forecast_auto = auto_model.predict(n_periods=len(test))

print(f'Auto ARIMA MAE: {mean_absolute_error(test, forecast_auto):.2f}')

三、Prophet：Facebook 的开源预测利器

Prophet 是 Facebook 开源的时间序列预测工具，特别适合处理包含强季节性和多节假日效应的商业数据。它采用加法模型，将趋势、季节性、节假日效应和噪声分别建模后叠加。

Prophet 的核心优势在于：

• 无需手动调参，默认参数在大多数场景下就能给出不错的结果
• 能自动处理缺失值和异常值
• 支持自定义节假日效应
• 同时建模日、周、年多种季节性
• 输出丰富的可解释性组件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from prophet import Prophet

import pandas as pd



# Prophet 要求列名必须为 ds 和 y

prophet_df = df.reset_index().rename(columns={'date': 'ds', 'value': 'y'})



train_p = prophet_df.iloc[:train_size]

test_p = prophet_df.iloc[train_size:]



# 创建模型并添加节假日

model = Prophet(

    yearly_seasonality=True,

    weekly_seasonality=True,

    daily_seasonality=False,

    changepoint_prior_scale=0.05,  # 控制趋势灵活度

    seasonality_prior_scale=10     # 控制季节性强度

)



# 添加中国节假日（自定义）

holidays = pd.DataFrame({

    'holiday': 'spring_festival',

    'ds': pd.to_datetime(['2023-01-22', '2024-02-10', '2025-01-29']),

    'lower_window': -7,   # 节前7天开始影响

    'upper_window': 7,    # 节后7天结束影响

})

model = Prophet(holidays=holidays)

model.fit(train_p)



# 预测

future = model.make_future_dataframe(periods=len(test_p), freq='D')

forecast_p = model.predict(future)



# 查看预测组件

fig = model.plot_components(forecast_p)

fig.savefig('prophet_components.png', dpi=150)

Prophet 的

是最关键的超参数：值越大，趋势越灵活（可能过拟合）；值越小，趋势越平滑（可能欠拟合）。推荐在 0.01 到 0.5 之间网格搜索。

四、深度学习方法：LSTM 时序建模

当数据量较大、非线性关系复杂时，深度学习模型往往能超越传统统计方法。LSTM（长短期记忆网络）通过门控机制有效缓解了 RNN 的梯度消失问题，成为时间序列建模的经典深度学习方案。

数据预处理与滑动窗口

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import torch

import torch.nn as nn

from torch.utils.data import Dataset, DataLoader

from sklearn.preprocessing import MinMaxScaler



# 归一化（LSTM对输入scale敏感）

scaler = MinMaxScaler(feature_range=(-1, 1))

scaled_data = scaler.fit_transform(df[['value']].values)



# 滑动窗口构建数据集

class TimeSeriesDataset(Dataset):

    def __init__(self, data, window_size=30, pred_len=1):

        self.data = data

        self.window_size = window_size

        self.pred_len = pred_len



    def __len__(self):

        return len(self.data) - self.window_size - self.pred_len + 1



    def __getitem__(self, idx):

        x = self.data[idx:idx + self.window_size]

        y = self.data[idx + self.window_size:idx + self.window_size + self.pred_len]

        return torch.FloatTensor(x), torch.FloatTensor(y)



WINDOW = 30  # 用过去30天预测未来1天

dataset = TimeSeriesDataset(scaled_data, window_size=WINDOW)

train_dataset = torch.utils.data.Subset(dataset, range(0, train_size - WINDOW))

test_dataset = torch.utils.data.Subset(dataset, range(train_size - WINDOW, len(dataset)))



train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)

test_loader = DataLoader(test_dataset, batch_size=1, shuffle=False)

LSTM 模型定义与训练

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class LSTMModel(nn.Module):

    def __init__(self, input_dim=1, hidden_dim=64, num_layers=2, output_dim=1, dropout=0.2):

        super().__init__()

        self.hidden_dim = hidden_dim

        self.num_layers = num_layers



        self.lstm = nn.LSTM(

            input_dim, hidden_dim, num_layers,

            batch_first=True, dropout=dropout

        )

        self.fc = nn.Linear(hidden_dim, output_dim)



    def forward(self, x):

        # x shape: (batch, seq_len, input_dim)

        out, (hn, cn) = self.lstm(x)

        # 取最后一个时间步的输出

        out = self.fc(out[:, -1, :])

        return out



device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

model = LSTMModel(hidden_dim=64, num_layers=2, dropout=0.2).to(device)

criterion = nn.MSELoss()

optimizer = torch.optim.Adam(model.parameters(), lr=0.001, weight_decay=1e-5)

scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(

    optimizer, mode='min', factor=0.5, patience=5

)



# 训练循环

EPOCHS = 50

for epoch in range(EPOCHS):

    model.train()

    train_loss = 0

    for batch_x, batch_y in train_loader:

        batch_x, batch_y = batch_x.to(device), batch_y.to(device)

        optimizer.zero_grad()

        outputs = model(batch_x)

        loss = criterion(outputs, batch_y)

        loss.backward()

        torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)

        optimizer.step()

        train_loss += loss.item()



    train_loss /= len(train_loader)

    scheduler.step(train_loss)



    if (epoch + 1) % 10 == 0:

        print(f'Epoch [{epoch+1}/{EPOCHS}], Loss: {train_loss:.6f}')

训练完成后进行预测并反归一化：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
model.eval()

predictions = []

with torch.no_grad():

    for batch_x, _ in test_loader:

        batch_x = batch_x.to(device)

        pred = model(batch_x)

        predictions.append(pred.cpu().numpy())



predictions = np.array(predictions).reshape(-1, 1)

predictions = scaler.inverse_transform(predictions)

y_true = scaler.inverse_transform(scaled_data[train_size:])



lstm_mae = mean_absolute_error(y_true[:len(predictions)], predictions)

print(f'LSTM MAE: {lstm_mae:.2f}')

五、多变量预测与多步预测

实际业务中，我们往往需要利用多个相关变量来预测目标值。例如，预测电力负荷时，除了历史负荷数据，温度、湿度和节假日信息都能提供额外信号。

多变量 LSTM 输入构造

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
# 假设有多个特征列

features = df[['value', 'temperature', 'humidity', 'is_holiday']].values

scaler_multi = MinMaxScaler(feature_range=(-1, 1))

scaled_multi = scaler_multi.fit_transform(features)



# 多变量数据集

class MultiVarDataset(Dataset):

    def __init__(self, data, target_col=0, window_size=30, pred_len=1):

        self.data = data

        self.target_col = target_col

        self.window_size = window_size

        self.pred_len = pred_len



    def __len__(self):

        return len(self.data) - self.window_size - self.pred_len + 1



    def __getitem__(self, idx):

        x = self.data[idx:idx + self.window_size]         # 所有特征

        y = self.data[idx + self.window_size:idx + self.window_size + self.pred_len, self.target_col]

        return torch.FloatTensor(x), torch.FloatTensor(y)



# 模型只需修改 input_dim

model_multi = LSTMModel(input_dim=4, hidden_dim=64, num_layers=2).to(device)

多步预测：Seq2Seq 与 Teacher Forcing

当需要一次预测未来多个时间步时，Seq2Seq（编码器-解码器）架构是更合适的选择：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class Encoder(nn.Module):

    def __init__(self, input_dim, hidden_dim, num_layers=2, dropout=0.2):

        super().__init__()

        self.lstm = nn.LSTM(input_dim, hidden_dim, num_layers,

                            batch_first=True, dropout=dropout)



    def forward(self, x):

        outputs, (hidden, cell) = self.lstm(x)

        return hidden, cell





class Decoder(nn.Module):

    def __init__(self, output_dim, hidden_dim, num_layers=2, dropout=0.2):

        super().__init__()

        self.output_dim = output_dim

        self.lstm = nn.LSTM(output_dim, hidden_dim, num_layers,

                            batch_first=True, dropout=dropout)

        self.fc = nn.Linear(hidden_dim, output_dim)



    def forward(self, x, hidden, cell):

        # x shape: (batch, 1, output_dim) - 单步输入

        output, (hidden, cell) = self.lstm(x, (hidden, cell))

        pred = self.fc(output)

        return pred, hidden, cell





class Seq2Seq(nn.Module):

    def __init__(self, encoder, decoder, device):

        super().__init__()

        self.encoder = encoder

        self.decoder = decoder

        self.device = device



    def forward(self, src, trg, teacher_forcing_ratio=0.5):

        batch_size = src.shape[0]

        trg_len = trg.shape[1]

        outputs = torch.zeros(batch_size, trg_len, 1).to(self.device)



        hidden, cell = self.encoder(src)

        # 解码器第一步输入：序列最后一个值

        decoder_input = src[:, -1:, 0:1]



        for t in range(trg_len):

            decoder_output, hidden, cell = self.decoder(decoder_input, hidden, cell)

            outputs[:, t:t+1, :] = decoder_output

            # Teacher forcing: 以一定概率用真实值作为下一步输入

            teacher_force = random.random() < teacher_forcing_ratio

            decoder_input = trg[:, t:t+1, :] if teacher_force else decoder_output



        return outputs

Teacher Forcing 在训练时用真实值作为解码器输入，加快收敛速度；推理时则完全依赖模型自身的预测值逐步展开。训练过程中逐步降低

（如从 0.5 退火到 0.1）能有效缓解暴露偏差问题。

六、Transformer 用于时序预测：Informer 与 PatchTST

近年来 Transformer 架构在时间序列领域也取得了显著进展。原始 Transformer 直接应用于时序数据存在计算复杂度高和缺乏局部归纳偏置等问题，研究者提出了多种改进方案。

以 PatchTST 为例，其核心思路是将时间序列分割为多个 patch，每个 patch 类似 Vision Transformer 中的图像块，然后通过 Transformer 建模 patch 间关系：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
import torch

import torch.nn as nn



class PatchTST(nn.Module):

    def __init__(self, seq_len, patch_len=16, stride=8, d_model=128,

                 n_heads=4, n_layers=3, dropout=0.1):

        super().__init__()

        self.patch_len = patch_len

        self.stride = stride

        self.num_patches = (seq_len - patch_len) // stride + 1



        # 线性投影将 patch 映射到 d_model 维度

        self.patch_embedding = nn.Linear(patch_len, d_model)

        self.pos_encoding = nn.Parameter(torch.randn(1, self.num_patches, d_model))



        encoder_layer = nn.TransformerEncoderLayer(

            d_model=d_model, nhead=n_heads, dim_feedforward=d_model * 4,

            dropout=dropout, batch_first=True

        )

        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers=n_layers)



        # 预测头

        self.head = nn.Sequential(

            nn.Flatten(start_dim=1),

            nn.Linear(self.num_patches * d_model, 256),

            nn.GELU(),

            nn.Dropout(dropout),

            nn.Linear(256, 1)

        )



    def forward(self, x):

        # x: (batch, seq_len)

        batch_size = x.shape[0]

        patches = []

        for i in range(self.num_patches):

            start = i * self.stride

            patches.append(x[:, start:start + self.patch_len])

        # (batch, num_patches, patch_len)

        patches = torch.stack(patches, dim=1)



        # 投影 + 位置编码

        emb = self.patch_embedding(patches) + self.pos_encoding



        # Transformer 编码

        out = self.transformer(emb)



        # 预测

        return self.head(out).squeeze(-1)

七、模型评估与选型策略

时间序列预测的评估比常规机器学习任务更需要谨慎，因为数据的时间顺序不可打乱。以下是正确的交叉验证方案：

滚动预测验证（Rolling Forecast）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
from sklearn.model_selection import TimeSeriesSplit



# 使用 TimeSeriesSplit 做时序交叉验证

tscv = TimeSeriesSplit(n_splits=5, test_size=30)



scores = []

for fold, (train_idx, val_idx) in enumerate(tscv.split(df)):

    train_data = df['value'].iloc[train_idx]

    val_data = df['value'].iloc[val_idx]



    model = SARIMAX(train_data, order=(1,1,1), seasonal_order=(1,1,1,12))

    result = model.fit(disp=False)

    pred = result.forecast(steps=len(val_data))



    mae = mean_absolute_error(val_data, pred)

    scores.append(mae)

    print(f'Fold {fold+1}: MAE={mae:.2f}')



print(f'\n平均MAE: {np.mean(scores):.2f} ± {np.std(scores):.2f}')

常用评估指标对比

1
2
3
4
5
6
7
8
9
def smape(y_true, y_pred):

    denominator = (np.abs(y_true) + np.abs(y_pred)) / 2

    return np.mean(np.abs(y_true - y_pred) / (denominator + 1e-8)) * 100



def mase(y_true, y_pred, y_train, seasonality=1):

    # 朴素预测的MAE作为基准

    naive_pred = y_train.shift(seasonality).dropna()

    naive_mae = mean_absolute_error(y_train[seasonality:], naive_pred)

    return mean_absolute_error(y_true, y_pred) / naive_mae

八、实战选型建议与最佳实践

根据数据量和问题复杂度，以下是推荐的技术选型路线：

• 数据量小（<1000条）、有明确季节性：首选 SARIMA 或 Prophet。模型轻量、可解释性强。
• 中等数据量、多变量影响显著：先尝试 LightGBM 配合时序特征工程（滑动统计量、滞后特征），效果往往出乎意料的好。
• 大数据量、复杂非线性关系：LSTM 或 PatchTST。注意做好归一化和窗口大小调优。
• 长序列多步预测：Informer 或 Autoformer，效率优势明显。

特征工程增强：滞后特征与滑动窗口统计量

在使用 LightGBM 等树模型做时序预测时，特征工程是关键：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def create_features(df, target_col, lags=[1,7,14,30], windows=[7,14,30]):

    df = df.copy()

    # 滞后特征

    for lag in lags:

        df[f'lag_{lag}'] = df[target_col].shift(lag)



    # 滑动窗口统计量

    for w in windows:

        df[f'rolling_mean_{w}'] = df[target_col].shift(1).rolling(w).mean()

        df[f'rolling_std_{w}'] = df[target_col].shift(1).rolling(w).std()

        df[f'rolling_min_{w}'] = df[target_col].shift(1).rolling(w).min()

        df[f'rolling_max_{w}'] = df[target_col].shift(1).rolling(w).max()



    # 时间特征

    df['month'] = df.index.month

    df['day_of_week'] = df.index.dayofweek

    df['day_of_year'] = df.index.dayofyear

    df['is_weekend'] = (df.index.dayofweek >= 5).astype(int)



    return df



feat_df = create_features(df, 'value')

feat_df = feat_df.dropna()

print(feat_df.head())

集成策略：模型融合提升鲁棒性

将统计模型、机器学习模型和深度学习模型进行加权融合，往往能获得比单一模型更好的效果：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 各模型预测结果

pred_sarima = sarima_forecast

pred_prophet = prophet_forecast

pred_lstm = lstm_forecast



# 简单加权融合（权重可在验证集上搜索）

weights = [0.3, 0.3, 0.4]  # SARIMA, Prophet, LSTM

ensemble_pred = (

    weights[0] * pred_sarima +

    weights[1] * pred_prophet +

    weights[2] * pred_lstm

)



# 或使用 Stacking

from sklearn.linear_model import Ridge

stacking_X = np.column_stack([pred_sarima, pred_prophet, pred_lstm])

meta_model = Ridge(alpha=1.0)

meta_model.fit(stacking_X, y_true)

final_pred = meta_model.predict(stacking_X)

总结

时间序列预测没有银弹，模型选择应基于数据特征和业务需求。实践中建议遵循以下原则：先从简单的统计模型建立 baseline，再逐步引入复杂模型；始终使用时序交叉验证而非随机划分；注重特征工程的重要性，好的特征往往比复杂模型更有效；关注模型的可解释性，让业务方理解预测结果的来源。通过本文介绍的方法组合，你能构建一套从经典到前沿的完整时间序列预测工具链，应对绝大多数实际预测场景。