Titanic Survival Prediction

https://www.kaggle.com/competitions/titanic

Titanic - Machine Learning from Disaster | Kaggle

오늘은 캐글에 있는 가장 기본 문제인 타이타닉 데이터셋을 이용하여서 머신러닝을 공부를 해 보겠습니다.

 

타이타닉에는 여러 데이터셋이 존재합니다.

Survived	생존 여부 0 = No, 1 = Yes
pclass	티켓 등급 1 = 1st, 2 = 2nd, 3 = 3rd
Sex	성별
Age	나이
Sibsp	함께 탑승한 형제자매, 배우자의 수
Parch	함꼐 탑승한 부모, 자식의 수
Ticket	티켓 번호
Fare	운임
Cabin	객실 번호
Embarked	탑승 항구 C = Cherbourg, Q = Queenstown, S = Southampton

 

학습 데이터셋에는 Survived가 존재하고 평가 데이터셋에는 존재하지않아 우리는 학습을 통하여 Survived를 판별 할 수 있는 코드를 만들어야 합니다.

---

from torch.utils.data import Dataset
# Define a custom dataset for Titanic data
class TitanicDataset(Dataset):
    def __init__(self, data, labels=None, mode='train'):
        self.data = data
        self.labels = labels
        self.mode = mode
    def __len__(self):
        return len(self.data)
    def __getitem__(self, idx):
        if self.mode == 'train':
            x = self.data[idx]
            y = self.labels[idx]
            return x, y
        else:
            x = self.data[idx]
            return x

from torch.utils.data import Dataset

- torch.utils.data모듈에서 Dataset 클래스를 가져옵니다. 이 클래스는 데이터셋 및 데이터 로딩 유틸리티를 다루는 Pytorch 모듈입니다.

 

def __init__(self, data, labels=None, mode='train')

- TitanicDataset클래스의 생성자 메서드입니다. 이 메서드는 클래스의 새 인스턴스를 초기화하고 세 개의 매개변수를 사용합니다.

data: 데이터셋의 입력 데이터

labels: 데이터에 해당하는 레이블

mode: 데이터셋의 모드

 

def __len__(self)

- 데이터셋의 길이를 가져오는데 사용됩니다.

 

def __getitem__(self, idx)

- 주어진 인덱스 idx에 해당하는 데이터 샘플을 가져오는데 사용됩니다

모드가 train인지 text인지 확인 후 조건데 해당하는 코드를 실행합니다.

import pandas as pd

# Read the CSV file into a DataFrame
df_train = pd.read_csv('train.csv')
df_test = pd.read_csv('test.csv')

# Display the DataFrame
print(df_train.head(3))
print(df_test.head(3))

train.csv, test.csv파일을 불러옵니다 이후 각 처음 3개 행을 출력합니다.

 

import numpy as np
def get_data_from_df(df):
    x = np.zeros([len(df), 4])
    
    # Sex
    df["Sex"] = df["Sex"].fillna("male")
    x[:, 0] = df["Sex"].map( {"male": 0, "female": 1} ).astype(float)
    
    # Pclass
    df["Pclass"] = df["Pclass"].fillna(3)
    x[:, 1] = df["Pclass"].map( {3: 0, 2: 1, 1: 2} ).astype(float)
    
    # Fare
    df["Fare"] = df["Fare"].fillna(df["Fare"].mean())
    x[:, 2] = (df["Fare"] - df["Fare"].mean()) / df["Fare"].std()
    
    # Embarked
    # fillna("S") : missing data에 대한 처리
    x[:, 3] = df["Embarked"].fillna("S").map( {"S": 0, "Q": 1, "C": 2} ).astype(float)
    return x

데이터셋을 설정합니다, 데이터셋은 Sex, Pclass, Fare, Embarked를 사용했습니다.

Sex

- male이면 0 female이면 1로 표기합니다. 결측값은 male로 표기합니다.

Pclass

- 3이면 0 2이면 1 1이면 2로 표기합니다. 결측값은 3으로 표기합니다.

Fare

- 결측값은 평균으로 정의하고 나머지 값들은 표준화를 시켜줍니다. 원래값에서 평균을 빼준 뒤 원래값으로 나누어 표기합니다.

Embarked

- S는 0 Q는 1 C는 2로 표기합니다. 결측값은 S로 생각합니다.

x_train = get_data_from_df(df_train)
y_train = df_train["Survived"].values.reshape(-1, 1)

x_test = get_data_from_df(df_test)

먼저 훈련 데이터에서 입력 데이터(x_train)를 추출합니다.

이후 훈련 데이터에서 레이블(y_train)을 추출합니다.

테스트 데이터에서 입력 데이터(x_text)를 추출합니다.

from torch.utils.data import DataLoader

train_dataset = TitanicDataset(data=x_train, labels=y_train, mode='train')
test_dataset = TitanicDataset(data=x_test, mode='test')

train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False)

먼저 훈련 데이터셋과 테스트 데이터셋을 생성해줍니다.

 

train_loader는 훈련 데이터셋을 미니배치로 나누어주는 DataLoader 객체입니다.

batch_size 매개변수는 미니배치의 크기를 지정하고, shuffle=True는 에폭마다 데이터셋을 섞는 것을 나타냅니다.

 

test_loader는 테스트 데이터셋을 미니배치로 나누어주는 DataLoader 객체입니다.

shuffle=False로 설정되어 있어 테스트 데이터셋은 섞이지 않습니다.

import torch.nn as nn

