083 종합 프로젝트 - 캐글 대회 도전 (3) 튜닝

키워드: 캐글, Kaggle, 튜닝, 앙상블

개요

이전 글에서 구축한 모델을 더욱 정교하게 튜닝합니다. FLAML의 고급 튜닝 기법, 커스텀 탐색 공간, 앙상블 방법을 적용하여 모델 성능을 극대화합니다.

실습 환경

• Python 버전: 3.11 권장
• 필요 패키지: flaml[automl]

pip install flaml[automl] pandas numpy scikit-learn

데이터 준비 (이전 글 연속)

import numpy as np
import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
from flaml import AutoML, tune

np.random.seed(42)
n_samples = 10000

# 083 데이터 생성 (이전 글과 동일)
data = {
    'customer_id': range(1, n_samples + 1),
    'age': np.random.randint(18, 70, n_samples),
    'gender': np.random.choice(['M', 'F'], n_samples),
    'tenure': np.random.randint(1, 72, n_samples),
    'monthly_charges': np.random.uniform(20, 100, n_samples),
    'total_charges': np.random.uniform(100, 5000, n_samples),
    'num_products': np.random.randint(1, 5, n_samples),
    'has_phone_service': np.random.choice([0, 1], n_samples, p=[0.1, 0.9]),
    'has_internet_service': np.random.choice([0, 1], n_samples, p=[0.2, 0.8]),
    'has_streaming': np.random.choice([0, 1], n_samples, p=[0.4, 0.6]),
    'num_support_tickets': np.random.poisson(2, n_samples),
    'avg_monthly_usage': np.random.uniform(10, 500, n_samples),
    'contract_type': np.random.choice(['month', 'year', '2year'], n_samples, p=[0.5, 0.3, 0.2]),
    'payment_method': np.random.choice(['credit', 'bank', 'electronic'], n_samples),
}
df = pd.DataFrame(data)

churn_prob = 0.1 + 0.3 * (df['tenure'] < 12).astype(int) + \
             0.2 * (df['num_support_tickets'] > 3).astype(int) + \
             0.1 * (df['contract_type'] == 'month').astype(int) - \
             0.1 * (df['monthly_charges'] < 50).astype(int)
df['churn'] = np.random.binomial(1, np.clip(churn_prob, 0, 1))

# 083 특성 엔지니어링
df['avg_charge_per_month'] = df['total_charges'] / (df['tenure'] + 1)
df['service_count'] = df['has_phone_service'] + df['has_internet_service'] + df['has_streaming']
df['ticket_per_tenure'] = df['num_support_tickets'] / (df['tenure'] + 1)
df['is_new_customer'] = (df['tenure'] < 12).astype(int)

# 083 인코딩
for col in ['gender', 'contract_type', 'payment_method']:
    df[col] = LabelEncoder().fit_transform(df[col])

# 083 데이터 분할
feature_cols = [c for c in df.columns if c not in ['customer_id', 'churn']]
X = df[feature_cols]
y = df['churn']

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)
X_train, X_val, y_train, y_val = train_test_split(X_train, y_train, test_size=0.2, random_state=42, stratify=y_train)

print(f"학습: {X_train.shape}, 검증: {X_val.shape}, 테스트: {X_test.shape}")

1. 커스텀 탐색 공간 정의

# 083 모델별 상세 탐색 공간
custom_hp = {
    'lgbm': {
        'n_estimators': {'domain': tune.randint(100, 500)},
        'num_leaves': {'domain': tune.randint(20, 100)},
        'max_depth': {'domain': tune.randint(3, 12)},
        'learning_rate': {'domain': tune.loguniform(0.01, 0.2)},
        'min_child_samples': {'domain': tune.randint(5, 50)},
        'subsample': {'domain': tune.uniform(0.6, 1.0)},
        'colsample_bytree': {'domain': tune.uniform(0.6, 1.0)},
        'reg_alpha': {'domain': tune.loguniform(1e-4, 1.0)},
        'reg_lambda': {'domain': tune.loguniform(1e-4, 1.0)},
    },
    'xgboost': {
        'n_estimators': {'domain': tune.randint(100, 500)},
        'max_depth': {'domain': tune.randint(3, 10)},
        'learning_rate': {'domain': tune.loguniform(0.01, 0.2)},
        'min_child_weight': {'domain': tune.randint(1, 10)},
        'subsample': {'domain': tune.uniform(0.6, 1.0)},
        'colsample_bytree': {'domain': tune.uniform(0.6, 1.0)},
        'reg_alpha': {'domain': tune.loguniform(1e-4, 1.0)},
        'reg_lambda': {'domain': tune.loguniform(1e-4, 1.0)},
    },
    'rf': {
        'n_estimators': {'domain': tune.randint(100, 300)},
        'max_depth': {'domain': tune.randint(5, 20)},
        'min_samples_split': {'domain': tune.randint(2, 20)},
        'min_samples_leaf': {'domain': tune.randint(1, 10)},
        'max_features': {'domain': tune.choice(['sqrt', 'log2', None])},
    }
}

