LSTM을 활용한 삼성 주가 예측 스크립트

삼성전자 주가 데이터는 yahoo finance에서 csv파일로 다운받아서 활용할 수 있다.

개발과정 - 시계열 데이터 분석 및 예측

1. 데이터 로드 및 분포 확인

df = pd.read_csv(), df.describe(), df.hist(), plot() 등

2. 데이터 전처리

outlier / missing value 확인 후 대체(또는 삭제) 처리

데이터 정규화(normalization) / 표준화(standardization)

딥러닝을 위한 feature column / label column 정의

3. 데이터 생성

window size 설정 후 feature / label 시계열 데이터 생성

학습 데이터 생성. 이때 입력 데이터는 (batch_size, time_steps, input_dims)형태의 3차원 텐서로 생성

순환 신경망 모델 구축 및 학습

 

데이터 로드 및 분포확인

import tensorflow as tf
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, DENSE, Dropout

raw_df = pd.read_csv('./005930.ks_3MA_5MA.csv') # 야후에서 다운받은 삼성전자 주가 정보
raw_df.head()

plt.figure(figsize=(7, 4))

plt.title('SAMSUNG ELECTRONIC STOCK PRICE')
plt.xlabel('period(dat)')
plt.ylabel('price(won)')
plt.grid()

plt.plot(raw_df['Adj Close'], label = 'Adj Close', color='b')
plt.legend(loc='best')

plt.show()

데이터 전처리 1 (outlier, missing value)

outlier 확인 - 통계적으로 비정상적으로 크거나 작은 데이터인 outlier(특이값)는 딥러닝 학습을 하기 위해서는 적절한 값으로 바꾸거나 삭제하는 등의 처리가 반드시 필요함

Missing value 확인(결측치) - 특정 데이터가 누락된 것을 말하며, outlier와 마찬가지로 이러한 missing value를 제거하거나 적절한 값으로 대체하는 등의 처리가 필요함

raw_df['Volume'] = raw.df['Volume'].replace(0, np.nan) # Volume 값 0을 NaN으로 모두 대체
for col in raw_df.columns: # 각 column에 0 개수 확인
	missing_rows = (raw_df.loc[raw_df[col]==0].shape[0]
    print(col. + ': '+str(missing_rows))
raw_df.isnull().sum()
raw_df=raw_df.dropna() # 모든 Missing value 삭제
raw_df.isnull().sum()

데이터 전처리 2 (정규화)

딥러닝 학습이 잘되기 위해서는 정규화 작업이 필요함. 즉, 날짜를 나타내는 Date 항목을 제외한 숫자로 표현되는 모든 Column에 대해서 -~1 값으로 정규화 수행

from sklearn.preprocessing import MinMasScaler
Scaler = MinMaxScaler()
scale_cols=['Open', 'High', 'Low', 'Close', 'Adj Close', '3MA', '5MA', 'Volume'] # 정규화 대상 column 정의
scaled_df=scaler.fit_transform(raw_df[scale_cols]) # 정규화 수행
print(type(scaled_df), '\n') # 리턴값은 numpy
scaled_df = pd.DataFrame(scaled_df, columns=scale_cols) # 정규화된 새로운 DataFrame 생성
print(scale_df)

데이터 전처리 3 (feature column / label column 정의)

딥러닝 학습을 위한 입력데이터 feature column, 정답데이터 label column 정의 후 numpy로 변환하여 데이터 전처리 과정을 완료함

feature_cols = ['3MA', '5MA', 'Adj Close'] # Feature 정의(입력 데이터)
label_cols = ['Adj Close'] # label 정의 (정답 데이터)
label_df = pd.DataFrame(scaled_df, columns = label_cols)
feature_df - pd.DataFrame(scaled_df, columns = feature_cols)
print(feature_df)
print(label_df)

label_np = label_df.to_numpy() # 딥러닝 학습을 위해 DataFrame -> Numpy로 변환
feature_np = feature_df.to_numpy() # 딥러닝 학습을 위해 DataFrame -> Numpy로 변환

데이터 생성 - 입력 데이터 feature / 정답 데이터 label

[1] 학습 데이터 X, Y 생성

window_size = 40
X, Y = make_sequence_dataset(feature_np, label_np, window_size)
# Numpy로 주어지는 시계열 데이터로 부터 window_size에 맞게 RNN 입력데이터, 정답 데이터를 생성함.
print(X.shape, Y.shape)

def make_sequence_dataset(feature, label, window_size):
	feature_list = []
    label_list = []
    for i in rance(len(feature)-window_size):
    	feature_list.append(feature[i:i+window_size])
        label_list.append(label[i+window_size])
       
    return np.array(feature_list), np.array(label_list)

데이터 분리 및 모델 구축

[2] 트레이닝 데이터 / 테스트 데이터 분리

split = -200

x_train = X[0:split]
y_train = Y[0:split]

x_test = X[split:]
y_test = Y[split:]

print(x_train.shape, y_train.shape)
print(x_test.shape, y_test.shape)

[3] LSTM 모델 구축

model = Sequential()
model.add(LSTM(128, activation = 'tanh', input_shape=x_train[0].shape))
model.add(Dense(1, activation='linear')
model.summary()

모델 컴파일 및 학습

[4] 모델 컴파일 및 학습

from tensorflow.keras.callbacks import EarlyStopping
model.compile(loss='mse', optimizer='Adam', metrics=['mae'])
early_stop=EarlyStopping(monitor='val_loss', patience=5)
model.fit(x_train, y_train, validation_data=(x_test, y_test, epochs=100, batch_size=16, callbacks=[early_stop])

[5] 삼성 주가 예측

pred = model.predict(x_test)
plt.figure(figsize=12, 6)
plt.title('3MA+5MA+Adj*Close, window_size=40')
plt.xlabel('period')
plt.ylabel('adj close')
plt.plot(y_test, label='actual')
plt.plot(pred, label='prediction')
plt.grid()
plt.legend(loc='best')
plt.show()