DNN을 이용한 분류모델 (with python)

import pandas as pd
import math
import tensorflow as tf
import matplotlib.pyplot as plt
import numpy as np
ALL_DATA_BY_USE = pd.read_csv("StockReturn.csv",encoding="euc-kr")
ALL_DATA_BY_USE

먼저 이러한 데이터가 있는 주가데이터를 범주가 여러개로 분류하려고 한다면,

 

use_columns = list(set(ALL_DATA_BY_USE.columns)- set(["date"]))
print(use_columns)

target = "target"

이처럼 사용할 컬럼이 타켓(target)컬럼을 제외하고 나머지 컬럼은 학습데이터로 사용한다.

그다음 더미변환을 해주려고 했으나 날짜가 object형으로 인식되는 바람에 저런꼴이 낫다. 날짜 컬럼은 사용하지 않도록 한다.

X_data = ALL_DATA_BY_USE[list(set(use_columns) - {target})]

Y_data = ALL_DATA_BY_USE[[target]]
ALL_DATA_BY_USE = pd.concat([X_data,Y_data],axis = 1)

위처럼 사용할 컬럼만 지정해주면 저렇게 보인다.

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)



LE = LabelEncoder()
Y_data = Y_data.astype('str') # @@ 강제로 str 형식으로 타입변환.
Y_data[target] = LE.fit_transform(Y_data[target])

해당 부분은 사이킷런으로 데이터의 스케일링을 해주는 부분인다. Scaler에 minmax가 아닌 StandardScaler 나 PowerTransformer 등을 이용해도 된다. 필자는 주가 데이터기 때문에 정규분포나 파워트렌스폼을 사용하는 것을 좋아하진 않는다. 주가데이터에 이상치는 없을것이니.

a = math.floor(Y_data.shape[0]*0.2)

y_train = Y_data[:-a]
y_test = Y_data[-a:]

x_train = X_data[:-a]
x_test = X_data[-a:]

해당 부분은 데이터들을 8:2비율로 나눈것이다. 교차검증을 위해서라고 보면 된다. 학습셋과 test셋을 나누었다.

 

# 여기서 케라스에서 카테고리 분류.
to_categorical = tf.keras.utils.to_categorical # https://www.tensorflow.org/api_docs/python/tf/keras/utils/to_categorical

y_train = to_categorical(y_train, Y_data[target].nunique(),dtype='int32')
y_test = to_categorical(y_test, Y_data[target].nunique(),dtype='int32')

다중 분류를 위해서는 해당부분이 꼭 필요하다.

이런식으로 바꾸어 주는 부분이다. 자세한 내용은 텐서플로우 패키지 내용을 참고하자.( https://www.tensorflow.org/api_docs/python/tf/keras/utils/to_categorical )

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

from tensorflow.keras.losses import Huber
from tensorflow.keras.activations import elu,gelu,swish,tanh,softmax
from tensorflow.keras.optimizers import Adam,RMSprop
from keras.layers import Input, LSTM, Dense,GRU

from keras.layers import Flatten
from keras.layers import Dropout
from tensorflow.keras import optimizers
from keras import layers

from keras import losses

# relu 은닉 층으로 학습
# sigmoid 0~1 분류 문제
# softmax 클래스 분류 문제

# # 활성화 함수 Sigmoid , Tanh ,ReLU, Leaky ReLU, PReLU, ELU,  Maxout
elu = tf.nn.elu
# leaky_relu = tf.nn.leaky_relu # leaky_relu 
# activation = elu # elu #leaky_relu#"relu"#"tanh" #"relu"=>0~1 시그모이드랑 함꼐

꼭 전부 import 할 필요는 없지만 내가 딥러닝 모델 만들때마다 쓰는 부분이다. 

model = keras.models.Sequential()
# 첫번째 은닉층
# "swish"
model.add(layers.Dense(64, activation=gelu, input_shape=(x_train.shape[1],), name='Hidden-1'))
# 드롭아웃 계층 Dropout(p) p라는 확률로 출력 노드의 신호를 보내다 말다 함, 과적합 방지
# model.add(layers.Dropout(0.2))
# 두번째 은닉층
model.add(layers.Dense(32, activation=gelu, name='Hidden-2'))
# 드롭아웃 계층 Dropout(p) p라는 확률로 출력 노드의 신호를 보내다 말다 함, 과적합 방지
# model.add(layers.Dropout(0.2))
# 출력층
model.add(layers.Dense(Y_data[target].nunique(), activation='softmax'))
# model.compile(loss='categorical_crossentropy', optimizer='adam', metrics=['accuracy'])
# model.compile(loss='categorical_crossentropy', optimizer=Adam(0.001), metrics=['accuracy'])


model.compile(optimizer=RMSprop(learning_rate = 0.00001),loss='categorical_crossentropy',metrics=['accuracy'])

4번쨰 줄에 인풋 쉐이프는 저런식으로 써주면 새로운 데이터가 들어와도 문제없이 학습된다. 다들 왜 매번 저부분을 새로 쓰는지 모르겠다. 

그리고 마지막 (model.add) 부분만 softmax함수로 나오게 하면 다지분류가 되는것이다!!

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

model_path = 'C:\\Users\\Happy\\Desktop\\help'
filename = os.path.join(model_path,'DNN.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를 판단하여 모델을 저장합니다.
                            )

내가 주로 쓰는 모델인데 earlystopping을 써서 15번째 학습동안 val loss의 개선이 없다면 멈추게 만들어 두었다.

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

해당 부분은 내가 데이터를 학습하면서 val loss가 최저인 부분에 모델을 저장한다.(좋은모델)

# 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()

해당 부분으로 얼마나 모델이 좋은지 확인한다. 

여기서 loss가 가장 적은곳의 모델이 저장되었을 것이다.

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

모델을 불러와서 얼마나 정확한지 한번 봐준다.

predictions = model.predict(x_test,use_multiprocessing =True) #예측값(학습)
actual = np.asarray(y_test)

pred = []
act = []
for i in range(len(predictions)):
    pred.append(np.argmax(predictions[i]))
    act.append(np.argmax(actual[i]))
    
pred = pd.DataFrame(pred,columns = ["예측값"]) #.apply(lambda x : (x * (MAX_val - MIN_val) + MIN_val)) # 원래 값 복원
actual = pd.DataFrame(act,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 accuracy_score, f1_score, r2_score , mean_absolute_error, mean_squared_error
print("accuracy_score",accuracy_score(actual["실제값"],pred["예측값"]))
pred.to_csv("DNN_Pred.csv")

분류모델이니 정확도 측정지표는 혼동행렬이 되겠다. mae나 mse는 무시하고 f1 score나 정확도, 정밀도, 오분류율 등을 측정할수 있다.

 

plt 부분이 주석처리된 이유는분류모델이기 떄문이다.

여기서 만약 predictions에서 오류가 난다면 predictions = LE.inverse_transform(predictions)을 추가해주면 될것이다.

 

LSTM 을 이용한 분류모델은 추후에 올리겠다.. 귀찾ㄶ다 

질문있으시면 댓글로 달아주세요 ~