print("커스텀 탐색 공간 정의 완료")
for model, params in custom_hp.items():
    print(f"\n{model}: {len(params)}개 파라미터")

2. 확장된 시간으로 AutoML

# 083 더 긴 시간 예산으로 학습
automl_tuned = AutoML()
automl_tuned.fit(
    X_train, y_train,
    task="classification",
    metric="roc_auc",
    time_budget=180,  # 3분
    estimator_list=['lgbm', 'xgboost', 'rf'],
    custom_hp=custom_hp,
    n_jobs=-1,
    seed=42,
    verbose=2
)

print(f"\n=== 튜닝된 FLAML 결과 ===")
print(f"최적 모델: {automl_tuned.best_estimator}")
print(f"최적 설정: {automl_tuned.best_config}")

3. 개별 모델 튜닝

from sklearn.metrics import roc_auc_score

# 083 각 모델별 최적 튜닝
models = {}

# 083 LightGBM 튜닝
automl_lgbm = AutoML()
automl_lgbm.fit(
    X_train, y_train,
    task="classification",
    metric="roc_auc",
    time_budget=60,
    estimator_list=['lgbm'],
    custom_hp={'lgbm': custom_hp['lgbm']},
    verbose=0
)
models['lgbm'] = automl_lgbm

# 083 XGBoost 튜닝
automl_xgb = AutoML()
automl_xgb.fit(
    X_train, y_train,
    task="classification",
    metric="roc_auc",
    time_budget=60,
    estimator_list=['xgboost'],
    custom_hp={'xgboost': custom_hp['xgboost']},
    verbose=0
)
models['xgboost'] = automl_xgb

# 083 Random Forest 튜닝
automl_rf = AutoML()
automl_rf.fit(
    X_train, y_train,
    task="classification",
    metric="roc_auc",
    time_budget=60,
    estimator_list=['rf'],
    custom_hp={'rf': custom_hp['rf']},
    verbose=0
)
models['rf'] = automl_rf

# 083 각 모델 성능 비교
print("\n=== 개별 모델 성능 비교 ===")
for name, model in models.items():
    val_proba = model.predict_proba(X_val)[:, 1]
    auc = roc_auc_score(y_val, val_proba)
    print(f"{name}: AUC = {auc:.4f}")

4. 앙상블 구축

# 083 단순 평균 앙상블
def simple_ensemble_predict_proba(models, X):
    """모델 예측 평균"""
    probas = []
    for name, model in models.items():
        proba = model.predict_proba(X)[:, 1]
        probas.append(proba)
    return np.mean(probas, axis=0)

# 083 검증 세트로 앙상블 평가
val_proba_ensemble = simple_ensemble_predict_proba(models, X_val)
ensemble_auc = roc_auc_score(y_val, val_proba_ensemble)
print(f"\n앙상블 (평균): AUC = {ensemble_auc:.4f}")

# 083 가중 평균 앙상블
def weighted_ensemble(models, X, weights):
    """가중 평균 앙상블"""
    probas = []
    for (name, model), weight in zip(models.items(), weights):
        proba = model.predict_proba(X)[:, 1]
        probas.append(proba * weight)
    return np.sum(probas, axis=0)

# 083 검증 AUC 기반 가중치
weights_raw = []
for name, model in models.items():
    val_proba = model.predict_proba(X_val)[:, 1]
    weights_raw.append(roc_auc_score(y_val, val_proba))

weights = np.array(weights_raw) / sum(weights_raw)
print(f"\n모델별 가중치: {dict(zip(models.keys(), weights.round(3)))}")

val_proba_weighted = weighted_ensemble(models, X_val, weights)
weighted_auc = roc_auc_score(y_val, val_proba_weighted)
print(f"앙상블 (가중 평균): AUC = {weighted_auc:.4f}")

5. 스태킹 앙상블

from sklearn.linear_model import LogisticRegression

# 083 메타 특성 생성
def create_meta_features(models, X):
    """메타 모델용 특성 생성"""
    meta_features = []
    for name, model in models.items():
        proba = model.predict_proba(X)[:, 1]
        meta_features.append(proba)
    return np.column_stack(meta_features)

# 083 메타 특성 생성
X_train_meta = create_meta_features(models, X_train)
X_val_meta = create_meta_features(models, X_val)
X_test_meta = create_meta_features(models, X_test)

