基於seq2seq的時間序列預測實驗

github.com

下面將進行若干組的實驗。方法是每組實驗生成一系列的隨機時間序列數據作為訓練數據，時間序列數據的長度是seq_length * 2， 我們用seq2seq模型的encoder學習出前半段（長度為seq_length）序列的pattern，然後用decoder的輸出作為對後半段數據的預測，我們希望decoder的輸出越接近後半段的真實數據越好。

實驗一：平穩時間序列

如下圖，前半段數據與後半段數據的長度相等，而且後半段的高度比前半段總是高出1。看過大量的這種類型的數據後，你能識別出「後半段的高度比前半段總是高出1」這種模式進而準確預測出後半段的數據嗎？看起來這是很簡單的模式識別問題，之所以簡單，是因為前半段和後半段的數據都是常量，只是後半段比前半段大一個常量罷了。

為了做實驗，我們需要生成大量的這種數據，並搭建seq2seq模型(使用tensorflow)，下面貼出主要的代碼。

生成總長度為30，即前半段、後半段長度均為15，後半段高度總比前半段高度大1的隨機時間序列數據，

def generate_x_y_data_two_freqs(batch_size, seq_length): batch_x = [] batch_y = [] for _ in range(batch_size): amp_rand = random.random() a = np.asarray([amp_rand] * seq_length) b = np.asarray([amp_rand + 1] * seq_length) sig = np.r_[a,b] x1 = sig[:seq_length] y1 = sig[seq_length:] x_ = np.array([x1]) y_ = np.array([y1]) x_, y_ = x_.T, y_.T batch_x.append(x_) batch_y.append(y_) batch_x = np.array(batch_x) batch_y = np.array(batch_y) batch_x = np.array(batch_x).transpose((1, 0, 2)) batch_y = np.array(batch_y).transpose((1, 0, 2)) return batch_x, batch_ydef generate_x_y_data_v2(batch_size): return generate_x_y_data_two_freqs(batch_size, seq_length=15)

用tensorflow搭建seq2seq，其中encoder和decoder使用相同結構(但不share weights)的雙層GRU。這塊的代碼是直接使用上文提到的github項目代碼，

tf.reset_default_graph()# sess.close()sess = tf.InteractiveSession()layers_stacked_count = 2with tf.variable_scope(Seq2seq): # Encoder: inputs enc_inp = [ tf.placeholder(tf.float32, shape=(None, input_dim), name="inp_{}".format(t)) for t in range(seq_length) ] # Decoder: expected outputs expected_sparse_output = [ tf.placeholder(tf.float32, shape=(None, output_dim), name="expected_sparse_output_".format(t)) for t in range(seq_length) ] # Give a "GO" token to the decoder. # Note: we might want to fill the encoder with zeros or its own feedback rather than with "+ enc_inp[:-1]" dec_inp = [ tf.zeros_like(enc_inp[0], dtype=np.float32, name="GO") ] + enc_inp[:-1] # Create a `layers_stacked_count` of stacked RNNs (GRU cells here). cells = [] for i in range(layers_stacked_count): with tf.variable_scope(RNN_{}.format(i)): cells.append(tf.nn.rnn_cell.GRUCell(hidden_dim)) # cells.append(tf.nn.rnn_cell.BasicLSTMCell(...)) cell = tf.nn.rnn_cell.MultiRNNCell(cells) # Here, the encoder and the decoder uses the same cell, HOWEVER, # the weights arent shared among the encoder and decoder, we have two # sets of weights created under the hood according to that functions def. dec_outputs, dec_memory = tf.nn.seq2seq.basic_rnn_seq2seq( enc_inp, dec_inp, cell ) # For reshaping the output dimensions of the seq2seq RNN: w_out = tf.Variable(tf.random_normal([hidden_dim, output_dim])) b_out = tf.Variable(tf.random_normal([output_dim])) # Final outputs: with linear rescaling for enabling possibly large and unrestricted output values. output_scale_factor = tf.Variable(1.0, name="Output_ScaleFactor") reshaped_outputs = [output_scale_factor*(tf.matmul(i, w_out) + b_out) for i in dec_outputs] # Training loss and optimizerwith tf.variable_scope(Loss): # L2 loss output_loss = 0 for _y, _Y in zip(reshaped_outputs, expected_sparse_output): output_loss += tf.reduce_mean(tf.nn.l2_loss(_y - _Y)) # L2 regularization (to avoid overfitting and to have a better generalization capacity) reg_loss = 0 for tf_var in tf.trainable_variables(): if not ("Bias" in tf_var.name or "Output_" in tf_var.name): reg_loss += tf.reduce_mean(tf.nn.l2_loss(tf_var)) loss = output_loss + lambda_l2_reg * reg_losswith tf.variable_scope(Optimizer): optimizer = tf.train.RMSPropOptimizer(learning_rate, decay=lr_decay, momentum=momentum) train_op = optimizer.minimize(loss) def train_batch(batch_size): """ Training step that optimizes the weights provided some batch_size X and Y examples from the dataset. """ X, Y = generate_x_y_data(isTrain=True, batch_size=batch_size) feed_dict = {enc_inp[t]: X[t] for t in range(len(enc_inp))} feed_dict.update({expected_sparse_output[t]: Y[t] for t in range(len(expected_sparse_output))}) _, loss_t = sess.run([train_op, loss], feed_dict) return loss_tdef test_batch(batch_size): """ Test step, does NOT optimizes. Weights are frozen by not doing sess.run on the train_op. """ X, Y = generate_x_y_data(isTrain=False, batch_size=batch_size) feed_dict = {enc_inp[t]: X[t] for t in range(len(enc_inp))} feed_dict.update({expected_sparse_output[t]: Y[t] for t in range(len(expected_sparse_output))}) loss_t = sess.run([loss], feed_dict) return loss_t[0]# Trainingtrain_losses = []test_losses = []sess.run(tf.global_variables_initializer())for t in range(nb_iters+1): train_loss = train_batch(batch_size) train_losses.append(train_loss) if t % 10 == 0: # Tester test_loss = test_batch(batch_size) test_losses.append(test_loss) print("Step {}/{}, train loss: {}, TEST loss: {}".format(t, nb_iters, train_loss, test_loss))print("Fin. train loss: {}, TEST loss: {}".format(train_loss, test_loss))

