从股票价格到传感器数据：手把手教你用Python处理多变量时间序列预测（CNN实战）

从股票价格到传感器数据：手把手教你用Python处理多变量时间序列预测（CNN实战）

金融市场的波动和工业设备的运转状态看似毫不相关，但它们都有一个共同点——都能通过多变量时间序列数据进行建模预测。当我们需要同时分析开盘价、收盘价和成交量，或是监控温度、湿度和压力等多个传感器指标时，传统单变量预测方法就显得力不从心。本文将带您从零开始构建一个能够处理这类复杂数据的CNN模型，通过Python代码实战演示如何将原始数据转化为模型可用的3D张量格式。

1. 理解多变量时间序列数据的独特挑战

多变量时间序列数据就像一部交响乐，每个变量都是不同的乐器，只有协调一致才能奏出完整的旋律。在金融领域，一支股票可能同时包含开盘价、最高价、最低价、收盘价和成交量等多个维度；在工业物联网场景中，一台设备可能同时采集温度、振动、电流等多个传感器读数。

这类数据与单变量时间序列相比有三个显著特点：

1. 变量间存在相关性：不同变量之间往往不是独立的，比如股票成交量增加通常会伴随价格波动
2. 采样频率可能不同：某些传感器可能每秒采集一次数据，而另一些可能每分钟才采集一次
3. 预测目标多样化：我们可能需要预测单个变量，也可能需要预测多个变量的未来值

数据准备示例表格：

提示：在实际项目中，建议使用Pandas的DataFrame来处理这类表格数据，它提供了丰富的时间序列操作功能

2. 构建高效的数据预处理管道

原始的多变量时间序列数据通常以CSV格式存储，我们需要将其转换为适合CNN模型处理的3D张量格式。这个转换过程需要考虑以下几个关键因素：

• 滑动窗口大小：决定每个样本包含多少个时间步的历史数据
• 特征工程：是否需要标准化、差分处理或添加衍生特征
• 缺失值处理：如何填补传感器可能丢失的数据点

下面是一个完整的Python数据处理函数，它能够将二维表格数据转换为CNN需要的三维格式：

import numpy as np import pandas as pd from sklearn.preprocessing import MinMaxScaler def prepare_multivariate_data(dataframe, window_size, target_cols, scale=True): """ 将多变量时间序列DataFrame转换为CNN所需的3D格式 参数: dataframe: 包含时间序列的Pandas DataFrame window_size: 滑动窗口大小(时间步数) target_cols: 需要预测的目标列名列表 scale: 是否进行归一化处理 返回: X: 形状为(样本数, 窗口大小, 特征数)的3D数组 y: 对应的目标值 scaler: 用于反向转换的缩放器对象 """ # 数据归一化 if scale: scaler = MinMaxScaler() scaled_data = scaler.fit_transform(dataframe) dataframe = pd.DataFrame(scaled_data, columns=dataframe.columns) else: scaler = None # 初始化空列表存储样本 X, y = [], [] # 滑动窗口生成样本 for i in range(len(dataframe) - window_size): # 获取窗口内的所有特征作为输入 window = dataframe.iloc[i:i+window_size].values # 获取窗口后的目标值作为输出 targets = dataframe.iloc[i+window_size][target_cols].values X.append(window) y.append(targets) return np.array(X), np.array(y), scaler

实际应用示例：

# 假设df是我们的原始DataFrame，包含温度、湿度、压力三列 # 我们想用过去5个时间步预测下一个时间步的所有三个变量 X, y, scaler = prepare_multivariate_data( dataframe=df, window_size=5, target_cols=['温度','湿度','压力'] ) print(f"X形状: {X.shape}") # 例如 (1000, 5, 3) print(f"y形状: {y.shape}") # 例如 (1000, 3)

3. 设计适合多变量预测的CNN架构

与传统用于图像处理的2D CNN不同，时间序列数据需要使用1D CNN。对于多变量预测，我们有几种不同的架构选择：

3.1 单输出架构

