딥러닝을 응용한 환율예측으로 가상화폐 차익거래 기회 백테스팅 (1)

먼저 환율 예측에 과거 논문 기반으로 쓰인 

데이터를 다운하여 한 폴더 내부에 넣는다.

# 사용할 모든 파일 리스트로 읽어오기
import os
import numpy as np
import pandas as pd
import math #math 모듈을 먼저 import해야 한다.    
import matplotlib.pyplot as plt

os_file_list = os.listdir('C:/Users/Happy/Desktop/논문용/krwusdtdata/전처리된파일')

read_path = "C:\\Users\\Happy\\Desktop\\논문용\\krwusdtdata\\전처리된파일\\"


# 딕셔너리 안에 모든 데이터 프레임 저장
All_data = {}

for files in os_file_list:
    print(files)
    df= pd.read_csv(read_path+files, encoding = "cp949")
    All_data[files[:-4]] = df

key_list = list(All_data.keys())
key_list = list(set(key_list) - {"BTC_KRW","bin_data"})
key_list

이렇게 한 파일 경로 위치에 있는 모든 csv파일들을 불러온 후에 All_data라는 딕셔너리 내부에 파일명을 이름으로 하여 딕셔너리를 만든다. 한마디로 딕셔너리 내부에 데이터들을 저장하였다.

# All_data , key_list
for i in range(len(key_list)):
    print(key_list[i])
    print(All_data[key_list[i]].info())
#     All_data[key_list[i]] = All_data[key_list[i]]

 

 

데이터들이 이상없는지 확인 한번 해준다.

time 컬럼들이 모드 object형식인 것을 알수 있다. 해당 형식을 바꾸기 위해 함수를 하나 정의해주었다.

def set_index(df):
    df['time'] = pd.to_datetime(df['time'], format='%Y-%m-%d', errors='raise')
    df = df.set_index('time',drop=False)
    return df
    
Use_data = {}
# All_data , key_list
for i in range(len(key_list)):
    Use_data[key_list[i]] = All_data[key_list[i]].copy() # [["time","Open"]]

for i in range(len(key_list)):
    print(key_list[i],i)
    Use_data[key_list[i]] = set_index(Use_data[key_list[i]])

# 모든 컬럼명을 데이터 프레임 이름으로 바꿈
for i in range(len(key_list)):
    col_list = list(Use_data[key_list[i]].columns)
    col_list = list(set(col_list) - {"time"})
    for j in range(len(col_list)):
        print(key_list[i]+"_"+col_list[j])
        Use_data[key_list[i]] = Use_data[key_list[i]].rename(columns={col_list[j]:key_list[i]+"_"+col_list[j]})

그리고 모든 컬럼명에 데이터 프레임 이름을 붙이고 결합하여 하나의 데이터 프레임으로 다시 만들었다. 이렇게 해야 모든 데이터를 사용할 수 있기 때문.

 

# 종가 컬럼 제거.
col_list_2 = list(ALL_DATA_BY_USE.columns)
use_list= []
for i in col_list_2:
    if(i.find("Close") != -1):
        pass
    else:
        use_list.append(i)
ALL_DATA_BY_USE = ALL_DATA_BY_USE[use_list]

여기서 종가 컬럼을 제거한 이유는 LSTM을 이용할 경우에 종가에 가중치가 매우 커질 수 있다. 다음날 환율을 예측할 것인데 종가 컬럼이 들어갈 경우에 시가 == 종가 인 경우가 발생할 것이기 때문이다. 쉽게 말하면 실제로 해당 모델을 돌린다면 진입시점을 시가로 할것인데 그날 종가는 알 수 없기 때문이다.

 

# 이렇게 설정하면 다음날 얼마나 오르고 내리는지 예측하게 됨.
ALL_DATA_BY_USE["KRW_USDT_High - KRW_USDT_Open"] = ALL_DATA_BY_USE["KRW_USDT_High"] - ALL_DATA_BY_USE["KRW_USDT_Open"] # 변동성
ALL_DATA_BY_USE["KRW_USDT_Open_shift_1"] = ALL_DATA_BY_USE[["KRW_USDT_Open"]].shift(-1) #다음날 시작가 데이터 까지 가지고 시작

해당 코드는 변동성에 대한 컬럼을 새로 만들어서 주었고

