058 시계열 특성 엔지니어링

키워드: 특성 엔지니어링, feature engineering, lag 특성

개요

시계열 예측에서 특성 엔지니어링은 모델 성능에 결정적인 영향을 미칩니다. 이 글에서는 시계열 데이터에서 유용한 특성을 추출하는 다양한 기법을 알아봅니다.

실습 환경

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

pip install pandas numpy scikit-learn matplotlib

기본 데이터 준비

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

# 058 복잡한 패턴을 가진 시계열
np.random.seed(42)
n_days = 730

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

# 058 다중 패턴
trend = np.linspace(100, 200, n_days)
weekly = 15 * np.sin(np.arange(n_days) * 2 * np.pi / 7)
yearly = 30 * np.sin(np.arange(n_days) * 2 * np.pi / 365)
monthly = 10 * np.sin(np.arange(n_days) * 2 * np.pi / 30)
noise = np.random.randn(n_days) * 10

# 058 특별 이벤트 효과
event_effect = np.zeros(n_days)
for i, date in enumerate(dates):
    if date.month == 12:  # 12월 프리미엄
        event_effect[i] = 30
    if date.dayofweek >= 5:  # 주말
        event_effect[i] += 10

values = trend + weekly + yearly + monthly + noise + event_effect

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

print("원본 데이터:")
print(df.head())

날짜 기반 특성

기본 날짜 특성

def add_date_features(df, date_col='date'):
    """기본 날짜 특성 추가"""
    df = df.copy()

    df['year'] = df[date_col].dt.year
    df['month'] = df[date_col].dt.month
    df['day'] = df[date_col].dt.day
    df['dayofweek'] = df[date_col].dt.dayofweek
    df['dayofyear'] = df[date_col].dt.dayofyear
    df['weekofyear'] = df[date_col].dt.isocalendar().week.astype(int)
    df['quarter'] = df[date_col].dt.quarter
    df['is_month_start'] = df[date_col].dt.is_month_start.astype(int)
    df['is_month_end'] = df[date_col].dt.is_month_end.astype(int)
    df['is_quarter_start'] = df[date_col].dt.is_quarter_start.astype(int)
    df['is_quarter_end'] = df[date_col].dt.is_quarter_end.astype(int)
    df['is_year_start'] = df[date_col].dt.is_year_start.astype(int)
    df['is_year_end'] = df[date_col].dt.is_year_end.astype(int)

    return df

df_features = add_date_features(df)
print("\n날짜 특성:")
print(df_features[['date', 'month', 'dayofweek', 'quarter']].head())

주기적 인코딩

def add_cyclical_features(df, date_col='date'):
    """주기적 특성 (사인/코사인 인코딩)"""
    df = df.copy()

    # 월 (12개월 주기)
    df['month_sin'] = np.sin(2 * np.pi * df[date_col].dt.month / 12)
    df['month_cos'] = np.cos(2 * np.pi * df[date_col].dt.month / 12)

    # 요일 (7일 주기)
    df['dow_sin'] = np.sin(2 * np.pi * df[date_col].dt.dayofweek / 7)
    df['dow_cos'] = np.cos(2 * np.pi * df[date_col].dt.dayofweek / 7)

    # 일 (월별, 대략 30일 주기)
    df['day_sin'] = np.sin(2 * np.pi * df[date_col].dt.day / 31)
    df['day_cos'] = np.cos(2 * np.pi * df[date_col].dt.day / 31)

    # 연중 일 (365일 주기)
    df['doy_sin'] = np.sin(2 * np.pi * df[date_col].dt.dayofyear / 365)
    df['doy_cos'] = np.cos(2 * np.pi * df[date_col].dt.dayofyear / 365)

    return df

df_features = add_cyclical_features(df_features)

# 058 시각화
fig, axes = plt.subplots(2, 2, figsize=(12, 8))

axes[0, 0].scatter(df_features['month_sin'], df_features['month_cos'], c=df_features['month'], cmap='hsv')
axes[0, 0].set_title('Month Cyclical Encoding')
axes[0, 0].set_xlabel('sin(month)')
axes[0, 0].set_ylabel('cos(month)')

axes[0, 1].scatter(df_features['dow_sin'], df_features['dow_cos'], c=df_features['dayofweek'], cmap='hsv')
axes[0, 1].set_title('Day of Week Cyclical Encoding')