訓練中training loss下降的很快，

預測效果如何呢？下面給出兩張預測效果圖，

可以發現模型的預測效果很好，能夠準確預測後半段的數據。

實驗二：鋸齒波時間序列

生成總長度30，前、後半段各長15的鋸齒波，鋸齒波每三個點為一小段，下一段為上一段的反轉。後半段的高度是前半段的2倍，效果圖如下，

生成數據代碼如下，

def generate_x_y_data_two_freqs(batch_size, seq_length): batch_x = [] batch_y = [] for _ in range(batch_size): amp_rand = random.random() sig = [] flag = 1 for _ in range(seq_length / 3): sig += [amp_rand * flag] * 3 flag = -flag flag = 1 for _ in range(seq_length / 3): sig += [amp_rand * flag * 2] * 3 flag = -flag sig = np.asarray(sig) x1 = sig[:seq_length] y1 = sig[seq_length:] x_ = np.array([x1]) y_ = np.array([y1]) x_, y_ = x_.T, y_.T batch_x.append(x_) batch_y.append(y_) batch_x = np.array(batch_x) batch_y = np.array(batch_y) batch_x = np.array(batch_x).transpose((1, 0, 2)) batch_y = np.array(batch_y).transpose((1, 0, 2)) return batch_x, batch_ydef generate_x_y_data_v2(batch_size): return generate_x_y_data_two_freqs(batch_size, seq_length=15)

模型訓練的loss如下，

可見，loss很快下降到一個很小的值。

下面給出幾張預測效果圖，

實驗三：正弦波時間序列

下圖是一個正弦波圖像。

正弦波的數學表達式為 ，其中 為振幅， 為相位， 為偏距， 為角速度。那麼深度學習模型能學習出這些參數嗎？讓我們以結果說話。

同樣的，先生成大量的隨機正弦波

def generate_x_y_data_two_freqs(batch_size, seq_length): batch_x = [] batch_y = [] for _ in range(batch_size): offset_rand = random.random() * 2 * math.pi freq_rand = (random.random() - 0.5) / 1.5 * 15 + 0.5 amp_rand = random.random() + 0.1 sig = amp_rand * np.sin(np.linspace( seq_length / 15.0 * freq_rand * 0.0 * math.pi + offset_rand, seq_length / 15.0 * freq_rand * 3.0 * math.pi + offset_rand, seq_length * 2 ) ) x1 = sig[:seq_length] y1 = sig[seq_length:] x_ = np.array([x1]) y_ = np.array([y1]) x_, y_ = x_.T, y_.T batch_x.append(x_) batch_y.append(y_) batch_x = np.array(batch_x) batch_y = np.array(batch_y) batch_x = np.array(batch_x).transpose((1, 0, 2)) batch_y = np.array(batch_y).transpose((1, 0, 2)) return batch_x, batch_ydef generate_x_y_data_v2(batch_size): return generate_x_y_data_two_freqs(batch_size, seq_length=15)

訓練過程loss如下

可以看到loss也是下降的比較快。

下面給出幾張預測效果，

同樣預測的效果很好！

實驗四：正弦波與餘弦波的疊加

振幅、周期等參數均不同的正弦波與餘弦波相疊加，下圖是一個疊加效果圖，

可以看到正弦和餘弦波疊加後產生的圖可能跟正弦、餘弦差的比較遠，所以這可能對深度學習提出了挑戰。

