Spaceship Titanic

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

Spaceship Titanic | Kaggle

오늘은 캐글의 문제 중 하나인Spaceshipt Titanic에 대해 정리해 보겠습니다.

해당 문제는 Titanic문제와 매우 유사하며 각 피쳐들만 변화된 형태로 볼 수 있겠습니다.

각 피쳐를 이용하여 평가 데이터에서 해당 승객이 Transported했는지를 알아보는 문제입니다.

 

학습 데이터

승객에 대한 개인 정보(약 8,700명)

• PasengerId: 승객 ID
• HomePlanet: 출발 행성(거주지)
• CryoSleep: CryoSleep 여부
• Cabin: 객실 번호(Deck/Num/Side)
• Destination: 승객의 목적지
• Age: 승객의 나이
• VIP: VIP 서비스 이용 여부
• RoomService, FoodCourt, ShoppingMall, Spa, VRDeck: 해당 서비스를 위해 지불한 비용
• Name: 승객의 이름
• Transported: 다른 차원으로 이동했는지 여부

평가 데이터

승객에 대한 개인 정보(약 8,700명)

• PasengerId: 승객 ID
• HomePlanet: 출발 행성(거주지)
• CryoSleep: CryoSleep 여부
• Cabin: 객실 번호(Deck/Num/Side)
• Destination: 승객의 목적지
• Age: 승객의 나이
• VIP: VIP 서비스 이용 여부
• RoomService, FoodCourt, ShoppingMall, Spa, VRDeck: 해당 서비스를 위해 지불한 비용
• Name: 승객의 이름
• Transported: 다른 차원으로 이동했는지 여부 ← 예측대상

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

위의 코드는 커스텀 데이터셋을 정의하는 부분입니다.

생성자를 생성해주고 길이변환, 그리고 인덱스를 통해 데이터를 가져올 수 있는 함수를 만들어 줍니다.

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

훈련 데이터와 평가 데이터를 각각 임포트 해 줍니다.

import numpy as np

def get_data_from_df(df):
  x = np.zeros([len(df), 4])

  # HomePlanet
  df["HomePlanet"] = df["HomePlanet"].fillna("Earth")
  x[:, 0] = df["HomePlanet"].map( {"Earth": 0, "Europa": 1, "Mars": 2} ).astype(float)

  # CryoSleep
  df["CryoSleep"] = df["CryoSleep"].fillna(False)
  x[:, 1] = df["CryoSleep"].astype(int)

  # VIP
  df["VIP"] = df["VIP"].fillna(False)
  x[:, 2] = df["VIP"].astype(int)

  # RoomService
  df["RoomService"] = df["RoomService"].fillna(0)
  x[:, 3] = (df["RoomService"] - df["RoomService"].mean()) / df["RoomService"].std()


  return x

각 피쳐를 정의해 줍니다.

HomePlanet

- 결측치는 Earth로 지정해줍니다. Earth는 0, Europa는 1, Mars는 2로 정의해 줍니다.

CyroSleep

- 결측치는 false로 지정해줍니다. true와 false에 따라 0과 1로 지정해줍니다.

VIP

- 결측치는 false로 지정해줍니다. 이 또한 true와 false에 따라 0과 1로 지정해줍니다.

RoomService

- 결측치는 0으로 지정해줍니다. 값들은 표준편차로 정의해 줍니다.

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

get_data_from_df함수를 통하여 피쳐를 추출하고 정규화한 뒤 이를 모델에 입력해 줍니다.

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)

데이터셋을 생성해 줍니다. batch_size와 shuffle의 여부를 나타냅니다.

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

다층 퍼셉트론(MLP)를 정의하여 줍니다. ReLU활성화 함수와 Sigmoid활성화 함수를 이용합니다.

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)

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

은닉층의 크기는 64 출력의 크기는 1(이진 분류 문제이므로) 정의해 줍니다.

손실 함수는 이진 분류 문제이므로 이진 교차 엔트로피(BCE)를 사용해주고 옵티마이저는 Adam을 사용합니다.

num_epochs = 50

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

위의 코드는 모델을 학습하는 반복문입니다.

반복할 에포크 수를 정의해주고 주어진 에포크만큼 반복합니다.

손실을 계산하고 예측한 값과 실제 레이블을 비교하여 정확도를 계산합니다.

 

이후 역전파를 수행하여 파라미터를 업데이트합니다.

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.bool().tolist()
    
total_pred = np.array(total_pred).flatten()
# Save the predictions to a CSV file
df_test['Transported'] = total_pred
df_test[['PassengerId', 'Transported']].to_csv('submission.csv', index=False)

이후 제출 양식에 맞게 csv파일을 생성한 뒤 제출해 줍니다.