파이썬/Tensorflow
RNN_주식 예측
easysheep
2023. 2. 3. 22:34
RNN 을 사용하여 시계열 데이터인 주식을 예측 해보자
import pandas as pd
import matplotlib.pyplot as plt
# 테스트 데이터와 트레인 데이터 홀드 아웃 할 때 사용하는 함수
from sklearn.model_selection import train_test_split
import tensorflow as tf
from tensorflow.keras import models
from tensorflow.keras import layers
from tensorflow.keras.models import Model
import sklearn as sns
# 사용할 변수 정의
batch_size = 100
path = "/content/drive/MyDrive/TensorFlow/dataset/stock_daily.csv"
df = pd.read_csv(path,header=1)불러온 데이터의 값은 다음과 같다.
df.info()
<class 'pandas.core.frame.DataFrame'>
RangeIndex: 732 entries, 0 to 731
Data columns (total 5 columns):
# Column Non-Null Count Dtype
--- ------ -------------- -----
0 # Open 732 non-null float64
1 High 732 non-null float64
2 Low 732 non-null float64
3 Volume 732 non-null int64
4 Close 732 non-null float64
dtypes: float64(4), int64(1)
memory usage: 28.7 KB- open : 시작가
- high : 제일 높았을 때
- low: 제일 낮았을 때
- Volume : 거래량
- close : 마감가
이 값중 예측 하려는 값은 Close 이므로 따로 저장한다.
dataset=df.copy()
ori_y=dataset.pop("Close")
ori_x=dataset.copy()
# x_dim = 4 (속성 개수 )
# 시계열 데이터이므로 shuffle= False
# train_test_split을 이용하여 train, test,validation 데이터를 분리 해준다.
train_x ,test_x ,train_y,test_y = train_test_split(ori_x ,ori_y , test_size = 0.2 , shuffle= False)
train_x ,val_x ,train_y,val_y = train_test_split(train_x ,train_y , test_size = 0.1 , shuffle=False)train 데이터를 확인하면 다음과 같이 Volume 값과 그 외의 값들이 크기의 차이가 많이 나는 것을 알 수있다.
이 데이터를 그냥 쓰면 Volume 값에 영향을 많이 받게 되므로 min_max 정규화를 실시한다.
# train_x의 평균 , 표준 편차, 최대, 최소값 등을 df형태로 저장한다.
train_x_des = train_x.describe()
# train_x_des를 tranpose(행,렬 변환) 해준다.
train_x_des = train_x_des.T
print(train_x_des)
count mean std min 25% \
# Open 526.0 6.856973e+02 94.555220 507.252283 6.103425e+02
High 526.0 6.912186e+02 94.682919 511.092313 6.165950e+02
Low 526.0 6.797701e+02 94.269379 501.202274 6.000700e+02
Volume 526.0 1.911786e+06 998970.828437 527200.000000 1.317550e+06
50% 75% max
# Open 7.162900e+02 7.665200e+02 8.378100e+02
High 7.215700e+02 7.699375e+02 8.419500e+02
Low 7.081200e+02 7.580900e+02 8.283500e+02
Volume 1.664050e+06 2.134750e+06 1.116490e+07min_max 정규화 및 TimeseriesGenerator 생성
# min_max 정규화
def min_max_norm(x):
return ((x - dataset_des['min'])/(dataset_des['max'] - dataset_des['min']))
# 기본 정규화
def standarc_norm(x):
return (x - dataset_des['mean']) / dataset_des['std']
# https://www.tensorflow.org/api_docs/python/tf/keras/preprocessing/sequence/TimeseriesGenerator 참고
# tf.keras.preprocessing.sequence.TimeseriesGenerator(data, targets, length, sampling_rate=1, stride=1, start_index=0, end_index=None, shuffle=False, reverse=False, batch_size=128)
# tf.keras.preprocessing.sequence.TimeseriesGenerator 에 넣기 위해 데이터를 list 또는 Numpy로 변환 , 변환하는 과정에서 정규화 실시
# min_max 정규화
min_max_norm_train_x = min_max_norm(train_x)
min_max_norm_test_x = min_max_norm(test_x)
min_max_norm_val_x = min_max_norm(val_x)
# list로 변환
min_max_norm_train_x = min_max_norm_train_x.values.tolist()
min_max_norm_test_x = min_max_norm_test_x.values.tolist()
min_max_norm_val_x = min_max_norm_val_x.values.tolist()
data_gen_train = \
tf.keras.preprocessing.sequence.TimeseriesGenerator(min_max_norm_train_x ,
train_y.values.tolist() ,
length = 4 ,
batch_size = batch_size)
data_gen_test = \
tf.keras.preprocessing.sequence.TimeseriesGenerator(min_max_norm_test_x ,
test_y.values.tolist() ,
length = 4 ,
batch_size = batch_size)
data_gen_val = \
tf.keras.preprocessing.sequence.TimeseriesGenerator(min_max_norm_val_x ,
val_y.values.tolist() ,
length = 4 ,
batch_size = batch_size)이제 모델을 생성하는 데 activation_function 을 2가지를 사용할 것 이다.
하나는 tanh 다른 하나는 relu이다.
# pytorch 와 유사한 방법인 Subclassing API 로 구현
class RnnModel(Model):
def __init__(self , rnn_activation_function):
# 부모 클래스의 생성자 실행
super(RnnModel , self).__init__()
# SimpleRNN(units)
self.rnn_layer = layers.SimpleRNN(128 , activation = rnn_activation_function)
self.dense_list = [layers.Dense(int(128/ 2**i ),activation = 'relu') if i != 4 else layers.Dense(1,activation = None) for i in range(1,5) ]
# Functional 모델 처럼 정의
def call(self , x):
x = self.rnn_layer(x)
for dense in self.dense_list:
x = dense(x)
return x
# 모델 요약을 출력
def summary(self):
inputs = layers.Input((5,4))
Model(inputs,self.call(inputs)).summary()
rnn_model_tanh = RnnModel('tanh')
rnn_model_relu = RnnModel('relu')
rnn_model_tanh.build(input_shape = (None,5,4))
rnn_model_relu.build(input_shape = (None,5,4))
# 모델 요약
rnn_model_tanh.summary()
rnn_model_relu.summary()
# 손실 함수 mse
loss_function=tf.keras.losses.mean_squared_error
#정규화기(최적화 알고리즘) Adam
optimizer=tf.keras.optimizers.Adam(learning_rate=0.001)
# 평가 지표 mae
metrics=tf.keras.metrics.mean_absolute_error
rnn_model_tanh.compile(loss=loss_function,
optimizer=optimizer,
metrics=[metrics])
rnn_model_relu.compile(loss=loss_function,
optimizer=optimizer,
metrics=[metrics])생성된 2개의 모델은 다음과 같다.
Model: "model_6"
_________________________________________________________________
Layer (type) Output Shape Param #
=================================================================
input_7 (InputLayer) [(None, 5, 4)] 0
simple_rnn_6 (SimpleRNN) (None, 128) 17024
dense_24 (Dense) (None, 64) 8256
dense_25 (Dense) (None, 32) 2080
dense_26 (Dense) (None, 16) 528
dense_27 (Dense) (None, 1) 17
=================================================================
Total params: 27,905
Trainable params: 27,905
Non-trainable params: 0
_________________________________________________________________
Model: "model_7"
_________________________________________________________________
Layer (type) Output Shape Param #
=================================================================
input_8 (InputLayer) [(None, 5, 4)] 0
simple_rnn_7 (SimpleRNN) (None, 128) 17024
dense_28 (Dense) (None, 64) 8256
dense_29 (Dense) (None, 32) 2080
dense_30 (Dense) (None, 16) 528
dense_31 (Dense) (None, 1) 17
=================================================================
Total params: 27,905
Trainable params: 27,905
Non-trainable params: 0
_________________________________________________________________모델을 위에서 정의한 TimeseriesGenerator 를 이용하여 훈련시킨다.
# 인수 참고 https://keras.io/ko/models/sequential/
# 모델 훈련
history_tanh = rnn_model_tanh.fit(
data_gen_train, # 데이터 셋
validation_data=data_gen_val,
steps_per_epoch=len(train_x)/batch_size, # 한 세대의 종료를 선언하고 다음 세대를 시작하기까지 단계(샘플 배치)의 총 개수
epochs=100, # 반복 횟수
validation_freq = 1
)
# 모델 훈련
history_relu = rnn_model_relu.fit(
data_gen_train,
validation_data=data_gen_val,
steps_per_epoch=len(train_x)/batch_size,
epochs=100,
validation_freq = 1
)만들어진 모델을 평가한다.
# data_gen_test에서 test x,y 데이터를 가지고 온다.
test_data_x, test_data_y=data_gen_test[0]
# 예상한 값을 test_data_y와 비교해 주시기 위해 평탄화(1차원)해준다.
prediction_tanh_y=rnn_model_tanh.predict(test_data_x).flatten()
prediction_relu_y=rnn_model_relu.predict(test_data_x).flatten()
test_y=test_data_y.flatten()
# plot.으로 나타내기 위해 다음과 같이한다.
time = range(1, len(test_y) + 1)
plt.plot(time, test_y, 'r', label='ture')
plt.plot(time, prediction_tanh_y, 'b', label='tanh')
plt.title('stock prediction')
plt.xlabel('time')
plt.ylabel('value')
plt.legend()
plt.show()
plt.plot(time, test_y, 'r', label='ture')
plt.plot(time, prediction_relu_y, 'g', label='relu')
plt.title('stock prediction')
plt.xlabel('time')
plt.ylabel('value')
plt.legend()
plt.show()
loss_tanh = history_tanh.history['loss']
loss_relu = history_relu.history['loss']
epochs = range(1, len(loss_tanh) + 1)
plt.plot(epochs, loss_tanh, 'bo', label='loss_tanh')
plt.plot(epochs, loss_relu, 'ro', label='loss_relu')
plt.title('relu and tanh loss')
plt.legend()
plt.show()
mae_tanh = history_tanh.history['mean_absolute_error']
mae_relu = history_relu.history['mean_absolute_error']
plt.plot(epochs, mae_tanh, 'bo', label='mae_tanh')
plt.plot(epochs, mae_relu, 'ro', label='mae_relu')
plt.title('relu and tanh mae')
plt.legend()
plt.show()
역시 tanh의 gradient vanishing 문제점을 해결 하려고 나온 relu 답게 더 빠른 훈련 속도를 보인다. 하지만 실제 값과 차이는 tanh보다 더 크다. 다음에는 Leaky ReLU를 사용해 봐야겠다.