Tensorflow LSTM 예시 코드(LSTM, RNN, ecg.csv)

1. 기본세팅

1.1. 라이브러리

import numpy as np 
import pandas as pd
import tensorflow as tf
import matplotlib.pyplot as plt

from sklearn.preprocessing import MinMaxScaler

from tensorflow.keras import layers
from tensorflow.keras.models import Model
from sklearn.metrics import accuracy_score, precision_score, recall_score
from sklearn.model_selection import train_test_split

import os

1.2 GPU 세팅

os.environ["CUDA_VISIBLE_DEVICES"]="0"
gpus = tf.config.experimental.list_physical_devices('GPU')
if gpus:
    try:
        tf.config.experimental.set_memory_growth(gpus[0], True)
    except RuntimeError as e:
        print(e)

2. Data Load

컬럼 0~139 까지는 시간에 따른 심전도 데이터
컬럼 140은 정상/비정상 class값

# Download the dataset
data = pd.read_csv('http://storage.googleapis.com/download.tensorflow.org/data/ecg.csv', header=None)
data.head()

데이터를 X/Y로 분할

# 마지막 컬럼 전까지는 X로
X = data.values[:,:-1]
# 마지막 컬럼은 Y로
Y = data.values[:,-1]

print('X shape :',X.shape) #=> X shape : (4998, 140)
print('Y shape :',Y.shape) #=> Y shape : (4998,)
# 데이터의 개수가 4998개이며 X는 140개의 피처로 이루어져있다.

3. Data Preprocessing(데이터 전처리)

3.1 데이터 Train/Test 분할

데이터의 개수가 약 5천개로 적기때문에 해당 글에서는 Train/Validation/Test로 나누지 않고 Train/Test로 나눈다.

이후 Validation 부분에서는 Train set을 사용한다.

x_train, x_test, y_train, y_test = train_test_split(X,Y,test_size=0.2,random_state=7, stratify=Y)
print('Train shape :',x_train.shape) #=> Train shape : (3998, 140)
print('Test shape :',x_test.shape) #=> Test shape : (1000, 140)

stratify는 균등하게 분배하기 위한 조건을 의미한다.

위 코드에서는 stratify=Y를 주어서 Class값이 Train/Test에 골고루 분배되도록 하였다.

아래 코드를 통해서 심전도 정상/비정상 class가 골고루 분배됐음을 확인할수 있다.

from collections import Counter
print(Counter(y_train)) #=> Counter({True: 2335, False: 1663})
print(Counter(y_test)) #=> Counter({True: 584, False: 416})

3.2 Scaling

scaler = MinMaxScaler()

x_train = scaler.fit_transform(x_train)
x_test = scaler.transform(x_test) # test set에는 transform만 사용하기

3.3 데이터 타입 변경하기

3.3.1 X 데이터 변경하기(입력 데이터, 독립변수)

x_train = tf.cast(x_train, tf.float32)
x_test = tf.cast(x_test, tf.float32)

3.3.2 Y 데이터 변경하기(목표값, Class, Label)

원본에서는 0과 1로 되어 있는데, 알아보기 쉽게하기 위하여 True/False로 변경한다.

0과1은 다음과 같이 변환된다. (0 -> False / 1 -> True)

y_train = y_train.astype(bool)
y_test = y_test.astype(bool)

4. 정상/비정상 시각화해보기

4.1 정상 데이터 시각화해보기

normal_x_train = x_train[y_train]

fig, ax = plt.subplots(ncols=3, nrows=2, figsize=(9,6))
fig.suptitle("Normal ECG")
ax = ax.ravel() # ax가 3*2차원이기에 for에 1차원으로 만들어 넣기위함
for idx, ax in enumerate(ax):
    ax.grid()
    ax.plot(np.arange(len(normal_x_train[idx])), normal_x_train[idx])
plt.show()

4.2 비정상 데이터 시각화해보기

anomalous_x_train = x_train[~y_train]

