时间序列预测数据

import inspect from typing import List from tensorflow.keras import backend as K, Model, Input, optimizers from tensorflow.keras import layers from tensorflow.keras.layers import Activation, SpatialDropout1D, Lambda from tensorflow.keras.layers import Layer, Conv1D, Dense, BatchNormalization, LayerNormalization def is_power_of_two(num: int): return num != 0 and ((num & (num - 1)) == 0) def adjust_dilations(dilations: list): if all([is_power_of_two(i) for i in dilations]): return dilations else: new_dilations = [2 ** i for i in dilations] return new_dilations class ResidualBlock(Layer): def __init__(self, dilation_rate: int, nb_filters: int, kernel_size: int, padding: str, activation: str = 'relu', dropout_rate: float = 0, kernel_initializer: str = 'he_normal', use_batch_norm: bool = False, use_layer_norm: bool = False, **kwargs): """Defines the residual block for the WaveNet TCN Args: x: The previous layer in the model training: boolean indicating whether the layer should behave in training mode or in inference mode dilation_rate: The dilation power of 2 we are using for this residual block nb_filters: The number of convolutional filters to use in this block kernel_size: The size of the convolutional kernel padding: The padding used in the convolutional layers, 'same' or 'causal'. activation: The final activation used in o = Activation(x + F(x)) dropout_rate: Float between 0 and 1. Fraction of the input units to drop. kernel_initializer: Initializer for the kernel weights matrix (Conv1D). use_batch_norm: Whether to use batch normalization in the residual layers or not. use_layer_norm: Whether to use layer normalization in the residual layers or not. kwargs: Any initializers for Layer class. """ self.dilation_rate = dilation_rate self.nb_filters = nb_filters self.kernel_size = kernel_size self.padding = padding self.activation = activation self.dropout_rate = dropout_rate self.use_batch_norm = use_batch_norm self.use_layer_norm = use_layer_norm self.kernel_initializer = kernel_initializer self.layers = [] self.layers_outputs = [] self.shape_match_conv = None self.res_output_shape = None self.final_activation = None super(ResidualBlock, self).__init__(**kwargs) def _add_and_activate_layer(self, layer): """Helper function for building layer Args: layer: Appends layer to internal layer list and builds it based on the current output shape of ResidualBlocK. Updates current output shape. """ self.layers.append(layer) self.layers[-1].build(self.res_output_shape) self.res_output_shape = self.layers[-1].compute_output_shape(self.res_output_shape) def build(self, input_shape): with K.name_scope(self.name): # name scope used to make sure weights get unique names self.layers = [] self.res_output_shape = input_shape for k in range(2): name = 'conv1D_{}'.format(k) with K.name_scope(name): # name scope used to make sure weights get unique names self._add_and_activate_layer(Conv1D(filters=self.nb_filters, kernel_size=self.kernel_size, dilation_rate=self.dilation_rate, padding=self.padding, name=name, kernel_initializer=self.kernel_initializer)) with K.name_scope('norm_{}'.format(k)): if self.use_batch_norm: self._add_and_activate_layer(BatchNormalization()) elif self.use_layer_norm: self._add_and_activate_layer(LayerNormalization()) self._add_and_activate_layer(Activation('relu')) self._add_and_activate_layer(SpatialDropout1D(rate=self.dropout_rate)) if self.nb_filters != input_shape[-1]: # 1x1 conv to match the shapes (channel dimension). name = 'matching_conv1D' with K.name_scope(name): # make and build this layer separately because it directly uses input_shape self.shape_match_conv = Conv1D(filters=self.nb_filters, kernel_size=1, padding='same', name=name, kernel_initializer=self.kernel_initializer) else: name = 'matching_identity' self.shape_match_conv = Lambda(lambda x: x, name=name) with K.name_scope(name): self.shape_match_conv.build(input_shape) self.res_output_shape = self.shape_match_conv.compute_output_shape(input_shape) self.final_activation = Activation(self.activation) self.final_activation.build(self.res_output_shape) # probably isn't necessary # this is done to force Keras to add the layers in the list to self._layers for layer in self.layers: self.__setattr__(layer.name, layer) self.__setattr__(self.shape_match_conv.name, self.shape_match_conv) self.__setattr__(self.final_activation.name, self.final_activation) super(ResidualBlock, self).build(input_shape) # done to make sure self.built is set True def call(self, inputs, training=None): """ Returns: A tuple where the first element is the residual model tensor, and the second is the skip connection tensor. """ x = inputs self.layers_outputs = [x] for layer in self.layers: training_flag = 'training' in dict(inspect.signature(layer.call).parameters) x = layer(x, training=training) if training_flag else layer(x) self.layers_outputs.append(x) x2 = self.shape_match_conv(inputs) self.layers_outputs.append(x2) res_x = layers.add([x2, x]) self.layers_outputs.append(res_x) res_act_x = self.final_activation(res_x) self.layers_outputs.append(res_act_x) return [res_act_x, x] def compute_output_shape(self, input_shape): return [self.res_output_shape, self.res_output_shape] class TCN(Layer): """Creates a TCN layer. Input shape: A tensor of shape (batch_size, timesteps, input_dim). Args: nb_filters: The number of filters to use in the convolutional layers. Can be a list. kernel_size: The size of the kernel to use in each convolutional layer. dilations: The list of the dilations. Example is: [1, 2, 4, 8, 16, 32, 64]. nb_stacks : The number of stacks of residual blocks to use. padding: The padding to use in the convolutional layers, 'causal' or 'same'. use_skip_connections: Boolean. If we want to add skip connections from input to each residual blocK. return_sequences: Boolean. Whether to return the last output in the output sequence, or the full sequence. activation: The activation used in the residual blocks o = Activation(x + F(x)). dropout_rate: Float between 0 and 1. Fraction of the input units to drop.