콘텐츠로 이동
Data Prep
상세

Time Series Decomposition

시계열 분해 기법과 실무 적용 가이드.

개요

시계열 분해(Decomposition)는 시계열 데이터를 구성 요소(추세, 계절성, 잔차)로 분리하여 패턴을 이해하고 예측에 활용하는 기법이다.

구분 설명
목적 시계열 구성 요소 분리
주요 성분 Trend, Seasonality, Residual
적용 분야 수요 예측, 재고 관리, 이상 탐지
대표 기법 Classical, STL, MSTL, Prophet

기본 모델

덧셈 모델 (Additive)

\[Y_t = T_t + S_t + R_t\]
  • 계절 변동폭이 일정할 때 사용
  • 예: 일정한 범위의 온도 변화

곱셈 모델 (Multiplicative)

\[Y_t = T_t \times S_t \times R_t\]
  • 계절 변동폭이 추세에 비례할 때 사용
  • 예: 매출 증가에 따른 계절 변동 확대

모델 선택 기준

┌─────────────────────────────────────────────────────────┐
│                모델 선택 가이드                          │
├─────────────────────────────────────────────────────────┤
│                                                         │
│  시계열 데이터 확인                                     │
│         │                                               │
│         ▼                                               │
│  계절 변동폭이 추세와 함께 변화하는가?                   │
│         │                                               │
│     ┌───┴───┐                                          │
│     │       │                                          │
│    Yes     No                                          │
│     │       │                                          │
│     ▼       ▼                                          │
│  곱셈 모델  덧셈 모델                                   │
│                                                         │
│  또는 log 변환 후 덧셈 모델 적용                        │
│  log(Y) = log(T) + log(S) + log(R)                     │
│                                                         │
└─────────────────────────────────────────────────────────┘

분해 기법

1. Classical Decomposition

가장 기본적인 이동평균 기반 분해:

from statsmodels.tsa.seasonal import seasonal_decompose
import pandas as pd
import matplotlib.pyplot as plt

# 데이터 로드 (월별 항공 승객 수)
df = pd.read_csv('airline_passengers.csv', parse_dates=['Month'], index_col='Month')

# 덧셈 분해
result_add = seasonal_decompose(df['Passengers'], model='additive', period=12)

# 곱셈 분해
result_mul = seasonal_decompose(df['Passengers'], model='multiplicative', period=12)

# 시각화
fig, axes = plt.subplots(4, 1, figsize=(12, 10))
result_mul.observed.plot(ax=axes[0], title='Original')
result_mul.trend.plot(ax=axes[1], title='Trend')
result_mul.seasonal.plot(ax=axes[2], title='Seasonal')
result_mul.resid.plot(ax=axes[3], title='Residual')
plt.tight_layout()

한계: - 시작/끝 부분 결측 (이동평균으로 인해) - 단일 계절성만 처리 - Robust하지 않음 (이상치에 민감)

2. STL (Seasonal and Trend decomposition using Loess)

LOESS 평활화를 사용한 유연한 분해:

from statsmodels.tsa.seasonal import STL

# STL 분해
stl = STL(df['Passengers'], period=12, robust=True)
result = stl.fit()

# 결과 확인
print(f"Trend: {result.trend.head()}")
print(f"Seasonal: {result.seasonal.head()}")
print(f"Residual: {result.resid.head()}")

# 시각화
result.plot()
plt.show()

STL 파라미터:

파라미터 설명 권장값
period 계절 주기 데이터 특성에 따라
seasonal 계절 평활 윈도우 7 (홀수)
trend 추세 평활 윈도우 자동 계산
robust 이상치 robust True
seasonal_deg LOESS 차수 1

3. MSTL (Multiple Seasonal-Trend decomposition)

다중 계절성 처리:

from statsmodels.tsa.seasonal import MSTL

# 시간별 데이터: 일간(24) + 주간(24*7) 계절성
mstl = MSTL(hourly_data, periods=[24, 24*7])
result = mstl.fit()

