如何基于Keras构建多模态输入RNN实现时间序列分类任务

我们需要先把DataFrame的长表结构转换成模型能接收的多输入格式，核心是按time_series_id分组提取每个序列的50个时间步特征：

• 先对分类标签做编码，把A/B转换为0/1的数值格式
• 按序列ID分组后，必须按timestamp排序保证时间步顺序正确
• 每个模态的输入最终转换为(样本数, 50, 对应特征维度)的三维数组

import numpy as np
import pandas as pd
from sklearn.preprocessing import LabelEncoder
from sklearn.model_selection import train_test_split

# 标签编码
le = LabelEncoder()
df['time_series_label'] = le.fit_transform(df['time_series_label'])

# 按序列ID分组处理
groups = df.groupby('time_series_id')
input1_list, input2_list, input3_list, label_list = [], [], [], []

for id_val, group in groups:
    # 必须按时间戳排序，保证时序顺序正确
    group = group.sort_values('timestamp')
    input1_list.append(np.vstack(group['input1'].values)) # 单样本形状(50,768)
    input2_list.append(np.vstack(group['input2'].values)) # 单样本形状(50,25)
    input3_list.append(np.vstack(group['input3'].values)) # 单样本形状(50,496)
    label_list.append(group['time_series_label'].iloc[0]) # 同个ID标签统一取第一个即可

# 转换为模型输入格式
X1 = np.array(input1_list) # 总形状(样本总数, 50, 768)
X2 = np.array(input2_list) # 总形状(样本总数, 50, 25)
X3 = np.array(input3_list) # 总形状(样本总数, 50, 496)
y = np.array(label_list) # 总形状(样本总数,)

# 划分训练集测试集
X1_train, X1_test, X2_train, X2_test, X3_train, X3_test, y_train, y_test = train_test_split(
    X1, X2, X3, y, test_size=0.2, random_state=42
)

因为是多输入结构，所以使用Keras函数式API搭建网络：每个模态先经过专属的特征提取层，拼接后再输入LSTM层提取时序特征，最后输出分类结果。

from tensorflow.keras.models import Model
from tensorflow.keras.layers import Input, Conv1D, LSTM, Dense, Concatenate, Dropout

# 定义三个输入层，形状统一为(时间步长, 单步特征维度)
input1 = Input(shape=(50, 768), name='input1')
input2 = Input(shape=(50, 25), name='input2')
input3 = Input(shape=(50, 496), name='input3')

# 模态1专属特征提取
x1 = Conv1D(filters=64, kernel_size=3, activation='relu')(input1)
x1 = Dropout(0.2)(x1)

# 模态2专属特征提取
x2 = Conv1D(filters=32, kernel_size=3, activation='relu')(input2)
x2 = Dropout(0.2)(x2)

# 模态3专属特征提取
x3 = Conv1D(filters=64, kernel_size=3, activation='relu')(input3)
x3 = Dropout(0.2)(x3)

# 沿特征维度拼接三个模态的输出
concat_feat = Concatenate(axis=-1)([x1, x2, x3])

# 时序特征提取
lstm_feat = LSTM(64)(concat_feat)
lstm_feat = Dropout(0.3)(lstm_feat)

# 二分类输出头
output = Dense(1, activation='sigmoid', name='output')(lstm_feat)

# 组装并编译模型
model = Model(inputs=[input1, input2, input3], outputs=output)
model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])

# 查看模型结构
model.summary()

如果不需要卷积层，也可以把Conv1D替换为Dense层做单步特征映射，效果差异可根据实际数据集调整。

直接传入三个模态的训练数据即可：

history = model.fit(
    [X1_train, X2_train, X3_train],
    y_train,
    batch_size=8,
    epochs=20,
    validation_data=([X1_test, X2_test, X3_test], y_test)
)

训练过程中可以根据验证集效果调整卷积核数量、LSTM隐藏层大小、dropout比例、学习率等超参数。

新的50步时序数据按步骤1的方式预处理为三个输入数组后，即可直接调用模型预测：

# 单条新数据预测示例，new_x1/new_x2/new_x3为预处理后的单序列，形状分别为(50,768)、(50,25)、(50,496)
new_x1 = np.expand_dims(new_x1, axis=0) # 增加batch维度，变为(1,50,768)
new_x2 = np.expand_dims(new_x2, axis=0)
new_x3 = np.expand_dims(new_x3, axis=0)

pred_prob = model.predict([new_x1, new_x2, new_x3], verbose=0)[0][0]
pred_label = 'A' if pred_prob < 0.5 else 'B'
print(f"预测结果：{pred_label}")

内容的提问来源于stack exchange，提问作者Cranjis