본문으로 건너뛰기

060 다변량 시계열 예측

키워드: 다변량, multivariate, 외생 변수

개요​

다변량 시계열 예측은 여러 관련 변수를 함께 활용하여 예측하는 방법입니다. 날씨, 프로모션, 경제 지표 등 외부 요인을 포함하면 예측 정확도를 높일 수 있습니다.

실습 환경​

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

다변량 데이터 생성​

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from flaml import AutoML
from sklearn.metrics import mean_absolute_error, r2_score

# 060 다변량 시계열 데이터 생성
np.random.seed(42)
n_days = 730

dates = pd.date_range('2022-01-01', periods=n_days, freq='D')

# 060 기본 변수들
temperature = 20 + 10 * np.sin(np.arange(n_days) * 2 * np.pi / 365) + np.random.randn(n_days) * 3
humidity = 60 + 20 * np.sin(np.arange(n_days) * 2 * np.pi / 365 + np.pi/4) + np.random.randn(n_days) * 5
promotion = np.random.choice([0, 1], n_days, p=[0.85, 0.15])
holiday = np.zeros(n_days)
for i, date in enumerate(dates):
    if date.month == 12 and date.day >= 20:
        holiday[i] = 1
    if date.month == 1 and date.day <= 3:
        holiday[i] = 1

# 060 타겟 변수 (매출) - 다른 변수들에 의존
trend = np.linspace(1000, 1500, n_days)
temp_effect = 5 * (temperature - 20)  # 온도 효과
promo_effect = promotion * 200  # 프로모션 효과
holiday_effect = holiday * 300  # 휴일 효과
weekly = 100 * np.sin(np.arange(n_days) * 2 * np.pi / 7)  # 주간 패턴
noise = np.random.randn(n_days) * 50

sales = trend + temp_effect + promo_effect + holiday_effect + weekly + noise
sales = np.maximum(sales, 500)  # 최소 매출

# 060 DataFrame 생성
df = pd.DataFrame({
    'date': dates,
    'sales': sales,
    'temperature': temperature,
    'humidity': humidity,
    'promotion': promotion,
    'holiday': holiday
})

print("다변량 시계열 데이터:")
print(df.head())
print(f"\n데이터 shape: {df.shape}")
print(f"기간: {df['date'].min()} ~ {df['date'].max()}")

데이터 시각화​

fig, axes = plt.subplots(3, 2, figsize=(14, 12))

# 060 매출 추이
axes[0, 0].plot(df['date'], df['sales'])
axes[0, 0].set_title('Sales')

# 060 온도
axes[0, 1].plot(df['date'], df['temperature'])
axes[0, 1].set_title('Temperature')

# 060 습도
axes[1, 0].plot(df['date'], df['humidity'])
axes[1, 0].set_title('Humidity')

# 060 프로모션
axes[1, 1].scatter(df['date'], df['promotion'], s=5, alpha=0.5)
axes[1, 1].set_title('Promotion Days')

# 060 매출 vs 온도
axes[2, 0].scatter(df['temperature'], df['sales'], alpha=0.3)
axes[2, 0].set_xlabel('Temperature')
axes[2, 0].set_ylabel('Sales')
axes[2, 0].set_title('Sales vs Temperature')

# 060 프로모션 효과
promo_sales = df.groupby('promotion')['sales'].mean()
axes[2, 1].bar(['No Promo', 'Promo'], promo_sales.values)
axes[2, 1].set_title('Average Sales by Promotion')

plt.tight_layout()
plt.show()

단변량 vs 다변량 비교​

단변량 예측 (기준선)​

# 060 날짜 특성 + Lag 특성만 사용
def create_univariate_features(df):
    df = df.copy()

    # 날짜 특성
    df['dayofweek'] = df['date'].dt.dayofweek
    df['month'] = df['date'].dt.month
    df['dayofyear'] = df['date'].dt.dayofyear

    # Lag 특성
    for lag in [1, 7, 14, 30]:
        df[f'lag_{lag}'] = df['sales'].shift(lag)

    # 이동 평균
    for window in [7, 14, 30]:
        df[f'rolling_mean_{window}'] = df['sales'].shift(1).rolling(window).mean()

    return df.dropna()

df_uni = create_univariate_features(df)

# 060 분할
train_size = int(len(df_uni) * 0.8)
train_uni = df_uni.iloc[:train_size]
test_uni = df_uni.iloc[train_size:]

feature_cols_uni = [col for col in df_uni.columns if col not in ['date', 'sales', 'temperature', 'humidity', 'promotion', 'holiday']]

X_train_uni = train_uni[feature_cols_uni]
y_train_uni = train_uni['sales']
X_test_uni = test_uni[feature_cols_uni]
y_test_uni = test_uni['sales']

# 060 단변량 모델
automl_uni = AutoML()
automl_uni.fit(
    X_train_uni, y_train_uni,
    task="regression",
    time_budget=60,
    split_type="time",
    verbose=0
)