fig, ax = plt.subplots(ncols=3, nrows=2, figsize=(9,6))
fig.suptitle("anomalous ECG")
ax = ax.ravel() # ax가 3*2차원이기에 for에 1차원으로 만들어 넣기위함
for idx, ax in enumerate(ax):
    ax.grid()
    ax.plot(np.arange(len(anomalous_x_train[idx])), anomalous_x_train[idx])
plt.show()

5. LSTM(RNN) 모델 생성 및 학습

LSTM 및 RNN의 모델 구조는 아래와 같다.

아래 그림에서 붉은색 박스가 LSTM모델에 입력될 Sequence(입력데이터)를 의미하며,

우리에게는 심전도 0~139까지가 Sequence에 해당된다.

Sequence의 각 값은 개별 Input Vector로 변경되며, 우리는 각각의 심박수값이 Input Vector에 해당한다.

5.1 LSTM(RNN)에 맞게 입력데이터 차원 변경

현재 우리는 Sequence정보를 1차원 array로 가지고 있기 때문에 각각의 심박수값을 Input Vector로 만들어주어야 한다.

따라서 expand_dims()를 사용하여 차원정보를 변경해준다.

x_train_ex = tf.expand_dims(x_train, axis=2)
x_test_ex = tf.expand_dims(x_test, axis=2)

print('원본 차원 정보 :',x_train.shape) #=> 원본 차원 정보 : (3998, 140)
print('변경된 차원 정보 :',x_train_ex.shape) #=> 변경된 차원 정보 : (3998, 140, 1)

# (3998, 140, 1)는 각각 아래의 의미를 가진다.
# (데이터 개수, Sequence 길이, Input Vector 길이)

아래의 코드로도 동일한 효과를 낼수 있다.

#x_train_ex2 = tf.reshape(x_train,(x_train.shape[0],x_train.shape[1],1))
#x_test_ex2 = tf.reshape(x_test,(x_test.shape[0],x_test.shape[1],1))
#print('변경된 차원 정보 :',x_train_ex2.shape)
#=> 변경된 차원 정보 : (3998, 140, 1)

5.2 LSTM 모델 생성

Bidirectional은 양방향 LSTM을 구현할때 사용된다.

return_sequences=True는 LSTM을 여러층 쌓을 때 모든 Squence의 hidden state를 출력하게 하는 설정이다.

model = tf.keras.Sequential([
    layers.LSTM(100, return_sequences=True, input_shape=(x_train_ex.shape[1], x_train_ex.shape[2])),
    layers.Dropout(0.25),
    layers.Bidirectional(layers.LSTM(100)),
    layers.Dropout(0.25),
    layers.Dense(1, activation='sigmoid')
])

동일하게 .add를 통해서도 구현 가능하다.

# model = tf.keras.Sequential()
# model.add(layers.LSTM(100, return_sequences=True, input_shape=(x_train_ex.shape[1], x_train_ex.shape[2])))
# model.add(layers.Dropout(0.25))
# model.add(layers.Bidirectional(layers.LSTM(100)))
# model.add(layers.Dropout(0.25))
# model.add(layers.Dense(1, activation='sigmoid'))

또한 LSTM이 아니라 RNN으로 구현하고 싶다면 layers.RNN을 사용하면 된다.

# model = tf.keras.Sequential([
#     layers.RNN(100, return_sequences=True, input_shape=(x_train_ex.shape[1], x_train_ex.shape[2])),
#     layers.Dropout(0.25),
#     layers.Bidirectional(layers.RNN(100)),
#     layers.Dropout(0.25),
#     layers.Dense(1, activation='sigmoid')
# ])

모델의 요약정보를 보고싶다면 아래의 명령어로 수행 가능하다.

model.summary()

위 요약정보를 통해서 다시 모델을 살펴보자면 아래와 같다.

<LSTM>

"return_sequences=True"이기 때문에 모든 Squence(140)에 대해서 hidden state가 출력되었다.