axes[1, 0].plot(df_features['month_sin'][:365], label='sin')
axes[1, 0].plot(df_features['month_cos'][:365], label='cos')
axes[1, 0].set_title('Month Encoding over Year')
axes[1, 0].legend()

axes[1, 1].plot(df_features['dow_sin'][:30], label='sin')
axes[1, 1].plot(df_features['dow_cos'][:30], label='cos')
axes[1, 1].set_title('Day of Week Encoding over Month')
axes[1, 1].legend()

plt.tight_layout()
plt.show()

비즈니스 특성

def add_business_features(df, date_col='date'):
    """비즈니스 관련 특성"""
    df = df.copy()

    # 주말/평일
    df['is_weekend'] = (df[date_col].dt.dayofweek >= 5).astype(int)

    # 월초/월말 (급여일 등)
    df['is_payday'] = ((df[date_col].dt.day == 25) |
                       (df[date_col].dt.day == 10)).astype(int)

    # 특정 계절
    df['is_summer'] = df[date_col].dt.month.isin([6, 7, 8]).astype(int)
    df['is_winter'] = df[date_col].dt.month.isin([12, 1, 2]).astype(int)

    # 연말/연초
    df['is_year_end_period'] = ((df[date_col].dt.month == 12) &
                                 (df[date_col].dt.day >= 20)).astype(int)

    return df

df_features = add_business_features(df_features)
print("\n비즈니스 특성:")
print(df_features[['date', 'is_weekend', 'is_payday', 'is_summer']].head(30))

Lag 특성

기본 Lag 특성

def add_lag_features(df, target_col='value', lags=[1, 2, 3, 7, 14, 21, 28, 30, 60, 90, 365]):
    """Lag 특성 추가"""
    df = df.copy()

    for lag in lags:
        df[f'lag_{lag}'] = df[target_col].shift(lag)

    return df

df_features = add_lag_features(df_features)
print("\nLag 특성:")
print(df_features[['date', 'value', 'lag_1', 'lag_7', 'lag_30']].head(35))

Lag 차이 특성

def add_lag_diff_features(df, target_col='value'):
    """Lag 차이 특성"""
    df = df.copy()

    # 전일 대비 변화
    df['diff_1'] = df[target_col].diff(1)

    # 전주 대비 변화
    df['diff_7'] = df[target_col].diff(7)

    # 전월 대비 변화
    df['diff_30'] = df[target_col].diff(30)

    # 전년 대비 변화
    df['diff_365'] = df[target_col].diff(365)

    # 변화율
    df['pct_change_1'] = df[target_col].pct_change(1)
    df['pct_change_7'] = df[target_col].pct_change(7)

    return df

df_features = add_lag_diff_features(df_features)
print("\nLag 차이 특성:")
print(df_features[['date', 'value', 'diff_1', 'diff_7', 'pct_change_1']].iloc[30:35])

이동 통계 특성

이동 평균과 표준편차

def add_rolling_features(df, target_col='value', windows=[3, 7, 14, 30, 60, 90]):
    """이동 통계 특성"""
    df = df.copy()

    for window in windows:
        # 평균 (과거 데이터만 사용하기 위해 shift)
        df[f'rolling_mean_{window}'] = df[target_col].shift(1).rolling(window=window).mean()
        df[f'rolling_std_{window}'] = df[target_col].shift(1).rolling(window=window).std()
        df[f'rolling_min_{window}'] = df[target_col].shift(1).rolling(window=window).min()
        df[f'rolling_max_{window}'] = df[target_col].shift(1).rolling(window=window).max()
        df[f'rolling_median_{window}'] = df[target_col].shift(1).rolling(window=window).median()

    return df

df_features = add_rolling_features(df_features)
print("\n이동 통계 특성:")
print(df_features[['date', 'value', 'rolling_mean_7', 'rolling_std_7']].iloc[30:35])

지수 이동 평균

def add_ewm_features(df, target_col='value', spans=[7, 14, 30]):
    """지수 이동 평균 특성"""
    df = df.copy()

    for span in spans:
        df[f'ewm_mean_{span}'] = df[target_col].shift(1).ewm(span=span).mean()
        df[f'ewm_std_{span}'] = df[target_col].shift(1).ewm(span=span).std()

    return df