# 각 계절성 확인
print(result.seasonal.shape)  # (n, 2) - 두 개의 계절성

# 시각화
fig, axes = plt.subplots(4, 1, figsize=(12, 12))
axes[0].plot(result.observed, label='Observed')
axes[1].plot(result.trend, label='Trend', color='red')
axes[2].plot(result.seasonal['seasonal_24'], label='Daily Seasonal')
axes[3].plot(result.seasonal['seasonal_168'], label='Weekly Seasonal')
plt.tight_layout()

4. Prophet

Facebook의 시계열 분해 + 예측 프레임워크:

from prophet import Prophet

# 데이터 준비 (ds, y 컬럼 필요)
df_prophet = df.reset_index().rename(columns={'Month': 'ds', 'Passengers': 'y'})

# 모델 학습
model = Prophet(
    yearly_seasonality=True,
    weekly_seasonality=False,
    daily_seasonality=False
)
model.fit(df_prophet)

# 성분 분해 시각화
fig = model.plot_components(model.predict(df_prophet))

Prophet 성분: - trend: 비선형 추세 (changepoint 포함) - yearly: 연간 계절성 (푸리에 급수) - weekly: 주간 계절성 - holidays: 공휴일 효과

5. 푸리에 기반 분해

import numpy as np
from scipy import fft

def fourier_decomposition(signal, n_components=10):
    """푸리에 변환 기반 분해"""
    # FFT
    freqs = fft.fft(signal)
    n = len(signal)

    # 주요 주파수 성분 추출
    magnitudes = np.abs(freqs)
    top_indices = np.argsort(magnitudes)[-n_components:]

    # 재구성
    reconstructed = np.zeros(n, dtype=complex)
    for idx in top_indices:
        reconstructed[idx] = freqs[idx]

    # Inverse FFT
    seasonal = np.real(fft.ifft(reconstructed))
    trend = signal - seasonal

    return trend, seasonal

# 적용
trend, seasonal = fourier_decomposition(df['Passengers'].values, n_components=5)

성분별 분석

Trend 분석

def analyze_trend(trend_component):
    """추세 성분 분석"""
    analysis = {}

    # 추세 방향
    slope = np.polyfit(range(len(trend_component)), trend_component, 1)[0]
    analysis['direction'] = 'increasing' if slope > 0 else 'decreasing'
    analysis['slope'] = slope

    # 추세 강도 (전체 분산 중 추세가 설명하는 비율)
    total_var = np.var(trend_component)
    analysis['trend_strength'] = total_var / np.var(df['Passengers'])

    # 변화점 탐지
    analysis['changepoints'] = detect_changepoints(trend_component)

    return analysis

def detect_changepoints(trend, threshold=2):
    """변화점 탐지 (CUSUM 기반)"""
    diff = np.diff(trend)
    z_scores = (diff - np.mean(diff)) / np.std(diff)
    changepoints = np.where(np.abs(z_scores) > threshold)[0]
    return changepoints

Seasonality 분석

def analyze_seasonality(seasonal_component, period):
    """계절성 분석"""
    analysis = {}

    # 계절 패턴 요약
    seasonal_pattern = seasonal_component[:period]
    analysis['pattern'] = seasonal_pattern
    analysis['peak'] = np.argmax(seasonal_pattern)
    analysis['trough'] = np.argmin(seasonal_pattern)
    analysis['amplitude'] = np.max(seasonal_pattern) - np.min(seasonal_pattern)

    # 계절성 강도
    # F = max(0, 1 - Var(R) / Var(S+R))
    resid_var = np.var(result.resid)
    seasonal_resid_var = np.var(seasonal_component + result.resid)
    analysis['strength'] = max(0, 1 - resid_var / seasonal_resid_var)

    return analysis

Residual 분석

from scipy import stats

