7 P2 프로젝트: 자전거 대여 수요 예측

import os
os.environ["TF_CPP_MIN_LOG_LEVEL"] = "2"

import tensorflow as tf
import keras
import torch

print("TensorFlow:", tf.__version__)
print("Keras:", keras.__version__)
print("PyTorch:", torch.__version__)
print("TensorFlow GPU:", tf.config.list_physical_devices("GPU"))
print("PyTorch CUDA:", torch.cuda.is_available())

from pathlib import Path
import zipfile
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

url = "https://archive.ics.uci.edu/static/public/275/bike+sharing+dataset.zip"

archive_path = keras.utils.get_file(
    fname="bike_sharing_dataset.zip",
    origin=url,
)

data_dir = Path(archive_path).parent / "bike_sharing_dataset"
data_dir.mkdir(exist_ok=True)

with zipfile.ZipFile(archive_path) as z:
    z.extractall(data_dir)

hour_path = next(data_dir.rglob("hour.csv"))
bike = pd.read_csv(hour_path)

bike.head()

print("데이터 크기:", bike.shape)
print()
print(bike.info())

bike["datetime"] = pd.to_datetime(bike["dteday"]) + pd.to_timedelta(bike["hr"], unit="h")
bike = bike.sort_values("datetime").reset_index(drop=True)

n = len(bike)
train_end = int(n * 0.70)
val_end = int(n * 0.85)

train_df = bike.iloc[:train_end].copy()
val_df = bike.iloc[train_end:val_end].copy()
test_df = bike.iloc[val_end:].copy()

print("train:", train_df.shape)
print("validation:", val_df.shape)
print("test:", test_df.shape)

print()
print("train 기간:", train_df["datetime"].min(), "~", train_df["datetime"].max())
print("validation 기간:", val_df["datetime"].min(), "~", val_df["datetime"].max())
print("test 기간:", test_df["datetime"].min(), "~", test_df["datetime"].max())

from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OneHotEncoder, StandardScaler

target = "cnt"

categorical_features = [
    "season", "yr", "mnth", "hr",
    "holiday", "weekday", "workingday", "weathersit"
]

numeric_features = [
    "temp", "atemp", "hum", "windspeed"
]

feature_columns = categorical_features + numeric_features

X_train_raw = train_df[feature_columns]
X_val_raw = val_df[feature_columns]
X_test_raw = test_df[feature_columns]

y_train = train_df[target].to_numpy(dtype="float32")
y_val = val_df[target].to_numpy(dtype="float32")
y_test = test_df[target].to_numpy(dtype="float32")

preprocessor = ColumnTransformer([
    ("categorical", OneHotEncoder(handle_unknown="ignore", sparse_output=False),
     categorical_features),
    ("numeric", StandardScaler(), numeric_features),
])

X_train = preprocessor.fit_transform(X_train_raw).astype("float32")
X_val = preprocessor.transform(X_val_raw).astype("float32")
X_test = preprocessor.transform(X_test_raw).astype("float32")

print("X_train:", X_train.shape)
print("X_val:", X_val.shape)
print("X_test:", X_test.shape)

from keras import layers

# TODO: 기준 모델을 구성한다.
keras_base = keras.Sequential([
    # layers.Dense(..., activation="relu"),
    # ...
    # layers.Dense(1)
])

# TODO: 입력 shape에 맞게 모델을 build한다.
# keras_base.build(input_shape=(None, X_train.shape[1]))
# keras_base.summary()

# TODO: 모델을 compile하고 훈련한다.
#
# keras_base.compile(
#     optimizer="adam",
#     loss="mse",
#     metrics=[keras.metrics.RootMeanSquaredError(name="rmse")]
# )
#
# history_base = keras_base.fit(
#     X_train,
#     y_train,
#     epochs=...,
#     batch_size=...,
#     validation_data=(X_val, y_val),
#     verbose=1,
# )

# TODO: 개선 모델과 필요한 callback을 구성한다.
#
# improved_model = keras.Sequential([
#     ...
# ])
#
# improved_model.compile(...)
#
# history_improved = improved_model.fit(
#     X_train,
#     y_train,
#     validation_data=(X_val, y_val),
#     ...
# )

import torch
from torch import nn
from torch.utils.data import TensorDataset, DataLoader

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

train_dataset = TensorDataset(
    torch.tensor(X_train, dtype=torch.float32),
    torch.tensor(y_train, dtype=torch.float32).reshape(-1, 1)
)
val_dataset = TensorDataset(
    torch.tensor(X_val, dtype=torch.float32),
    torch.tensor(y_val, dtype=torch.float32).reshape(-1, 1)
)
test_dataset = TensorDataset(
    torch.tensor(X_test, dtype=torch.float32),
    torch.tensor(y_test, dtype=torch.float32).reshape(-1, 1)
)

