[파이토치] Cross-Validation

In k-fold cross-validation, the original sample is randomly partitioned into k equal sized subsamples

• Of the k subsamples, a single subsample is retained as the validation data
• The remaining k-1 subsamples are used as training data
• The cross-validation process is then repeated k times with each of the k samples used exactly once as the validation data
• 학습을 위한 데이터가 적은 경우 유용한 방법론이다.

 

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 # Regression 문제의 평가를 위해 MSE(Mean Squared Error)를 불러온다.

# Plot

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

df = pd.read_csv('./data/reg.csv', index_col=[0])

# 데이터를 넘파이 배열로 만들기

X = df.drop('Price', axis=1).to_numpy() # 데이터프레임에서 타겟값(Price)을 제외하고 넘파이 배열로 만들기

Y = df['Price'].to_numpy().reshape((-1,1)) # 데이터프레임 형태의 타겟값을 넘파이 배열로 만들기

# 텐서 데이터로 변환하는 클래스(3강 참고)

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

X_train, X_test, Y_train, Y_test = train_test_split(X, Y, test_size=0.7)

trainset = TensorData(X_train, Y_train) # trainDataLoader는 for문 내에서 선언한다.

testset = TensorData(X_test, Y_test)

testloader = torch.utils.data.DataLoader(testset, batch_size=32, shuffle=False)

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

kfold = KFold(n_splits=3, shuffle=True) # k는 3으로 설정
criterion = nn.MSELoss()

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() # 넘파이 배열로 변경

    rmse = np.sqrt(mean_squared_error(predictions, actual)) # sklearn을 이용하여 RMSE 계산

    model.train() # 학습 중간에 평가를 했지때문에 모델을 eval()로 변환했다면 train()으로 재선언해야한다.

    return rmse  

 

# 이번 예시에서는 상관없으나 평가 시에는 정규화 기술을 배제하여 온전한 모델로 평가를 해야한다. 따라서 .eval()을 사용한다.

# 즉, 드랍아웃이나 배치 정규화 등과 같이 학습 시에만 사용하는 기술들이 적용 된 모델은 평가 시에는 비활성화 해야하며 학습 시 .train()을 사용한다.

validation_loss = []

for fold, (train_idx, val_idx) in enumerate(kfold.split(trainset)):

    # fold는 1,2,3의 값을 가질것이고, train_idx와 val_idx는 각각 인덱스 값을 가진다.

    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) 

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

trainloader = torch.utils.data.DataLoader(trainset, batch_size=32, shuffle=False)

train_rmse = evaluation(trainloader) # train 데이터의 RMSE

test_rmse = evaluation(testloader) # test 데이터의 RMSE

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

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