교차 검증 (Cross-Validation)

보통 머신러닝에서는 train, validation, test 로 데이터를 나눠서 학습을 진행함.
(train은 모델 학습, validation은 파라미터 튜닝, test는 평가에 사용됨)

다만 많은 데이터를 요구하는 머신러닝 학습과정에서 데이터 수가 적으면 모델의 학습이 정상적으로 이루어지지 않음.

예를 들어, 데이터셋의 크기가 작은 경우 데이터를 단순히 train, validation으로 이분하여 나눠버리면 '유의미한 정보'가 validation에 들어간 경우에는 모델 학습이 정상적으로 이루어지지 않을 수 있음.

K-fold validation은 '재표본을 추출'해서 모델을 학습하는 방법임. 우선 train 데이터를 'k'개로 나눔. 이후 한 번씩 돌아가면서(iteration) k개 만큼 나눈 데이터중 한 개를 test data로 사용하고 나머지는 train으로 사용함.

이런 경우 모든 데이터를 train에 사용할 수 있어 모델의 오버피팅을 방지하고 한정된 데이터셋으로도 안정적인 학습을 할 수 있음.

(사실 딥러닝에서는 교차검증을 거의 사용하지 않음, 계산량과 자원 문제로 인해)

1. 라이브러리 (k-fold validation은 sklearn으로 쉽게 사용 가능)

import pandas as pd # 데이터프레임 형태를 다룰 수 있는 라이브러리
import numpy as np
from sklearn.model_selection import train_test_split # 전체 데이터를 학습 데이터와 평가 데이터로 분할

# ANN
import torch
from torch import nn, optim # torch 내의 세부적인 기능
from torch.utils.data import DataLoader, Dataset # 데이터를 모델에 사용
import torch.nn.functional as F # torch 내의 세부적인 기능

# Cross Validation
from sklearn.model_selection import KFold

# Loss
from sklearn.metrics import mean_squared_error # MSE(Mean Squared Error)

# Plot
import matplotlib.pyplot as plt # 시각화 도구

이런 데이터가 있다고 가정하면 (변수 13개, 타겟 1개)

2. 데이터 불러오고 텐서로 변환

# 데이터를 넘파이 배열로
X = df.drop('Price', axis=1).to_numpy() # 타겟값(Price)을 제외하고 넘파이 배열로 
Y = df['Price'].to_numpy().reshape((-1,1)) # 타겟값을 넘파이 배열로 만들기

텐서로 변환

class TensorData(Dataset):

    def __init__(self, x_data, y_data):
        self.x_data = torch.FloatTensor(x_data)
        self.y_data = torch.FloatTensor(y_data)
        self.len = self.y_data.shape[0]

    def __getitem__(self, index):

        return self.x_data[index], self.y_data[index] 

    def __len__(self):
        return self.len

train, test 분할

X_train, X_test, Y_train, Y_test = train_test_split(X, Y, test_size=0.7)
trainset = TensorData(X_train, Y_train)
testset = TensorData(X_test, Y_test)
testloader = torch.utils.data.DataLoader(testset, batch_size=32, shuffle=False)

3. 간단한 회귀 모델 생성

class Regressor(nn.Module):
    def __init__(self):
        super().__init__() # 모델 연산 정의
        self.fc1 = nn.Linear(13, 50, bias=True) # 입력층(13) -> 은닉층1(50)으로 가는 연산
        self.fc2 = nn.Linear(50, 30, bias=True) # 은닉층1(50) -> 은닉층2(30)으로 가는 연산
        self.fc3 = nn.Linear(30, 1, bias=True) # 은닉층2(30) -> 출력층(1)으로 가는 연산
    
    def forward(self, x):
        x = self.fc1(x) 
        x = self.fc2(x) 
        x = self.fc3(x) 
      
        return x
        
Regressor()

#result
Regressor(
  (fc1): Linear(in_features=13, out_features=50, bias=True)
  (fc2): Linear(in_features=50, out_features=30, bias=True)
  (fc3): Linear(in_features=30, out_features=1, bias=True)
)

4. K-fold 및 criterion 만들기 (k=3)

