1D CNN登革热病例时间序列预测：预测曲线与实际不匹配问题排查

核心问题 #

你当前的代码把1D CNN当成普通监督学习模型使用——直接用单时间步的协变量预测同期的登革热病例数，完全没利用时间序列的时序依赖关系。这就是指标数值看起来还行，但预测曲线和实际走势完全不匹配的根本原因：模型只是拟合单步特征与目标的静态关联，没学习到病例数随时间变化的趋势、周期等关键模式。

另外代码还存在两个小问题：

• 绘图时用到的train_dates和test_dates变量未定义，运行会报错。
• 1D CNN的输入维度设计未适配时序窗口，无法捕捉时序特征。

修正方案 #

1. 重构时序输入：创建滑动窗口 #

时间序列预测的核心是用过去k个时间步的所有特征（含历史病例数），预测下一个时间步的病例数。需实现滑动窗口函数，将原始数据转换为这种格式。

2. 修复未定义变量 #

提取训练集和测试集的日期索引，用于绘图。

3. 调整模型结构 #

适配滑动窗口后的输入维度，增强模型对时序特征的捕捉能力。

修正后的完整代码 #

import pandas as pd
import numpy as np
from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import mean_squared_error, mean_absolute_error
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Conv1D, MaxPooling1D, Flatten, Dense, Dropout, LeakyReLU
from tensorflow.keras.optimizers import Adam
from tensorflow.keras.callbacks import EarlyStopping
import matplotlib.pyplot as plt

# 加载数据
file_path = 'https://gist.githubusercontent.com/JishanAhmed2019/a7666a3651d27bf03dc93e63aac896b0/raw/f93f28aeaa41418689744a4fbb5bde29114f9872/Dengue.csv'
data = pd.read_csv(file_path, index_col='Date', sep='\t', parse_dates=True)

# 定义滑动窗口函数：用过去look_back步数据预测未来1步
def create_sequences(data, look_back=7):
    X, y = [], []
    for i in range(len(data) - look_back):
        X.append(data[i:(i + look_back), :])
        y.append(data[i + look_back, -1])  # 最后一列是目标变量Dhaka_Dengue
    return np.array(X), np.array(y)

# 数据拆分与缩放
split_fraction = 0.85
split_point = int(len(data) * split_fraction)
train_data = data.iloc[:split_point]
test_data = data.iloc[split_point - 7:]  # 保留前7步用于创建测试集的第一个窗口

# 整体缩放（因为需要滑动窗口，先缩放再拆分窗口，注意测试集用训练集的scaler）
scaler = MinMaxScaler()
train_scaled = scaler.fit_transform(train_data)
test_scaled = scaler.transform(test_data)

# 创建时序窗口，这里用7天的历史数据预测下一天
look_back = 7
X_train, y_train = create_sequences(train_scaled, look_back)
X_test, y_test = create_sequences(test_scaled, look_back)

# 拆分验证集
val_fraction = 0.15
val_split_point = int(len(X_train) * (1 - val_fraction))
X_train_final, y_train_final = X_train[:val_split_point], y_train[:val_split_point]
X_val, y_val = X_train[val_split_point:], y_train[val_split_point:]

# 定义1D CNN模型，适配窗口输入维度
model = Sequential([
    Conv1D(64, 3, padding='same', activation=LeakyReLU(alpha=0.1), input_shape=(X_train_final.shape[1], X_train_final.shape[2])),
    MaxPooling1D(2, padding="same"),
    Conv1D(32, 3, padding='same', activation=LeakyReLU(alpha=0.1)),
    MaxPooling1D(2, padding="same"),
    Flatten(),
    Dense(32, activation=LeakyReLU(alpha=0.1)),
    Dropout(0.2),
    Dense(1)
])

# 编译模型
model.compile(optimizer=Adam(learning_rate=0.001), loss='mean_squared_error', metrics=['mae'])

# 早停回调
early_stopping = EarlyStopping(monitor='val_loss', patience=10, verbose=1, mode='min')

# 训练模型
history = model.fit(
    X_train_final, y_train_final,
    epochs=500,
    batch_size=32,
    verbose=1,
    validation_data=(X_val, y_val),
    callbacks=[early_stopping]
)

