반응형
RNN,LSTM,GRU
1. RNN
이전의 데이터가 다음 데이터에 영향을 준다.
시계열 분석에 유용하다
2. LSTM
RNN의 단기기억문제 해결책으로 고안됨 (시간이 많이 지난 데이터의 영향력이 너무 낮아지는 문제)
3. GRU
LSTM구조보다 간결하게 변경함, 빠른 속도와 유사한 성능
평가 함수
### Evaluation of 1 pair of set
def evaluation(Y_real, Y_pred, graph_on=False):
loss_length = len(Y_real.values.flatten()) - len(Y_pred)
if loss_length != 0:
Y_real = Y_real[loss_length:]
if graph_on == True:
pd.concat([Y_real, pd.DataFrame(Y_pred, index=Y_real.index, columns=['prediction'])], axis=1).plot(kind='line', figsize=(20,6),
xlim=(Y_real.index.min(),Y_real.index.max()),
linewidth=3, fontsize=20)
plt.title('Time Series of Target', fontsize=20)
plt.xlabel('Index', fontsize=15)
plt.ylabel('Target Value', fontsize=15)
MAE = abs(Y_real.values.flatten() - Y_pred).mean()
MSE = ((Y_real.values.flatten() - Y_pred)**2).mean()
MAPE = (abs(Y_real.values.flatten() - Y_pred)/Y_real.values.flatten()*100).mean()
Score = pd.DataFrame([MAE, MSE, MAPE], index=['MAE', 'MSE', 'MAPE'], columns=['Score']).T
Residual = pd.DataFrame(Y_real.values.flatten() - Y_pred, index=Y_real.index, columns=['Error'])
return Score, Residual
# Score_tr, Residual_tr = evaluation(Y_train, pred_tr_reg1, graph_on=True)
### Evaluation of train/test pairs
def evaluation_trte(Y_real_tr, Y_pred_tr, Y_real_te, Y_pred_te, graph_on=False):
Score_tr, Residual_tr = evaluation(Y_real_tr, Y_pred_tr, graph_on=graph_on)
Score_te, Residual_te = evaluation(Y_real_te, Y_pred_te, graph_on=graph_on)
Score_trte = pd.concat([Score_tr, Score_te], axis=0)
Score_trte.index = ['Train', 'Test']
return Score_trte, Residual_tr, Residual_te
RNN을 이용한 분석
- 필요 라이브러리 다운
from keras.models import Sequential, Model, load_model
from keras.layers import Input, Dense, Activation, Flatten, Dropout
from keras.layers import SimpleRNN, LSTM, GRU
- 데이터 다운 및 파라미터 설정
# Data Loading
location = './Data/Cryptocurrency/Bitcoin.csv'
raw_all = pd.read_csv(location, index_col='Date')
raw_all.index = pd.to_datetime(raw_all.index)
# Parameters
criteria = '2020-01-01'
scaler = preprocessing.MinMaxScaler()
sequence = 60
batch_size = 32
epoch = 10
verbose = 1
dropout_ratio = 0
- 데이터 분할
# Feature Engineering
## Train & Test Split
train = raw_all.loc[raw_all.index < criteria,:]
test = raw_all.loc[raw_all.index >= criteria,:]
print('Train_size:', train.shape, 'Test_size:', test.shape)
- 스케일링
train의 값으로 test의 스케일링일 수행함. (따라서 test는 0~1 값이 아니다)
train과 test를 따로 스케일링 할 수도 있다.
train_scaled = scaler.fit_transform(train)
test_scaled = scaler.transform(test)
- 차원 세팅(3차원으로 만든다)
## X / Y Split
X_train, Y_train = [], []
for index in range(len(train_scaled) - sequence):
X_train.append(train_scaled[index: index + sequence])
Y_train.append(train_scaled[index + sequence])
X_test, Y_test = [], []
for index in range(len(test_scaled) - sequence):
X_test.append(test_scaled[index: index + sequence])
Y_test.append(test_scaled[index + sequence])
<example>
- 데이터 자체의 형태 확인
## Retype and Reshape
#3차원으로 만들어야함
X_train, Y_train = np.array(X_train), np.array(Y_train)
X_test, Y_test = np.array(X_test), np.array(Y_test)
print('X_train:', X_train.shape, 'Y_train:', Y_train.shape)
print('X_test:', X_test.shape, 'Y_test:', Y_test.shape)
<example> - train
- 모델링
# RNN
model = Sequential()
#return seqeunce : 계산된 파라미터를 다시 받아올것인가?
model.add(SimpleRNN(128, input_shape=(X_train.shape[1], X_train.shape[2]), return_sequences=True, activation='relu'))
model.add(Dropout(dropout_ratio))
model.add(SimpleRNN(256, return_sequences=True, activation="relu"))
model.add(Dropout(dropout_ratio))
model.add(SimpleRNN(128, return_sequences=True, activation="relu"))
model.add(Dropout(dropout_ratio))
model.add(SimpleRNN(64, return_sequences=True, activation="relu"))
model.add(Dropout(dropout_ratio))
model.add(Flatten())
model.add(Dense(1))
model.compile(optimizer='adam', loss='mean_squared_error')
model.summary()
model_fit = model.fit(X_train, Y_train,
batch_size=batch_size, epochs=epoch,
verbose=verbose)
plt.plot(pd.DataFrame(model_fit.history))
plt.grid(True)
plt.show()
- 예측값
# prediction
Y_train_pred = model.predict(X_train)
Y_test_pred = model.predict(X_test)
- 평가
# evaluation
result = model.evaluate(X_test, Y_test_pred)
if scaler != []:
Y_train = scaler.inverse_transform(Y_train)
Y_train_pred = scaler.inverse_transform(Y_train_pred)
Y_test = scaler.inverse_transform(Y_test)
Y_test_pred = scaler.inverse_transform(Y_test_pred)
Score_RNN, Residual_tr, Residual_te = evaluation_trte(pd.DataFrame(Y_train), Y_train_pred.flatten(),
pd.DataFrame(Y_test), Y_test_pred.flatten(), graph_on=True)
display(Score_RNN)
LSTM을 이용한 분석
# Data Loading
location = './Data/Cryptocurrency/Bitcoin.csv'
raw_all = pd.read_csv(location, index_col='Date')
raw_all.index = pd.to_datetime(raw_all.index)
# Parameters
criteria = '2020-01-01'
scaler = preprocessing.MinMaxScaler()
sequence = 60
batch_size = 32
epoch = 10
verbose = 1
dropout_ratio = 0
# Feature Engineering
## Train & Test Split
train = raw_all.loc[raw_all.index < criteria,:]
test = raw_all.loc[raw_all.index >= criteria,:]
print('Train_size:', train.shape, 'Test_size:', test.shape)
## Scaling
train_scaled = scaler.fit_transform(train)
test_scaled = scaler.transform(test)
## X / Y Split
X_train, Y_train = [], []
for index in range(len(train_scaled) - sequence):
X_train.append(train_scaled[index: index + sequence])
Y_train.append(train_scaled[index + sequence])
X_test, Y_test = [], []
for index in range(len(test_scaled) - sequence):
X_test.append(test_scaled[index: index + sequence])
Y_test.append(test_scaled[index + sequence])
## Retype and Reshape
X_train, Y_train = np.array(X_train), np.array(Y_train)
X_test, Y_test = np.array(X_test), np.array(Y_test)
print('X_train:', X_train.shape, 'Y_train:', Y_train.shape)
print('X_test:', X_test.shape, 'Y_test:', Y_test.shape)
# LSTM
model = Sequential()
model.add(LSTM(128, input_shape=(X_train.shape[1], X_train.shape[2]), return_sequences=True, activation='relu'))
model.add(Dropout(dropout_ratio))
model.add(LSTM(256, return_sequences=True, activation="relu"))
model.add(Dropout(dropout_ratio))
model.add(LSTM(128, return_sequences=True, activation="relu"))
model.add(Dropout(dropout_ratio))
model.add(LSTM(64, return_sequences=False, activation="relu"))
model.add(Dropout(dropout_ratio))
model.add(Dense(1))
model.compile(optimizer='adam', loss='mean_squared_error')
model.summary()
model_fit = model.fit(X_train, Y_train,
batch_size=batch_size, epochs=epoch,
verbose=verbose)
plt.plot(pd.DataFrame(model_fit.history))
plt.grid(True)
plt.show()
# prediction
Y_train_pred = model.predict(X_train)
Y_test_pred = model.predict(X_test)
# evaluation
result = model.evaluate(X_test, Y_test_pred)
if scaler != []:
Y_train = scaler.inverse_transform(Y_train)
Y_train_pred = scaler.inverse_transform(Y_train_pred)
Y_test = scaler.inverse_transform(Y_test)
Y_test_pred = scaler.inverse_transform(Y_test_pred)
Score_LSTM, Residual_tr, Residual_te = evaluation_trte(pd.DataFrame(Y_train), Y_train_pred.flatten(),
pd.DataFrame(Y_test), Y_test_pred.flatten(), graph_on=True)
display(Score_LSTM)
# error analysis
# error_analysis(Residual_te, ['Error'], pd.DataFrame(X_train.reshape(X_train.shape[0], X_train.shape[1])), graph_on=True)
GRU를 이용한 분석
# Data Loading
location = './Data/Cryptocurrency/Bitcoin.csv'
raw_all = pd.read_csv(location, index_col='Date')
raw_all.index = pd.to_datetime(raw_all.index)
# Parameters
criteria = '2020-01-01'
scaler = preprocessing.MinMaxScaler()
sequence = 60
batch_size = 32
epoch = 10
verbose = 1
dropout_ratio = 0
# Feature Engineering
## Train & Test Split
train = raw_all.loc[raw_all.index < criteria,:]
test = raw_all.loc[raw_all.index >= criteria,:]
print('Train_size:', train.shape, 'Test_size:', test.shape)
## Scaling
train_scaled = scaler.fit_transform(train)
test_scaled = scaler.transform(test)
## X / Y Split
X_train, Y_train = [], []
for index in range(len(train_scaled) - sequence):
X_train.append(train_scaled[index: index + sequence])
Y_train.append(train_scaled[index + sequence])
X_test, Y_test = [], []
for index in range(len(test_scaled) - sequence):
X_test.append(test_scaled[index: index + sequence])
Y_test.append(test_scaled[index + sequence])
## Retype and Reshape
X_train, Y_train = np.array(X_train), np.array(Y_train)
X_test, Y_test = np.array(X_test), np.array(Y_test)
print('X_train:', X_train.shape, 'Y_train:', Y_train.shape)
print('X_test:', X_test.shape, 'Y_test:', Y_test.shape)
# GRU
model = Sequential()
model.add(GRU(128, input_shape=(X_train.shape[1], X_train.shape[2]), return_sequences=True, activation='relu'))
model.add(Dropout(dropout_ratio))
model.add(GRU(256, return_sequences=True, activation="relu"))
model.add(Dropout(dropout_ratio))
model.add(GRU(128, return_sequences=True, activation="relu"))
model.add(Dropout(dropout_ratio))
model.add(GRU(64, return_sequences=False, activation="relu")) #마지막 layer에 return sequence=False는 flatten효과
model.add(Dropout(dropout_ratio))
model.add(Dense(1))
model.compile(optimizer='adam', loss='mean_squared_error')
model.summary()
model_fit = model.fit(X_train, Y_train,
batch_size=batch_size, epochs=epoch,
verbose=verbose)
plt.plot(pd.DataFrame(model_fit.history))
plt.grid(True)
plt.show()
# prediction
Y_train_pred = model.predict(X_train)
Y_test_pred = model.predict(X_test)
# evaluation
result = model.evaluate(X_test, Y_test_pred)
if scaler != []:
Y_train = scaler.inverse_transform(Y_train)
Y_train_pred = scaler.inverse_transform(Y_train_pred)
Y_test = scaler.inverse_transform(Y_test)
Y_test_pred = scaler.inverse_transform(Y_test_pred)
Score_GRU, Residual_tr, Residual_te = evaluation_trte(pd.DataFrame(Y_train), Y_train_pred.flatten(),
pd.DataFrame(Y_test), Y_test_pred.flatten(), graph_on=True)
display(Score_GRU)
# error analysis
error_analysis(Residual_te, ['Error'], pd.DataFrame(X_train.reshape(X_train.shape[0], X_train.shape[1])), graph_on=True)반응형