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을 권장한다.