y_pred_uni = automl_uni.predict(X_test_uni)
mae_uni = mean_absolute_error(y_test_uni, y_pred_uni)

print(f"단변량 모델 MAE: {mae_uni:.2f}")

다변량 예측​

# 060 외생 변수 포함
def create_multivariate_features(df):
    df = df.copy()

    # 날짜 특성
    df['dayofweek'] = df['date'].dt.dayofweek
    df['month'] = df['date'].dt.month
    df['dayofyear'] = df['date'].dt.dayofyear

    # Lag 특성 (타겟)
    for lag in [1, 7, 14, 30]:
        df[f'lag_{lag}'] = df['sales'].shift(lag)

    # 이동 평균 (타겟)
    for window in [7, 14, 30]:
        df[f'rolling_mean_{window}'] = df['sales'].shift(1).rolling(window).mean()

    # 외생 변수 Lag (온도)
    for lag in [0, 1, 7]:
        df[f'temp_lag_{lag}'] = df['temperature'].shift(lag)

    # 외생 변수 이동 평균
    df['temp_rolling_7'] = df['temperature'].shift(1).rolling(7).mean()

    return df.dropna()

df_multi = create_multivariate_features(df)

# 060 분할
train_size = int(len(df_multi) * 0.8)
train_multi = df_multi.iloc[:train_size]
test_multi = df_multi.iloc[train_size:]

feature_cols_multi = [col for col in df_multi.columns if col not in ['date', 'sales']]

X_train_multi = train_multi[feature_cols_multi]
y_train_multi = train_multi['sales']
X_test_multi = test_multi[feature_cols_multi]
y_test_multi = test_multi['sales']

# 060 다변량 모델
automl_multi = AutoML()
automl_multi.fit(
    X_train_multi, y_train_multi,
    task="regression",
    time_budget=60,
    split_type="time",
    verbose=0
)

y_pred_multi = automl_multi.predict(X_test_multi)
mae_multi = mean_absolute_error(y_test_multi, y_pred_multi)

print(f"다변량 모델 MAE: {mae_multi:.2f}")
print(f"개선: {(mae_uni - mae_multi) / mae_uni * 100:.1f}%")

특성 중요도 분석​

# 060 다변량 모델의 특성 중요도
if hasattr(automl_multi.best_model, 'feature_importances_'):
    importance = automl_multi.best_model.feature_importances_
    importance_df = pd.DataFrame({
        'feature': feature_cols_multi,
        'importance': importance
    }).sort_values('importance', ascending=False)

    plt.figure(figsize=(10, 8))
    plt.barh(range(len(importance_df)), importance_df['importance'].values[::-1])
    plt.yticks(range(len(importance_df)), importance_df['feature'].values[::-1])
    plt.xlabel('Importance')
    plt.title('Feature Importance (Multivariate Model)')
    plt.tight_layout()
    plt.show()

    print("\n상위 10개 중요 특성:")
    print(importance_df.head(10).to_string(index=False))

외생 변수 예측 처리​

외생 변수의 미래 값​

# 060 문제: 예측 시점에 외생 변수의 미래 값이 필요
print("외생 변수 처리 전략:")
print("-" * 50)

strategies = {
    '전략': ['예측 사용', '최근 값 사용', '평균 사용', '계절적 예측'],
    '설명': [
        '외생 변수도 예측 모델 구축',
        'Lag된 외생 변수만 사용',
        '과거 평균값으로 대체',
        '작년 같은 시기 값 사용'
    ],
    '장점': [
        '가장 정확할 수 있음',
        '별도 예측 불필요',
        '간단함',
        '계절성 반영'
    ],
    '단점': [
        '추가 모델 필요',
        '정보 손실',
        '정확도 낮음',
        '작년 데이터 필요'
    ]
}

print(pd.DataFrame(strategies).to_string(index=False))

Lag 외생 변수만 사용​

# 060 Lag된 외생 변수만 사용 (데이터 누출 방지)
def create_safe_multivariate_features(df, forecast_horizon=1):
    """예측 시점에 알 수 있는 외생 변수만 사용"""
    df = df.copy()

    # 날짜 특성 (예측 시점에 알 수 있음)
    df['dayofweek'] = df['date'].dt.dayofweek
    df['month'] = df['date'].dt.month

    # 타겟 Lag
    for lag in [1, 7, 14, 30]:
        df[f'lag_{lag}'] = df['sales'].shift(lag)

    # 외생 변수는 최소 forecast_horizon 만큼 Lag
    for lag in range(forecast_horizon, forecast_horizon + 7):
        df[f'temp_lag_{lag}'] = df['temperature'].shift(lag)
        df[f'humidity_lag_{lag}'] = df['humidity'].shift(lag)

    # 프로모션과 휴일은 미리 계획된 경우 현재/미래 값도 사용 가능
    df['promotion_planned'] = df['promotion']  # 계획된 프로모션
    df['holiday_known'] = df['holiday']  # 알려진 휴일

    return df.dropna()

