# 基于PyTorch的CNN-BiLSTM时序预测全流程实现

时序预测在金融、气象、能源等领域具有广泛应用价值。本文将详细介绍如何使用PyTorch构建CNN-BiLSTM混合模型，实现从数据预处理、模型构建到部署应用的完整流程。

## 混合架构设计思路

CNN-BiLSTM模型结合了卷积神经网络提取局部特征和双向长短时记忆网络捕捉长时序依赖的优势。CNN层能有效提取时间序列的局部模式和趋势，BiLSTM层则能从前向和后向两个方向捕捉序列的时间依赖关系。

```python

import torch

import torch.nn as nn

import numpy as np

import pandas as pd

from sklearn.preprocessing import MinMaxScaler

# 数据滑窗处理类

class TimeSeriesDataset(torch.utils.data.Dataset):

    def __init__(self, data, window_size, forecast_horizon=1):

        """

        初始化时间序列数据集

        data: 输入数据

        window_size: 滑动窗口大小

        forecast_horizon: 预测步长

        """

        self.data = data

        self.window_size = window_size

        self.forecast_horizon = forecast_horizon

        

    def __len__(self):

        return len(self.data) - self.window_size - self.forecast_horizon + 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.forecast_horizon]

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

```

## 数据预处理与特征工程

```python

class DataPreprocessor:

    def __init__(self, scaler_type='minmax'):

        if scaler_type == 'minmax':

            self.scaler = MinMaxScaler()

        

    def prepare_data(self, file_path, window_size=60, test_ratio=0.2):

        """

        准备时序数据

        """

        # 读取数据

        df = pd.read_csv(file_path)

        

        # 选择特征列（假设最后一列为目标变量）

        feature_columns = df.columns[:-1]

        target_column = df.columns[-1]

        

        # 数据标准化

        features_scaled = self.scaler.fit_transform(df[feature_columns])

        target_scaled = self.scaler.fit_transform(df[[target_column]])

        

        # 合并特征和目标

        data_scaled = np.hstack([features_scaled, target_scaled])

        

        # 划分训练集和测试集

        train_size = int(len(data_scaled) * (1 - test_ratio))

        train_data = data_scaled[:train_size]

        test_data = data_scaled[train_size:]

        

        return train_data, test_data

    

    def create_data_loaders(self, data, window_size, batch_size=32):

        """创建数据加载器"""

        dataset = TimeSeriesDataset(data, window_size)

        dataloader = torch.utils.data.DataLoader(

            dataset, 

            batch_size=batch_size, 

            shuffle=True

        )

        return dataloader

```

## CNN-BiLSTM模型构建

```python

class CNNBiLSTMModel(nn.Module):

    def __init__(self, input_size, hidden_size, num_layers, output_size, kernel_size=3):

        super(CNNBiLSTMModel, self).__init__()

        

        # 卷积层用于提取局部特征

        self.conv1d = nn.Conv1d(

            in_channels=input_size,

            out_channels=hidden_size,

            kernel_size=kernel_size,

            padding='same'

        )

        

        # 批归一化层

        self.batch_norm = nn.BatchNorm1d(hidden_size)

        

        # ReLU激活函数

        self.relu = nn.ReLU()

        

        # 双向LSTM层

        self.bilstm = nn.LSTM(

            input_size=hidden_size,

            hidden_size=hidden_size,

            num_layers=num_layers,

            batch_first=True,

            bidirectional=True

        )

        

        # Dropout层防止过拟合

        self.dropout = nn.Dropout(0.3)

        

        # 全连接输出层

        self.fc = nn.Linear(hidden_size * 2, output_size)  # *2因为双向LSTM

        

    def forward(self, x):

        # x形状: (batch_size, seq_len, input_size)

        

        # 调整维度用于卷积层

        x = x.permute(0, 2, 1)  # (batch_size, input_size, seq_len)

        

        # CNN处理

        conv_out = self.conv1d(x)

        conv_out = self.batch_norm(conv_out)

        conv_out = self.relu(conv_out)

        

        # 调整维度用于LSTM

        conv_out = conv_out.permute(0, 2, 1)  # (batch_size, seq_len, hidden_size)

        

        # BiLSTM处理

        lstm_out, _ = self.bilstm(conv_out)

        

        # 应用dropout

        lstm_out = self.dropout(lstm_out)

        <"r4.j9k5.org.cn"><"k6.j9k5.org.cn"><"p3.j9k5.org.cn">

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

        last_output = lstm_out[:, -1, :]

        

        # 全连接层

        output = self.fc(last_output)

        

        return output

```