# 083 메타 모델 학습
meta_model = LogisticRegression(random_state=42)
meta_model.fit(X_train_meta, y_train)

# 083 스태킹 평가
stacking_proba = meta_model.predict_proba(X_val_meta)[:, 1]
stacking_auc = roc_auc_score(y_val, stacking_proba)
print(f"\n스태킹 앙상블: AUC = {stacking_auc:.4f}")

# 083 메타 모델 계수
print(f"메타 모델 계수: {dict(zip(models.keys(), meta_model.coef_[0].round(3)))}")

6. 최종 성능 비교

# 083 테스트 세트 최종 평가
results = []

# 083 개별 모델
for name, model in models.items():
    proba = model.predict_proba(X_test)[:, 1]
    auc = roc_auc_score(y_test, proba)
    results.append({'모델': name, 'Test AUC': auc})

# 083 앙상블
test_proba_avg = simple_ensemble_predict_proba(models, X_test)
results.append({'모델': 'Ensemble (Avg)', 'Test AUC': roc_auc_score(y_test, test_proba_avg)})

test_proba_weighted = weighted_ensemble(models, X_test, weights)
results.append({'모델': 'Ensemble (Weighted)', 'Test AUC': roc_auc_score(y_test, test_proba_weighted)})

test_proba_stacking = meta_model.predict_proba(X_test_meta)[:, 1]
results.append({'모델': 'Stacking', 'Test AUC': roc_auc_score(y_test, test_proba_stacking)})

# 083 결과 정리
results_df = pd.DataFrame(results).sort_values('Test AUC', ascending=False)
print("\n=== 최종 테스트 성능 비교 ===")
print(results_df.to_string(index=False))

7. 최적 모델 선택 및 저장

import joblib

# 083 최고 성능 모델 선택
best_method = results_df.iloc[0]['모델']
print(f"\n최고 성능 방법: {best_method}")

# 083 모델 저장
if 'Stacking' in best_method:
    # 스태킹 저장
    ensemble_package = {
        'base_models': models,
        'meta_model': meta_model,
        'method': 'stacking'
    }
elif 'Weighted' in best_method:
    ensemble_package = {
        'base_models': models,
        'weights': weights,
        'method': 'weighted'
    }
else:
    ensemble_package = {
        'base_models': {best_method: models[best_method]},
        'method': 'single'
    }

joblib.dump(ensemble_package, 'best_model_package.pkl')
print("모델 저장 완료: best_model_package.pkl")

8. 캐글 제출 파일 생성

def create_submission(model_package, X_test, test_ids=None):
    """캐글 제출 파일 생성"""

    if model_package['method'] == 'stacking':
        base_models = model_package['base_models']
        meta_model = model_package['meta_model']
        X_meta = create_meta_features(base_models, X_test)
        predictions = meta_model.predict_proba(X_meta)[:, 1]

    elif model_package['method'] == 'weighted':
        predictions = weighted_ensemble(
            model_package['base_models'],
            X_test,
            model_package['weights']
        )
    else:
        model = list(model_package['base_models'].values())[0]
        predictions = model.predict_proba(X_test)[:, 1]

    if test_ids is None:
        test_ids = range(len(predictions))

    submission = pd.DataFrame({
        'id': test_ids,
        'churn_probability': predictions
    })

    return submission

# 083 제출 파일 생성
submission = create_submission(ensemble_package, X_test, range(len(X_test)))
submission.to_csv('submission.csv', index=False)
print("\n제출 파일 생성: submission.csv")
print(submission.head())

튜닝 팁 정리

tips = {
    '전략': ['시간 예산', '탐색 공간', '앙상블', '검증'],
    '권장': [
        '초기 30분 → 확장 1-2시간',
        '모델별 맞춤 탐색 공간',
        '상위 3-5개 모델 앙상블',
        'Stratified K-Fold 사용'
    ],
    '주의': [
        '과도한 시간은 과적합 위험',
        '너무 넓은 공간은 비효율',
        '앙상블은 다양성이 핵심',
        '리더보드 과적합 주의'
    ]
}

print("\n=== 캐글 튜닝 팁 ===")
print(pd.DataFrame(tips).to_string(index=False))

정리

• 커스텀 탐색 공간: 모델별 상세 파라미터 정의
• 개별 튜닝: 각 모델 최적화 후 앙상블
• 앙상블: 평균, 가중 평균, 스태킹
• 제출 파일: 캐글 형식으로 생성
• 앙상블이 단일 모델보다 일반적으로 우수

다음 글 예고

다음 글에서는 추천 시스템 기초 프로젝트를 진행합니다. FLAML을 활용한 추천 모델 구축 방법을 알아봅니다.

---

FLAML AutoML 마스터 시리즈 #083