[PyTorch] Cross-Validation(교차 검증)

기본적인 모델의 학습과 평가 과정은 학습 데이터로 학습하고 평가 데이터로 성능을 확인한다. 그리고 평가 데이터에서 가장 성능이 높은 모델을 고르게 된다. 

 

하지만 이때 과적합이 발생할 수 있기에 이를 보완하기 위해 검증 데이터를 사용할 수 있다. 전체 데이터를 학습/검증/평가 데이터로  나누고 학습 데이터로 모델 학습 -> 검증 데이터로 모델 평가 및 최적화 -> 평가 데이터로 최종 평가를 수행한다.

 

여기서 k-Fold Cross-Validation(k겹 교차 검증)은 학습 데이터 전체를 사용하면서 검증할 수 있는 방법으로 머신러닝 분야에서 매우 널리 쓰이는 검증 방법이다.  학습 데이터를 k개로 나누어 1개는 검증 데이터로, 나머지 k-1개는 학습 데이터로 사용한다. 따라서 k번의 검증 과정이 필요하기 때문에 느린 것이 단점이다.

 

각각 k번의 검증을 한 뒤, k개의 검증 성능의 평균을 내어 점수를 산정해 모델의 성능을 평가한다. 

 

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split  # 데이터 나누기
import torch
from torch import nn, optim  # nn: 신경망 레이어, optim: 최적화 알고리즘
from torch.utils.data import Dataset  # 데이터셋, 미니배치 로더
import torch.nn.functional as F  # ReLU 등 다양한 함수 제공
from sklearn.model_selection import KFold  # 교차 검증
from sklearn.metrics import mean_squared_error  # RMSE 계산에 사용

# 1. GPU 장치 설정 (cuda 사용 가능하면 GPU, 아니면 CPU)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")


# 2. 사용자 정의 Dataset 클래스
class TensorData(Dataset):
    def __init__(self, x_data, y_data):
        """
        x_data, y_data: numpy 배열 형태
        -> FloatTensor로 변환 후, device(GPU/CPU)에 올림
        """
        self.x_data = torch.FloatTensor(x_data).to(device)
        self.y_data = torch.FloatTensor(y_data).to(device)
        self.len = self.y_data.shape[0]  # 데이터 개수 (행 개수)

    def __getitem__(self, index):
        """
        인덱스(index)에 해당하는 (입력, 출력) 쌍을 반환
        """
        return self.x_data[index], self.y_data[index]

    def __len__(self):
        """
        전체 데이터셋의 샘플 개수 반환
        """
        return self.len


# 3. 모델 정의: 회귀 모델 (nn.Module 상속)
class Regressor(nn.Module):
    def __init__(self):
        super().__init__()
        # fully-connected 레이어 정의
        # 입력 특징 차원: 13, 은닉층1: 50개, 은닉층2: 30개, 출력층: 1개
        self.fc1 = nn.Linear(13, 50, bias=True)
        self.fc2 = nn.Linear(50, 30, bias=True)
        self.fc3 = nn.Linear(30, 1, bias=True)

    def forward(self, x):
        """
        순전파(Forward):
        입력 x를 차례대로 레이어에 통과시키며, 활성화 함수 ReLU 적용
        """
        x = self.fc1(x)
        x = self.fc2(x)
        x = self.fc3(x)
        return x


# 4. 평가 함수: RMSE 계산
def evaluation(dataloader):
    """
    dataloader를 사용해 미니배치로 데이터를 불러온 뒤,
    모델의 예측값(outputs)과 실제값(values)을 모두 모아 RMSE 계산.
    """
    predictions = torch.tensor([], dtype=torch.float).to(device)
    actual = torch.tensor([], dtype=torch.float).to(device)

    with torch.no_grad():  # 평가 시에는 기울기 계산(역전파) 비활성화
        model.eval()  # 모델을 평가 모드로 설정 (드롭아웃 등 비활성화)
        for data in dataloader:
            inputs, values = data  # 입력과 타깃값
            outputs = model(inputs)  # 모델 예측

            # 예측값과 실제값을 축적
            predictions = torch.cat((predictions, outputs), 0)  # 예측값 누적
            actual = torch.cat((actual, values), 0)  # 실제값 누적

    # GPU 텐서를 CPU로 옮긴 뒤, numpy 배열로 변환
    predictions = predictions.cpu().numpy()
    actual = actual.cpu().numpy()

    # 평균 제곱 오차(MSE) -> RMSE
    rmse = np.sqrt(mean_squared_error(predictions, actual))
    model.train()  # 모델을 다시 학습 모드로 설정
    return rmse


# 5. 데이터 준비: CSV 파일 로드
df = pd.read_csv("./pytorch/data/reg.csv", index_col=[0])
# 예시로 'Price'라는 열이 종속변수(타겟), 나머지는 특징(입력)

# x: 입력 특징들, y: 타겟(Price)
x = df.drop("Price", axis=1).to_numpy()  # DataFrame -> numpy 배열
y = df["Price"].to_numpy().reshape(-1, 1)  # (N,) -> (N,1) 형태로 변환