df_safe = create_safe_multivariate_features(df, forecast_horizon=1)
print(f"안전한 특성 데이터 shape: {df_safe.shape}")

다단계 다변량 예측​

class MultivariateForecastPipeline:
    """다변량 시계열 예측 파이프라인"""

    def __init__(self, time_budget=60):
        self.time_budget = time_budget
        self.model = None
        self.feature_cols = None

    def create_features(self, df, include_exog=True):
        df = df.copy()

        # 날짜 특성
        df['dayofweek'] = df['date'].dt.dayofweek
        df['month'] = df['date'].dt.month
        df['is_weekend'] = (df['dayofweek'] >= 5).astype(int)

        # 타겟 Lag
        for lag in [1, 7, 14, 30]:
            df[f'sales_lag_{lag}'] = df['sales'].shift(lag)

        # 이동 통계
        for window in [7, 14]:
            df[f'sales_rolling_{window}'] = df['sales'].shift(1).rolling(window).mean()

        if include_exog:
            # 외생 변수 (Lag 사용)
            for lag in [1, 2, 3, 7]:
                df[f'temp_lag_{lag}'] = df['temperature'].shift(lag)

            # 알려진 이벤트 (현재 값 사용 가능)
            df['is_promotion'] = df['promotion']
            df['is_holiday'] = df['holiday']

        return df.dropna()

    def fit(self, df):
        df_features = self.create_features(df)

        self.feature_cols = [col for col in df_features.columns
                            if col not in ['date', 'sales', 'temperature', 'humidity', 'promotion', 'holiday']]

        X = df_features[self.feature_cols]
        y = df_features['sales']

        self.model = AutoML()
        self.model.fit(
            X, y,
            task="regression",
            time_budget=self.time_budget,
            split_type="time",
            verbose=0
        )

        return self

    def predict(self, df):
        df_features = self.create_features(df)
        X = df_features[self.feature_cols]
        return self.model.predict(X)

# 060 사용
pipeline = MultivariateForecastPipeline(time_budget=60)
pipeline.fit(train_multi.assign(date=df['date'].iloc[:len(train_multi)]))

# 060 (실제 사용 시에는 predict 메서드 구현 필요)

결과 비교​

# 060 시각화
fig, axes = plt.subplots(2, 1, figsize=(14, 10))

dates_test = test_uni['date'] if 'date' in test_uni.columns else df['date'].iloc[-len(test_uni):]

# 060 예측 비교
axes[0].plot(range(len(y_test_uni)), y_test_uni.values, label='Actual', linewidth=2)
axes[0].plot(range(len(y_pred_uni)), y_pred_uni, label=f'Univariate (MAE: {mae_uni:.1f})', alpha=0.8)
axes[0].plot(range(len(y_pred_multi)), y_pred_multi, label=f'Multivariate (MAE: {mae_multi:.1f})', alpha=0.8)
axes[0].legend()
axes[0].set_title('Univariate vs Multivariate Prediction')
axes[0].set_xlabel('Time Index')
axes[0].set_ylabel('Sales')

# 060 오차 비교
errors_uni = y_test_uni.values - y_pred_uni
errors_multi = y_test_multi.values - y_pred_multi

axes[1].hist(errors_uni, bins=30, alpha=0.5, label='Univariate')
axes[1].hist(errors_multi, bins=30, alpha=0.5, label='Multivariate')
axes[1].axvline(x=0, color='r', linestyle='--')
axes[1].legend()
axes[1].set_title('Error Distribution')
axes[1].set_xlabel('Error')

plt.tight_layout()
plt.show()

# 060 요약
print("\n최종 비교:")
print("-" * 40)
print(f"단변량 MAE: {mae_uni:.2f}")
print(f"다변량 MAE: {mae_multi:.2f}")
print(f"개선율: {(mae_uni - mae_multi) / mae_uni * 100:.1f}%")
print(f"단변량 R²: {r2_score(y_test_uni, y_pred_uni):.4f}")
print(f"다변량 R²: {r2_score(y_test_multi, y_pred_multi):.4f}")

정리​

  • 다변량 시계열: 타겟 외에 관련 변수를 함께 활용
  • 외생 변수: 온도, 프로모션, 휴일 등 외부 요인
  • 데이터 누출 주의: 외생 변수도 적절한 Lag 사용
  • 계획 가능한 변수: 프로모션, 휴일은 미래 값 사용 가능
  • 외생 변수 추가로 예측 정확도 향상 기대

다음 글 예고​

다음 글에서는 시계열 프로젝트 - 주가 예측에 대해 알아보겠습니다. 실제 금융 시계열 예측 프로젝트를 진행합니다.

FLAML AutoML 마스터 시리즈 #060