kfold = KFold(n_splits=3, shuffle=True)
criterion = nn.MSELoss()

5. 평가(evaluation)

def evaluation(dataloader):
    
    predictions = torch.tensor([], dtype=torch.float) # 예측값을 저장하는 텐서
    actual = torch.tensor([], dtype=torch.float) # 실제값을 저장하는 텐서
        
    with torch.no_grad():
        model.eval() # 평가를 할 때에는 반드시 .eval()  사용
        for data in dataloader:
            inputs, values = data
            outputs = model(inputs)

            predictions = torch.cat((predictions, outputs), 0) # cat을 통해 예측값을 누적
            actual = torch.cat((actual, values), 0) # cat을 통해 실제값을 누적
    
    predictions = predictions.numpy() # 넘파이 배열로 변경
    actual = actual.numpy() # 넘파이 배열로 변경 (sklearn 사용해야되니까)
    rmse = np.sqrt(mean_squared_error(predictions, actual)) # sklearn을 이용하여 RMSE 계산
    model.train()
    return rmse  

# 평가 시 .eval()을 사용해야 하는 이유
# 이번 예시에서는 상관없으나 평가 시에는 정규화 기술(dropout 등)을 배제하여 온전한 모델로 평가를 해야함
# 따라서 .eval()을 사용
# 즉, 드랍아웃이나 배치 정규화 등과 같이 학습 시에만 사용하는 기술들이 적용 된 모델은 평가 시에는 비활성화 해야됨
# 학습 시 .train()을 사용

6. k-fold validation 을 이용한 학습

validation_loss = []

for fold, (train_idx, val_idx) in enumerate(kfold.split(trainset)): # 위의 k-fold class 사용
    
    
    train_subsampler = torch.utils.data.SubsetRandomSampler(train_idx) # index 생성
    val_subsampler = torch.utils.data.SubsetRandomSampler(val_idx) # index 생성
    
    # sampler를 이용한 DataLoader 정의
    trainloader = torch.utils.data.DataLoader(trainset, batch_size=32, sampler=train_subsampler) # 해당하는 index 추출
    valloader = torch.utils.data.DataLoader(trainset, batch_size=32, sampler=val_subsampler)
    
    # 모델
    model = Regressor()
    optimizer = optim.Adam(model.parameters(), lr=0.001, weight_decay=1e-7)
    
    for epoch in range(400): # 400번 학습을 진행한다.

        for data in trainloader: # 무작위로 섞인 32개 데이터가 있는 배치가 하나 씩 들어온다.

            inputs, values = data # data에는 X, Y가 들어있다.

            optimizer.zero_grad() # 최적화 초기화

            outputs = model(inputs) # 모델에 입력값 대입 후 예측값 산출
            loss = criterion(outputs, values) # 손실 함수 계산
            loss.backward() # 손실 함수 기준으로 역전파 설정 
            optimizer.step() # 역전파를 진행하고 가중치 업데이트

    train_rmse = evaluation(trainloader) # 학습 데이터의 RMSE
    val_rmse = evaluation(valloader)
    print("k-fold", fold," Train Loss: %.4f, Validation Loss: %.4f" %(train_rmse, val_rmse)) 
    validation_loss.append(val_rmse)

validation_loss = np.array(validation_loss)
mean = np.mean(validation_loss)
std = np.std(validation_loss)
print("Validation Score: %.4f, ± %.4f" %(mean, std))

# return
k-fold 0  Train Loss: 0.0945, Validation Loss: 0.1605
k-fold 1  Train Loss: 0.1118, Validation Loss: 0.1503
k-fold 2  Train Loss: 0.1240, Validation Loss: 0.1077
Validation Score: 0.1395, ± 0.0229

test !

trainloader = torch.utils.data.DataLoader(trainset, batch_size=32, shuffle=False)
train_rmse = evaluation(trainloader) # 학습 데이터의 RMSE
test_rmse = evaluation(testloader) # 시험 데이터의 RMSE

print("Train RMSE: %.4f" %train_rmse)
print("Test RMSE: %.4f" %test_rmse)

#result
Train RMSE: 0.1189
Test RMSE: 0.1279