# 生成预测并反缩放
train_predictions = model.predict(X_train_final)
test_predictions = model.predict(X_test)

# 反缩放需要构造完整特征维度，再提取目标列
def inverse_transform_predictions(predictions, scaler, look_back, original_data):
    # 创建全零数组，填充预测值到目标列位置
    dummy_array = np.zeros((len(predictions), original_data.shape[1]))
    dummy_array[:, -1] = predictions.flatten()
    return scaler.inverse_transform(dummy_array)[:, -1]

train_predictions_inverse = inverse_transform_predictions(train_predictions, scaler, look_back, train_data)
y_train_inverse = inverse_transform_predictions(y_train_final.reshape(-1,1), scaler, look_back, train_data)
test_predictions_inverse = inverse_transform_predictions(test_predictions, scaler, look_back, test_data)
y_test_inverse = inverse_transform_predictions(y_test.reshape(-1,1), scaler, look_back, test_data)

# 获取对应日期
train_dates = train_data.index[look_back:val_split_point+look_back]
val_dates = train_data.index[val_split_point+look_back:]
test_dates = test_data.index[look_back:]

# 计算指标
train_rmse = np.sqrt(mean_squared_error(y_train_inverse, train_predictions_inverse))
val_rmse = np.sqrt(mean_squared_error(inverse_transform_predictions(y_val.reshape(-1,1), scaler, look_back, train_data), 
                                      inverse_transform_predictions(model.predict(X_val), scaler, look_back, train_data)))
test_rmse = np.sqrt(mean_squared_error(y_test_inverse, test_predictions_inverse))

train_mae = mean_absolute_error(y_train_inverse, train_predictions_inverse)
val_mae = mean_absolute_error(inverse_transform_predictions(y_val.reshape(-1,1), scaler, look_back, train_data), 
                              inverse_transform_predictions(model.predict(X_val), scaler, look_back, train_data))
test_mae = mean_absolute_error(y_test_inverse, test_predictions_inverse)

print("训练集 RMSE:", train_rmse, "MAE:", train_mae)
print("验证集 RMSE:", val_rmse, "MAE:", val_mae)
print("测试集 RMSE:", test_rmse, "MAE:", test_mae)

# 绘制损失曲线
plt.figure(figsize=(10, 4))
plt.plot(history.history['loss'], label='训练损失')
plt.plot(history.history['val_loss'], label='验证损失')
plt.title('训练与验证损失')
plt.legend()
plt.show()

# 绘制训练集预测vs实际
plt.figure(figsize=(14, 5))
plt.plot(train_dates, y_train_inverse, label='实际病例数')
plt.plot(train_dates, train_predictions_inverse, label='预测病例数')
plt.title('训练集实际vs预测')
plt.ylabel('登革热病例数')
plt.xlabel('日期')
plt.xticks(rotation=45)
plt.legend()
plt.tight_layout()
plt.show()

# 绘制测试集预测vs实际
plt.figure(figsize=(14, 5))
plt.plot(test_dates, y_test_inverse, label='实际病例数')
plt.plot(test_dates, test_predictions_inverse, label='预测病例数')
plt.title('测试集实际vs预测')
plt.ylabel('登革热病例数')
plt.xlabel('日期')
plt.xticks(rotation=45)
plt.legend()
plt.tight_layout()
plt.show()

# 绘制完整实际序列
plt.figure(figsize=(14, 7))
plt.plot(data.index, data['Dhaka_Dengue'], label='实际登革热病例数', color='blue')
plt.title('登革热病例数时间序列')
plt.xlabel('日期')
plt.ylabel('病例数')
plt.legend()
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()

关键说明 #

• 滑动窗口：用过去7天的所有特征（包括历史病例数）预测下一天的病例数，让模型学习时序依赖。
• 缩放逻辑：因为滑动窗口需要连续数据，先对训练集整体缩放，再用同一scaler处理测试集，避免数据泄露。
• 模型调整：增加了一层Conv1D和全连接层，提升模型对时序特征的捕捉能力。

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