[딥러닝-케라스] 케라스 자동차 연비 예측 모델

1. modules import

import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt

import tensorflow as tf
from tensorflow.keras.layers import Dense, Input
from tensorflow.keras.optimizers import RMSprop
from tensorflow.keras.models import Model
from tensorflow.keras.utils import get_file, plot_model

 

 

2. 데이터 로드

# 해당 주소의 데이터를 다운로드
dataset_path = get_file("auto-mpg.data", "http://archive.ics.uci.edu/ml/machine-learning-databases/auto-mpg/auto-mpg.data")

# 열 이름 지정
column_names = ['MPG', 'Cylinders', 'Displacement', 'Horsepower', 'Weight','Acceleration','Model Year', 'Origin']

# 지정된 열이름을 사용하여 데이터를 판다스 데이터프레임 형식으로 로드
raw_dataset = pd.read_csv(dataset_path, names = column_names,
                          na_values = '?', comment = '\t',
                          sep = ' ', skipinitialspace = True)

 

 

3. 데이터 확인

# raw data바로 사용하지 않고 copy()하여 사용
dataset = raw_dataset.copy()
dataset

 

 

4. 데이터 전처리

• 해당 데이터는 일부 데이터가 누락되어 있음

dataset.isna().sum()

# 출력 결과
MPG             0
Cylinders       0
Displacement    0
Horsepower      6
Weight          0
Acceleration    0
Model Year      0
Origin          0
dtype: int64

• 누락된 행 삭제

# Horsepower에 6개의 결측값이 있으므로 결측값 제거
dataset = dataset.dropna()

• 'Origin' 범주형 데이터
  • 원-핫 인코딩 진행

origin = dataset.pop('Origin')
dataset['USA'] = (origin == 1) * 1.0
dataset['Europe'] = (origin == 2) * 1.0
dataset['Japan'] = (origin == 3) * 1.0
dataset

 

 

4-1. 검증 데이터셋 생성

# train 데이터로 전체 데이터의 0.8을 추출
# 전체 데이터에서 train 데이터를 drop시킨 나머지를 test 데이터로 지정
train_dataset = dataset.sample(frac = 0.8, random_state = 0)
test_dataset = dataset.drop(train_dataset.index)

 

 

4-2. 데이터 조사

sns.pairplot(train_dataset[['MPG', 'Cylinders', 'Displacement', 'Horsepower', 'Weight']], diag_kind = 'kde')

# 데이터의 통계정보
train_stats = train_dataset.describe()
# MPG는 정답이기 때문에 통계 정보에서 제외
train_stats.pop("MPG")
train_stats = train_stats.transpose()
train_stats

 

 

4-3. 데이터의 특성과 레이블 분리

train_labels = train_dataset.pop('MPG')
test_labels = test_dataset.pop('MPG')

 

 

4-4. 데이터 정규화

# 통계정보에서 평균과 표준편차를 가져와서 각 데이터 값에서 평균을 빼고 표준편차로 나눠 정규화
def normalization(x):
    return (x - train_stats['mean']) / train_stats['std']

normed_train_data = normalization(train_dataset)
normed_test_data = normalization(test_dataset)

 

 

5. 모델 구성

def build_model():
    input = Input(shape = len(train_dataset.keys()), name = 'input')
    hidden1 = Dense(64,activation = 'relu', name = 'dense1')(input)
    hidden2 = Dense(64, activation = 'relu', name = 'dense2')(hidden1)
    output = Dense(1, name = 'output')(hidden2)

    model = Model(inputs = [input], outputs = [output])

    model.compile(loss = 'mse',
                  optimizer = RMSprop(0.001),
                  metrics = ['mae', 'mse'])
    
    return model

model = build_model()
model.summary()

# 출력 결과
Model: "model"
_________________________________________________________________
 Layer (type)                Output Shape              Param #   
=================================================================
 input (InputLayer)          [(None, 9)]               0         
                                                                 
 dense1 (Dense)              (None, 64)                640       
                                                                 
 dense2 (Dense)              (None, 64)                4160      
                                                                 
 output (Dense)              (None, 1)                 65        
                                                                 