df_features = add_ewm_features(df_features)
print("\n지수 이동 평균:")
print(df_features[['date', 'rolling_mean_7', 'ewm_mean_7']].iloc[30:35])

고급 특성

추세 관련 특성

def add_trend_features(df, target_col='value', window=30):
    """추세 관련 특성"""
    df = df.copy()

    # 현재값과 이동평균의 차이
    rolling_mean = df[target_col].shift(1).rolling(window=window).mean()
    df['deviation_from_mean'] = df[target_col].shift(1) - rolling_mean

    # 이동평균 대비 비율
    df['ratio_to_mean'] = df[target_col].shift(1) / rolling_mean

    # 추세 기울기 (간단 버전)
    df['trend_slope'] = df[target_col].shift(1).rolling(window=window).apply(
        lambda x: np.polyfit(range(len(x)), x, 1)[0] if len(x) == window else np.nan
    )

    return df

df_features = add_trend_features(df_features)
print("\n추세 특성:")
print(df_features[['date', 'deviation_from_mean', 'ratio_to_mean', 'trend_slope']].iloc[60:65])

통계적 특성

def add_statistical_features(df, target_col='value', window=30):
    """통계적 특성"""
    df = df.copy()

    rolling = df[target_col].shift(1).rolling(window=window)

    # 왜도
    df['rolling_skew'] = rolling.skew()

    # 첨도
    df['rolling_kurt'] = rolling.kurt()

    # 범위
    df['rolling_range'] = rolling.max() - rolling.min()

    # 변동계수
    df['rolling_cv'] = rolling.std() / rolling.mean()

    return df

df_features = add_statistical_features(df_features)

전체 특성 파이프라인

def create_all_ts_features(df, target_col='value', date_col='date'):
    """전체 시계열 특성 생성"""
    df = df.copy()

    # 날짜 특성
    df = add_date_features(df, date_col)
    df = add_cyclical_features(df, date_col)
    df = add_business_features(df, date_col)

    # Lag 특성
    df = add_lag_features(df, target_col, lags=[1, 2, 3, 7, 14, 21, 28, 30])
    df = add_lag_diff_features(df, target_col)

    # 이동 통계
    df = add_rolling_features(df, target_col, windows=[7, 14, 30])
    df = add_ewm_features(df, target_col, spans=[7, 14, 30])

    # 고급 특성
    df = add_trend_features(df, target_col, window=14)
    df = add_statistical_features(df, target_col, window=14)

    # 결측치 제거
    df = df.dropna()

    return df

df_final = create_all_ts_features(df)
print(f"\n최종 데이터 shape: {df_final.shape}")
print(f"생성된 특성 수: {df_final.shape[1] - 2}")  # date, value 제외

# 058 특성 목록
feature_cols = [col for col in df_final.columns if col not in ['date', 'value']]
print(f"\n특성 목록 ({len(feature_cols)}개):")
for i, col in enumerate(feature_cols[:20], 1):
    print(f"  {i}. {col}")
print("  ...")

특성 중요도 확인

from flaml import AutoML
from sklearn.model_selection import train_test_split

# 058 데이터 분할
train_size = int(len(df_final) * 0.8)
train = df_final.iloc[:train_size]
test = df_final.iloc[train_size:]

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

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

# 058 특성 중요도
if hasattr(automl.best_model, 'feature_importances_'):
    importance = automl.best_model.feature_importances_
    importance_df = pd.DataFrame({
        'feature': feature_cols,
        'importance': importance
    }).sort_values('importance', ascending=False)

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

정리

• 날짜 특성: 월, 요일, 분기 등 + 사인/코사인 인코딩
• Lag 특성: 과거 값 참조 (데이터 누출 주의)
• 이동 통계: 평균, 표준편차, 최대/최소
• 추세 특성: 이동평균 대비 편차, 기울기
• 비즈니스 특성: 주말, 휴일, 계절 등
• 특성 생성 시 shift(1) 사용하여 데이터 누출 방지

다음 글 예고

다음 글에서는 시계열 교차 검증에 대해 알아보겠습니다. 시계열 특성에 맞는 교차 검증 전략을 다룹니다.

---

FLAML AutoML 마스터 시리즈 #058