[AI 스터디] Section 9 : RNN

RNN 특징

1. Sequence Data 에 특화 : 순서가 있는 데이터를 처리하기 위해 특화된 모델
2. ‘기억’ 능력을 갖고 있음
   • 순서대로 들어오는 것을 기억하다가 다음 순서로 들어온 것의 결과는 무엇일지 예측
     ⇒ 맥락에 맞게 기억이 이어지고, 예측이 진행
   • 네트워크의 기억 : 지금까지의 입력 데이터를 요약한 정보
     ⇒ 맥락에 맞지 않은 내용이 들어온 경우 (새로운 입력이 들어옴) : 그때마다 네트워크는 기억을 조금씩 수정
3. 입력을 모두 처리하고 난 후 네트워크에게 남겨진 기억은 시퀀스 전체를 요약하는 정보로 압축
   ⇒ 사람의 시퀀스 정보 처리 방식과 비슷, 기억을 바탕으로 새로운 단어 이해
4. 새로운 단어마다 과정 반복 ⇒ Recurrent (순환적)

RNN 구조

: 순서대로 펼쳐 놓으면 가중치를 공유하는 매우 딥한 뉴럴네트워크

\( X_t \) : 시간에 따라 들어가는 시계열 정보 ⇒ 각 타임스텝마다 밸류가 존재 / unfold 표시 : 하나로 되어있는 것 (왼쪽) / 오른쪽 : 펼쳐 놓은 모습 (weight 공유 가능)

 

⇒ Internal State : \( h_t = tanh(W_hh_{t-1} + W_xx_t) \) → 새로운 기억을 만드는 것 (tanh이므로 (-1, 1))

    Output : \( O_t =softmax(W_oh_t) \) → 다중 분류 문제인 경우

1. 과정
   • 타임스텝이 전부 한번에 입력
   • 첫번쨰 타임 스텝 \( X_1 \) : 의 밸류를 이용한 새로운 기억을 만듬
   • 두번째 타임 스텝 : \( X_1 \) 의 기억과  \(  X_2 \) 가 합쳐져 새로운 기억이 만들어짐
   • 마지막 타임스텝 : 앞에서 계속 압축된 기억과 마지막 밸류가 합쳐져 결과를 생성
2. 예시 : 오늘 가격이 60원, 내일 70원이면, 모레 가격은 ?
3. 특징
   • 타임스텝별 아웃풋도 출력 가능
   • 항상 같은 가중치를 가진다.
4. 학습 : BPTT (Backpropagation Through Time)로 파라미터 학습
   • 타임스텝으로 계속 연결되므로 변형된 Backpropagation으로 훈련됨
     ⇒ 타임 스텝 만큼 계속 연결되는 딥한 뉴럴 네트워크의 형태를 띄기 때문
5. Simple (Vanilla) RNN 의 구조
   → 단기기억에 특화된 모델로, 원리를 이해하는 용도로 사용되고, 실전에선 잘 사용하지 않음
   

   

   
   
   1. Hidden State Update : \( H_t = \phi(H_{t-1}W_{hh} + X_tW_{xh} + b_h) \)
   2. Observation Update : \( o_t = H_tW_{hq} + b_q \)
   3. weight 가 학습되어짐
      \( W_{hh} \) : state 가중치 매트릭스
      \( W_{hq} \) : 출력 가중치 매트릭스
      \( W_{xh} \) : 입력 가중치 매트릭스,
   4. 입력 : \( X_t \) 타임 시퀀스 데이터
      → 윈도우 사이즈 만큼의 개수가 들어감
          ( 윈도우사이즈 = 3 ⇒ 인풋 : 타임스텝 1 2 3 (3개) )
   5. 기억 생성 : 입력 → tanh 통과
      (타임 스텝 마다 과정을 반복해서, 마지막 타임스텝이 될 때 까지)
   6. 출력 : 마지막 타임 스텝까지의 최종 기억

 

LSTM

: 원래 있는 단기 기억에 장기기억을 하나 추가해, 장기기억이 쭉 연결된 형태

   ⇒ 타임스텝이 길어져도, 옛날 기억을 덜 잊도록

 

tanh : simple RNN과 동일한 원리 / gate : 3개 추가됨 / 0~1 : 시그모이드

 

내부 구조

→ 게이트의 값은 역전파로 최종 예측에 도움이 되도록 스스로 조정됨

→ weight를 simple RNN보다 3개 추가하여 학습하는데 있어서 더 정교한 학습이 가능

1. Input : 이전 step의 hidden + 새로운 데이터 → 새로운 cell 상태 후보
   $$\tilde{C}^t = \tanh(W_c[a^{t-1}, x^t] + b_c)$$
   
   
2. Update Gate : Input을 어느 정도 받아들일지 결정 (0-무시, 1-전체 )
   → Input이 아웃풋 예측에 도움이 되는지에 따라 장기기억으로 보낼지 말지 결정
   $$\Gamma_u = \sigma(W_u[a^{t-1}, x^t] + b_u)$$
   
   
