Early Stopping과 대규모 탐색 전략

from hyperopt import fmin, tpe, hp, STATUS_OK, STATUS_FAIL
from hyperopt import SparkTrials
from xgboost import XGBClassifier
from sklearn.model_selection import cross_val_score
import mlflow
import numpy as np

# 글로벌 최적값 추적
best_score = float('-inf')
no_improve_count = 0
MAX_NO_IMPROVE = 20  # 20회 연속 개선 없으면 중단

def objective(params):
    global best_score, no_improve_count

    with mlflow.start_run(nested=True):
        mlflow.log_params(params)

        model = XGBClassifier(
            **params,
            use_label_encoder=False,
            eval_metric="logloss",
            early_stopping_rounds=10,  # XGBoost 내부 Early Stopping
            n_estimators=500  # 최대 500 라운드, 실제로는 조기 종료
        )

        # XGBoost의 early_stopping_rounds 활용
        model.fit(
            X_train, y_train,
            eval_set=[(X_val, y_val)],
            verbose=False
        )

        score = model.best_score
        mlflow.log_metric("best_score", score)
        mlflow.log_metric("best_iteration", model.best_iteration)

        if score > best_score:
            best_score = score
            no_improve_count = 0
        else:
            no_improve_count += 1

        return {"loss": -score, "status": STATUS_OK}

import optuna

def objective(trial):
    params = {
        "max_depth": trial.suggest_int("max_depth", 3, 12),
        "learning_rate": trial.suggest_float("learning_rate", 0.01, 0.3, log=True),
        "n_estimators": trial.suggest_int("n_estimators", 100, 1000, step=50),
    }

    model = XGBClassifier(**params, use_label_encoder=False, eval_metric="logloss")

    # K-Fold 교차 검증의 각 Fold에서 중간 결과 보고
    scores = []
    for fold_idx, (train_idx, val_idx) in enumerate(kfold.split(X, y)):
        model.fit(X[train_idx], y[train_idx])
        score = model.score(X[val_idx], y[val_idx])
        scores.append(score)

        # 중간 결과 보고 → Pruner가 판단
        trial.report(np.mean(scores), fold_idx)

        # Pruner가 "유망하지 않음"으로 판단하면 조기 중단
        if trial.should_prune():
            raise optuna.TrialPruned()

    return np.mean(scores)

study = optuna.create_study(
    direction="maximize",
    pruner=optuna.pruners.MedianPruner(
        n_startup_trials=10,    # 최소 10회는 완주
        n_warmup_steps=2,       # 최소 2 Fold는 실행
        interval_steps=1        # 매 Fold마다 가지치기 판단
    )
)
study.optimize(objective, n_trials=200)

# 가지치기로 실제 완주한 시행 수 확인
complete_trials = [t for t in study.trials if t.state == optuna.trial.TrialState.COMPLETE]
print(f"200회 중 {len(complete_trials)}회만 완주 → {200 - len(complete_trials)}회 조기 중단으로 비용 절감")

Hyperopt를 선택하세요:
  ✅ Databricks ML Runtime에서 바로 사용하고 싶을 때
  ✅ SparkTrials로 쉬운 분산 실행이 필요할 때
  ✅ 추가 라이브러리 설치를 피하고 싶을 때

Optuna를 선택하세요:
  ✅ 가지치기(Pruning)로 비용을 절감하고 싶을 때
  ✅ 다목적 최적화가 필요할 때 (정확도 + 추론 시간 동시 최적화)
  ✅ 더 직관적인 API와 시각화를 원할 때
  ✅ 조건부 탐색 공간이 복잡할 때

둘 다 사용:
  ✅ Hyperopt로 빠른 초기 탐색 → Optuna로 세밀한 최적화

Stage 1: Coarse Search (거친 탐색) — Random Search, 50~100회
  목표: 유망한 영역 식별
  방법: 넓은 범위에서 Random Search
  결과: 상위 10% 결과의 파라미터 범위 추출

Stage 2: Fine Search (세밀한 탐색) — Bayesian (TPE), 100~200회
  목표: 유망 영역 내에서 최적값 탐색
  방법: Stage 1에서 좁혀진 범위로 TPE 실행
  결과: 최적 파라미터 후보 5~10개

Stage 3: Validation (검증) — Grid Search, 10~50회
  목표: 최종 후보들의 안정성 검증
  방법: 후보 파라미터의 미세 변동에 대한 성능 안정성 확인
  결과: 가장 안정적인 최종 파라미터 선택

import numpy as np
from hyperopt import fmin, tpe, hp, rand, SparkTrials

# Stage 1: Random Search (넓은 범위)
coarse_space = {
    "max_depth": hp.choice("max_depth", range(2, 20)),
    "learning_rate": hp.loguniform("lr", np.log(0.001), np.log(1.0)),
    "n_estimators": hp.choice("n_est", range(50, 1000, 50)),
    "min_child_weight": hp.uniform("mcw", 0.1, 20),
    "subsample": hp.uniform("ss", 0.3, 1.0),
}

coarse_trials = SparkTrials(parallelism=8)
coarse_best = fmin(fn=objective, space=coarse_space,
                   algo=rand.suggest,  # Random Search
                   max_evals=80, trials=coarse_trials)

# Stage 1 결과 분석: 상위 10%의 파라미터 범위 추출
top_results = sorted(coarse_trials.results, key=lambda x: x['loss'])[:8]
# → max_depth: 5~9, lr: 0.01~0.1, n_estimators: 200~500 ...

# Stage 2: Bayesian Search (좁은 범위)
fine_space = {
    "max_depth": hp.choice("max_depth", [5, 6, 7, 8, 9]),
    "learning_rate": hp.loguniform("lr", np.log(0.01), np.log(0.1)),
    "n_estimators": hp.choice("n_est", [200, 300, 400, 500]),
    "min_child_weight": hp.uniform("mcw", 1, 8),
    "subsample": hp.uniform("ss", 0.6, 0.95),
}

fine_trials = SparkTrials(parallelism=4)  # 병렬도 줄여서 TPE 효과 극대화
fine_best = fmin(fn=objective, space=fine_space,
                 algo=tpe.suggest,  # Bayesian (TPE)
                 max_evals=150, trials=fine_trials)