주기적인 Sin함수의 RNN(Recurrent Neural Network) 머신 러닝 학습과 예측

 

 

 

 

파이선 코딩을 배우는 초보자도 머신 러닝에 독학으로 한번 도전해 보자.

머신 러닝을 배우려는 파이선 코딩 초보자들이 가지게 될 확률이 높은 의문점들을

하나하나 찾아내어 실제 풀어보고 결과를 확인해볼 수 있는 머신 러닝!

인터넷 교보문고에서 450페이지19900원에 판매중입니다.

 

 

 

 

 

 

 

 

 

 

 

________________________________________________________________________________________________________________________

 

 

 

RNN 머신 러닝 학습코드는 랜덤한 증권 가격을 학습 후 학습 기간 이 후의 구간에서도 랜덤한 예측 값을 잘 보여 주었다. 그와 반대로 RNN을 사용하여 랜덤이 아닌 주기적으로 변동되는 Sin 함수 그래프를 예측해 보자.

Sin 함수 예측 문제를 풀어보기 위해서는 엑셀 데이터 csv 파일 준비가 필요하다. 첫 번째 컬럼에서 0.0∼10,0 사이의 구간을 1000 등분한다. 엑셀에서 일정한 값 만큼 증가되는 값을 준비하려면 첫 번째 와 두 번째 cell에 값을 쓴 후 이 두cell 을 활성화 하여 아래로 1000번 cell까지 내리면 된다. 이 값에 대한 sin(6x)의 값을 두 번째 컬럼에 준비한다. 그 요령을 두 번째 cell에 ‘=sin(6*A1)’ 과 같이 코딩 후 이 cell을 활성화 시켜 1000번째 cell 까지 내리면 된다.

RNN 코드는 증권예측 코드와 동일하나 아래와 같이 파라메터들을 바꿔 주어야 한다. 증권가격 예측에서는 data_dim = 5 였지만 이 문제에서는 x 좌표와 Sin 함수 값 2개이 컬럼 데이터가 필요하므로 data_dim = 2 로 설정한다. 각 컬럼별 데이터 포인트 수는 알기 쉽게 1000개로 하였다.

train_size를 결정함에 있어서 10% 범위를 설정한다. Sin 함수는 주기 함수이기 때문에 현재 위 그래프에 의하면 9개 반 의 주기를 포함하고 있는데 대략 1개 주기를 대상으로 학습하고 웨이트 값을 결정하게 된다.

결과적으로 나머지 8개 반의 Sin 함수의 주기적인 거동을 얼마나 비슷하게 예측하는냐는 다소 아날로그적인 감이 드는 문제이다. 즉 관찰 포인트는 주기와 진폭 에러로 두자. sequence_length 값이 커짐에 따라 오차가 많이 줄어들며 주기 자체는 정확하게 예측 한다는 것을 알 수 있다. 스펙트럼 차원에서 보면 주파수가 일정하다는 점을 RNN 이 파악하고 있다는 것이 될 것이다.

이 예제를 돌려 보기 위한 코드는 다운받아 사용하기 바란다. 단 엑셀 데이터 파일과 파이선 코드를 같은 폴더에 두도록 한다. 아울러 파이선 코드 다운 로드 시에 indentation 훼손 여부를 반드시 확인해 보기 바란다.

 

Sin  함수 예측을 위한 엑셀 CSV 데이터

data_sin_01.csv

0.02MB

 

#rnn_esin_prediction.py

'''
This script shows how to predict sine function using a basic RNN
'''
import tensorflow as tf
import numpy as np
import matplotlib
import os

tf.reset_default_graph()
tf.set_random_seed(777)  # reproducibility

if "DISPLAY" not in os.environ:
    # remove Travis CI Error
    matplotlib.use('Agg')

import matplotlib.pyplot as plt

def MinMaxScaler(data):
    ''' Min Max Normalization

    Parameters
    ----------
    data : numpy.ndarray
        input data to be normalized
        shape: [Batch size, dimension]

    Returns
    ----------
    data : numpy.ndarry
        normalized data
        shape: [Batch size, dimension]

    References
    ----------
    .. [1] http://sebastianraschka.com/Articles/2014_about_feature_scaling.html

    '''
    numerator = data - np.min(data, 0)
    denominator = np.max(data, 0) - np.min(data, 0)
    # noise term prevents the zero division
    return numerator / (denominator + 1e-7)

# train Parameters
seq_length = 7
iterations = 500
data_dim = 2
hidden_dim = 10
output_dim = 1
learning_rate = 0.01

# Open, High, Low, Volume, Close
xy = np.loadtxt('data_sin_01.csv', delimiter=',')
xy = xy[::-1]  # reverse order (chronically ordered)
xy = MinMaxScaler(xy)
x = xy
y = xy[:, [-1]]  # Close as label

# build a dataset
dataX = []
dataY = []
for i in range(0, len(y) - seq_length):
    _x = x[i:i + seq_length]
    _y = y[i + seq_length]  # Next close price
    print(_x, "->", _y)
    dataX.append(_x)
    dataY.append(_y)

# train/test split
train_size = int(len(dataY) * 0.1)
test_size = len(dataY) - train_size
trainX, testX = np.array(dataX[0:train_size]), np.array(
    dataX[train_size:len(dataX)])
trainY, testY = np.array(dataY[0:train_size]), np.array(
    dataY[train_size:len(dataY)])

# input place holders
X = tf.placeholder(tf.float32, [None, seq_length, data_dim])
Y = tf.placeholder(tf.float32, [None, 1])

# build a LSTM network
cell = tf.contrib.rnn.BasicLSTMCell(
    num_units=hidden_dim, state_is_tuple=True, activation=tf.tanh)
outputs, _states = tf.nn.dynamic_rnn(cell, X, dtype=tf.float32)
Y_pred = tf.contrib.layers.fully_connected(
    outputs[:, -1], output_dim, activation_fn=None)

# cost/loss
loss = tf.reduce_sum(tf.square(Y_pred - Y))  # sum of the squares
# optimizer
optimizer = tf.train.AdamOptimizer(learning_rate)
train = optimizer.minimize(loss)

# RMSE
targets = tf.placeholder(tf.float32, [None, 1])
predictions = tf.placeholder(tf.float32, [None, 1])
rmse = tf.sqrt(tf.reduce_mean(tf.square(targets - predictions)))

with tf.Session() as sess:
    init = tf.global_variables_initializer()
    sess.run(init)

    # Training step
    for i in range(iterations):
        _, step_loss = sess.run([train, loss], feed_dict={
                                X: trainX, Y: trainY})
        print("[step: {}] loss: {}".format(i, step_loss))

    # Test step
    test_predict = sess.run(Y_pred, feed_dict={X: testX})
    rmse_val = sess.run(rmse, feed_dict={
                    targets: testY, predictions: test_predict})
    print("RMSE: {}".format(rmse_val))

    # Plot predictions
    plt.plot(testY)
    plt.plot(test_predict)
    plt.xlabel("Time Period")
    plt.ylabel("Stock Price")
    plt.show()

data_sin_01.csv

0.02MB