## 模型训练与验证

```python

class ModelTrainer:

    def __init__(self, model, device='cuda' if torch.cuda.is_available() else 'cpu'):

        self.model = model.to(device)

        self.device = device

        self.criterion = nn.MSELoss()

        self.optimizer = torch.optim.Adam(model.parameters(), lr=0.001)

        self.scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(

            self.optimizer, mode='min', patience=10

        )

        

    def train_epoch(self, train_loader):

        """训练一个epoch"""

        self.model.train()

        total_loss = 0

        

        for batch_idx, (data, target) in enumerate(train_loader):

            data, target = data.to(self.device), target.to(self.device)

            

            self.optimizer.zero_grad()

            output = self.model(data)

            loss = self.criterion(output, target)

            loss.backward()

            

            # 梯度裁剪防止梯度爆炸

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

            

            self.optimizer.step()

            total_loss += loss.item()

            

        return total_loss / len(train_loader)

    

    def validate(self, val_loader):

        """验证模型"""

        self.model.eval()

        total_loss = 0

        

        with torch.no_grad():

            for data, target in val_loader:

                data, target = data.to(self.device), target.to(self.device)

                output = self.model(data)

                loss = self.criterion(output, target)

                total_loss += loss.item()

                

        return total_loss / len(val_loader)

    

    def train(self, train_loader, val_loader, epochs=100):

        """完整训练过程"""

        train_losses = []

        val_losses = []

        

        for epoch in range(epochs):

            train_loss = self.train_epoch(train_loader)

            val_loss = self.validate(val_loader)

            

            train_losses.append(train_loss)

            val_losses.append(val_loss)

            

            # 调整学习率

            self.scheduler.step(val_loss)

            

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

                print(f'Epoch [{epoch+1}/{epochs}], '

                      f'Train Loss: {train_loss:.6f}, '

                      f'Val Loss: {val_loss:.6f}')

                

        return train_losses, val_losses

```

## 模型评估与可视化

```python

import matplotlib.pyplot as plt

from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score

class ModelEvaluator:

    def __init__(self, model, scaler):

        self.model = model

        self.scaler = scaler

        

    def evaluate(self, test_loader, device):

        """评估模型性能"""

        self.model.eval()

        predictions = []

        actuals = []

        

        with torch.no_grad():

            for data, target in test_loader:

                data = data.to(device)

                output = self.model(data)

                

                # 反标准化

                output_np = output.cpu().numpy()

                target_np = target.numpy()

                

                # 如果使用了标准化，需要反标准化

                if self.scaler is not None:

                    output_np = self.scaler.inverse_transform(output_np)

                    target_np = self.scaler.inverse_transform(target_np)

                

                predictions.append(output_np)

                actuals.append(target_np)

        

        predictions = np.vstack(predictions)

        actuals = np.vstack(actuals)

        

        # 计算评估指标

        mse = mean_squared_error(actuals, predictions)

        mae = mean_absolute_error(actuals, predictions)

        r2 = r2_score(actuals, predictions)

        

        print(f'MSE: {mse:.4f}')

        print(f'MAE: {mae:.4f}')

        print(f'R² Score: {r2:.4f}')

        

        return predictions, actuals, {'MSE': mse, 'MAE': mae, 'R2': r2}

    

    def plot_results(self, predictions, actuals, title="预测结果对比"):

        """可视化预测结果"""

        plt.figure(figsize=(12, 6))

        

        # 只展示前100个样本

        sample_size = min(100, len(predictions))

        plt.plot(actuals[:sample_size], label='实际值', alpha=0.7)

        plt.plot(predictions[:sample_size], label='预测值', alpha=0.7)

        

        plt.title(title)

        plt.xlabel('时间步')

        plt.ylabel('数值')

        plt.legend()

        plt.grid(True, alpha=0.3)

        plt.show()

```

