시계열 예측은 데이터를 분석하고 미래를 예측하는 데 중요한 역할을 합니다. 이번 Part 2에서는 고전적 시계열 모델부터 머신러닝 기반 모델까지 다양한 예측 기법들을 다룰 예정입니다. Part 1에서 시계열 데이터의 구성 요소와 전처리 방법을 익혔다면, 이제 이 데이터를 기반으로 예측을 수행하는 방법을 알보겠습니다.
이전 글 보기: 데이터로 보는 트렌드: 시계열 데이터의 비밀(1)
Part 2: 시계열 예측 모델링 기초
고전적 시계열 모델 이해하기
시계열 예측에서 가장 기본이 되는 고전적 모델들은 여전히 많은 실무 현장에서 활용되고 있습니다. 이들 모델은 간단하면서도 강력한 성능을 자랑하며, 예측의 첫 걸음으로 적합합니다.
기본 시계열 모델
- 이동평균 (Moving Average): 과거 데이터의 평균을 사용하여 미래를 예측하는 가장 기본적인 방법입니다. 예를 들어, 주간 판매량을 예측할 때 과거 4주간의 평균을 사용할 수 있습니다.
- 지수평활 (Exponential Smoothing): 최근 데이터에 더 높은 가중치를 부여하여 빠르게 변화하는 패턴을 더 잘 포착하는 방법입니다. α(알파) 파라미터를 통해 최근 데이터의 중요도를 조절할 수 있습니다.
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from statsmodels.tsa.holtwinters import ExponentialSmoothing
# 샘플 데이터 생성
np.random.seed(42)
dates = pd.date_range(start='2023-01-01', end='2023-12-31', freq='D')
sales = 100 + np.random.normal(0, 10, len(dates)) + \
np.sin(np.arange(len(dates)) * 2 * np.pi / 30) * 20 # 계절성 추가
data = pd.Series(sales, index=dates, name='Sales')
def plot_moving_average(data, windows=[7, 14, 30]):
"""이동평균 시각화"""
plt.figure(figsize=(12, 6))
plt.plot(data.index, data.values, label='Original', alpha=0.5)
for window in windows:
ma = data.rolling(window=window).mean()
plt.plot(data.index, ma,
label=f'{window}-day Moving Average',
alpha=0.8)
plt.title('Moving Average Example')
plt.legend()
plt.grid(True)
plt.show()
def plot_exponential_smoothing(data, alphas=[0.2, 0.5, 0.8]):
"""지수평활 시각화"""
plt.figure(figsize=(12, 6))
plt.plot(data.index, data.values, label='Original', alpha=0.5)
for alpha in alphas:
exp_smoothed = pd.Series(index=data.index)
exp_smoothed[0] = data[0]
for t in range(1, len(data)):
exp_smoothed[t] = alpha * data[t] + (1 - alpha) * exp_smoothed[t-1]
plt.plot(data.index, exp_smoothed,
label=f'Exponential Smoothing (α={alpha})',
alpha=0.8)
plt.title('Exponential Smoothing Example')
plt.legend()
plt.grid(True)
plt.show()
# 시각화 실행
plot_moving_average(data)
plot_exponential_smoothing(data)
# 성능 비교
def compare_models(data):
results = pd.DataFrame(index=['RMSE'])
# 7일 이동평균
ma7 = data.rolling(window=7).mean()
results['MA(7)'] = np.sqrt(np.mean((data - ma7.shift(1))**2))
# 지수평활 (α=0.3)
exp_smooth = pd.Series(index=data.index)
exp_smooth[0] = data[0]
alpha = 0.3
for t in range(1, len(data)):
exp_smooth[t] = alpha * data[t] + (1 - alpha) * exp_smooth[t-1]
results['Exp Smoothing(0.3)'] = np.sqrt(np.mean((data - exp_smooth.shift(1))**2))
return results.round(2)
print("\nModel Performance Comparison:")
print(compare_models(data))ARIMA와 SARIMA 모델링
고전적 모델들이 직관적이고 간단하지만, 보다 정교한 예측을 위해서는 ARIMA(AutoRegressive Integrated Moving Average)와 SARIMA(Seasonal ARIMA) 모델이 유용합니다. 이들은 시계열 데이터의 자기상관성과 계절성 패턴을 반영할 수 있어, 더욱 정밀한 예측을 가능하게 합니다.
- ARIMA: 데이터의 자기회귀(AR), 이동평균(MA), 차분(I)을 결합하여 비정상성을 제거하고 예측을 수행합니다.
ARIMA 모델의 주요 파라미터:
- p(자기회귀 차수) : 과거 값들의 영향을 얼마나 고려할지
- d(차분 차수) : 시계열을 정상화하기 위한 차분 횟수
- q(이동평균 차수) : 과거 오차항의 영향을 얼마나 고려할지
from statsmodels.tsa.arima.model import ARIMA
from statsmodels.tsa.stattools import adfuller
def check_stationarity(data):
"""정상성 검정"""
result = adfuller(data)
print('ADF Statistic:', result[0])
print('p-value:', result[1])
print('\nCritical values:')
for key, value in result[4].items():
print(f'\t{key}: {value}')
def fit_arima(data, order=(1,1,1)):
"""ARIMA 모델 적용"""
model = ARIMA(data, order=order)
results = model.fit()
# 모델 성능 지표
print("\nModel Statistics:")
print(f'AIC: {results.aic:.2f}')
print(f'BIC: {results.bic:.2f}')
# 예측 시각화
plt.figure(figsize=(12, 6))
plt.plot(data.index, data.values, label='Original')
plt.plot(data.index, results.fittedvalues, label='ARIMA Fitted')
plt.title(f'ARIMA{order} Model Fitting')
plt.legend()
plt.grid(True)
plt.show()
return results
# ARIMA 모델 적용
print("Stationarity Test:")
check_stationarity(data)
arima_results = fit_arima(data)- SARIMA: ARIMA의 확장된 형태로, 계절성을 반영할 수 있는 모델입니다. 연간, 월간, 주간 등의 주기적 패턴이 있는 데이터에 특히 유용합니다.
AIC와 BIC: 모델 선택의 기준
모델을 선택할 때는 객관적인 평가 지표가 필요합니다. AIC와 BIC는 모델의 적합도와 복잡성을 평가하는 데 중요한 역할을 합니다.
- AIC (Akaike Information Criterion): 모델의 적합도를 평가하면서 과적합을 방지하는 페널티 항을 포함하여, 값이 작을수록 더 좋은 모델을 의미합니다.
- BIC (Bayesian Information Criterion): AIC와 유사하지만 더 강한 페널티를 부여하여 모델 선택에 있어 더 보수적인 기준을 제공합니다. 샘플 크기를 고려한 보정이 이루어집니다.
import pandas as pd
import numpy as np
from statsmodels.tsa.arima.model import ARIMA
def compare_arima_models(data, p_range, d_range, q_range):
"""
다양한 ARIMA 모델의 AIC와 BIC를 비교
Parameters:
-----------
data : Series
시계열 데이터
p_range : range
AR 차수 범위
d_range : range
차분 차수 범위
q_range : range
MA 차수 범위
Returns:
--------
DataFrame
각 모델의 파라미터와 평가 지표
"""
results = []
for p in p_range:
for d in d_range:
for q in q_range:
try:
model = ARIMA(data, order=(p, d, q))
fitted = model.fit()
results.append({
'p': p,
'd': d,
'q': q,
'AIC': fitted.aic,
'BIC': fitted.bic,
})
except:
continue
results_df = pd.DataFrame(results)
return results_df.sort_values('AIC')
# 사용 예시
if __name__ == "__main__":
# 샘플 데이터 생성
np.random.seed(42)
dates = pd.date_range(start='2023-01-01', end='2023-12-31', freq='D')
data = pd.Series(np.random.normal(0, 1, len(dates)) + \
np.sin(np.arange(len(dates)) * 2 * np.pi / 30),
index=dates)
# 다양한 ARIMA 모델 비교
models_comparison = compare_arima_models(
data,
p_range=range(0, 3),
d_range=range(0, 2),
q_range=range(0, 3)
)
print("Top 5 models by AIC:")
print(models_comparison.head())
print("\nBest model parameters:")
best_model = models_comparison.iloc[0]
print(f"ARIMA({best_model['p']},{best_model['d']},{best_model['q']})")이 두 지표를 함께 고려하면 단순히 데이터에 적합한 모델 뿐 아니라 일반화 가능성이 높은 모델을 선택할 수 있습니다.
LightGBM을 활용한 시계열 예측
고전적 시계열 모델들이 유용하지만, 머신러닝 기반 모델은 대규모 데이터셋을 다룰 때 더 큰 효과를 발휘합니다. 특히 LightGBM은 시계열 예측에서도 뛰어난 성능을 보여주고 있습니다.
LightGBM의 주요 특징
LightGBM은 그라디언트 부스팅 트리(Gradient Boosting Tree) 기반의 모델로, 다음과 같은 장점을 제공합니다:
- 빠른 학습 속도와 높은 예측 정확도
- 대규모 데이터 처리 능력
- 효율적인 메모리 사용
시계열 데이터를 위한 특성 엔지니어링
LightGBM을 시계열 예측에 적용할 때는 다음과 같은 특성들을 활용합니다:
- 시간 정보 특성: 시간, 요일, 월, 분기, 연도 등
- 시차(Lag) 특성: 과거 1일, 7일, 14일, 30일 등
- 통계적 특성: 이동 평균, 이동 표준편차 등
import lightgbm as lgb
import pandas as pd
import numpy as np
from sklearn.model_selection import TimeSeriesSplit
from sklearn.metrics import mean_squared_error, mean_absolute_error
import matplotlib.pyplot as plt
class TimeSeriesMLPredictor:
def __init__(self, data, target_col='value'):
self.data = data
self.target_col = target_col
self.model = None
self.feature_importance = None
def create_features(self, df):
"""시계열 특성 생성"""
df = df.copy()
# 시간 관련 특성
df['hour'] = df.index.hour
df['dayofweek'] = df.index.dayofweek
df['quarter'] = df.index.quarter
df['month'] = df.index.month
df['year'] = df.index.year
df['dayofyear'] = df.index.dayofyear
# 시차 특성
for lag in [1, 7, 14, 30]:
df[f'lag_{lag}'] = df[self.target_col].shift(lag)
# 이동 평균
for window in [7, 14, 30]:
df[f'rolling_mean_{window}'] = df[self.target_col].rolling(window=window).mean()
df[f'rolling_std_{window}'] = df[self.target_col].rolling(window=window).std()
return df
def prepare_data(self):
"""데이터 준비"""
df = self.create_features(self.data)
# NaN 제거
df = df.dropna()
# 특성과 타겟 분리
feature_columns = [col for col in df.columns if col != self.target_col]
X = df[feature_columns]
y = df[self.target_col]
return X, y
def train(self, params=None):
"""모델 학습"""
if params is None:
params = {
'objective': 'regression',
'metric': 'rmse',
'num_leaves': 31,
'learning_rate': 0.05,
'feature_fraction': 0.9
}
X, y = self.prepare_data()
# 시계열 교차 검증
tscv = TimeSeriesSplit(n_splits=5)
models = []
cv_scores = []
for fold, (train_idx, val_idx) in enumerate(tscv.split(X), 1):
X_train, X_val = X.iloc[train_idx], X.iloc[val_idx]
y_train, y_val = y.iloc[train_idx], y.iloc[val_idx]
train_set = lgb.Dataset(X_train, y_train)
val_set = lgb.Dataset(X_val, y_val)
model = lgb.train(
params,
train_set,
valid_sets=[train_set, val_set],
num_boost_round=1000,
early_stopping_rounds=50,
verbose_eval=False
)
models.append(model)
# 검증 세트에 대한 예측
predictions = model.predict(X_val)
rmse = np.sqrt(mean_squared_error(y_val, predictions))
cv_scores.append(rmse)
print(f'Fold {fold} - RMSE: {rmse:.2f}')
print(f'\nMean RMSE: {np.mean(cv_scores):.2f} (+/- {np.std(cv_scores):.2f})')
# 최종 모델 선택 (가장 좋은 성능의 모델)
best_model_idx = np.argmin(cv_scores)
self.model = models[best_model_idx]
# 특성 중요도 저장
self.feature_importance = pd.DataFrame({
'feature': X.columns,
'importance': self.model.feature_importance('gain')
}).sort_values('importance', ascending=False)
def plot_feature_importance(self, top_n=10):
"""특성 중요도 시각화"""
plt.figure(figsize=(10, 6))
importance_df = self.feature_importance.head(top_n)
plt.barh(importance_df['feature'], importance_df['importance'])
plt.title(f'Top {top_n} Most Important Features')
plt.xlabel('Feature Importance (gain)')
plt.tight_layout()
plt.show()
# 사용 예시
if __name__ == "__main__":
# 샘플 데이터 생성
dates = pd.date_range(start='2023-01-01', end='2023-12-31', freq='H')
ts = pd.Series(
np.random.normal(100, 10, len(dates)) + \
np.sin(np.arange(len(dates)) * 2 * np.pi / 24) * 20,
index=dates,
name='value'
)
ts = pd.DataFrame(ts)
# 모델 학습 및 시각화
ml_predictor = TimeSeriesMLPredictor(ts)
ml_predictor.train()
ml_predictor.plot_feature_importance()마치며
이번 Part 2에서는 시계열 예측 모델링의 기본적인 기법들을 살펴보았습니다. 고전적인 통계 모델부터 최신 머신러닝 기법까지, 각각의 장단점과 적용 방법을 이해하고, 실무에서 시계열 예측을 어떻게 활용할 수 있을지에 대해 알아보았습니다.
다음 Part 3에서는 예측 모델의 성능을 지속적으로 관리하는 방법을 다룰 예정입니다. 예측 모델은 시간이 지나면서 정확도가 저하될 수 있기 때문에, 성능 변화를 탐지하고 적절한 재학습 주기를 설정하는 방법을 소개해드리겠습니다.