[PYTHON] Ich ließ RNN Sin Wave lernen und versuchte vorherzusagen

0. Grob gesagt

• Implementierung eines einfachen RNN (Recurrent Neural Network) mit TensorFlow.
• Unter Verwendung von RNN wurde die Sinuswelle trainiert, um "sin (t + 1)" (nächster Schritt) aus "sin (t)" vorherzusagen.
• Durch Verketten der Ausgabeergebnisse von RNN konnten wir die Vorhersage von "sin (t + n)" (mehrere Schritte) realisieren.
• LSTM (Long Short-Term Memory) wurde für die RNN-Zelle verwendet.

*** Hinzugefügt am 27. Mai 2016: ***

Ich schrieb die Fortsetzung "Ich ließ RNN Sinuswellen lernen und sagte voraus: Hyperparameteranpassung".

1. Über TensorFlow, RNN, LSTM

Ich werde es grob weglassen. Ich denke, dass das TensorFlow-Tutorial und die Artikel, auf die verwiesen wird, hilfreich sein werden.

• Recurrent Neural Networks
• Understanding LSTM Networks -- colah's blog
• Übersicht über das LSTM-Netzwerk - Qiita (Übersetzung des obigen Artikels)

2. Vorbereitung der Trainingsdaten

Eine Sinuswelle mit 50 Schritten pro Zyklus wurde für 100 Zyklen mit insgesamt 5.000 Schritten erzeugt und als Trainingsdaten verwendet. Darüber hinaus haben wir zwei Arten von Trainingsdaten vorbereitet: kein Lärm und kein Lärm.

Die Trainingsdaten bestehen aus einem Paar von "sin (t)" (sin-Wert zum Zeitpunkt t) und "sin (t + 1)" (sin-Wert zum Zeitpunkt t + 1). Einzelheiten zur Generierung von Trainingsdaten finden Sie in der ipynb-Datei (IPython Notebook). (Abgesehen davon war ich überrascht, dass die "ipynb" -Datei auf GitHub in der Vorschau angezeigt wurde.)

2.1 Kein Lärm

train_data/normal.ipynb

2.2. Laut

train_data/noised.ipynb

3. Lernen / Vorhersage

Dieses Mal lernen und prognostizieren wir mit einem Code. Der Quellcode ist im Anhang am Ende des Satzes aufgeführt.

3.1. Prozessablauf

Der Lern- und Vorhersagefluss ist wie folgt.

1. Lernen mit Trainingsdaten
2. Sagen Sie "sin (t + 1)" anhand der Anfangsdaten (dem Beginn der Trainingsdaten) voraus.
3. Vorhersage von "sin (t + 2)" unter Verwendung der vorhergesagten "sin (t + 1)" 4.3 Wiederholen Sie 3

3.2. Netzwerkkonfiguration

Ich habe ein Netzwerk namens "Eingangsschicht-versteckte Schicht-RNN-Zellen-Ausgangsschicht" verwendet. Wir haben auch LSTM für die RNN-Zelle verwendet.

3.3. Hyperparameter

Die zum Lernen und Vorhersagen verwendeten Hyperparameter sind wie folgt.

4. Vorhersageergebnis

Die folgende Abbildung zeigt die Vorhersageergebnisse. Die Legende lautet wie folgt.

• Schwarze gepunktete Linie: Trainingsdaten
• Durchgehend blau: Anfangsdaten
• Durchgehend grün: Vorhersagedaten

4.1. Kein Lärm

Eine solche Wellenform wird ausgegeben. Die Gesamtamplitude ist flach, die Eckpunkte sind verzerrt und die Frequenz ist etwas niedriger. Informationen zu bestimmten Werten finden Sie unter basic / output.ipynb.

4.2. Laut

Die Amplitude ist noch flacher und die Frequenz ist etwas höher als ohne Rauschen. Es scheint auch, dass die in den Trainingsdaten enthaltene Rauschkomponente reduziert wurde. Siehe noised / output.ipynb für bestimmte Werte.

5. Zukunftspläne

Ich möchte versuchen, die Netzwerkkonfiguration und die Hyperparameter zu ändern, um zu sehen, welche Vorhersageergebnisse erzielt werden.

*** Hinzugefügt am 27. Mai 2016: ***

Ich schrieb die Fortsetzung "Ich ließ RNN Sinuswellen lernen und sagte voraus: Hyperparameteranpassung".

Anhang: Quellcode

Der Quellcode für die geräuschlose Version ist unten dargestellt. Den Quellcode der lauten Version finden Sie in GitHub. Die verrauschte Version und die verrauschte Version unterscheiden sich nur im Namen der Eingabedatei.

• Rauschfreie Version: basic / rnn.py (Code siehe unten)
• Laute Version: noised / rnn.py

rnn.py

import tensorflow as tf
from tensorflow.models.rnn import rnn, rnn_cell
import numpy as np
import random

def make_mini_batch(train_data, size_of_mini_batch, length_of_sequences):
    inputs  = np.empty(0)
    outputs = np.empty(0)
    for _ in range(size_of_mini_batch):
        index   = random.randint(0, len(train_data) - length_of_sequences)
        part    = train_data[index:index + length_of_sequences]
        inputs  = np.append(inputs, part[:, 0])
        outputs = np.append(outputs, part[-1, 1])
    inputs  = inputs.reshape(-1, length_of_sequences, 1)
    outputs = outputs.reshape(-1, 1)
    return (inputs, outputs)