## 模型部署与应用

```python

class ModelDeployer:

    def __init__(self, model_path=None):

        self.model = None

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

        

        if model_path:

            self.load_model(model_path)

    <"m7.j9k5.org.cn"><"a9.j9k5.org.cn"><"d2.j9k5.org.cn">

    def save_model(self, model, path, scaler=None):

        """保存模型和预处理对象"""

        torch.save({

            'model_state_dict': model.state_dict(),

            'scaler': scaler

        }, path)

        print(f"模型已保存到: {path}")

    

    def load_model(self, path):

        """加载模型"""

        checkpoint = torch.load(path, map_location=self.device)

        

        # 假设我们知道模型结构

        model = CNNBiLSTMModel(

            input_size=checkpoint.get('input_size', 10),

            hidden_size=checkpoint.get('hidden_size', 64),

            num_layers=checkpoint.get('num_layers', 2),

            output_size=checkpoint.get('output_size', 1)

        )

        

        model.load_state_dict(checkpoint['model_state_dict'])

        model.to(self.device)

        model.eval()

        

        self.model = model

        self.scaler = checkpoint.get('scaler')

        

        print("模型加载完成")

        return model

    

    def predict(self, input_data):

        """实时预测"""

        if self.model is None:

            raise ValueError("模型未加载")

        

        self.model.eval()

        with torch.no_grad():

            # 预处理输入数据

            if self.scaler:

                input_data = self.scaler.transform(input_data)

            

            input_tensor = torch.FloatTensor(input_data).unsqueeze(0).to(self.device)

            prediction = self.model(input_tensor)

            

            # 反标准化

            if self.scaler:

                prediction = self.scaler.inverse_transform(

                    prediction.cpu().numpy()

                )

            

            return prediction[0]

    

    def create_web_api(self, app):

        """创建Web API接口（示例）"""

        from flask import request, jsonify

        

        @app.route('/predict', methods=['POST'])

        def predict_endpoint():

            try:

                data = request.json['data']

                prediction = self.predict(data)

                return jsonify({'prediction': prediction.tolist()})

            except Exception as e:

                return jsonify({'error': str(e)}), 400

```

## 完整工作流程示例

```python

def main():

    # 1. 数据准备

    preprocessor = DataPreprocessor()

    train_data, test_data = preprocessor.prepare_data(

        'time_series_data.csv', 

        window_size=60

    )

    

    # 2. 创建数据加载器

    train_loader = preprocessor.create_data_loaders(train_data, window_size=60)

    test_loader = preprocessor.create_data_loaders(test_data, window_size=60)

    

    # 3. 初始化模型

    model = CNNBiLSTMModel(

        input_size=train_data.shape[1],

        hidden_size=64,

        num_layers=2,

        output_size=1

    )

    

    # 4. 训练模型

    trainer = ModelTrainer(model)

    train_losses, val_losses = trainer.train(train_loader, test_loader, epochs=50)

    

    # 5. 评估模型

    evaluator = ModelEvaluator(model, preprocessor.scaler)

    predictions, actuals, metrics = evaluator.evaluate(test_loader, trainer.device)

    evaluator.plot_results(predictions, actuals)

    

    # 6. 保存模型

    deployer = ModelDeployer()

    deployer.save_model(model, 'cnn_bilstm_model.pth', preprocessor.scaler)

    

    return model, metrics

if __name__ == "__main__":

    main()

```

## 总结与优化建议

本文介绍了基于PyTorch的CNN-BiLSTM时序预测模型的完整实现流程。该混合模型能够有效提取时间序列的局部特征和长时依赖关系，适用于各种时序预测任务。

优化方向：

1. **特征工程**：结合领域知识添加更有意义的特征

2. **模型结构**：尝试不同层数、隐藏单元数的组合

3. **注意力机制**：可加入注意力层提高重要时间步的关注度

4. **集成学习**：结合多个模型的预测结果

5. **超参数调优**：使用网格搜索或贝叶斯优化寻找最佳参数

实际应用中应根据具体业务场景和数据特点进行调整，特别注意过拟合问题，通过正则化、早停等策略保证模型泛化能力。模型部署时需考虑实时性和资源消耗的平衡。