如何创建特征批次大小不同的TensorFlow时间序列滑动窗口数据集

需求说明 #

• 输入为长度Z的Pandas DataFrame，最终生成的数据集总长度可按需调整（若允许不足批次则为Z - horizon，严格按批次则为Z - horizon - batch_size + 1）
• 输入特征（inputs）的批次大小为batch_size，每个批次包含batch_size个连续的历史滑动窗口（每个窗口长度为horizon）
• 目标特征（targets）的批次大小固定为1，每个批次对应1个当前滑动窗口（长度为horizon）
• 核心逻辑：用batch_size个连续的历史窗口，预测紧接着的1个目标窗口，滑动步长为1遍历数据集

修改后的代码实现 #

import tensorflow as tf
import pandas as pd

horizon = 5
batch_size = 10
columns = {
    "inputs": ["input_1", "input_2"],
    "targets": ["target_1"],
}
batch_options = {
    "drop_remainder": True,
    "deterministic": True,
}

# 构造示例数据
d = range(100)
df = pd.DataFrame(data={'input_1': d, 'input_2': d, 'target_1': d})

# 拆分输入和目标特征的切片
input_slices = df[columns["inputs"]].astype("float32")
target_slices = df[columns["targets"]].astype("float32")

# 生成输入滑动窗口数据集：每个元素是(horizon, input_features)
input_window_ds = (
    tf.data.Dataset.from_tensor_slices(input_slices)
    .window(horizon, shift=1, drop_remainder=True)
    .flat_map(lambda x: x.batch(horizon, **batch_options))
)

# 生成目标滑动窗口数据集：每个元素是(horizon, target_features)
target_window_ds = (
    tf.data.Dataset.from_tensor_slices(target_slices)
    .window(horizon, shift=1, drop_remainder=True)
    .flat_map(lambda x: x.batch(horizon, **batch_options))
)

# 将输入窗口数据集按batch_size分组，得到每个元素是(batch_size, horizon, input_features)
input_batch_ds = input_window_ds.batch(batch_size, **batch_options)

# 对齐目标窗口：目标窗口对应输入批次的下一个窗口，跳过前batch_size-1个目标窗口
target_aligned_ds = target_window_ds.skip(batch_size - 1)

# 合并输入批次和目标窗口，得到最终数据集
final_ds = tf.data.Dataset.zip((input_batch_ds, target_aligned_ds))

# 验证输出形状
for inputs, target in final_ds.take(1):
    print(f"Inputs形状: {inputs.shape}")  # 输出：(10, 5, 2)
    print(f"Target形状: {target.shape}")  # 输出：(5, 1)，若需(1,5,1)可添加target = tf.expand_dims(target, 0)

关键步骤解释 #

1. 拆分窗口数据集：分别为inputs和targets生成独立的滑动窗口数据集，每个元素是单个horizon长度的序列窗口。
2. 输入窗口批次化：将输入窗口按batch_size分组，得到包含连续历史窗口的输入批次。
3. 时间对齐处理：由于输入批次是前batch_size个历史窗口，对应的目标窗口是第batch_size个窗口，因此跳过前batch_size-1个目标窗口，保证输入与目标的时间对应关系。
4. 合并数据集：通过tf.data.Dataset.zip将输入批次和对应目标窗口合并，得到符合要求的数据集结构。

调整总长度为Z - horizon的方案 #

若需要严格保证数据集总长度为Z - horizon，可将batch_options["drop_remainder"]设为False，允许输入批次包含不足batch_size的窗口：

batch_options["drop_remainder"] = False
input_batch_ds = input_window_ds.batch(batch_size, **batch_options)
final_ds = tf.data.Dataset.zip((input_batch_ds, target_window_ds))

此方案下，前batch_size-1个输入批次的窗口数量会不足batch_size，需根据业务场景判断是否可接受。

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