이때 hidden state의 길이는 100이다.

<Dropout>

25%의 가중치를 랜덤하게 제거한다.

이때 차원정보는 변하지 않는다.

<Bidirectional LSTM>

먼저, "return_sequences=False"이기 때문에 Squence의 마지막 hidden state만 출력된다.

양방향 LSTM을 사용했기 때문에 순방향 hidden state와 역방향 hidden state가 출력되어 200길이의 hidden state가 생성된다.

<Dropout_1>

25%의 가중치를 랜덤하게 제거한다.

이때 차원정보는 변하지 않는다.

<Dense>

입력받은 길이 200의 Vector로부터 길이 1의 Vector를 출력한다.

예측 목표가 정상/비정상을 판단하는 binary classification 문제이기 때문이다.

 

5.3 모델 컴파일

binary classification 문제이기 때문에 'binary_crossentropy'를 사용한다.

model.compile(loss = 'binary_crossentropy',
              optimizer=tf.keras.optimizers.Adam(learning_rate=0.001))

5.4 모델 학습

아래 코드에서 주의할점은 다음과 같다.

LSTM 입력이기 때문에 차원변경을 통해서 x_train_ex를 만들었다 x_train을 넣는 실수를 범하지 말자!!

해당 예시에서는 데이터의 개수가 작아서 validation set을 따로 만들지 않아서 Validation_data에 train set을 넣었다.

training_record = model.fit(x_train_ex, y_train,
                    epochs=30,
                    batch_size=128,
                    validation_data=(x_train_ex, y_train),
                    shuffle=True)

5.5 Learning Curve 시각화

epoch이 진행됨에 따라서 Train 및 Validation의 성능을 확인할 수 있다.

그런데 우리는 validation set에 train set을 입력했는데 왜 두 그래프가 다른지 의아할 수 있다.

그 이유는 train할때 batch 별로 역전파를 수행하기에 같은 epoch여도 batch마다 사용된 가중치값이 다르다.

validation은 해당 epoch의 모든 batch에 대한 역전파가 완료된 상태에서 진행되기 때문에,

같은 데이터여도 다른 결과값을 가질수 있다.

fig, ax = plt.subplots()
plt.plot(training_record.history["loss"], label="Training Loss")
plt.plot(training_record.history["val_loss"], label="Validation Loss")
plt.legend()
fig.suptitle("Loss")
plt.show()

6. 평가

6.1 평가하기

binary classification 문제이기 때문에 출력된 값은 True일 확률값이 출력된다.

즉, 0.56이라는 결과가 나왔을 경우 "True(심전도 정상)일 확률이 56%이다"라고 해석 가능하다.

(마지막 Dense Layer에서 sigmoid를 취해주었기 때문에 0~1 범위의 값을 가진다.)

(softmax가 아닌 sigmoid를 사용한 이유는 binary classification 문제이기 때문이다.)

pred_proba = model.predict(x_test_ex)

6.2 Class 판단하기

현재 pred_proba에 들어있는 값은 True일 확률이지 True/False로 판단된 값이 아니다.

따라서 우리는 특정 Threshold 값을 정하여 True/False를 판단해주어야한다.

# 2차원인 pred_proba를 1차원으로 변경해준다 (1000,1)=>(1000)
pred_proba_1d = pred_proba.reshape(-1)

# 임계치 이상이면 True 미만이면 False를 부여한다.
threshold = 0.5
pred = (pred_proba_1d >= threshold)

6.3 성능 계산하기

# Compute the metrics
accuracy_test_rnn= accuracy_score(y_test, pred)
print(f'Accuracy: {accuracy_test_rnn}')
#=> Accuracy: 0.988

precision_test_rnn=precision_score(y_test, pred)
print(f'Precision = {round(precision_test_rnn,3)}')
#=> Precision = 0.99

recall_test_rnn=recall_score(y_test, pred)
print(f'Recall = {round(recall_test_rnn,3)}')
#=> Recall = 0.99