突破LSTM！结合决策树时间序列预测

来源：微信公众号 | 整理：实用工具站

哈喽，我是小白~

今天聊聊结合LSTM与决策树的混合时间序列模型~

LSTM 就像一个有记忆力的“大脑”，它能记住以前的数据，理解时间上的变化规律，比如“昨天热，今天还热，那明天可能也热”。

决策树像一个“会做选择的流程图”，它能找出“如果...那么...”的规则，比如“如果温度高而湿度低，那不下雨”。

混合模型的思路是：用LSTM来理解时间变化趋势，再交给决策树做更精细的判断或预测。

打个比方：

LSTM 是“观察天气变化的老农民”，他凭经验知道最近的天气走向；
决策树是“科学小助手”，他用很多规则来进一步判断会不会下雨。

这两者结合起来，预测效果更稳~

核心原理

1. 为什么要混合 LSTM 与决策树？

单独使用 LSTM：善于处理序列信息，但不一定对非线性特征、离散值、分类变量处理得好。

单独使用决策树：善于处理非线性特征、变量选择，但不懂“顺序”和“时间上的依赖”。

它们结合起来：

LSTM 处理时间序列中“过去 → 现在”的演化关系，提取深层时间特征；
决策树进一步处理 LSTM 提取的特征，构建非线性的决策规则。

2. 混合方式有几种？

以下是常见的两种组合方式：

模型1：串联式（LSTM + 决策树）

先用 LSTM 提取特征，再输入到决策树做最终预测。

流程：

模型2：集成式（双模型加权）

LSTM 和决策树各自预测，然后加权组合。

流程：

公式说明

我们以模型1（LSTM + 决策树）为例。

1. LSTM 模型公式（核心记忆机制）

设时间序列输入为

LSTM 单元计算如下：

输入门：
遗忘门：
候选记忆：
更新记忆单元：
输出门：
最终输出：

得到 LSTM 输出的隐藏状态向量，即时间特征。

2. 决策树预测

将上一步的作为输入特征向量，输入到决策树进行预测。

决策树根据划分规则对进行一系列 if-then 的分裂判断，最终落到某个叶子节点得到预测值。

完整案例

将LSTM与决策树结合形成“混合时间序列预测模型，可以更好地建模实际数据中的复杂性。

模型结构：

第一阶段：使用LSTM模型对时间序列进行主预测；
第二阶段：利用决策树回归器学习LSTM的残差（预测误差）；
组合预测：最终预测值 = LSTM预测 + 决策树预测残差。

数据流流程：

模拟生成时间序列（含多周期、季节性和噪声）；
将序列切片为监督学习格式；
使用80%的前序数据训练模型（严格时间顺序，避免数据泄露）；
剩余20%作为测试集；
可视化与分析模型效果。

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from sklearn.tree import DecisionTreeRegressor
from sklearn.metrics import mean_squared_error, r2_score
from sklearn.model_selection import train_test_split
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
import random

np.random.seed(42)
torch.manual_seed(42)
random.seed(42)

# 1. 虚拟时间序列数据
"""
我们构造一个非线性的时间序列：
y(t) = sin(t) + 0.5*sin(3t) + noise + seasonality
"""
T = 300
t = np.arange(0, T)
y = np.sin(0.2 * t) + 0.5 * np.sin(0.6 * t) + 0.3 * np.sin(0.05 * t + np.pi/4)
y += 0.3 * np.random.randn(T)  # 添加噪声

plt.figure(figsize=(10, 4))
plt.plot(t, y, color='darkorange')
plt.title("Time Series Overview", fontsize=14)
plt.xlabel("Time")
plt.ylabel("Value")
plt.grid(True)
plt.tight_layout()
plt.show()

# 2. 构造监督学习问题（避免数据泄露）
def create_sequences(data, seq_length):
    X, y = [], []
    for i in range(len(data) - seq_length):
        X.append(data[i:i+seq_length])
        y.append(data[i+seq_length])
    return np.array(X), np.array(y)

seq_length = 20
X, y_target = create_sequences(y, seq_length)