.shift(-1) 을 함으로써 다음날 시작가 데이터까지 가지고 학습하게 만들었다.

# index가 time의 경우 안좋기 때문에 인덱스 변환.
ALL_DATA_BY_USE = ALL_DATA_BY_USE.reset_index(drop = True)
ALL_DATA_BY_USE

이부분은 학습시 timedelta 형식 데이터는 오류를 반환하기 때문, + 인덱스가 날짜 형식의 경우 인식을 못하기 떄문에 인덱스를 버렷다.

# 일단 다음날 시가를 예측해보자.!!
use_columns = list(ALL_DATA_BY_USE.columns)
print(use_columns)
target = "KRW_USDT_Open"#"KRW_USDT_High - KRW_USDT_Open" # "KRW_USDT_Open"
ALL_DATA_BY_USE = ALL_DATA_BY_USE.dropna()
ALL_DATA_BY_USE[[target]]

여기서 target 을 지정해주었는데 이런식으로 코드를 짜면 같은데이터를 가지고 학습할 컬럼의 target만 바꾸면 되게 하기 위해서 만들었다. 즉 변동성 예측이나 종가 예측도 쉽게 가능하다는 말이다.

from statsmodels.tsa.seasonal import seasonal_decompose


target_col = seasonal_decompose(ALL_DATA_BY_USE[target], model = 'additive' ,period = 500, extrapolate_trend = 1)

fig = plt.figure()
fig = target_col.plot()
fig.set_size_inches(15,12)

ALL_DATA_BY_USE[target+'col_trend'] = target_col.trend
ALL_DATA_BY_USE[target+'col_seasonal'] = target_col.seasonal
ALL_DATA_BY_USE[target+'col_resid'] = target_col.resid

.시계열 분해를 하여 파생변수를 추가하였다.

 

 

ALL_DATA_BY_USE.columns

이걸 해두는 이유는 해당 모델을 다시 사용할때 컬럼 순서가 바뀌어버리면 LSTM모델이 잘못 인식하기 때문에 컬럼 순서 확인을 위해 복사해두고 사용할때

 

 

이런 식으로 사용하면 된다. 이 글은 다음 글에 포스팅 하겠다.

ALL_DATA_BY_USE = ALL_DATA_BY_USE[col_index]

인덱스 그대로 설정해주고.

 

# X_data = ALL_DATA_BY_USE[list(set(use_columns) - {target})]
X_data = ALL_DATA_BY_USE
Y_data = ALL_DATA_BY_USE[[target]]
ALL_DATA_BY_USE = pd.concat([X_data,Y_data],axis = 1).

X, Y(예측할 값.) 를 나누어 준다.

from sklearn.preprocessing import MinMaxScaler,PowerTransformer,LabelEncoder,StandardScaler

X_col = list(X_data.columns)
Scaler = MinMaxScaler()
Scaler.fit(X_data)
X_data = Scaler.transform(X_data)
#     df = Scaler_X.fit_transform(df)
X_data = pd.DataFrame(X_data, columns = X_col)

위 방식을 으로 스케일링을 해준다. 스케일링이 없으면 LSTM모델의 예측율이 현저히 떨어진다. 

# 최대최소 정규화
Y_col = list(Y_data.columns)

MAX_val = float(max(Y_data.values))
MIN_val = float(min(Y_data.values))
# 변환(man,max 변환을 또해주면 기존값을 잃어버림.)
Y_data[Y_col] = Y_data[Y_col].apply(lambda x : (x - MIN_val)/(MAX_val - MIN_val))
# 역변환
# Y_data[Y_col] = Y_data[Y_col].apply(lambda x : (x * (MAX_val - MIN_val) + MIN_val))

MAX,MIN_val을 굳이 변수를 사용하여 저장해둔 이유는 나중에 역변환 할경우 Sclaer.invers_transform 하면 되지만

재정의 해주는 경우에는 리셋되기 때문에 함수를 한번더 만들어서 수동으로 역변환할수 있게 만들었다.

 

여기서부터는 LSTM을 위한 전처리다. 머신러닝을 사용할것이면 이 부분 부터는 다른 포스팅을 참조하길 바란다.

