059 시계열 교차 검증

키워드: 교차 검증, time series split, walk-forward

개요

시계열 데이터에서는 일반적인 K-Fold 교차 검증을 사용할 수 없습니다. 미래 데이터가 학습에 포함되면 데이터 누출이 발생하기 때문입니다. 이 글에서는 시계열에 적합한 교차 검증 전략을 알아봅니다.

실습 환경

• Python 버전: 3.11 권장
• 필요 패키지: scikit-learn, pandas, numpy

pip install scikit-learn pandas numpy matplotlib

왜 K-Fold를 사용하면 안 되는가?

데이터 누출 문제

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from sklearn.model_selection import KFold, TimeSeriesSplit

# 059 시계열 데이터
np.random.seed(42)
n = 100
dates = pd.date_range('2023-01-01', periods=n, freq='D')
values = np.cumsum(np.random.randn(n)) + np.arange(n) * 0.1

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

# 059 K-Fold 시각화
fig, axes = plt.subplots(2, 1, figsize=(14, 8))

# 059 K-Fold (잘못된 방법)
kfold = KFold(n_splits=5, shuffle=True, random_state=42)

ax = axes[0]
for fold, (train_idx, val_idx) in enumerate(kfold.split(df)):
    ax.scatter(train_idx, [fold] * len(train_idx), c='blue', alpha=0.5, s=10, label='Train' if fold == 0 else '')
    ax.scatter(val_idx, [fold] * len(val_idx), c='red', s=20, label='Val' if fold == 0 else '')
ax.set_title('K-Fold (Wrong for Time Series) - Future data in training!')
ax.set_xlabel('Time Index')
ax.set_ylabel('Fold')
ax.legend()

# 059 TimeSeriesSplit (올바른 방법)
tscv = TimeSeriesSplit(n_splits=5)

ax = axes[1]
for fold, (train_idx, val_idx) in enumerate(tscv.split(df)):
    ax.scatter(train_idx, [fold] * len(train_idx), c='blue', alpha=0.5, s=10, label='Train' if fold == 0 else '')
    ax.scatter(val_idx, [fold] * len(val_idx), c='red', s=20, label='Val' if fold == 0 else '')
ax.set_title('TimeSeriesSplit (Correct) - Only past data in training')
ax.set_xlabel('Time Index')
ax.set_ylabel('Fold')
ax.legend()

plt.tight_layout()
plt.show()

print("K-Fold 문제점:")
print("  - 미래 데이터(빨간점 뒤의 파란점)가 학습에 포함됨")
print("  - 실제 예측 상황과 다름")
print("  - 과적합된 성능 추정")

TimeSeriesSplit

기본 사용법

from sklearn.model_selection import TimeSeriesSplit

# 059 TimeSeriesSplit 상세
tscv = TimeSeriesSplit(n_splits=5)

print("TimeSeriesSplit 분할:")
print("-" * 50)
for fold, (train_idx, val_idx) in enumerate(tscv.split(df), 1):
    train_start, train_end = train_idx[0], train_idx[-1]
    val_start, val_end = val_idx[0], val_idx[-1]

    print(f"Fold {fold}:")
    print(f"  Train: index {train_start}-{train_end} ({len(train_idx)} samples)")
    print(f"  Val:   index {val_start}-{val_end} ({len(val_idx)} samples)")

고정 테스트 크기

# 059 테스트 크기 고정
tscv_fixed = TimeSeriesSplit(n_splits=5, test_size=10)

print("\n고정 테스트 크기 (test_size=10):")
for fold, (train_idx, val_idx) in enumerate(tscv_fixed.split(df), 1):
    print(f"  Fold {fold}: Train={len(train_idx)}, Val={len(val_idx)}")

Gap 추가

# 059 Gap (예측과 학습 사이 간격)
tscv_gap = TimeSeriesSplit(n_splits=5, test_size=10, gap=5)

print("\nGap 포함 (gap=5):")
for fold, (train_idx, val_idx) in enumerate(tscv_gap.split(df), 1):
    train_end = train_idx[-1]
    val_start = val_idx[0]
    print(f"  Fold {fold}: Train ends at {train_end}, Val starts at {val_start}, Gap={val_start-train_end-1}")

Walk-Forward Validation

확장 윈도우 (Expanding Window)

def expanding_window_cv(n_samples, n_splits=5, min_train_size=50, test_size=10):
    """확장 윈도우 교차 검증"""
    splits = []

    # 각 폴드의 테스트 시작 위치
    test_start_positions = np.linspace(min_train_size, n_samples - test_size, n_splits, dtype=int)

    for test_start in test_start_positions:
        train_idx = np.arange(0, test_start)
        test_idx = np.arange(test_start, min(test_start + test_size, n_samples))
        splits.append((train_idx, test_idx))

    return splits

# 059 확장 윈도우
expanding_splits = expanding_window_cv(len(df), n_splits=5, min_train_size=30, test_size=10)

print("Expanding Window:")
for fold, (train_idx, val_idx) in enumerate(expanding_splits, 1):
    print(f"  Fold {fold}: Train={len(train_idx)}, Val={len(val_idx)}")

# 059 시각화
fig, ax = plt.subplots(figsize=(14, 4))
for fold, (train_idx, val_idx) in enumerate(expanding_splits):
    ax.barh(fold, len(train_idx), left=0, color='blue', alpha=0.6)
    ax.barh(fold, len(val_idx), left=train_idx[-1]+1, color='red', alpha=0.6)
