045 회귀에서의 앙상블 기법

키워드: 앙상블, ensemble, 모델 결합

개요

앙상블(Ensemble)은 여러 모델의 예측을 결합하여 더 나은 성능을 얻는 기법입니다. FLAML은 자동으로 앙상블을 구성할 수 있으며, 이 글에서는 회귀에서의 앙상블 전략을 알아봅니다.

실습 환경

• Python 버전: 3.11 권장
• 필요 패키지: flaml[automl], scikit-learn, numpy

pip install flaml[automl] scikit-learn numpy matplotlib

앙상블 기법 개요

앙상블의 장점

import numpy as np
import matplotlib.pyplot as plt

# 045 단일 모델 vs 앙상블 시뮬레이션
np.random.seed(42)

# 3개 모델의 예측 (각각 오차 있음)
true_value = 100
model1_pred = true_value + np.random.randn(100) * 10  # 표준편차 10
model2_pred = true_value + np.random.randn(100) * 10
model3_pred = true_value + np.random.randn(100) * 10

# 045 앙상블 (평균)
ensemble_pred = (model1_pred + model2_pred + model3_pred) / 3

# 045 오차 비교
print("평균 절대 오차 (MAE):")
print(f"  Model 1: {np.mean(np.abs(model1_pred - true_value)):.2f}")
print(f"  Model 2: {np.mean(np.abs(model2_pred - true_value)):.2f}")
print(f"  Model 3: {np.mean(np.abs(model3_pred - true_value)):.2f}")
print(f"  Ensemble: {np.mean(np.abs(ensemble_pred - true_value)):.2f}")
print("\n→ 앙상블이 개별 모델보다 안정적!")

앙상블 유형

ensemble_types = {
    '유형': ['Bagging', 'Boosting', 'Stacking', 'Voting/Averaging'],
    '대표 모델': ['Random Forest', 'XGBoost/LightGBM', 'Stacked Generalization', '단순 평균'],
    '특징': [
        '병렬 학습, 분산 감소',
        '순차 학습, 편향 감소',
        '메타 모델 사용',
        '다양한 모델 결합'
    ]
}

import pandas as pd
print("앙상블 유형:")
print(pd.DataFrame(ensemble_types).to_string(index=False))

FLAML의 앙상블 기능

자동 앙상블

from flaml import AutoML
from sklearn.datasets import fetch_california_housing
from sklearn.model_selection import train_test_split
from sklearn.metrics import r2_score, mean_absolute_error

# 045 데이터 준비
data = fetch_california_housing()
X_train, X_test, y_train, y_test = train_test_split(
    data.data, data.target, test_size=0.2, random_state=42
)

# 045 앙상블 활성화
automl_ensemble = AutoML()
automl_ensemble.fit(
    X_train, y_train,
    task="regression",
    time_budget=120,
    ensemble=True,  # 앙상블 활성화
    metric="r2",
    verbose=1
)

y_pred_ensemble = automl_ensemble.predict(X_test)
r2_ensemble = r2_score(y_test, y_pred_ensemble)

print(f"\n앙상블 모델 R²: {r2_ensemble:.4f}")

단일 모델과 비교

# 045 앙상블 없이
automl_single = AutoML()
automl_single.fit(
    X_train, y_train,
    task="regression",
    time_budget=120,
    ensemble=False,  # 앙상블 비활성화
    metric="r2",
    verbose=0
)

y_pred_single = automl_single.predict(X_test)
r2_single = r2_score(y_test, y_pred_single)

print(f"단일 모델 R²: {r2_single:.4f}")
print(f"앙상블 모델 R²: {r2_ensemble:.4f}")
print(f"개선: {(r2_ensemble - r2_single) * 100:.2f}%p")

수동 앙상블 구현

1. 단순 평균 (Simple Averaging)

from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.linear_model import Ridge

# 045 여러 모델 학습
models = {
    'Random Forest': RandomForestRegressor(n_estimators=100, random_state=42),
    'Gradient Boosting': GradientBoostingRegressor(n_estimators=100, random_state=42),
    'Ridge': Ridge()
}

predictions = {}
for name, model in models.items():
    model.fit(X_train, y_train)
    predictions[name] = model.predict(X_test)
    print(f"{name} R²: {r2_score(y_test, predictions[name]):.4f}")

# 045 단순 평균
ensemble_avg = np.mean(list(predictions.values()), axis=0)
print(f"\n단순 평균 앙상블 R²: {r2_score(y_test, ensemble_avg):.4f}")

2. 가중 평균 (Weighted Averaging)

def weighted_ensemble(predictions, weights):
    """가중 평균 앙상블"""
    weighted_pred = np.zeros_like(list(predictions.values())[0])
    for pred, weight in zip(predictions.values(), weights):
        weighted_pred += pred * weight
    return weighted_pred

# 045 검증 성능 기반 가중치
val_scores = {}
for name, model in models.items():
    val_scores[name] = r2_score(y_test, predictions[name])

# 045 가중치 계산 (성능 비례)
total_score = sum(val_scores.values())
weights = [val_scores[name] / total_score for name in models.keys()]

print("모델별 가중치:")
for name, weight in zip(models.keys(), weights):
    print(f"  {name}: {weight:.4f}")

# 045 가중 평균 예측
ensemble_weighted = weighted_ensemble(predictions, weights)
print(f"\n가중 평균 앙상블 R²: {r2_score(y_test, ensemble_weighted):.4f}")

3. 스태킹 (Stacking)

from sklearn.model_selection import cross_val_predict