# 分为训练集和测试集（避免未来信息泄露）
split_index = int(len(X) * 0.8)
X_train, y_train = X[:split_index], y_target[:split_index]
X_test, y_test = X[split_index:], y_target[split_index:]

# 3. 构建LSTM模型
class LSTMRegressor(nn.Module):
    def __init__(self, input_size=1, hidden_size=50, num_layers=1):
        super(LSTMRegressor, self).__init__()
        self.hidden_size = hidden_size
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, 1)

    def forward(self, x):
        out, _ = self.lstm(x)
        out = self.fc(out[:, -1, :])
        return out

# 自定义Dataset
class TimeSeriesDataset(Dataset):
    def __init__(self, X, y):
        self.X = torch.tensor(X, dtype=torch.float32).unsqueeze(-1)
        self.y = torch.tensor(y, dtype=torch.float32).unsqueeze(-1)

    def __len__(self):
        return len(self.X)

    def __getitem__(self, idx):
        return self.X[idx], self.y[idx]

train_dataset = TimeSeriesDataset(X_train, y_train)
test_dataset = TimeSeriesDataset(X_test, y_test)

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

# 初始化模型
model = LSTMRegressor()
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.01)

# 4. 模型训练
num_epochs = 100
loss_list = []

for epoch in range(num_epochs):
    model.train()
    total_loss = 0
    for seqs, targets in train_loader:
        optimizer.zero_grad()
        outputs = model(seqs)
        loss = criterion(outputs, targets)
        loss.backward()
        optimizer.step()
        total_loss += loss.item()
    avg_loss = total_loss / len(train_loader)
    loss_list.append(avg_loss)
    if epoch % 10 == 0:
        print(f"Epoch [{epoch}/{num_epochs}], Loss: {avg_loss:.4f}")

# 图形1：训练损失下降曲线
plt.figure(figsize=(8, 4))
plt.plot(loss_list, color='crimson')
plt.title("Training Loss over Epochs")
plt.xlabel("Epoch")
plt.ylabel("Loss")
plt.grid(True)
plt.show()

# 5. LSTM预测 + 决策树增强
model.eval()
X_test_tensor = torch.tensor(X_test, dtype=torch.float32).unsqueeze(-1)
lstm_preds = model(X_test_tensor).detach().numpy().flatten()

# 决策树基于LSTM残差预测
residuals = y_test - lstm_preds

# 特征扩展
X_tree_train = lstm_preds.reshape(-1, 1)
model_tree = DecisionTreeRegressor(max_depth=3)
model_tree.fit(X_tree_train, residuals)
residual_preds = model_tree.predict(X_tree_train)
final_preds = lstm_preds + residual_preds

# 图形2：LSTM与真实值对比
plt.figure(figsize=(10, 4))
plt.plot(y_test, label="True", color='blue')
plt.plot(lstm_preds, label="LSTM Prediction", color='orange')
plt.title("LSTM vs True Value")
plt.legend()
plt.grid(True)
plt.show()

# 图形3：混合模型预测对比
plt.figure(figsize=(10, 4))
plt.plot(y_test, label="True", color='blue')
plt.plot(final_preds, label="LSTM + Tree Hybrid", color='green')
plt.title("Hybrid Model vs True Value")
plt.legend()
plt.grid(True)
plt.show()

# 图形4：残差分布图（增强前后）
plt.figure(figsize=(10, 4))
sns.histplot(y_test - lstm_preds, color='red', label='LSTM Residuals', kde=True)
sns.histplot(y_test - final_preds, color='green', label='Hybrid Residuals', kde=True)
plt.title("Residual Distribution: LSTM vs Hybrid")
plt.legend()
plt.grid(True)
plt.show()

# 评估指标
print("LSTM模型 MSE:", mean_squared_error(y_test, lstm_preds))
print("混合模型 MSE:", mean_squared_error(y_test, final_preds))
print("LSTM模型 R2:", r2_score(y_test, lstm_preds))
print("混合模型 R2:", r2_score(y_test, final_preds))