train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=256, shuffle=False)
test_loader = DataLoader(test_dataset, batch_size=256, shuffle=False)

print("device:", device)

def train_regression_model(model, train_loader, val_loader, epochs=50,
                           weight_decay=0.0, patience=None):
    model = model.to(device)
    loss_fn = nn.MSELoss()
    optimizer = torch.optim.Adam(
        model.parameters(),
        weight_decay=weight_decay
    )

    history = {
        "loss": [], "val_loss": [],
        "rmse": [], "val_rmse": []
    }

    best_val_loss = float("inf")
    best_state = None
    wait = 0

    for epoch in range(epochs):
        model.train()
        train_sse = 0.0
        train_n = 0

        for x_batch, y_batch in train_loader:
            x_batch = x_batch.to(device)
            y_batch = y_batch.to(device)

            optimizer.zero_grad()
            pred = model(x_batch)
            loss = loss_fn(pred, y_batch)
            loss.backward()
            optimizer.step()

            train_sse += ((pred - y_batch) ** 2).sum().item()
            train_n += len(x_batch)

        model.eval()
        val_sse = 0.0
        val_n = 0

        with torch.no_grad():
            for x_batch, y_batch in val_loader:
                x_batch = x_batch.to(device)
                y_batch = y_batch.to(device)

                pred = model(x_batch)
                val_sse += ((pred - y_batch) ** 2).sum().item()
                val_n += len(x_batch)

        train_mse = train_sse / train_n
        val_mse = val_sse / val_n

        history["loss"].append(train_mse)
        history["val_loss"].append(val_mse)
        history["rmse"].append(np.sqrt(train_mse))
        history["val_rmse"].append(np.sqrt(val_mse))

        if patience is not None:
            if val_mse < best_val_loss:
                best_val_loss = val_mse
                best_state = {
                    k: v.detach().cpu().clone()
                    for k, v in model.state_dict().items()
                }
                wait = 0
            else:
                wait += 1
                if wait >= patience:
                    break

    if best_state is not None:
        model.load_state_dict(best_state)

    return model, history

# TODO: 기준 모델을 구성한다.
torch_base = nn.Sequential(
    # nn.Linear(X_train.shape[1], ...),
    # nn.ReLU(),
    # ...
    # nn.Linear(..., 1),
)

# TODO: 기준 모델을 훈련한다.
# torch_base, torch_history_base = train_regression_model(
#     torch_base,
#     train_loader,
#     val_loader,
#     epochs=...
# )

# TODO: 개선 모델을 구성하고 훈련한다.
#
# torch_improved = nn.Sequential(
#     ...
# )
#
# torch_improved, torch_history_improved = train_regression_model(
#     torch_improved,
#     train_loader,
#     val_loader,
#     epochs=...,
#     weight_decay=...,
#     patience=...
# )

# Keras
keras_base_history = history_base.history
keras_improved_history = history_improved.history

# PyTorch
pytorch_base_history = torch_history_base
pytorch_improved_history = torch_history_improved

# TODO: 훈련·검증 RMSE를 시각화한다.
#
# plt.figure(figsize=(7, 4))
# plt.plot(base_history["rmse"], label="Base training RMSE")
# plt.plot(base_history["val_rmse"], label="Base validation RMSE")
# plt.xlabel("Epoch")
# plt.ylabel("RMSE")
# plt.legend()
# plt.show()
#
# 개선 모델도 같은 방식으로 확인한다.

# Keras 최종 모델
keras_final_model = ...

test_mse, keras_test_rmse = keras_final_model.evaluate(
    X_test, y_test, verbose=0
)

print("Keras test MSE :", test_mse)
print("Keras test RMSE:", keras_test_rmse)

# PyTorch 최종 모델
pytorch_final_model = ...
pytorch_final_model.eval()

predictions = []

with torch.no_grad():
    for x_batch, _ in test_loader:
        pred = pytorch_final_model(x_batch.to(device))
        predictions.append(pred.cpu().numpy())

pytorch_predictions = np.concatenate(predictions).reshape(-1)
pytorch_test_rmse = np.sqrt(np.mean((pytorch_predictions - y_test) ** 2))

print("PyTorch test RMSE:", pytorch_test_rmse)

# TODO: 최종 모델의 테스트 예측값을 final_predictions에 저장한다.
#
# Keras 예:
# final_predictions = final_model.predict(X_test, verbose=0).reshape(-1)
#
# PyTorch에서는 위 테스트 평가에서 만든 predictions를 사용할 수 있다.

# TODO: 실제값과 예측값 일부를 비교한다.
#
# comparison = pd.DataFrame({
#     "actual": y_test,
#     "predicted": final_predictions,
# })
#
# comparison["error"] = comparison["actual"] - comparison["predicted"]
# comparison["abs_error"] = comparison["error"].abs()
#
# comparison.head(20)