当所有输入变量共同影响单个输出变量时，可以使用这种简单架构。例如，用温度、湿度和压力预测设备是否会发生故障（二分类问题）。

from tensorflow.keras.models import Sequential from tensorflow.keras.layers import Conv1D, MaxPooling1D, Flatten, Dense def build_single_output_cnn(input_shape): model = Sequential([ Conv1D(filters=64, kernel_size=3, activation='relu', input_shape=input_shape), MaxPooling1D(pool_size=2), Conv1D(filters=128, kernel_size=3, activation='relu'), Flatten(), Dense(100, activation='relu'), Dense(1, activation='sigmoid') # 二分类输出 ]) model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy']) return model

3.2 多输出架构

当需要同时预测多个相关变量时，可以使用多输出架构。例如，同时预测下一时间步的温度、湿度和压力。

from tensorflow.keras.models import Model from tensorflow.keras.layers import Input, concatenate def build_multi_output_cnn(input_shape, output_dims): inputs = Input(shape=input_shape) # 共享的特征提取层 conv1 = Conv1D(filters=64, kernel_size=3, activation='relu')(inputs) pool1 = MaxPooling1D(pool_size=2)(conv1) conv2 = Conv1D(filters=128, kernel_size=3, activation='relu')(pool1) flat = Flatten()(conv2) dense = Dense(100, activation='relu')(flat) # 多个输出层 outputs = [] for dim in output_dims: outputs.append(Dense(dim, activation='linear')(dense)) model = Model(inputs=inputs, outputs=outputs) model.compile(optimizer='adam', loss='mse') return model

架构选择指南：

4. 实战：股票价格多步预测案例

让我们通过一个具体的股票预测案例，将前面介绍的技术串联起来。假设我们有一个包含开盘价、最高价、最低价、收盘价和成交量的股票数据集，目标是预测未来3天的收盘价。

4.1 数据准备与特征工程

首先，我们加载数据并进行必要的预处理：

import yfinance as yf import pandas as pd # 下载苹果公司股票数据 data = yf.download('AAPL', start='2020-01-01', end='2023-01-01') # 添加技术指标 data['MA_10'] = data['Close'].rolling(window=10).mean() data['RSI_14'] = compute_rsi(data['Close'], 14) # 假设已实现RSI计算函数 # 删除缺失值 data = data.dropna() # 选择特征列和目标列 features = ['Open', 'High', 'Low', 'Close', 'Volume', 'MA_10', 'RSI_14'] target = ['Close'] # 划分训练测试集 train_size = int(len(data) * 0.8) train_data = data.iloc[:train_size] test_data = data.iloc[train_size:] # 准备数据 window_size = 10 X_train, y_train, scaler = prepare_multivariate_data(train_data[features], window_size, target) X_test, y_test, _ = prepare_multivariate_data(test_data[features], window_size, target, scale=False) y_test = scaler.transform(test_data[target].iloc[window_size:])

4.2 构建并训练预测模型

针对多步预测任务，我们设计一个序列到序列的CNN模型：

from tensorflow.keras.models import Sequential from tensorflow.keras.layers import Conv1D, MaxPooling1D, Flatten, Dense, RepeatVector, TimeDistributed def build_seq2seq_cnn(input_shape, output_steps): model = Sequential([ Conv1D(filters=64, kernel_size=3, activation='relu', input_shape=input_shape), MaxPooling1D(pool_size=2), Conv1D(filters=128, kernel_size=3, activation='relu'), Flatten(), Dense(100, activation='relu'), RepeatVector(output_steps), # 重复向量用于多步输出 TimeDistributed(Dense(1)) # 每个时间步一个输出 ]) model.compile(optimizer='adam', loss='mse') return model # 构建模型 model = build_seq2seq_cnn((window_size, len(features)), output_steps=3) model.fit(X_train, y_train, epochs=50, batch_size=32, validation_split=0.1)

4.3 模型评估与结果可视化

训练完成后，我们需要评估模型在测试集上的表现：

import matplotlib.pyplot as plt # 在测试集上预测 predictions = model.predict(X_test) # 反归一化 predictions = scaler.inverse_transform(predictions.reshape(-1, 1)).reshape(-1, 3) y_test_actual = scaler.inverse_transform(y_test.reshape(-1, 1)).reshape(-1, 3) # 绘制结果对比 plt.figure(figsize=(12, 6)) plt.plot(test_data.index[window_size:window_size+100], y_test_actual[:100, 0], label='实际值') plt.plot(test_data.index[window_size:window_size+100], predictions[:100, 0], label='预测值') plt.title('股票收盘价预测结果对比') plt.xlabel('日期') plt.ylabel('价格') plt.legend() plt.show()

性能优化技巧：

1. 调整滑动窗口大小：通过交叉验证找到最佳历史时间步数
2. 尝试不同的卷积核大小：较小的核适合捕捉短期模式，较大的核适合长期趋势
3. 添加注意力机制：让模型能够关注最重要的时间点和特征
4. 使用残差连接：帮助训练更深的网络而不出现梯度消失

5. 处理现实挑战：缺失值与频率不一致

真实世界的数据很少是完美的。当我们处理工业传感器数据时，常常会遇到以下问题：

• 数据缺失：某些传感器可能暂时离线
• 采样频率不一致：温度传感器每分钟采样一次，而振动传感器每秒采样多次
• 异常值：传感器可能偶尔产生不合理读数

5.1 处理缺失值的策略

def handle_missing_data(dataframe, strategy='interpolate'): """ 处理DataFrame中的缺失值 参数: dataframe: 输入数据 strategy: 处理策略，可选'interpolate'(插值), 'ffill'(前向填充), 'bfill'(后向填充) """ if strategy == 'interpolate': return dataframe.interpolate() elif strategy == 'ffill': return dataframe.ffill() elif strategy == 'bfill': return dataframe.bfill() else: raise ValueError(f"未知的缺失值处理策略: {strategy}")

5.2 解决采样频率不一致问题

对于不同频率的数据，我们有几种处理方式：

1. 上采样低频数据：使用插值方法增加低频数据的采样点
2. 下采样高频数据：对高频数据进行聚合（平均、最大、最小等）
3. 使用专门处理不规则序列的模型：如Time2Vec等特殊架构

def resample_data(dataframe, target_freq='1min'): """ 将数据重新采样到统一频率 参数: dataframe: 包含时间索引的DataFrame target_freq: 目标频率，如'1min', '1H'等 """ return dataframe.resample(target_freq).mean() # 使用平均值聚合

5.3 异常值检测与处理

def detect_and_handle_outliers(dataframe, threshold=3): """ 使用Z-score方法检测和处理异常值 参数: dataframe: 输入数据 threshold: Z-score阈值，超过此值视为异常 """ from scipy import stats df = dataframe.copy() z_scores = np.abs(stats.zscore(df)) # 用列中位数替换异常值 for col in df.columns: median = df[col].median() df[col][z_scores[col] > threshold] = median return df

6. 模型部署与生产环境考虑

将训练好的模型部署到生产环境需要考虑以下几个关键因素：

1. 实时预测需求：是否需要实时处理流数据
2. 模型更新频率：多长时间重新训练一次模型
3. 资源限制：部署环境的计算资源限制
4. 监控与日志：如何跟踪模型性能下降

部署架构示例：

[数据源] -> [流处理引擎] -> [特征工程] -> [模型服务] -> [结果存储] ↑ ↑ [数据质量监控] [模型性能监控]

6.1 使用TensorFlow Serving部署模型

# 保存模型为SavedModel格式 model.save('multivariate_cnn_model', save_format='tf') # 使用Docker启动TensorFlow Serving服务 # docker run -p 8501:8501 --name tf_serving \ # -v "$(pwd)/multivariate_cnn_model:/models/multivariate_cnn_model" \ # -e MODEL_NAME=multivariate_cnn_model -t tensorflow/serving

6.2 创建预测API服务

from flask import Flask, request, jsonify import numpy as np import pandas as pd import requests app = Flask(__name__) # 加载预处理对象和特征列表 # ... @app.route('/predict', methods=['POST']) def predict(): # 获取并预处理输入数据 raw_data = request.json df = pd.DataFrame(raw_data) processed_data = preprocess_input(df) # 调用TensorFlow Serving模型 response = requests.post( 'http://localhost:8501/v1/models/multivariate_cnn_model:predict', json={'instances': processed_data.tolist()} ) # 处理并返回预测结果 predictions = np.array(response.json()['predictions']) return jsonify({'predictions': predictions.tolist()}) if __name__ == '__main__': app.run(host='0.0.0.0', port=5000)

7. 进阶技巧与优化方向

当您掌握了多变量时间序列预测的基础后，可以考虑以下进阶方向提升模型性能：

7.1 混合架构：CNN与LSTM结合

from tensorflow.keras.models import Sequential from tensorflow.keras.layers import Conv1D, MaxPooling1D, LSTM, Dense def build_cnn_lstm_hybrid(input_shape): model = Sequential([ Conv1D(filters=64, kernel_size=3, activation='relu', input_shape=input_shape), MaxPooling1D(pool_size=2), LSTM(100, return_sequences=True), LSTM(50), Dense(1) ]) model.compile(optimizer='adam', loss='mse') return model

7.2 注意力机制增强

from tensorflow.keras.layers import Layer import tensorflow as tf class TemporalAttention(Layer): def __init__(self, units): super(TemporalAttention, self).__init__() self.W1 = Dense(units) self.W2 = Dense(units) self.V = Dense(1) def call(self, features): # 计算注意力分数 attention_hidden = tf.nn.tanh(self.W1(features) + self.W2(features)) score = self.V(attention_hidden) # 计算注意力权重 attention_weights = tf.nn.softmax(score, axis=1) # 应用权重 context_vector = attention_weights * features context_vector = tf.reduce_sum(context_vector, axis=1) return context_vector

7.3 超参数优化

使用Keras Tuner自动寻找最佳超参数组合：

import keras_tuner as kt def build_model(hp): model = Sequential() # 可调的卷积层参数 for i in range(hp.Int('num_layers', 1, 3)): model.add(Conv1D( filters=hp.Int(f'filters_{i}', min_value=32, max_value=256, step=32), kernel_size=hp.Int(f'kernel_{i}', min_value=2, max_value=5), activation='relu' )) if hp.Boolean(f'maxpool_{i}'): model.add(MaxPooling1D()) model.add(Flatten()) # 可调的密集层参数 for i in range(hp.Int('dense_layers', 1, 2)): model.add(Dense( units=hp.Int(f'dense_units_{i}', min_value=32, max_value=256, step=32), activation='relu' )) model.add(Dense(1)) model.compile( optimizer=hp.Choice('optimizer', ['adam', 'rmsprop']), loss='mse' ) return model tuner = kt.RandomSearch( build_model, objective='val_loss', max_trials=20, executions_per_trial=2, directory='tuning', project_name='multivariate_cnn' ) tuner.search(X_train, y_train, epochs=50, validation_split=0.2)

8. 实际应用中的经验分享

在多个工业项目中应用多变量时间序列预测模型后，我总结了以下几点实战经验：

1. 数据质量比模型复杂更重要：清洗良好的数据配合简单模型，往往比原始数据配合复杂模型效果更好
2. 合理设置评估指标：对于不同应用场景，选择合适的评估指标（如MAE、MAPE或自定义指标）
3. 考虑预测不确定性：在某些关键应用中，提供预测区间比单点预测更有价值
4. 模型可解释性：使用SHAP或LIME等工具解释模型预测，增加业务人员信任度
5. 持续监控与更新：建立模型性能下降的检测机制和定期更新流程

常见问题解决速查表：

多变量时间序列预测是一个强大但复杂的工具，需要数据科学、领域知识和工程实践的紧密结合。通过本文介绍的技术栈，您应该能够构建出适应各种场景的预测解决方案。记住，没有放之四海而皆准的最佳模型，只有最适合特定问题和数据的解决方案。