064 시계열 앙상블

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

개요

단일 모델보다 여러 모델을 결합한 앙상블이 더 안정적인 예측을 제공합니다. 이 글에서는 시계열 예측에서 앙상블 기법을 활용하는 방법을 알아봅니다.

실습 환경

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

pip install flaml[automl] pandas numpy matplotlib

앙상블이 필요한 이유

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from flaml import AutoML
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.linear_model import Ridge, Lasso
from sklearn.metrics import mean_absolute_error, mean_squared_error

# 064 시계열 데이터 생성
np.random.seed(42)
n_days = 500

dates = pd.date_range('2022-01-01', periods=n_days, freq='D')
trend = np.linspace(100, 200, n_days)
seasonal = 30 * np.sin(np.arange(n_days) * 2 * np.pi / 365)
weekly = 10 * np.sin(np.arange(n_days) * 2 * np.pi / 7)
noise = np.random.randn(n_days) * 15

values = trend + seasonal + weekly + noise

df = pd.DataFrame({
    'date': dates,
    'value': values
})

# 064 특성 생성
df['dayofweek'] = df['date'].dt.dayofweek
df['month'] = df['date'].dt.month
df['dayofyear'] = df['date'].dt.dayofyear
df['lag_1'] = df['value'].shift(1)
df['lag_7'] = df['value'].shift(7)
df['rolling_mean_7'] = df['value'].shift(1).rolling(7).mean()
df['rolling_std_7'] = df['value'].shift(1).rolling(7).std()

df = df.dropna()

# 064 분할
train_size = int(len(df) * 0.8)
train = df.iloc[:train_size]
test = df.iloc[train_size:]

feature_cols = ['dayofweek', 'month', 'dayofyear', 'lag_1', 'lag_7',
                'rolling_mean_7', 'rolling_std_7']

X_train = train[feature_cols]
y_train = train['value']
X_test = test[feature_cols]
y_test = test['value']

print(f"학습: {len(train)}, 테스트: {len(test)}")

기본 앙상블 방법

단순 평균 앙상블

# 064 개별 모델 학습
models = {
    'ridge': Ridge(alpha=1.0),
    'rf': RandomForestRegressor(n_estimators=100, random_state=42),
    'gbm': GradientBoostingRegressor(n_estimators=100, random_state=42)
}

predictions = {}
for name, model in models.items():
    model.fit(X_train, y_train)
    predictions[name] = model.predict(X_test)

# 064 개별 모델 성능
print("개별 모델 성능:")
for name, pred in predictions.items():
    mae = mean_absolute_error(y_test, pred)
    print(f"  {name}: MAE = {mae:.2f}")

# 064 단순 평균 앙상블
ensemble_avg = np.mean([predictions['ridge'], predictions['rf'],
                        predictions['gbm']], axis=0)

print(f"\n단순 평균 앙상블: MAE = {mean_absolute_error(y_test, ensemble_avg):.2f}")

가중 평균 앙상블

def weighted_ensemble(predictions_dict, weights):
    """가중 평균 앙상블"""
    weighted_pred = np.zeros(len(list(predictions_dict.values())[0]))

    for name, weight in weights.items():
        weighted_pred += predictions_dict[name] * weight

    return weighted_pred

# 064 검증 세트로 가중치 결정
val_size = int(len(X_train) * 0.2)
X_tr = X_train.iloc[:-val_size]
y_tr = y_train.iloc[:-val_size]
X_val = X_train.iloc[-val_size:]
y_val = y_train.iloc[-val_size:]

# 064 검증 세트 예측
val_predictions = {}
for name, model in models.items():
    model_copy = type(model)(**model.get_params())
    model_copy.fit(X_tr, y_tr)
    val_predictions[name] = model_copy.predict(X_val)

# 064 검증 MAE 기반 가중치 (역수)
val_maes = {name: mean_absolute_error(y_val, pred)
            for name, pred in val_predictions.items()}

total_inv = sum(1/mae for mae in val_maes.values())
weights = {name: (1/mae)/total_inv for name, mae in val_maes.items()}

print("검증 세트 기반 가중치:")
for name, w in weights.items():
    print(f"  {name}: {w:.3f}")

# 064 가중 평균 앙상블
ensemble_weighted = weighted_ensemble(predictions, weights)
print(f"\n가중 평균 앙상블: MAE = {mean_absolute_error(y_test, ensemble_weighted):.2f}")

스태킹 앙상블

기본 스태킹

from sklearn.model_selection import TimeSeriesSplit

