하이퍼파라미터 튜닝

import optuna
from optuna.samplers import TPESampler
from optuna.pruners import MedianPruner
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
from sklearn.model_selection import train_test_split
import numpy as np

# 샘플 데이터 생성
def create_sample_data(n_samples=1000, n_features=20):
    X = np.random.randn(n_samples, n_features).astype(np.float32)
    y = (X[:, 0] + X[:, 1] * 2 > 0).astype(np.int64)
    return train_test_split(X, y, test_size=0.2)

class TunableModel(nn.Module):
    """튜닝 가능한 하이퍼파라미터를 가진 모델"""
    def __init__(self, input_dim, hidden_dim, num_layers, dropout):
        super().__init__()
        layers = []
        in_dim = input_dim

        for i in range(num_layers):
            layers.append(nn.Linear(in_dim, hidden_dim))
            layers.append(nn.ReLU())
            layers.append(nn.Dropout(dropout))
            in_dim = hidden_dim

        layers.append(nn.Linear(hidden_dim, 2))
        self.network = nn.Sequential(*layers)

    def forward(self, x):
        return self.network(x)

def objective(trial: optuna.Trial) -> float:
    """Optuna 목적 함수 - 최소화할 값 반환"""
    # 하이퍼파라미터 샘플링
    hidden_dim = trial.suggest_int("hidden_dim", 32, 256, step=32)
    num_layers = trial.suggest_int("num_layers", 1, 4)
    dropout = trial.suggest_float("dropout", 0.1, 0.5)
    learning_rate = trial.suggest_float("learning_rate", 1e-5, 1e-2, log=True)
    batch_size = trial.suggest_categorical("batch_size", [16, 32, 64, 128])
    optimizer_name = trial.suggest_categorical("optimizer", ["Adam", "AdamW", "SGD"])

    # 데이터 준비
    X_train, X_val, y_train, y_val = create_sample_data()
    train_dataset = TensorDataset(torch.tensor(X_train), torch.tensor(y_train))
    val_dataset = TensorDataset(torch.tensor(X_val), torch.tensor(y_val))
    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
    val_loader = DataLoader(val_dataset, batch_size=batch_size)

    # 모델 생성
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    model = TunableModel(
        input_dim=20,
        hidden_dim=hidden_dim,
        num_layers=num_layers,
        dropout=dropout
    ).to(device)

    # 옵티마이저 설정
    if optimizer_name == "Adam":
        optimizer = optim.Adam(model.parameters(), lr=learning_rate)
    elif optimizer_name == "AdamW":
        optimizer = optim.AdamW(model.parameters(), lr=learning_rate)
    else:
        optimizer = optim.SGD(model.parameters(), lr=learning_rate, momentum=0.9)

    criterion = nn.CrossEntropyLoss()

    # 학습 루프
    n_epochs = 30
    for epoch in range(n_epochs):
        model.train()
        for batch_x, batch_y in train_loader:
            batch_x, batch_y = batch_x.to(device), batch_y.to(device)
            optimizer.zero_grad()
            output = model(batch_x)
            loss = criterion(output, batch_y)
            loss.backward()
            optimizer.step()

        # 검증
        model.eval()
        val_loss = 0.0
        correct = 0
        total = 0
        with torch.no_grad():
            for batch_x, batch_y in val_loader:
                batch_x, batch_y = batch_x.to(device), batch_y.to(device)
                output = model(batch_x)
                val_loss += criterion(output, batch_y).item()
                _, predicted = torch.max(output, 1)
                total += batch_y.size(0)
                correct += (predicted == batch_y).sum().item()

        accuracy = correct / total

        # 중간 결과 보고 (조기 종료 판단용)
        trial.report(accuracy, epoch)

        # 성능이 나쁘면 조기 종료 (pruning)
        if trial.should_prune():
            raise optuna.TrialPruned()

    return accuracy  # 최대화할 값

def run_hyperparameter_tuning():
    """하이퍼파라미터 튜닝 실행"""
    # 스터디 생성 (TPE 샘플러 + Median Pruner)
    study = optuna.create_study(
        direction="maximize",  # accuracy 최대화
        sampler=TPESampler(seed=42),
        pruner=MedianPruner(n_warmup_steps=5)
    )

    # 최적화 실행
    study.optimize(
        objective,
        n_trials=50,  # 총 50회 시도
        timeout=600,  # 최대 10분
        show_progress_bar=True
    )

    # 결과 출력
    print(f"\n{'='*50}")
    print(f"Best trial:")
    print(f"  Value (Accuracy): {study.best_trial.value:.4f}")
    print(f"  Params:")
    for key, value in study.best_trial.params.items():
        print(f"    {key}: {value}")

    # 하이퍼파라미터 중요도 분석
    print(f"\n하이퍼파라미터 중요도:")
    importances = optuna.importance.get_param_importances(study)
    for param, importance in importances.items():
        print(f"  {param}: {importance:.4f}")

    return study

# 분산 튜닝 (Ray Tune 예시)
def ray_tune_example():
    """Ray Tune을 사용한 분산 하이퍼파라미터 튜닝"""
    from ray import tune
    from ray.tune.schedulers import ASHAScheduler

    # 탐색 공간 정의
    config = {
        "hidden_dim": tune.choice([64, 128, 256]),
        "num_layers": tune.randint(1, 5),
        "learning_rate": tune.loguniform(1e-5, 1e-2),
        "batch_size": tune.choice([32, 64, 128]),
    }

    # ASHA 스케줄러 (조기 종료)
    scheduler = ASHAScheduler(
        max_t=100,
        grace_period=10,
        reduction_factor=3
    )

    # 튜닝 실행
    analysis = tune.run(
        objective,  # 동일한 목적 함수 사용 가능
        config=config,
        num_samples=100,
        scheduler=scheduler,
        resources_per_trial={"cpu": 2, "gpu": 0.5}
    )

    return analysis.best_config

if __name__ == "__main__":
    study = run_hyperparameter_tuning()

    # 최적 하이퍼파라미터로 최종 모델 학습
    best_params = study.best_trial.params
    print(f"\n최적 하이퍼파라미터로 최종 모델 학습...")
    print(f"Best params: {best_params}")