# Define the neural network architecture
class TitanicNet(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super(TitanicNet, self).__init__()
        self.fc1 = nn.Linear(input_size, hidden_size)
        self.relu = nn.ReLU()
        self.fc2 = nn.Linear(hidden_size, output_size)
        self.sigmoid = nn.Sigmoid()
        
    def forward(self, x):
        out = self.fc1(x)
        out = self.relu(out)
        out = self.fc2(out)
        out = self.sigmoid(out)
        return out

신경망을 구축하는 코드입니다.

self.fc1 = nn.Linear(input0size, hidden_size)

- 입력층과 은닉층 간의 선형 변환을 적용합니다

이후 활성화 함수인 ReLU를 이용합니다.

 

self.fc2 = nn.Linear(hidden_size, output_size)

- 은닉층과 출력층 간의 선형 변환을 적용합니다.

이후 활성화 함수인 Sigmoid를 사용합니다.

 

forward 함수

먼저 입력 데이터를 첫 번째 선형 변환과 활성화 함수를 거칩니다.

out = self.fc1(x)

out = self.relu(out)

이후 은닉층의 출력을 두 번째 선형 변환과 활성화 함수를 거칩니다.

out = self.fc2(out)

out = self.sigmoid(out)

최종 출력은 모델의 예측값 입니다.

import torch
import torch.optim as optim

# Initialize the model
input_size = len(x_train[0])
hidden_size = 64
output_size = 1
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

model = TitanicNet(input_size, hidden_size, output_size)
model = model.to(device)

# Define the loss function and optimizer
criterion = nn.BCELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)

PyTorch를 이용하여 모델을 초기화하고 손실 함수와 옵티마이저를 정의합니다.

 

input_size

- 입력 데이터의 특성 수에 해당합니다. 이 값은 x_train의 첫 번째 차원의 길이로 부터 가져옵니다.

hidden_size

- 은닉층의 뉴런 수로 사용자가 지정한 값으로 설정됩니다.

output_size

- 출력층의 뉴런 수로 이진 분류 문제이기 때문에 1로 설정합니다.

device

- 모델을 GPU또는 CPU로 설정합니다. CUDA가 사용 가능한 경우 cuda 아닌 경우 cpu로 설정합니다.

 

이후 TitanicNet클래스를 사용하여 모델을 초기화하고 .to(device)를 사용하여 모델을 지정한 장치로 이동합니다.

 

손실함수로는 이진 교차 엔트로피 손실 함수 nn.BCELoss()를 사용합니다.

옵티마이저로는 Adam옵티마이저를 사용하여 모델의 파라미터를 최적화합니다. 학습률은 0.001로 설정됩니다.

 

# Train the model
num_epochs = 10

total_correct, total_samples = 0, 0
losses = []
model.train()
for epoch in range(num_epochs):
    epoch_loss = 0 # 에폭별 손실 초기화
    for inputs, labels in train_loader:	# 각 미니배치에 대한 반복
        inputs = inputs.float().to(device) # 입력 데이터를 지정한 장치로 이동
        labels = labels.float().to(device) # 레이블을 지정한 장치로 이동
        
        outputs = model(inputs)				# 모델로부터 예측값 계산
        loss = criterion(outputs, labels)	# 손실 계산
        epoch_loss += loss.item() 			# 에폭별 손실 누적
        predicted = torch.round(outputs)	# 예측값을 반올림하여 이진 분류
        total_correct += (predicted == labels).sum().item()	# 정확한 예측 수 업데이트
        total_samples += labels.size(0)		# 전체 샘플 수 업데이트
        optimizer.zero_grad()				# 기울기 초기화
        loss.backward()						# 역전파 수행
        optimizer.step()					# 옵티마이저로 파라미터 업데이트
	losses.append(epoch_loss)				# 에폭별 손실 기록
    accuracy = total_correct / total_samples# 정확도 계산
    print(f"Epoch : {epoch: 2d}, Accuracy: {accuracy: .6f}")	# 현재 에폭의 정확도 출력

먼저 num_epochs변수에 에폭 수를 지정해줍니다.

이후 전체 정확한 예측 수(total_correct)와 전체 샘플 수(total_samples)를 초기화 해 줍니다.

 

이후 for문을 통하여 주어진 에폭 수 만큼 반복해줍니다.

이진 교차 엔트로피 손실 함수를 사용하여 손실을 계산하고, 역전파를 수행하여 모델 파라미터를 업데이트합니다.

이후 각 에폭마다 정확도를 계산하여 출력합니다.

 

import matplotlib.pyplot as plt

losses = np.array(losses)

plt.plot(losses)
plt.show()

손실을 그래프로 시각화합니다.

# Evaluate the model
total_pred = []

model.eval()
with torch.no_grad():
    for inputs in test_loader:
        inputs = inputs.float().to(device)
        outputs = model(inputs)
        predicted = torch.round(outputs)
        total_pred += predicted.int().tolist()

모델을 평가하고 테스트 데이터에 대한 예측을 생성합니다.

total_pred에 모든 예측값을 저장합니다.

 

각 테스트 미니배치에 대한 반복을 통하여 입력 데이터를 지정한 장치로 이동 후 모델로부터 예측값을 계산합니다.

이후 예측값을 반올림하여 이진 분류를 진행 후 에측값을 리스트에 추가해줍니다.

total_pred = np.array(total_pred).flatten()

# Save the predictions to a CSV file
df_test['Survived'] = total_pred
df_test[['PassengerId', 'Survived']].to_csv('submission.csv', index=False)

이후 submission.csv파일을 생성하여 캐글에 제출해줍니다.