def analyze_residuals(residuals):
    """잔차 분석 - 모델 적합도 확인"""
    analysis = {}

    # 정규성 검정
    stat, p_value = stats.shapiro(residuals.dropna())
    analysis['normality'] = {'statistic': stat, 'p_value': p_value}

    # 자기상관 검정 (Ljung-Box)
    from statsmodels.stats.diagnostic import acorr_ljungbox
    lb_result = acorr_ljungbox(residuals.dropna(), lags=10)
    analysis['autocorrelation'] = lb_result

    # 이상치 탐지
    z_scores = np.abs(stats.zscore(residuals.dropna()))
    analysis['outliers'] = np.where(z_scores > 3)[0]

    # 이분산성 확인
    residuals_clean = residuals.dropna()
    first_half = residuals_clean[:len(residuals_clean)//2]
    second_half = residuals_clean[len(residuals_clean)//2:]
    analysis['variance_change'] = np.var(second_half) / np.var(first_half)

    return analysis

실무 적용

1. 수요 예측

def demand_forecast_with_decomposition(sales_data, forecast_horizon=12):
    """분해 기반 수요 예측"""
    # STL 분해
    stl = STL(sales_data, period=12, robust=True)
    result = stl.fit()

    # 추세 예측 (선형 외삽)
    trend = result.trend.dropna()
    x = np.arange(len(trend))
    slope, intercept = np.polyfit(x, trend, 1)

    future_x = np.arange(len(trend), len(trend) + forecast_horizon)
    future_trend = slope * future_x + intercept

    # 계절성 반복
    seasonal = result.seasonal[-12:].values
    future_seasonal = np.tile(seasonal, forecast_horizon // 12 + 1)[:forecast_horizon]

    # 결합
    forecast = future_trend + future_seasonal

    return forecast, {'trend': future_trend, 'seasonal': future_seasonal}

2. 이상 탐지

def detect_anomalies_decomposition(data, period=12, threshold=3):
    """분해 기반 이상 탐지"""
    # 분해
    stl = STL(data, period=period, robust=True)
    result = stl.fit()

    # 잔차 기반 이상치 탐지
    residuals = result.resid.dropna()
    z_scores = np.abs((residuals - residuals.mean()) / residuals.std())

    anomalies = data.index[z_scores > threshold]

    return anomalies, {
        'residuals': residuals,
        'z_scores': z_scores,
        'threshold': threshold
    }

# 사용
anomalies, info = detect_anomalies_decomposition(df['Sales'])
print(f"이상치 탐지: {len(anomalies)}개")

3. 계절 조정 (Seasonal Adjustment)

def seasonal_adjustment(data, period=12):
    """계절 조정된 시계열 생성"""
    stl = STL(data, period=period)
    result = stl.fit()

    # 계절성 제거
    seasonally_adjusted = data - result.seasonal

    # 또는 곱셈 모델의 경우
    # seasonally_adjusted = data / result.seasonal

    return seasonally_adjusted

# YoY 비교에 활용
adjusted = seasonal_adjustment(df['Revenue'])
yoy_growth = adjusted.pct_change(periods=12)

4. What-If 분석

def whatif_analysis(data, scenario='optimistic', growth_rate=0.1):
    """시나리오 기반 분석"""
    stl = STL(data, period=12)
    result = stl.fit()

    scenarios = {}

    # 기본 추세 외삽
    trend = result.trend.dropna()
    current_level = trend.iloc[-1]

    # 시나리오별 추세 조정
    if scenario == 'optimistic':
        future_trend = current_level * (1 + growth_rate) ** np.arange(1, 13)
    elif scenario == 'pessimistic':
        future_trend = current_level * (1 - growth_rate/2) ** np.arange(1, 13)
    else:  # baseline
        slope = np.polyfit(range(len(trend)), trend, 1)[0]
        future_trend = current_level + slope * np.arange(1, 13)

    # 계절성 추가
    seasonal = result.seasonal[-12:].values
    forecast = future_trend + seasonal

    return forecast

고급 기법

계절성 강도 자동 탐지

from scipy.signal import periodogram

def detect_seasonality(data, max_period=365):
    """계절성 주기 자동 탐지"""
    # 주파수 분석
    freqs, power = periodogram(data.dropna())

    # 상위 주기 추출
    periods = 1 / freqs[1:]  # 0 제외
    power = power[1:]

    # 유의미한 주기 필터링
    significant_periods = []
    threshold = np.mean(power) + 2 * np.std(power)

    for p, pw in zip(periods, power):
        if pw > threshold and p < max_period and p > 2:
            significant_periods.append((int(round(p)), pw))

    return sorted(significant_periods, key=lambda x: x[1], reverse=True)

# 사용
periods = detect_seasonality(df['Sales'])
print(f"탐지된 계절성: {periods}")
# 예: [(7, 0.45), (30, 0.32), (365, 0.28)]

동적 계절성

def dynamic_seasonality_analysis(data, period=12, window_size=36):
    """시간에 따른 계절성 변화 분석"""
    seasonal_patterns = []

    for start in range(0, len(data) - window_size, period):
        window = data[start:start + window_size]

        stl = STL(window, period=period)
        result = stl.fit()

        seasonal_patterns.append({
            'start': data.index[start],
            'pattern': result.seasonal[:period].values,
            'amplitude': result.seasonal.max() - result.seasonal.min()
        })

    return pd.DataFrame(seasonal_patterns)

# 계절성 변화 추세 확인
dynamic = dynamic_seasonality_analysis(df['Sales'])
plt.plot(dynamic['start'], dynamic['amplitude'])
plt.title('Seasonal Amplitude Over Time')

시각화 모범 사례

def plot_decomposition_dashboard(data, result, period=12):
    """분해 결과 대시보드"""
    fig, axes = plt.subplots(2, 2, figsize=(14, 10))

    # 1. 원본 + 추세
    axes[0, 0].plot(data.index, data.values, label='Original', alpha=0.7)
    axes[0, 0].plot(result.trend.index, result.trend.values, label='Trend', linewidth=2)
    axes[0, 0].set_title('Original Data with Trend')
    axes[0, 0].legend()

    # 2. 계절 패턴 (1주기)
    seasonal_pattern = result.seasonal[:period]
    axes[0, 1].bar(range(period), seasonal_pattern)
    axes[0, 1].set_title('Seasonal Pattern (1 Cycle)')
    axes[0, 1].axhline(0, color='red', linestyle='--')

    # 3. 잔차 분포
    axes[1, 0].hist(result.resid.dropna(), bins=30, edgecolor='black')
    axes[1, 0].set_title('Residual Distribution')

    # 4. 잔차 시계열 + 이상치
    residuals = result.resid.dropna()
    z_scores = np.abs((residuals - residuals.mean()) / residuals.std())
    anomalies = z_scores > 3

    axes[1, 1].plot(residuals.index, residuals.values)
    axes[1, 1].scatter(residuals.index[anomalies], residuals[anomalies], 
                       color='red', s=50, label='Anomalies')
    axes[1, 1].set_title('Residuals with Anomalies')
    axes[1, 1].legend()

    plt.tight_layout()
    return fig

요약

기법 특징 적합한 상황
Classical 간단, 빠름 기본 분석, 단일 계절성
STL Robust, 유연 이상치 있는 데이터
MSTL 다중 계절성 시간별/일별/주별 복합
Prophet 예측 통합 비즈니스 예측, 공휴일
Fourier 수학적 엄밀 주파수 분석 필요 시

시계열 분해는 데이터 이해, 예측, 이상 탐지의 기본이 되는 필수 기법이다. STL을 기본으로 사용하고, 다중 계절성이 필요하면 MSTL, 예측까지 필요하면 Prophet을 권장한다.