SKT FLY AI : 9일차 - RNN

RNN

---

순환 신경망

• 입력과 출력을 시퀀스 단위로 처리
  • 시퀀스란 문장 같이 단어가 나열된 것
  • 이러한 시퀀스들을 처리하기 위해 고안된 모델을 스퀀스 모델이라고 한다.
  • 그 중에서 RNN은 딥 러닝의 가장 기본적인 시퀀스 모델
• 은닉층에서 활성화 함수를 통해 결과를 내보내는 역할을 하는 노드를 셀이라고 한다.
  • 이 셀은 이전의 값을 기억, 일종의 메모리 역할을 수행(메모리 셀)
  • 은닉층의 메모리 셀에서 나온 값이 다음 은낙층의 메모리 셀에 입력 -> 이 값을 은닉 상태라고 이야기한다.
• 입력, 출력의 개수에 따라 형태 결정
  • 일대일 형태 (Vanilla Neural Network)
    • 한개의 입력에 대한 한 개의 출력을 생성하는 모델
    • 1:1 RNN은 간단한 기계학습 문제에 사용
  • 일대다 형태
    • 한개의 입력에 여러개의 출력 생성
    • 시계열 데이터의 예측, 감정분석, 번역 등에 사용
  • 다대일 형태
    • 여러 입력에 대한 한개의 출력을 생성하는 모델
    • N:1 형태의 RNN은 시계열 데이터의 예측, 상태감지, 경고 발생등에 사용
  • 다대다 형태
    • 여러개의 입력에 대해 여러개의 출력을 생성하는 모델
    • 복잡한 문제를 풀 수 있는 능력
    • 비디오 분류, 이미지 캡셔닝, 게임 API등 문제에 적용

CNN과 RNN

• CNN : 이미지와 같은 spatial data 기반
• RNN : 텍스트나 Time series data와 같은 sequential data 기반

### FinanceDataReader를 사용해서 삼성전자 주식을 받아옴
import FinanceDataReader as fdr
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns

samsung = fdr.DataReader('005930', '1998-09-01', '2022-10-30')
samsung.shape

### 2. 데이터 시각화
plt.figure(figsize=(12,6))
sns.lineplot(y=samsung['Close'], x=samsung.index)
plt.show()

### 3. 특성 추가
samsung['3MA'] = np.around(samsung['Close'].rolling(window=3).mean(), 0)
samsung['5MA'] = np.around(samsung['Close'].rolling(window=5).mean(), 0)
# 최고와 최저의 중간값을 특성으로 추가한다.
samsung['Mid'] = (samsung['High'] + samsung['Low'])/2
samsung.head()

### 이상치 처리
# 거래량 'Volume'이 0인 데이터는 이상치로 판단하고, Volume이 0인 행만 추출해서 확인한다.
samsung.loc[samsung['Volume']==0]

### Volume이 0인 값을 NaN값으로 변경해서 결측치로 처리
# 시계열 데이터는 연속성을 가져야 해서 결측치를 처리하면 안된다...
samsung['Volume'] = samsung['Volume'].replace(0, np.NaN)
samsung.isna().sum(axis=0)

### 결측치 처리
# 일단 해본다.
samsung = samsung.dropna()
samsung.isna().sum(axis=0)

### 데이터 정규화(최대-최소 정규화)
from sklearn.preprocessing import MinMaxScaler
import pandas as pd

sc = MinMaxScaler()
scale_cols = ['Close', '3MA', '5MA', 'Mid', 'Volume']
df_scaled = sc.fit_transform(samsung[scale_cols])

df_scaled = pd.DataFrame(df_scaled)
df_scaled.columns = scale_cols

print(df_scaled[:5])

### 시퀀스 데이터로 변형
# 시퀀스 형태로 변형
def make_sequence_dataset(feature, label, window_size):
  feature_list = []
  label_list = []
  for i in range(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)

import pandas as pd

# 데이터 프레임 생성
feature_cols = ['3MA', '5MA', 'Mid', 'Volume']
label_cols = ['Close']
npX = pd.DataFrame(df_scaled, columns=feature_cols).values
npY = pd.DataFrame(df_scaled, columns=label_cols).values
print(npX.shape, npY.shape)

### 윈도우 사이즈는 타임스텝의 길이로 설정
window_size = 10 # 5이상
X_data, Y_data = make_sequence_dataset(npX,npY, window_size)
print(X_data.shape, Y_data.shape)

### 테스트 데이터 분리
split = int(len(X_data)*0.8)
X_train = X_data[0:split]
y_train = Y_data[0:split]
X_test = X_data[split:]
y_test = Y_data[split:]
print(X_train.shape, y_train.shape)
print(X_test.shape, y_test.shape)

### 모델 만들기
from tensorflow import keras
from tensorflow.keras import layers

model = keras.Sequential()
model.add(layers.LSTM(64, activation='tanh', input_shape=(10,4))) # input_shape의 첫번째 인자는 window_size와 같아야 한다...
model.add(layers.Dense(16, activation='relu'))
model.add(layers.Dense(1))

model.summary()

### 모델 학습
from tensorflow.keras.losses import Huber
from keras.callbacks import EarlyStopping

model.compile(loss='mse', optimizer='adam', metrics=['mae'])
early_stop = EarlyStopping(monitor='val_loss', patience=10)

EPOCHS = 100
BATCH_SIZE = 16
history = model.fit(X_train, y_train,
                    epochs=EPOCHS,
                    batch_size=BATCH_SIZE,
                    validation_data=(X_test, y_test),
                    callbacks = [early_stop])

### 학습 곡선
import matplotlib.pyplot as plt

def plot_history(history):
  his_dict = history.history
  loss = his_dict['loss']
  val_loss = his_dict['val_loss']
  epochs = range(1, len(loss) + 1)
  fig = plt.figure(figsize = (12, 5))
  ax1 = fig.add_subplot(1, 2, 1)
  ax1.plot(epochs, loss, 'b-', label = 'train_loss')
  ax1.plot(epochs, val_loss, 'r-', label = 'val_loss')
  ax1.set_title('train and val loss')
  ax1.set_xlabel('epochs')
  ax1.set_ylabel('loss')
  ax1.legend()
  acc = his_dict['mae']
  val_acc = his_dict['val_mae']
  ax2 = fig.add_subplot(1, 2, 2)
  ax2.plot(epochs, acc, 'b-', label = 'train_mae')
  ax2.plot(epochs, val_acc, 'r-', label = 'val_mae')
  ax2.set_title('train and val mae')
  ax2.set_xlabel('epochs')
  ax2.set_ylabel('mae')
  ax2.legend()
  plt.show()

plot_history(history)

### 평가 및 예측
loss, mae = model.evaluate(X_test, y_test)

y_pred = model.predict(X_test)

for i in range(5):
  print('Close:', y_test[i], "Predict:", y_pred[i])

### 결과 시각화
plt.figure(figsize=(12, 6))
plt.ylabel('Close')
plt.xlabel('period')
plt.plot(y_test[20:], label='actual')
plt.plot(y_pred, label='prediction')
plt.grid()
plt.legend(loc='best')
plt.show()

### MAPE(평균 절대값 백분율 오차)
loss, mae = model.evaluate(X_test, y_test)
mape1 = np.sum(abs(y_test-y_pred)/y_test) / len(X_test)
print(mape1)