=================================================================
Total params: 4,865
Trainable params: 4,865
Non-trainable params: 0
_________________________________________________________________

plot_model(model)

 

 

6. 샘플 데이터 확인

sample_batch = normed_train_data[:10]
sample_result = model.predict(sample_batch)
sample_batch

• 정규화가 잘 된 데이터 값을 확인할 수 있음

 

 

7. 모델 학습

epochs = 1000
history = model.fit(normed_train_data, train_labels,
                    epochs = epochs, validation_split = 0.2)

 

 

8. 모델 학습 시각화

# 학습 과정에서 생겼던 각 반복의 loss값과 mae, mse 값을 모두 데이터프레임 형태로 저장
hist = pd.DataFrame(history.history)
hist['epoch'] = history.epoch
hist

# 위의 데이터프레임을 사용하여 하나의 함수로 시각화 생성
def plot_history(history):
    hist = pd.DataFrame(history.history)
    hist['epoch'] = history.epoch

    plt.figure(figsize = (12, 6))
    
    plt.subplot(1, 2, 1)
    plt.xlabel('Epochs')
    plt.ylabel('MPG Mean Absolute Error')
    plt.plot(hist['epoch'], hist['mae'], label = 'Train Error')
    plt.plot(hist['epoch'], hist['val_mae'], label = 'Val Error')
    plt.ylim([0, 5])
    plt.legend()

    plt.subplot(1, 2, 2)
    plt.xlabel('Epochs')
    plt.ylabel('MPG Mean Squared Error')
    plt.plot(hist['epoch'], hist['mse'], label = 'Train Error')
    plt.plot(hist['epoch'], hist['val_mse'], label = 'Val Error')
    plt.ylim([0, 20])
    plt.legend()

    plt.show()

plot_history(history)

• 검증데이터의 오차(Val Error)값이 mae, mse 값 모두 일정 값 이하로 더이상 떨어지지 않음
• 학습을 더 진행해봤자 검증데이터의 오차가 줄어들지 않으면 의미가 없고
  train 데이터의 오차만 줄어들어 둘 사이 간격이 벌어지면 오히려 모델이 train 데이터에 과대적합될 수 있음

 

 

9. EarlyStopping을 이용한 규제화

from tensorflow.keras.callbacks import EarlyStopping

model = build_model()

# 10번의 성능 향상을 보고 그 동안 성능 향상이 이뤄지지 않으면 stop
early_stop = EarlyStopping(monitor = 'val_loss', patience = 10)

history = model.fit(normed_train_data, train_labels, epochs = epochs,
                    validation_split = 0.2, callbacks = [early_stop])

• 1000번 다 반복되지 않고 91번째에서 성능 향상이 없다고 판단되어 학습 중지

plot_history(history)

 

 

10. 모델 평가

# test 데이터를 모델에 넣어 나온 loss와 mae, mse 값 저장
loss, mae, mse = model.evaluate(normed_test_data, test_labels, verbose = 2)
print(mae)

# 출력 결과
# 1.88정도의 mpg 오차내에서 예측
3/3 - 0s - loss: 5.7125 - mae: 1.8831 - mse: 5.7125 - 61ms/epoch - 20ms/step
1.8831140995025635

 

 

11. 학습된 모델을 통한 예측

# 예측
test_pred = model.predict(normed_test_data).flatten()

# 예측된 값과 실제 값의 산점도를 그려 선형성을 만족하는지 확인
plt.scatter(test_labels, test_pred)
plt.xlabel('True Values')
plt.ylabel('Predictions')
plt.axis('equal')
plt.axis('square')
plt.grid()
plt.xlim([0, plt.xlim()[1]])
plt.ylim([0, plt.ylim()[1]])
plt.plot([-50, 50], [-50, 50])
plt.show()

# 잘못 예측한 값은 어느정도 되는지 시각화
error = test_pred - test_labels
plt.hist(error, bins = 30)
plt.xlabel('Prediction Error')
plt.grid()
plt.ylabel('Count')
plt.show()