def create_stacking_features(models, X_train, y_train, X_test, n_splits=5):
    """스태킹을 위한 메타 특성 생성"""
    tscv = TimeSeriesSplit(n_splits=n_splits)

    # 학습 데이터 메타 특성
    train_meta = np.zeros((len(X_train), len(models)))

    for fold, (tr_idx, val_idx) in enumerate(tscv.split(X_train)):
        X_tr = X_train.iloc[tr_idx]
        y_tr = y_train.iloc[tr_idx]
        X_val = X_train.iloc[val_idx]

        for i, (name, model) in enumerate(models.items()):
            model_copy = type(model)(**model.get_params())
            model_copy.fit(X_tr, y_tr)
            train_meta[val_idx, i] = model_copy.predict(X_val)

    # 테스트 데이터 메타 특성
    test_meta = np.zeros((len(X_test), len(models)))

    for i, (name, model) in enumerate(models.items()):
        model.fit(X_train, y_train)
        test_meta[:, i] = model.predict(X_test)

    return train_meta, test_meta

# 064 스태킹 특성 생성
train_meta, test_meta = create_stacking_features(models, X_train, y_train, X_test)

# 064 결측치 제거 (초기 폴드)
valid_idx = ~np.any(train_meta == 0, axis=1)
train_meta_clean = train_meta[valid_idx]
y_train_clean = y_train.iloc[valid_idx]

# 064 메타 모델 학습
meta_model = Ridge(alpha=1.0)
meta_model.fit(train_meta_clean, y_train_clean)

# 064 스태킹 예측
stacking_pred = meta_model.predict(test_meta)

print(f"스태킹 앙상블: MAE = {mean_absolute_error(y_test, stacking_pred):.2f}")
print(f"\n메타 모델 가중치: {meta_model.coef_}")

FLAML을 메타 모델로 사용

# 064 FLAML을 메타 모델로
automl_meta = AutoML()
automl_meta.fit(
    train_meta_clean, y_train_clean.values,
    task="regression",
    time_budget=30,
    metric="mae",
    verbose=0
)

flaml_stacking_pred = automl_meta.predict(test_meta)
print(f"FLAML 스태킹: MAE = {mean_absolute_error(y_test, flaml_stacking_pred):.2f}")
print(f"메타 모델: {automl_meta.best_estimator}")

FLAML 다중 모델 앙상블

상위 N개 모델 앙상블

def get_top_n_models(automl, n=3):
    """FLAML에서 상위 N개 모델 추출"""
    # FLAML 검색 결과에서 상위 모델 정보 가져오기
    config_history = automl.config_history

    # 최고 성능 순 정렬
    sorted_configs = sorted(config_history.items(),
                           key=lambda x: x[1].get('val_loss', float('inf')))[:n]

    return sorted_configs

# 064 FLAML 학습
automl = AutoML()
automl.fit(
    X_train, y_train,
    task="regression",
    time_budget=60,
    metric="mae",
    verbose=0
)

print(f"FLAML 단일 최적 모델: MAE = {mean_absolute_error(y_test, automl.predict(X_test)):.2f}")

# 064 다양한 모델 유형으로 앙상블
model_types = ['lgbm', 'rf', 'xgboost', 'extra_tree']
ensemble_preds = []

for est in model_types:
    try:
        automl_single = AutoML()
        automl_single.fit(
            X_train, y_train,
            task="regression",
            time_budget=15,
            estimator_list=[est],
            metric="mae",
            verbose=0
        )
        pred = automl_single.predict(X_test)
        ensemble_preds.append(pred)
        print(f"  {est}: MAE = {mean_absolute_error(y_test, pred):.2f}")
    except Exception as e:
        print(f"  {est}: 학습 실패")

if ensemble_preds:
    multi_model_avg = np.mean(ensemble_preds, axis=0)
    print(f"\n다중 모델 평균 앙상블: MAE = {mean_absolute_error(y_test, multi_model_avg):.2f}")

시계열 전용 앙상블

예측 구간별 앙상블

def horizon_based_ensemble(models, X_test, y_test, test_dates, horizons=[7, 14, 30]):
    """예측 기간별 다른 모델 사용"""
    # 각 기간에 최적 모델 선택
    # 짧은 기간: Lag 특성 활용 모델
    # 긴 기간: 계절성 모델

    final_pred = np.zeros(len(X_test))

    for i in range(len(X_test)):
        days_ahead = (test_dates.iloc[i] - test_dates.iloc[0]).days

        if days_ahead < horizons[0]:  # 단기 (7일)
            model_pred = predictions['rf'][i]
        elif days_ahead < horizons[1]:  # 중기 (14일)
            model_pred = predictions['gbm'][i]
        else:  # 장기
            model_pred = predictions['ridge'][i]

        final_pred[i] = model_pred

    return final_pred

horizon_pred = horizon_based_ensemble(models, X_test, y_test, test['date'])
print(f"기간별 앙상블: MAE = {mean_absolute_error(y_test, horizon_pred):.2f}")

불확실성 기반 앙상블