3. Forget Gate : 이전 cell state를 어느 정도 기억할지 결정 (0-forget, 1-전체 기억)
   → 현재 하고 있는 기억이 아웃풋 예측에 도움이 되는지에 따라 기억할지 결정
   $$\Gamma_f = \sigma(W_f[a^{t-1}, x^t] + b_f)$$
4. Output Gate : Input을 어느 정도 다음 step으로 보낼지 결정
   $$\Gamma_o = \sigma(W_o[a^{t-1}, x^t] + b_o)$$
   
   
5. Cell State
   $$C^t = \Gamma_u \ast \tilde{C}^t + \Gamma_f \ast C^{t-1}$$
   
   
6. Hidden State
   $$a^t = \Gamma_o \ast \tanh(C^t)$$

Recurrent Neural Network I/O Overview

→ simple RNN, LSTM, GRU 인풋, 아웃풋은 동일

 

1. Input shape : 3 차원
   • 첫번째 차원 : Batch size
   • 두번째 차원 : Time step → 윈도우 사이즈
   • 세번째 차원 : Input features → dims • Univariate – one • Multivariate - many
2. EXAMPLE
   • 내가 배치를 한번에 64개씩 데이터를 묶어서 보낸다 → batch_size = 64
   • 윈도우 사이즈를 7로하여, 과거 7일의 주가를 보고 오늘 주가를 예측 하겠다. → time step = 7 / Input features = 1
   • 주식가격, 환율, 금리, 거래량을 사용하여 예측하겠다. → Input features = 4

---

실습 - LSTM을 이용한 수열 패턴 인식

데이터 소개 및 예측

1. input 은 0 ~ 99 까지의 연속된 숫자
2. target 은 (1 ~ 101) * 2

모델 구조

: 연속된 5 개의 숫자를 보고 다음 숫자를 알아맞추도록 LSTM을 이용한 모델 ⇒ ex) [[5], [6], [7], [8], [9]] → [20]

  [[35], [36], [37], [38], [39]]. → [80]

sequential data 생성

import numpy as np
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, LSTM

numbers = [[i] for i in range(105)]
numbers[:5] # [[0], [1], [2], [3], [4]]

# 순서가 있는 데이터를 만들어주는 것
# window size = 5
# 연속된 숫자가 있을때, 마지막 숫자의 2배가 되는 숫자를 맞춰라
data = []
target = []
for i in range(5, len(numbers)):
    data.append(numbers[i-5: i]) # 5부터 시작해서 i-5
    target.append(numbers[i][0] * 2)

print(data[5]) # [[5], [6], [7], [8], [9]]
print(target[5]) # 20

# 파이썬 List는 사용할 수 없으므로 -> numpy ndarray로 변환
data = np.array(data, dtype="float32")
target = np.array(target, dtype="float32")

data.shape, target.shape # ((100, 5, 1), (100,)) : 5 -> time step / 1 -> feature

LSTM 모델 : 스케일링 x

model = Sequential()
model.add(LSTM(16, input_shape=(5, 1)))
model.add(Dense(1)) # 연속된 숫자 알아맞추기 이므로 1

# 모델 컴파일
model.compile(optimizer='adam', loss='mae', metrics=['mae'])

# 모델 훈련
history = model.fit(data, target, epochs=500, validation_split=0.2)

# 간단해서 test data를 만듬
test_data = [[35], [36], [37], [38], [39]]
x = np.array(test_data, dtype="float32").reshape(1, 5, 1)

# 예측 : [[2950.1875]] => 구림
model.predict(x.reshape(1, 5, 1)) * 100 

LSTM 모델 : 스케일링 o

data = []
target = []
for i in range(5, len(numbers)):
    data.append(numbers[i-5: i]) # 5부터 시작해서 i-5
    target.append(numbers[i][0] * 2)

print(data[5]) # [[5], [6], [7], [8], [9]]
print(target[5]) # 20

# 파이썬 List는 사용할 수 없으므로 -> numpy ndarray로 변환
data = np.array(data, dtype="float32")
target = np.array(target, dtype="float32")

# Normalize
data = data / 100.
target = target / 100.

data.shape, target.shape # ((100, 5, 1), (100,)) : 5 -> time step / 1 -> feature

model = Sequential()
model.add(LSTM(16, input_shape=(5, 1)))
model.add(Dense(1)) # 연속된 숫자 알아맞추기 이므로 1

# 모델 컴파일
model.compile(optimizer='adam', loss='mae', metrics=['mae'])

# 모델 훈련
history = model.fit(data, target, epochs=500, validation_split=0.2)

# 간단해서 test data를 만듬
test_data = [[35], [36], [37], [38], [39]]
x = np.array(test_data, dtype="float32").reshape(1, 5, 1) /100