先生成訓練數據，

def generate_x_y_data_two_freqs(batch_size, seq_length): batch_x = [] batch_y = [] for _ in range(batch_size): offset_rand = random.random() * 2 * math.pi freq_rand = (random.random() - 0.5) / 1.5 * 15 + 0.5 amp_rand = random.random() + 0.1 sig = amp_rand * np.sin(np.linspace( seq_length / 15.0 * freq_rand * 0.0 * math.pi + offset_rand, seq_length / 15.0 * freq_rand * 3.0 * math.pi + offset_rand, seq_length * 2 ) ) offset_rand = random.random() * 2 * math.pi freq_rand = (random.random() - 0.5) / 1.5 * 15 + 0.5 amp_rand = 1.2 sig = amp_rand * np.cos(np.linspace( seq_length / 15.0 * freq_rand * 0.0 * math.pi + offset_rand, seq_length / 15.0 * freq_rand * 3.0 * math.pi + offset_rand, seq_length * 2 ) ) + sig x1 = sig1[:seq_length] y1 = sig1[seq_length:] x_ = np.array([x1]) y_ = np.array([y1]) x_, y_ = x_.T, y_.T batch_x.append(x_) batch_y.append(y_) batch_x = np.array(batch_x) batch_y = np.array(batch_y) batch_x = np.array(batch_x).transpose((1, 0, 2)) batch_y = np.array(batch_y).transpose((1, 0, 2)) return batch_x, batch_ydef generate_x_y_data_v2(batch_size): return generate_x_y_data_two_freqs(batch_size, seq_length=15)

訓練的loss如下，

可見相對於實驗三，實驗四的loss明顯變大了，下降速度也沒那麼快了～

下面是幾張預測效果圖，

可以看到，預測效果較之於實驗三有些下降，但也不算太糟糕，還是大致可以預測對曲線的形狀。可以想像出深度學習可以通過訓練數據從疊加曲線中剝離出正弦波和餘弦波，並估算二者的參數。

實驗五：正弦與餘弦的隨機疊加

在實驗四中，我們是把正弦波與餘弦波直接相加，在本實驗中，我們採取不同的方式進行疊加，即在每一點，我們把餘弦波加到正弦波上，或不把餘弦波加到正弦波上(只有正弦)，我們在每一個點上生成一個隨機數來控制相應的疊加操作，取0～1間的隨機數，如果隨機數大於0.5，則進行疊加操作，否則不進行疊加。

下面是一個隨機疊加效果圖，

可以想像，這種在每一個點上正弦和餘弦波隨機疊加的數據，應該很難學習的，甚至無法找到數據的pattern。

生成數據代碼如下

def generate_x_y_data_two_freqs(batch_size, seq_length): batch_x = [] batch_y = [] for _ in range(batch_size): offset_rand = random.random() * 2 * math.pi freq_rand = (random.random() - 0.5) / 1.5 * 15 + 0.5 amp_rand = random.random() + 0.1 sig = amp_rand * np.sin(np.linspace( seq_length / 15.0 * freq_rand * 0.0 * math.pi + offset_rand, seq_length / 15.0 * freq_rand * 3.0 * math.pi + offset_rand, seq_length * 2 ) ) offset_rand = random.random() * 2 * math.pi freq_rand = (random.random() - 0.5) / 1.5 * 15 + 0.5 amp_rand = 1.2 sig = np.asarray([np.sign(max(random.random()-0.5,0)) for _ in range(seq_length * 2)]) * amp_rand * np.cos(np.linspace( seq_length / 15.0 * freq_rand * 0.0 * math.pi + offset_rand, seq_length / 15.0 * freq_rand * 3.0 * math.pi + offset_rand, seq_length * 2 ) ) + sig x1 = sig[:seq_length] y1 = sig[seq_length:] x_ = np.array([x1]) y_ = np.array([y1]) x_, y_ = x_.T, y_.T batch_x.append(x_) batch_y.append(y_) batch_x = np.array(batch_x) batch_y = np.array(batch_y) batch_x = np.array(batch_x).transpose((1, 0, 2)) batch_y = np.array(batch_y).transpose((1, 0, 2)) return batch_x, batch_ydef generate_x_y_data_v2(batch_size): return generate_x_y_data_two_freqs(batch_size, seq_length=15)

訓練的loss如下，

不出所料，loss下不去，學習幾乎寸步難行～

貼幾張預測效果圖，預測已經很糟糕了。。。

總結：

本文做了五組實驗，隨機生成了大量的數據，用基於seq2seq的深度學習模型去學習時間序列數據的pattern。從實驗中我們發現，如果數據的生成是有規律的，那麼深度學習可以發現數據內部隱藏的pattern，而像第五組實驗，由於數據在每一個點上包含了太多的隨機性，所以時間序列數據沒有包含顯而易見的pattern，深度學習也無能無力。

推薦閱讀：