# 6. 학습/테스트 데이터 분할 (7:3 비율로)
x_train, x_test, y_train, y_test = train_test_split(x, y, test_size=0.7)

# 7. Dataset, DataLoader 생성
trainset = TensorData(x_train, y_train)
testset = TensorData(x_test, y_test)
testloader = torch.utils.data.DataLoader(testset, batch_size=32, shuffle=False)
# 테스트셋은 shuffle=False (평가 시 데이터 순서 유지)

# 7. 교차 검증 설정 및 손실 함수 정의
kfold = KFold(n_splits=3, shuffle=True)  # 데이터를 3개의 Fold로 나누어 교차 검증
criterion = nn.MSELoss()  # 손실 함수로 MSE(평균 제곱 오차) 사용

# 8. K-Fold 교차 검증 및 학습
validation_loss = []  # 각 Fold의 검증 손실값 저장
for fold, (train_idx, val_idx) in enumerate(kfold.split(trainset)):
    # Fold별 학습/검증 데이터 분리
    train_subsampler = torch.utils.data.SubsetRandomSampler(train_idx)
    val_subsampler = torch.utils.data.SubsetRandomSampler(val_idx)

    # Fold별 학습 및 검증용 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().to(device)  # 매 Fold마다 새로운 모델 생성
    optimizer = optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-7)
    # Adam 옵티마이저, weight_decay=1e-7: 가중치 감쇠(정규화)로 과적합 방지

    # 학습 루프
    for epoch in range(400):
        for data in trainloader:
            inputs, values = data
            optimizer.zero_grad()  # 기울기 초기화

            outputs = model(inputs)  # 순전파로 예측값 계산
            loss = criterion(outputs, values)  # 손실 계산
            loss.backward()  # 역전파로 기울기 계산
            optimizer.step()  # 가중치 업데이트

    # Fold별 RMSE 계산
    train_rmse = evaluation(trainloader)  # 학습 데이터 RMSE
    val_rmse = evaluation(valloader)  # 검증 데이터 RMSE
    print(
        f"K-Fold {fold}: Train RMSE = {train_rmse:.4f}, Validation RMSE = {val_rmse:.4f}"
    )
    validation_loss.append(val_rmse) # 검증 손실 저장

# 9. 교차 검증 결과 요약
validation_loss = np.array(validation_loss)
mean = np.mean(validation_loss) # 검증 손실 평균
std = np.std(validation_loss) # 검증 손실 표준편차
print(f"Validation RMSE: {mean:.4f} ± {std:.4f}")

# 10. 전체 데이터 학습 및 테스트 평가
trainloader = torch.utils.data.DataLoader(trainset, batch_size=32, shuffle=False)
train_rmse = evaluation(trainloader)  # 학습 데이터 RMSE
test_rmse = evaluation(testloader)  # 테스트 데이터 RMSE
print(f"Train RMSE: {train_rmse:.4f}")
print(f"Test RMSE: {test_rmse:.4f}")

K-Fold 0: Train RMSE = 0.1242, Validation RMSE = 0.1213
K-Fold 1: Train RMSE = 0.1121, Validation RMSE = 0.1388
K-Fold 2: Train RMSE = 0.1035, Validation RMSE = 0.1585
Validation RMSE: 0.1395 ± 0.0152
Train RMSE: 0.1245
Test RMSE: 0.1358

 

위 코드는 K-Fold 교차 검증을 통해 데이터를 3개의 Fold로 나누어 각각 k(1, 2, 3)번째 데이터를 제외한 나머지 데이터를 학습 데이터로 사용하고, k번째 데이터를 평가 데이터로 사용하여 손실을 계산한다. 

이 과정을 통해 각 학습/평가에 대한 손실 값을 비교하며, 이 비교 값이 작으면 모델이 과적합 없이 안정적으로 학습되었음을 의미한다. 

 

마지막에 전체 데이터를 평가함으로써, 최종 평가를 수행한다. 검증 과정은 일종의 중간 점검 역할을 하며, 검증 단계에서 좋은 결과가 나와도 최종 평가(Test RMSE)에서 더 나쁜 결과를 보일 수 있다.

 

위 결과를 기반으로 모델의 성능을 평가해 보면 각 Fold에서 학습 데이터의 RMSE가 검증 데이터의 RMSE보다 낮은데, 이는 모델이 학습 데이터에 더 잘 맞춰졌음을 의미하며, 검증 데이터에 대한 일반화 성능이 약간 부족함을 보인다. 또한, Validation RMSE의 표준편차가 매우 작으므로, 모델이 데이터 분할(Fold)에 관계없이 안정적인 성능을 보였음을 나타낸다. 

 

최종 평가에서 Test RMSE는 0.1358로 Validation RMSE와 유사한 수준이며, 모델이 테스트 데이터에 대해 적절히 일반화되었음을 보여준다. 또한 모델의 Train RMSE(0.1245)와 비교했을 때, 테스트 데이터에서의 성능 차이가 크지 않으므로, 안정적인 학습이 이루어졌다고 볼 수 있다.