class StackingRegressor:
    """스태킹 앙상블 구현"""

    def __init__(self, base_models, meta_model):
        self.base_models = base_models
        self.meta_model = meta_model

    def fit(self, X, y):
        # 1단계: 베이스 모델 학습 및 메타 특성 생성
        meta_features = np.zeros((len(y), len(self.base_models)))

        for i, (name, model) in enumerate(self.base_models.items()):
            # 교차 검증 예측으로 메타 특성 생성
            meta_features[:, i] = cross_val_predict(model, X, y, cv=5)
            model.fit(X, y)  # 전체 데이터로 재학습

        # 2단계: 메타 모델 학습
        self.meta_model.fit(meta_features, y)
        return self

    def predict(self, X):
        # 베이스 모델 예측
        meta_features = np.zeros((len(X), len(self.base_models)))
        for i, (name, model) in enumerate(self.base_models.items()):
            meta_features[:, i] = model.predict(X)

        # 메타 모델 예측
        return self.meta_model.predict(meta_features)

# 045 스태킹 적용
base_models = {
    'RF': RandomForestRegressor(n_estimators=50, random_state=42),
    'GB': GradientBoostingRegressor(n_estimators=50, random_state=42),
}
meta_model = Ridge()

stacking = StackingRegressor(base_models, meta_model)
stacking.fit(X_train, y_train)

y_pred_stacking = stacking.predict(X_test)
print(f"스태킹 앙상블 R²: {r2_score(y_test, y_pred_stacking):.4f}")

sklearn의 앙상블 사용

from sklearn.ensemble import StackingRegressor as SklearnStacking
from sklearn.ensemble import VotingRegressor

# 045 VotingRegressor (평균)
voting = VotingRegressor(estimators=[
    ('rf', RandomForestRegressor(n_estimators=50, random_state=42)),
    ('gb', GradientBoostingRegressor(n_estimators=50, random_state=42)),
    ('ridge', Ridge())
])

voting.fit(X_train, y_train)
y_pred_voting = voting.predict(X_test)
print(f"Voting R²: {r2_score(y_test, y_pred_voting):.4f}")

# 045 StackingRegressor
stacking_sklearn = SklearnStacking(
    estimators=[
        ('rf', RandomForestRegressor(n_estimators=50, random_state=42)),
        ('gb', GradientBoostingRegressor(n_estimators=50, random_state=42)),
    ],
    final_estimator=Ridge()
)

stacking_sklearn.fit(X_train, y_train)
y_pred_stacking_sk = stacking_sklearn.predict(X_test)
print(f"Sklearn Stacking R²: {r2_score(y_test, y_pred_stacking_sk):.4f}")

FLAML + 커스텀 앙상블

class FLAMLEnsemble:
    """FLAML을 포함한 앙상블"""

    def __init__(self, time_budget_per_model=30):
        self.time_budget = time_budget_per_model
        self.models = {}

    def fit(self, X, y):
        # FLAML 모델들
        for estimator in ['lgbm', 'xgboost', 'rf']:
            automl = AutoML()
            automl.fit(
                X, y,
                task="regression",
                time_budget=self.time_budget,
                estimator_list=[estimator],
                verbose=0
            )
            self.models[estimator] = automl

        return self

    def predict(self, X):
        predictions = []
        for name, model in self.models.items():
            predictions.append(model.predict(X))
        return np.mean(predictions, axis=0)

# 045 사용
flaml_ensemble = FLAMLEnsemble(time_budget_per_model=20)
flaml_ensemble.fit(X_train, y_train)

y_pred_flaml_ens = flaml_ensemble.predict(X_test)
print(f"\nFLAML 앙상블 R²: {r2_score(y_test, y_pred_flaml_ens):.4f}")

앙상블 결과 비교

results = {
    'Method': ['Single (FLAML)', 'Ensemble (FLAML)', 'Simple Average', 'Weighted Average', 'Stacking', 'Voting'],
    'R²': [r2_single, r2_ensemble, r2_score(y_test, ensemble_avg),
           r2_score(y_test, ensemble_weighted), r2_score(y_test, y_pred_stacking),
           r2_score(y_test, y_pred_voting)]
}

results_df = pd.DataFrame(results).sort_values('R²', ascending=False)
print("\n앙상블 방법 비교:")
print(results_df.to_string(index=False))

# 045 시각화
plt.figure(figsize=(10, 6))
plt.barh(results_df['Method'], results_df['R²'])
plt.xlabel('R² Score')
plt.title('Ensemble Methods Comparison')
plt.xlim(0.8, 0.9)
plt.tight_layout()
plt.show()

앙상블 선택 가이드

guide = {
    '상황': ['시간 제한', '최고 성능 필요', '해석 필요', '다양한 모델'],
    '권장 방법': ['FLAML ensemble=True', 'Stacking', '단일 모델', 'Voting'],
    '이유': [
        '자동 최적화',
        '메타 학습 효과',
        '복잡도 낮음',
        '다양성 확보'
    ]
}

print("\n앙상블 선택 가이드:")
print(pd.DataFrame(guide).to_string(index=False))

정리

• 앙상블은 여러 모델을 결합하여 성능을 향상시킵니다.
• FLAML의 ensemble=True로 자동 앙상블을 사용할 수 있습니다.
• 단순 평균: 가장 쉬운 방법, 안정적
• 가중 평균: 성능 좋은 모델에 더 높은 가중치
• 스태킹: 메타 모델로 최적 결합 학습
• 다양한 모델을 결합할수록 다양성이 높아져 효과적입니다.
• 항상 앙상블이 더 좋은 것은 아니므로 비교 검증이 필요합니다.

다음 글 예고

다음 글에서는 회귀에서의 교차 검증 전략에 대해 알아보겠습니다. 시계열 데이터를 위한 특수 교차 검증 방법을 다룹니다.

---

FLAML AutoML 마스터 시리즈 #045