def uncertainty_weighted_ensemble(predictions_dict, X_test, n_bootstrap=50):
    """불확실성 기반 가중 앙상블"""
    n_samples = len(X_test)
    n_models = len(predictions_dict)

    # 부트스트랩으로 예측 분산 추정
    variances = {}
    for name in predictions_dict.keys():
        variances[name] = np.var(list(predictions_dict.values()), axis=0) + 1e-6

    # 분산의 역수로 가중치
    weights = np.zeros((n_samples, n_models))
    for i, name in enumerate(predictions_dict.keys()):
        weights[:, i] = 1 / variances[name]

    # 정규화
    weights = weights / weights.sum(axis=1, keepdims=True)

    # 가중 평균
    final_pred = np.zeros(n_samples)
    for i, (name, pred) in enumerate(predictions_dict.items()):
        final_pred += pred * weights[:, i]

    return final_pred

uncertainty_pred = uncertainty_weighted_ensemble(predictions, X_test)
print(f"불확실성 기반 앙상블: MAE = {mean_absolute_error(y_test, uncertainty_pred):.2f}")

앙상블 성능 비교

# 064 모든 앙상블 방법 비교
results = {
    'Ridge': mean_absolute_error(y_test, predictions['ridge']),
    'Random Forest': mean_absolute_error(y_test, predictions['rf']),
    'GBM': mean_absolute_error(y_test, predictions['gbm']),
    'Simple Average': mean_absolute_error(y_test, ensemble_avg),
    'Weighted Average': mean_absolute_error(y_test, ensemble_weighted),
    'Stacking (Ridge)': mean_absolute_error(y_test, stacking_pred),
    'Stacking (FLAML)': mean_absolute_error(y_test, flaml_stacking_pred),
}

results_df = pd.DataFrame([
    {'Method': k, 'MAE': v} for k, v in results.items()
]).sort_values('MAE')

print("\n앙상블 방법 비교:")
print(results_df.to_string(index=False))

# 064 시각화
plt.figure(figsize=(10, 6))
colors = ['red' if 'Average' in m or 'Stacking' in m else 'blue'
          for m in results_df['Method']]
plt.barh(results_df['Method'], results_df['MAE'], color=colors)
plt.xlabel('MAE')
plt.title('Ensemble Methods Comparison (Red = Ensemble)')
plt.tight_layout()
plt.show()

실시간 앙상블 업데이트

class OnlineEnsemble:
    """온라인 학습 앙상블"""

    def __init__(self, models, decay=0.95):
        self.models = models
        self.weights = {name: 1/len(models) for name in models}
        self.decay = decay
        self.errors = {name: [] for name in models}

    def predict(self, X):
        predictions = {}
        for name, model in self.models.items():
            predictions[name] = model.predict(X)

        weighted_pred = sum(predictions[name] * self.weights[name]
                           for name in self.models)
        return weighted_pred, predictions

    def update(self, X, y_true):
        """실제값으로 가중치 업데이트"""
        _, predictions = self.predict(X)

        for name, pred in predictions.items():
            error = np.abs(y_true - pred)
            self.errors[name].append(error.mean())

        # 최근 오차 기반 가중치 업데이트
        recent_errors = {name: np.mean(errs[-10:]) if errs else 1.0
                        for name, errs in self.errors.items()}

        total_inv = sum(1/(e+1e-6) for e in recent_errors.values())
        for name in self.models:
            new_weight = (1/(recent_errors[name]+1e-6)) / total_inv
            self.weights[name] = (self.decay * self.weights[name] +
                                  (1-self.decay) * new_weight)

# 064 온라인 앙상블 시뮬레이션
online_ensemble = OnlineEnsemble(models)

online_preds = []
for i in range(0, len(X_test), 7):  # 주 단위 업데이트
    batch_X = X_test.iloc[i:i+7]
    batch_y = y_test.iloc[i:i+7]

    pred, _ = online_ensemble.predict(batch_X)
    online_preds.extend(pred)

    online_ensemble.update(batch_X, batch_y.values)

online_preds = np.array(online_preds[:len(y_test)])
print(f"\n온라인 앙상블: MAE = {mean_absolute_error(y_test, online_preds):.2f}")
print("최종 가중치:")
for name, w in online_ensemble.weights.items():
    print(f"  {name}: {w:.3f}")

앙상블 가이드

guide = {
    '상황': ['빠른 구현', '최고 성능', '해석 필요', '실시간'],
    '추천 방법': ['단순 평균', '스태킹 + FLAML', '가중 평균', '온라인 앙상블'],
    '장점': ['구현 간단', '성능 최적화', '가중치로 해석', '적응형'],
    '단점': ['최적화 없음', '복잡도 높음', '검증셋 필요', '구현 복잡']
}

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

정리

• 단순 평균: 구현 쉽고 안정적
• 가중 평균: 검증 세트로 가중치 결정
• 스태킹: 메타 모델로 결합, 최고 성능 가능
• FLAML 활용: 메타 모델 또는 다중 모델 유형
• 온라인 앙상블: 실시간 가중치 업데이트
• 다양성 있는 모델 조합이 핵심

다음 글 예고

다음 글에서는 시계열 파트 총정리를 통해 지금까지 배운 시계열 예측 기법들을 종합적으로 정리하겠습니다.

---

FLAML AutoML 마스터 시리즈 #064