# 지정해야할 초기변수들 
WINDOW_SIZE=64 # 64 #1024 # n일간 데이터를 기반으로 내일 데이터 예측 
BATCH_SIZE=4 # 16 # 32
day_offset = 0 # 예측할 미래 날짜 day_offset = 0 이면 하루 뒤

나는 이런 초매게 변수들을 처음에 지정해주는 편이다. day_offset은 하루뒤를 예측할것인지 이틀뒤를 예측할것인지이다.

def make_window_size_dataset(data, label, window_size, day_offset=0): # x값 , y값 , 윈도우 사이즈, 예측할 미래 날짜
    feature_list = []
    label_list = []
    for i in range(len(data) - window_size - day_offset):
        feature_list.append(np.array(data.iloc[i:i+window_size]))
        label_list.append(np.array(label.iloc[i+window_size + day_offset]))
    return np.array(feature_list), np.array(label_list)

x_WINDOW, y_WINDOW = make_window_size_dataset(X_data, Y_data, WINDOW_SIZE, day_offset)
x_WINDOW.shape,y_WINDOW.shape

윈도우 사이즈에 맞게 데이터셋을 설정해준다.

 

a = math.floor(y_WINDOW.shape[0]*0.1)
b = math.floor(y_WINDOW.shape[0]*0.2)

x_train_WINDOW = x_WINDOW[:-b]
x_validation_WINDOW = x_WINDOW[-b:-a]
x_test_WINDOW = x_WINDOW[-a:]

y_train_WINDOW = y_WINDOW[:-b]
y_validation_WINDOW = y_WINDOW[-b:-a]
y_test_WINDOW = y_WINDOW[-a:]

그다음 train, test셋을 나누어야 하는데 약 데이터셋으 20% 센트만 사용하였다.

from keras.models import Sequential
from keras.layers import Dense, LSTM, Conv1D, Lambda
from keras.callbacks import EarlyStopping, ModelCheckpoint
from keras.layers import LSTM, GRU
import keras

from tensorflow.keras.losses import Huber
from tensorflow.keras.optimizers import Adam
from keras.layers import Input, LSTM, Dense

from tensorflow.keras.activations import elu,gelu,swish,tanh,softmax
from tensorflow.keras.optimizers import Adam,RMSprop

from keras.layers import Flatten
from keras.layers import Dropout
from tensorflow.keras import optimizers
from keras import layers
import tensorflow as tf
from keras import losses

elu = tf.nn.elu

model = keras.models.Sequential([
    keras.layers.Input(shape=(x_train_WINDOW.shape[1] , x_train_WINDOW.shape[2])),
    keras.layers.LSTM(1024, return_sequences=True, name='LSTM_0'),
    keras.layers.LSTM(512, return_sequences=True, name='LSTM_1'),
    keras.layers.LSTM(256, return_sequences=True, name='LSTM_2'),
    keras.layers.LSTM(128, return_sequences=True, name='LSTM_3'),
    keras.layers.Flatten(),
    keras.layers.Dense(64, activation=elu),
    keras.layers.Dense(1, activation=elu)
#     keras.layers.Dense(1, activation="sigmoid")
])
model.compile(loss=Huber(), optimizer=Adam(0.000001), metrics=['mse']) #0.001이 기본 러닝 레이트

활성화 함수로써는 속도는 느리지만 좀 더 좋다는 elu를 사용하였고, 목적함수로써는 Adam을 사용하였다. 전부 스케일링을 해주었기 때문에 0.000001정도의 학습을 하였고 은닉층을 늘려서 조금 더 모델을 고도화 시켯다.

import os
from keras.callbacks import EarlyStopping, ModelCheckpoint
# earlystopping은 (patience 수)n번 epoch통안 val_loss 개선이 없다면 학습을 멈춥니다.
early_stop = EarlyStopping(monitor='val_loss', patience=10)

model_path = 'C:\\Users\\Happy\\Desktop\논문용\\krwusdtdata'
filename = os.path.join(model_path,'USDT_LSTM_환율예측.h5')