# 예측 : [[[80.26072] => 좋다 
model.predict(x.reshape(1, 5, 1)) * 100 

---

실습 - LSTM을 이용한 주식 가격 예측

데이터 : Yahoo finance data

모델 : Apple 주식의 가격 추세 예측 (지난 window-size 일의 역사적 가격을 사용하여 시간 t의 가격을 예측)

→ 추세를 파악

데이터 불러오기

!pip install yfinance

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import yfinance as yf

aapl = yf.download('AAPL', start='2018-01-01', end='2022-12-31', progress=False)

# 애플 주가 그래프 그리기
aapl.Close.plot()

데이터셋 분리 및 스케일링

hist = []
target = []
window = 3
close = aapl['Close'].values  # 종가

for i in range(len(close) - window): # 윈도우 개수만큼 남기도록
    x = close[i:i+window]
    y = close[i+window]
    hist.append(x)
    target.append(y)

close[:10] 
# array([[27.33250046],
#       [26.5625    ],
#       [26.56500053],
#       [26.9375    ],
#       [27.97249985],
#       [28.00250053],
#       [27.3125    ],
#       [27.55500031],
#       [27.45000076],
#       [26.70499992]])

hist[:5] # 하나씩 움직이며 윈도우 개수만큼 가짐
[array([[27.33250046],
        [26.5625    ],
        [26.56500053]]),
 array([[26.5625    ],
        [26.56500053],
        [26.9375    ]]),
 array([[26.56500053],
        [26.9375    ],
        [27.97249985]]),
 array([[26.9375    ],
        [27.97249985],
        [28.00250053]]),
 array([[27.97249985],
        [28.00250053],
        [27.3125    ]])]

# hist"의 각 요소는 window개 timestep의 list입니다. 
# => 1씩 증가하기 때문에 "hist"의 두 번째 요소의 마지막 항목은 "target"의 첫 번째 요소와 같아야 합니다.
# 또한 마지막 숫자가 같아야 합니다.
# print(close[-1])
# print(i+length)
# print(target[-1])

hist[1][-1] == target[0] # True

hist = np.array(hist)
target = np.array(target)
target = target.reshape(-1, 1) 
print(hist.shape) # 1835, 3 : 3일치의 데이터를 받아서
print(target.shape) # 1835, 1 : 다음날 하루의 종가를 알아맞추도록

# train/test split : 독립적이지 않으므로 shuffle 하면 안되므로 패키지 사용 불가
# 1098일의 데이터로 모델을 학습시키고 다음 100일의 데이터로 테스트하는 방식으로 데이터를 분할
split = len(hist) - 100     # 1735
X_train = hist[:split]      # 1735,3
X_test = hist[split:]       # 100,3
y_train = target[:split]    # 1735,1
y_test = target[split:]     # 100, 1

# 스케일링
sc1 = MinMaxScaler()
X_train_scaled = sc1.fit_transform(X_train)
X_test_scaled = sc1.transfrom(X_test)

sc2 = MinMaxScaler()
y_train_scaled = sc2.fit_transform(y_train)
y_test_scaled = sc2.transfrom(y_test)

# time sequence가 없으므로 차원이 하나 더 늘어야하므로 reshape -> 3D가 되어야함
X_train = X_train.reshape(-1, window, 1)
X_tes = X_test.reshape(-1, window, 1)

X_train.shape, X_test.shape # ((1735, 3, 1), (100, 3, 1)) : 배치 사이즈, 타임 시퀀스, 피처 개수

모델 생성

model = tf.keras.Sequential()
# 첫번째 LSTM과 두번째 LSTM이 연결되어야하므로 return_sequences를 주어야함
# 3개의 LSTM 셀을 쌓음
model.add(LSTM(units=64, return_sequences=True, input_shape=(window, 1), dropout=0.2))
model.add(LSTM(units=32, return_sequences=True, dropout=0.2))
model.add(LSTM(units=16, dropout=0.2))
# Dense 레이어와 연결
model.add(Dense(units=1))
model.add(Lambda(lambda x: x * 100))

# 컴파일
model.compile(optimizer='adam', loss='mean_squared_error')
# 예측
history = model.fit(X_train, y_train, epochs=100, batch_size=32)

# 훈련 동안의 loss 변화 시각화
plt.plot(history.history['loss'])
plt.legend(['Training Loss'])
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.show();

추세 예측

# 앞으로 100일동안의 주가 추이 예측
pred = model.predict(X_test)

# 예측과 실제 비교한 시각화
plt.figure(figsize=(12,6))
plt.plot(np.concatenate((y_train, y_test)), label='True')
plt.plot(np.concatenate((y_train, pred)), label='Predicted')
plt.title('Apple Stock Price Prediction')
plt.legend()
plt.show()

plt.figure(figsize=(8,4))
plt.plot(y_test, label='True')
plt.plot( pred, label='Predicted')
plt.title('Apple Stock Price Prediction')
plt.legend()
plt.show()