ax.set_xlabel('Time Index')
ax.set_ylabel('Fold')
ax.set_title('Expanding Window CV')
plt.tight_layout()
plt.show()

슬라이딩 윈도우 (Sliding Window)

def sliding_window_cv(n_samples, n_splits=5, train_size=50, test_size=10, step=10):
    """슬라이딩 윈도우 교차 검증"""
    splits = []

    for i in range(n_splits):
        train_start = i * step
        train_end = train_start + train_size
        test_start = train_end
        test_end = test_start + test_size

        if test_end > n_samples:
            break

        train_idx = np.arange(train_start, train_end)
        test_idx = np.arange(test_start, test_end)
        splits.append((train_idx, test_idx))

    return splits

# 059 슬라이딩 윈도우
sliding_splits = sliding_window_cv(len(df), n_splits=5, train_size=50, test_size=10, step=10)

print("\nSliding Window:")
for fold, (train_idx, val_idx) in enumerate(sliding_splits, 1):
    print(f"  Fold {fold}: Train={train_idx[0]}-{train_idx[-1]}, Val={val_idx[0]}-{val_idx[-1]}")

# 059 시각화
fig, ax = plt.subplots(figsize=(14, 4))
for fold, (train_idx, val_idx) in enumerate(sliding_splits):
    ax.barh(fold, len(train_idx), left=train_idx[0], color='blue', alpha=0.6)
    ax.barh(fold, len(val_idx), left=val_idx[0], color='red', alpha=0.6)
ax.set_xlabel('Time Index')
ax.set_ylabel('Fold')
ax.set_title('Sliding Window CV')
plt.tight_layout()
plt.show()

FLAML에서 시계열 CV

split_type 설정

from flaml import AutoML
from sklearn.datasets import make_regression

# 059 시계열 특성이 있는 데이터 (Lag 특성 포함)
X = df[['value']].copy()
X['lag_1'] = df['value'].shift(1)
X['lag_7'] = df['value'].shift(7)
X = X.dropna()
y = df['value'].iloc[7:].values

# 059 FLAML with time split
automl = AutoML()
automl.fit(
    X, y,
    task="regression",
    time_budget=30,
    split_type="time",  # 시계열 분할
    n_splits=5,
    metric="mae",
    verbose=1
)

print(f"\nFLAML 최적 모델: {automl.best_estimator}")

커스텀 CV 사용

from sklearn.model_selection import cross_val_score
from sklearn.ensemble import RandomForestRegressor

# 059 커스텀 CV로 평가
model = RandomForestRegressor(n_estimators=50, random_state=42)

# 059 TimeSeriesSplit 사용
tscv = TimeSeriesSplit(n_splits=5)
scores = cross_val_score(model, X, y, cv=tscv, scoring='neg_mean_absolute_error')

print("\nTimeSeriesSplit CV 결과:")
print(f"  MAE per fold: {-scores}")
print(f"  Mean MAE: {-scores.mean():.4f} (±{scores.std():.4f})")

CV 전략 비교

from sklearn.linear_model import Ridge

# 059 다양한 CV 전략 비교
strategies = {
    'K-Fold (wrong)': KFold(n_splits=5, shuffle=True, random_state=42),
    'TimeSeriesSplit': TimeSeriesSplit(n_splits=5),
    'TimeSeriesSplit (gap=5)': TimeSeriesSplit(n_splits=5, gap=5),
}

model = Ridge()
results = {}

for name, cv in strategies.items():
    try:
        scores = cross_val_score(model, X, y, cv=cv, scoring='neg_mean_absolute_error')
        results[name] = {
            'mean': -scores.mean(),
            'std': scores.std()
        }
    except Exception as e:
        results[name] = {'mean': None, 'std': None, 'error': str(e)}

print("\nCV 전략 비교:")
print("-" * 50)
for name, res in results.items():
    if res['mean']:
        print(f"{name}:")
        print(f"  MAE: {res['mean']:.4f} (±{res['std']:.4f})")
    else:
        print(f"{name}: Error - {res.get('error')}")

CV 선택 가이드

guide = {
    '상황': ['일반 시계열', '긴 시계열', '짧은 시계열', '개념 변화 우려'],
    'CV 방법': ['TimeSeriesSplit', 'Sliding Window', 'Expanding Window', 'Sliding Window (작은 window)'],
    'n_splits': ['5-10', '5-10', '3-5', '5-10'],
    '이유': [
        '표준 접근법',
        '계산 효율성, 최신 패턴 중요',
        '데이터 최대 활용',
        '최신 데이터에 집중'
    ]
}

print("\nCV 선택 가이드:")
print(pd.DataFrame(guide).to_string(index=False))

정리

• K-Fold는 시계열에 부적합: 미래 데이터 누출
• TimeSeriesSplit: sklearn 기본 제공, 확장 윈도우 방식
• Sliding Window: 고정 크기 학습 윈도우, 개념 변화에 강건
• Gap 설정: 예측과 학습 데이터 사이 간격으로 현실적 평가
• FLAML: split_type="time" 설정

다음 글 예고

다음 글에서는 다변량 시계열 예측에 대해 알아보겠습니다. 여러 변수를 활용한 시계열 예측 방법을 다룹니다.

---

FLAML AutoML 마스터 시리즈 #059