checkpoint = ModelCheckpoint(filename, #filepath
                             monitor='val_loss',#모델 저장시 기준이 되는 값 => val_loss는 loss가 가장 적을 때 저장
                             verbose=1, # 이게 1 이면 저장되었다고 표시됨
                             save_best_only=True, # True의 경우 학습 중 현 시점 가장 좋은 모델로 저장됨
                             save_weights_only=True, # True의 경우 모델 레이어 및 가중치도 저장됨
#                              save_freq = BATCH_SIZE, # 'epoch'을 사용할 경우, 매 에폭마다 모델이 저장됩니다. integer을 사용할 경우, 숫자만큼의 배치를 진행되면 모델이 저장됩니다.
                             mode='auto'# val_acc 인 경우, 정확도이기 때문에 클수록 좋습니다. 따라서 이때는 max를 입력해줘야합니다. 만약 val_loss 인 경우, loss 값이기 때문에 값이 작을수록 좋습니다. 따라서 이때는 min을 입력해줘야합니다. auto로 할 경우, 모델이 알아서 min, max를 판단하여 모델을 저장합니다.
                            )

얼리스타핑의 학습을 10번 정도로 두어 학습에 개선이 없다면 학습을 멈추게 하였다.

중간중간 val_loss가 최저점을 갱신하면(모델이 좋으면) 저장하게 하였다. 

 

history = model.fit(x_train_WINDOW, y_train_WINDOW, 
                                    epochs=1000, 
                                    batch_size=BATCH_SIZE,
                                    validation_data=(x_validation_WINDOW, y_validation_WINDOW),
                                    #validation_data=(x_test_WINDOW, y_test_WINDOW), 
                                    callbacks=[early_stop, checkpoint]) # 여기에 얼리스타핑 ,드롭아웃, L1,L2,엘라 규제 등 추가 가능
# model.load_weights(filename)

에포크를 1000 번을 두었고 얼마나 학습율이 좋은지 본다.

 

# summarize history for loss
plt.plot(history.history['loss'])
plt.plot(history.history['val_loss'])
plt.title('model loss')
plt.ylabel('mse')
plt.xlabel('epoch')
plt.legend(['train', 'val'], loc='upper left')
plt.show()

 

내 컴퓨터로 학습하기에는 매우 느리고 하여 사실 얼리스타핑 patien을 100으로 두었다. 그리고 확인을 해보니 약 100번 정도 학습시에 적절한 모델인것을 알 수 있다.

model.load_weights(filename) #저장된 최적 모델 불러옴 끄면 그냥 최종 모델 사용가능

그 다음 가중치를 로드하고,

pred = model.predict(x_test_WINDOW) #예측값(실제)
actual = np.asarray(y_test_WINDOW)

pred = pd.DataFrame(pred,columns = ["예측값"]).apply(lambda x : (x * (MAX_val - MIN_val) + MIN_val)) # 원래 값 복원
actual = pd.DataFrame(actual,columns = ["실제값"]).apply(lambda x : (x * (MAX_val - MIN_val) + MIN_val)) # 예측 값 복원

print(pred.shape, actual.shape)

i = 0
j = len(pred)

plt.figure(figsize=(12, 9))
plt.title('test_data')
plt.plot(actual[i:j], label='actual')
plt.plot(pred[i:j], label='prediction')
plt.legend()
plt.show()

from sklearn.metrics import mean_absolute_error, mean_squared_error
print("MSE",mean_squared_error(actual["실제값"],pred["예측값"]))
print("RMSE",mean_squared_error(actual["실제값"],pred["예측값"],squared=False))

MSE는 189 

RMSE는 13 정도 나왔다.

위 모습은 검증셋 데이터고

학습데이터만으로 보면

pred = model.predict(x_train_WINDOW) #예측값(학습)
actual = np.asarray(y_train_WINDOW)

pred = pd.DataFrame(pred,columns = ["예측값"]).apply(lambda x : (x * (MAX_val - MIN_val) + MIN_val)) # 원래 값 복원
actual = pd.DataFrame(actual,columns = ["실제값"]).apply(lambda x : (x * (MAX_val - MIN_val) + MIN_val)) # 예측 값 복원

print(pred.shape, actual.shape)

i = 0
j = len(pred)

plt.figure(figsize=(12, 9))
plt.title('train_data')
plt.plot(actual[i:j], label='actual')
plt.plot(pred[i:j], label='prediction')
plt.legend()
plt.show()

from sklearn.metrics import mean_absolute_error, mean_squared_error
print("MSE",mean_squared_error(actual["실제값"],pred["예측값"]))
print("RMSE",mean_squared_error(actual["실제값"],pred["예측값"],squared=False))

이 정도가 나온다.