Tensorflow 窗口时间序列数据的处理

时间序列数据处理数据集简介

数据源：

该数据集描述了一个时间序列下多个投资项目的 300 个匿名特征（“f_0”到“f_299”）和一个目标特征（“”）。目标特征需要根据后续时间节点的匿名特征进行预测。

本文的主要目标是构建特定长度的时间序列 RNN 网络的训练和测试集。

训练集、验证集和测试集的划分

由于给定的要求是预测后续时间点的目标特征，因此模型建立在过去的模式在未来仍然存在的基础上。因此，对于这样的模型，按时间划分训练集、验证集和测试集是合理的。时间序列号（“”）在数据集中给出，从 0 到 1219，共条形图。取其中的 2% 作为测试集，从时间序列 1201 到 1219。

窗口序列数据的获取与应用

解决这个问题的想法很简单。将数据集中的每个投资项目视为一个独立的时间序列，根据数据集的划分可以得到定长的时间序列数据，然后对划分的数据集使用滑动窗口的方法。

但在实际应用中会遇到一些问题。首先，滑动窗口法得到的时序数据具有更大的重复性。假设目标时间序列长度为20，如果将窗口序列数据集直接写入磁盘，将占用原始数据集近20倍的空间。

相应地，在训练过程中完全采用实时计算来获取窗口序列并不是一种可取的方法。每个epoch都会重复计算窗口序列的过程，计算函数的效率直接影响训练速度。

一个折中的方案是在原始数据集中只记录窗口序列中每个时间点的数据对应的序号，作为序号数据集写入磁盘。在训练过程中，通过读取原始数据集和序列号数据集生成一个batch。

由于RNN网络允许长度不定的时间序列作为输入，非矩阵形式的batch会影响输入的效率，所以未达到要求长度的窗口序列填充全零并插入一行全零为此目的进入原始数据集（注意：插入全零行需要预处理操作，例如规范化、规范化等）。

MIN_LEN = 20 # 最小窗口序列长度，低于该长度的窗口序列会被全零行填充
FEATURE_NUM = 300
ZERO_INDEX = 3141410 # 全零行序号
def form_indexes(data,time_range): # data：原数据集 time_range:时间序列范围
    id_list = sorted(data['investment_id'].unique())
    if 0 in id_list:
        id_list.remove(0)
    indexes_list = []
    for id in tqdm(id_list):
        
        sub_data = data[data['investment_id']==id].sort_values(by=['time_id'])
        time_list = tuple(sorted(sub_data['time_id'].unique()))
        for t in range(time_range[0],time_range[1]):
            if t in time_list:
                i_t = time_list.index(t)
                temp = list(sub_data[max(i_t-MIN_LEN+1,0):i_t+1].index.values)
                indexes = [ZERO_INDEX]*(MIN_LEN-len(temp)) + temp
    return indexes_list

训练前构建窗口序列数据训练集和测试集（验证集）

通过 tf.data 构建数据集的优势。方法是，函数只会在数据被使用（读或预读）时运行，不会占用太多内存，可以进行同时、批处理等操作。更容易完成。

train_indexset= pd.read_parquet('trainindex.parquet')
val_indexset= pd.read_parquet('valindex.parquet')
def gen_func(train_val_or_test): # 生成器函数
    if train_val_or_test == 1:
        for indexes in train_indexset.iterrows():
            features = data.iloc[indexes[1].values].values[:,4:]
            label = data.iloc[indexes[1].values[-1]]['target']
            yield (features,label)
    elif train_val_or_test == 2:
        for indexes in val_indexset.iterrows():
            features = data.iloc[indexes[1].values].values[:,4:]
            label = data.iloc[indexes[1].values[-1]]['target']
            yield (features,label)
    else:
        print("error input")
        raise ValueError
# 指定输出的形状和数据类型
featureSpec = tf.TensorSpec(
    shape=[MIN_LEN,FEATURE_NUM],
    dtype=tf.dtypes.float32,
    name=None
)
labelSpec = tf.TensorSpec(
    shape=[],
    dtype=tf.dtypes.float32,
    name=None
)
train_data = tf.data.Dataset.from_generator(generator=gen_func,args=[1] ,output_signature=(featureSpec,labelSpec))
val_data = tf.data.Dataset.from_generator(generator=gen_func,args=[2] ,output_signature=(featureSpec,labelSpec))

以下模型和超参数仅用于演示目的，并不具有指导意义。

MIN_LEN = 20
FEATURE_NUM = 300
BATCH_SIZE = 1000
EPOCH_NUM = 50 
def build_RNNmodel():
    model = tf.keras.models.Sequential(
        [
            tf.keras.layers.Masking(mask_value=0.,
                                    input_shape=(MIN_LEN, FEATURE_NUM)),
            tf.keras.layers.LSTM(1024,activation='tanh',
                                return_sequences=True,
                                dropout=0.5,
                                kernel_initializer=tf.initializers.TruncatedNormal(stddev=0.01),
                                ),
            tf.keras.layers.LSTM(256,activation='tanh',
                                dropout=0.5,
                                kernel_initializer=tf.initializers.TruncatedNormal(stddev=0.01),
                                ),
            tf.keras.layers.Dense(1,activation='relu')
        ]
    )
    return model

train_batchs = train_data.batch(batch_size=BATCH_SIZE).prefetch(BATCH_SIZE)
val_batchs = val_data.batch(batch_size=BATCH_SIZE).prefetch(BATCH_SIZE)
# 设置prefetch可以预读取后续批次数据提高运行速度
model = build_RNNmodel()
model.compile(loss='mae', optimizer=tf.keras.optimizers.Adam(0.0001))
history = model.fit(train_batchs,epochs=EPOCH_NUM,validation_data=val_batchs)

这里只取整体数据的一部分作为演示。每批有 1000 个窗口序列，每个 epoch 有 451 个批，运行一个 epoch 的时间约为 530 秒。