def make_prediction_initial(train_data, index, length_of_sequences):
    return train_data[index:index + length_of_sequences, 0]

train_data_path             = "../train_data/normal.npy"
num_of_input_nodes          = 1
num_of_hidden_nodes         = 2
num_of_output_nodes         = 1
length_of_sequences         = 50
num_of_training_epochs      = 2000
length_of_initial_sequences = 50
num_of_prediction_epochs    = 100
size_of_mini_batch          = 100
learning_rate               = 0.1
forget_bias                 = 1.0
print("train_data_path             = %s" % train_data_path)
print("num_of_input_nodes          = %d" % num_of_input_nodes)
print("num_of_hidden_nodes         = %d" % num_of_hidden_nodes)
print("num_of_output_nodes         = %d" % num_of_output_nodes)
print("length_of_sequences         = %d" % length_of_sequences)
print("num_of_training_epochs      = %d" % num_of_training_epochs)
print("length_of_initial_sequences = %d" % length_of_initial_sequences)
print("num_of_prediction_epochs    = %d" % num_of_prediction_epochs)
print("size_of_mini_batch          = %d" % size_of_mini_batch)
print("learning_rate               = %f" % learning_rate)
print("forget_bias                 = %f" % forget_bias)

train_data = np.load(train_data_path)
print("train_data:", train_data)

#Korrigieren Sie den Zufallszahlen-Startwert.
random.seed(0)
np.random.seed(0)
tf.set_random_seed(0)

optimizer = tf.train.GradientDescentOptimizer(learning_rate=learning_rate)

with tf.Graph().as_default():
    input_ph      = tf.placeholder(tf.float32, [None, length_of_sequences, num_of_input_nodes], name="input")
    supervisor_ph = tf.placeholder(tf.float32, [None, num_of_output_nodes], name="supervisor")
    istate_ph     = tf.placeholder(tf.float32, [None, num_of_hidden_nodes * 2], name="istate") #Benötigt zwei Werte pro Zelle.

    with tf.name_scope("inference") as scope:
        weight1_var = tf.Variable(tf.truncated_normal([num_of_input_nodes, num_of_hidden_nodes], stddev=0.1), name="weight1")
        weight2_var = tf.Variable(tf.truncated_normal([num_of_hidden_nodes, num_of_output_nodes], stddev=0.1), name="weight2")
        bias1_var   = tf.Variable(tf.truncated_normal([num_of_hidden_nodes], stddev=0.1), name="bias1")
        bias2_var   = tf.Variable(tf.truncated_normal([num_of_output_nodes], stddev=0.1), name="bias2")

        in1 = tf.transpose(input_ph, [1, 0, 2])         # (batch, sequence, data) -> (sequence, batch, data)
        in2 = tf.reshape(in1, [-1, num_of_input_nodes]) # (sequence, batch, data) -> (sequence * batch, data)
        in3 = tf.matmul(in2, weight1_var) + bias1_var
        in4 = tf.split(0, length_of_sequences, in3)     # sequence * (batch, data)

        cell = rnn_cell.BasicLSTMCell(num_of_hidden_nodes, forget_bias=forget_bias)
        rnn_output, states_op = rnn.rnn(cell, in4, initial_state=istate_ph)
        output_op = tf.matmul(rnn_output[-1], weight2_var) + bias2_var

    with tf.name_scope("loss") as scope:
        square_error = tf.reduce_mean(tf.square(output_op - supervisor_ph))
        loss_op      = square_error
        tf.scalar_summary("loss", loss_op)

    with tf.name_scope("training") as scope:
        training_op = optimizer.minimize(loss_op)

    summary_op = tf.merge_all_summaries()
    init = tf.initialize_all_variables()

    with tf.Session() as sess:
        saver = tf.train.Saver()
        summary_writer = tf.train.SummaryWriter("data", graph=sess.graph)
        sess.run(init)

        for epoch in range(num_of_training_epochs):
            inputs, supervisors = make_mini_batch(train_data, size_of_mini_batch, length_of_sequences)

            train_dict = {
                input_ph:      inputs,
                supervisor_ph: supervisors,
                istate_ph:     np.zeros((size_of_mini_batch, num_of_hidden_nodes * 2)),
            }
            sess.run(training_op, feed_dict=train_dict)

            if (epoch + 1) % 10 == 0:
                summary_str, train_loss = sess.run([summary_op, loss_op], feed_dict=train_dict)
                summary_writer.add_summary(summary_str, epoch)
                print("train#%d, train loss: %e" % (epoch + 1, train_loss))

        inputs  = make_prediction_initial(train_data, 0, length_of_initial_sequences)
        outputs = np.empty(0)
        states  = np.zeros((num_of_hidden_nodes * 2)),

        print("initial:", inputs)
        np.save("initial.npy", inputs)

        for epoch in range(num_of_prediction_epochs):
            pred_dict = {
                input_ph:  inputs.reshape((1, length_of_sequences, 1)),
                istate_ph: states,
            }
            output, states = sess.run([output_op, states_op], feed_dict=pred_dict)
            print("prediction#%d, output: %f" % (epoch + 1, output))

            inputs  = np.delete(inputs, 0)
            inputs  = np.append(inputs, output)
            outputs = np.append(outputs, output)

        print("outputs:", outputs)
        np.save("output.npy", outputs)

        saver.save(sess, "data/model")