시계열 데이터 분석에 필요한 개념들을 요약하고 정리하는 시리즈 - 3. 단일 시계열 모델 분석 (ARIMA) 편
# 라이브러리 불러오기
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
# 데이터 불러오기
df0 = pd.read_csv('Bindukuri_temperature.csv')
print(df0.shape)
display(df0.head(5))
display(df0.tail(5))
df0.info()
'''
<class 'pandas.core.frame.DataFrame'>
RangeIndex: 751 entries, 0 to 750
Data columns (total 2 columns):
# Column Non-Null Count Dtype
--- ------ -------------- -----
0 Date 751 non-null object
1 MeanTemp 751 non-null float64
dtypes: float64(1), object(1)
memory usage: 11.9+ KB
'''
ts0 = df0.copy()
ts0['Date'] = pd.to_datetime(ts0.Date, format='%Y-%m-%d')
ts0.info()
'''
<class 'pandas.core.frame.DataFrame'>
RangeIndex: 751 entries, 0 to 750
Data columns (total 2 columns):
# Column Non-Null Count Dtype
--- ------ -------------- -----
0 Date 751 non-null datetime64[ns]
1 MeanTemp 751 non-null float64
dtypes: datetime64[ns](1), float64(1)
memory usage: 11.9 KB
'''
ts1 = ts0.copy()
# 연도
ts1['Year'] = ts1.Date.dt.year
# 월
ts1['Month'] = ts1.Date.dt.month
# 일
ts1['Day'] = ts1.Date.dt.day
# 요일
ts1['Day_name'] = ts1.Date.dt.day_name()
ts1
# 필요한 column만 남기기
ts = ts1.loc[:, ['Date', 'MeanTemp']]
# 인덱스 설정 후 drop
ts.index = ts.Date
ts = ts.drop(columns="Date")
ts
# 시각화 함수 정의
def plot_ts(data, color, alpha, label):
plt.figure(figsize=(11,5))
plt.plot(data, color=color, alpha=alpha, label=label)
plt.title("Mean Temperature of Bindukuri")
plt.ylabel('Mean temperature')
plt.legend()
plt.show()
plot_ts(ts, 'blue', 0.25, 'Original')
from statsmodels.tsa.stattools import adfuller
adfuller(ts, autolag='AIC')
'''
(-1.4095966745887758,
0.5776668028526356,
11,
739,
{'1%': -3.439229783394421,
'5%': -2.86545894814762,
'10%': -2.5688568756191392},
2474.854578321788)
'''
def ADF_test(data):
# ADF 실시
results = adfuller(data, autolag='AIC')
# 통계량
s = results[0]
# p-value
p = results[1]
# 기각역
cv = results[4]
# 출력
print('-'*30)
print('Augemented Dickey-Fuller Test')
print('H0 : 단위근이 존재한다 (비정상 시계열)')
print('Ha : 단위근이 없다 (정상 시계열)')
print('Critical Values : {}'.format(cv))
print('-'*30)
print('Test Statistics : {:.4f}'.format(s))
print('p-value : {:.4f}'.format(p))
print('-'*30)
# ts 데이터로 ADF 실행해보기
ADF_test(ts)
'''
------------------------------
Augemented Dickey-Fuller Test
H0 : 단위근이 존재한다 (비정상 시계열)
Ha : 단위근이 없다 (정상 시계열)
Critical Values : {'1%': -3.439229783394421, '5%': -2.86545894814762, '10%': -2.5688568756191392}
------------------------------
Test Statistics : -1.4096
p-value : 0.5777
------------------------------
'''
# 영가설 기각 불가 : 단위근이 존재함. 비정상 시계열이다.
def plot_rolling(data, roll_size):
# 이동평균함수(rolling) - 평균, 표준편차
roll_mean = data.rolling(window=roll_size).mean()
roll_std = data.rolling(window=roll_size).std()
# 시각화
plt.figure(figsize=(11,5))
plt.plot(data, color='blue', alpha=0.25, label='Original')
plt.plot(roll_mean, color='black', label='Rolling mean')
plt.plot(roll_std, color='red', label='Rolling std')
plt.title('Rolling Mean & Standard Deviation')
plt.ylabel("Mean Temperature")
plt.legend()
plt.show()
# 함수 실행
plot_rolling(ts, 6)
.shift()
함수를 이용# ts에서 ts.shift(1)을 빼기 (1차 차분)
ts_diff = ts - ts.shift() # default=1임
# 시각화
plot_ts(ts_diff, color='blue', alpha=0.25, label='differencing')
plot_rolling(ts_diff, 6)
ADF_test(ts_diff.dropna())
'''
------------------------------
Augemented Dickey-Fuller Test
H0 : 단위근이 존재한다 (비정상 시계열)
Ha : 단위근이 없다 (정상 시계열)
Critical Values : {'1%': -3.439229783394421, '5%': -2.86545894814762, '10%': -2.5688568756191392}
------------------------------
Test Statistics : -11.6790
p-value : 0.0000
------------------------------
'''
.diff()
함수 이용# 차분
ts_diff2 = ts.diff().dropna()
# ADF 테스트
ADF_test(ts_diff2)
'''
------------------------------
Augemented Dickey-Fuller Test
H0 : 단위근이 존재한다 (비정상 시계열)
Ha : 단위근이 없다 (정상 시계열)
Critical Values : {'1%': -3.439229783394421, '5%': -2.86545894814762, '10%': -2.5688568756191392}
------------------------------
Test Statistics : -11.6790
p-value : 0.0000
------------------------------
'''
# 영가설 기각 : 단위근이 없다. 정상시계열이다.
# ACF and PACF
from statsmodels.tsa.stattools import acf, pacf
# ACF
acf_20 = acf(x=ts_diff2, nlags=20)
# PACF
pacf_20 = pacf(x=ts_diff2, nlags=20, method='ols')
# 95% 신뢰구간 계산하기
confidence = 1.96/np.sqrt(len(ts_diff2))
# 시각화
plt.figure(figsize=(8,4))
# ACF
plt.subplot(1,2,1)
plt.plot(acf_20)
plt.axhline(y=0, linestyle='--', alpha=0.5)
plt.axhline(y=-confidence, linestyle='--', alpha=0.5)
plt.axhline(y=confidence, linestyle='--', alpha=0.5)
plt.title('AutoCorrelation Function')
# PACF
plt.subplot(1,2,2)
plt.plot(pacf_20)
plt.axhline(y=0, linestyle='--', alpha=0.5)
plt.axhline(y=-confidence, linestyle='--', alpha=0.5)
plt.axhline(y=confidence, linestyle='--', alpha=0.5)
plt.title('Partial AutoCorrelation Function')
plt.tight_layout()
# ACF
acf_10 = acf(x=ts_diff2, nlags=10)
# PACF
pacf_10 = pacf(x=ts_diff2, nlags=10, method='ols')
# 95% 신뢰구간
confidence = 1.96/np.sqrt(len(ts_diff2))
# 시각화
plt.figure(figsize=(8,4))
plt.subplot(1,2,1)
plt.plot(acf_10)
plt.axhline(y=0, linestyle='--', alpha=0.5)
plt.axhline(y=-confidence, linestyle='--', alpha=0.5)
plt.axhline(y=confidence, linestyle='--', alpha=0.5)
plt.title('AutoCorrelation Function')
plt.subplot(1,2,2)
plt.plot(pacf_10)
plt.axhline(y=0, linestyle='--', alpha=0.5)
plt.axhline(y=-confidence, linestyle='--', alpha=0.5)
plt.axhline(y=confidence, linestyle='--', alpha=0.5)
plt.title('Partial AutoCorrelation Function')
plt.tight_layout()
# ARIMA(p, d, q)
# 최적 q값 (신뢰구간 최초 진입 시점)
# ACF (q, MA) = 3
# 최적 p값 (신뢰구간 최초 진입 시점)
# PACF (p, AR) = 6
# 예상되는 최적의 ARIMA 모델
# ARIMA(6, 1, 3)
# ARIMA
from statsmodels.tsa.arima.model import ARIMA
# index를 period로 변환해주어야 warning이 뜨지 않음
ts_copy = ts.copy()
ts_copy.index = pd.DatetimeIndex(ts_copy.index).to_period('D')
# 예측을 시작할 위치(이후 차분을 적용하기 때문에 맞추어주었음
start_idx = ts_copy.index[1]
# ARIMA(1,0,1)
model1 = ARIMA(ts_copy, order=(1,0,1))
# fit model
model1_fit = model1.fit()
# 전체에 대한 예측 실시
forecast1 = model1_fit.predict(start=start_idx)
from sklearn.metrics import mean_squared_error
def plot_and_error(data, forecast):
# MSE 계산
mse = mean_squared_error(data, forecast)
# 시각화
plt.figure(figsize=(11,5))
plt.plot(data, color='blue', alpha=0.25 , label='Original')
plt.plot(forecast, color='red', label='predicted')
plt.title("Time Series Forecast")
plt.ylabel("Mean Temperature")
plt.legend()
plt.show()
# MSE 출력
print('Mean Squared Error : {:.4f}'.format(mse))
plot_and_error(ts[1:], forecast1)
# 모델 파라미터 최적화 (p=6, d=1, q=3)
model2 = ARIMA(ts_copy, order=(6,1,3))
# fit
model2_fit = model2.fit()
# 예측
forecast2 = model2_fit.predict(start=start_idx)
# 시각화 및 MSE 연산
plot_and_error(ts[1:], forecast2)
pmdarima
라는 파이썬 라이브러리를 통해 Auto ARIMA를 실시하였습니다.# 기본 라이브러리 불러오기
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
# 데이터 불러오기 (첫 열을 datetime형태로 인덱스로 지정하여 불러오기)
df0 = pd.read_csv('stock_data.csv', index_col=0, parse_dates=True)
df0.info()
'''
<class 'pandas.core.frame.DataFrame'>
DatetimeIndex: 312 entries, 2017-01-03 to 2018-03-29
Data columns (total 2 columns):
# Column Non-Null Count Dtype
--- ------ -------------- -----
0 company_A 312 non-null float64
1 company_M 312 non-null float64
dtypes: float64(2)
memory usage: 7.3 KB
'''
print(df0.shape)
display(df0.head())
display(df0.tail())
df0.isnull().sum()
'''
company_A 0
company_M 0
dtype: int64
'''
stock_a = df0[['company_A']]
stock_a
train = stock_a['company_A'][:int(0.8*len(stock_a))]
test = stock_a['company_A'][int(0.8*len(stock_a)):]
train.plot(label='Train')
test.plot(label='Test')
plt.legend()
plt.show()
pmdarima
라이브러리를 통해 최적의 차분 수를 구함import pmdarima as pm
kpss_diffs = pm.arima.ndiffs(train, alpha=0.05, test='kpss', max_d=5)
adf_diffs = pm.arima.ndiffs(train, alpha=0.05, test='adf', max_d=5)
n_diffs = max(kpss_diffs, adf_diffs)
print(f"Optimized 'd' = {n_diffs}")
'''
Optimized 'd' = 1
'''
# 직접 지정해주어 모델링 실시
model = pm.auto_arima(y=train, # 데이터
d=n_diffs, # 차분 (d), 기본값 = None
start_p= 0, # 시작 p값, 기본값 = 2
max_p = 5, # p 최대값, 기본값 = 5
start_q= 0, # 시작 q값, 기본값 = 2
max_q = 5, # q 최대값, 기본값 = 5
m=1, # season의 주기, 기본값 = 1
seasonal=False, # sARIMA를 실시, 기본값 = True
stepwise=True, # stepwise algorithm, 기본값 = True
trace=True) # 각 step을 출력할지, 기본값 = False
'''
Performing stepwise search to minimize aic
ARIMA(0,1,0)(0,0,0)[0] intercept : AIC=967.793, Time=0.01 sec
ARIMA(1,1,0)(0,0,0)[0] intercept : AIC=969.462, Time=0.03 sec
ARIMA(0,1,1)(0,0,0)[0] intercept : AIC=969.432, Time=0.07 sec
ARIMA(0,1,0)(0,0,0)[0] : AIC=969.948, Time=0.01 sec
ARIMA(1,1,1)(0,0,0)[0] intercept : AIC=971.366, Time=0.14 sec
Best model: ARIMA(0,1,0)(0,0,0)[0] intercept
Total fit time: 0.266 seconds
'''
model2 = pm.auto_arima(train, d=1, seasonal=False, trace=True)
'''
Performing stepwise search to minimize aic
ARIMA(2,1,2)(0,0,0)[0] intercept : AIC=966.160, Time=0.33 sec
ARIMA(0,1,0)(0,0,0)[0] intercept : AIC=967.793, Time=0.01 sec
ARIMA(1,1,0)(0,0,0)[0] intercept : AIC=969.462, Time=0.03 sec
ARIMA(0,1,1)(0,0,0)[0] intercept : AIC=969.432, Time=0.04 sec
ARIMA(0,1,0)(0,0,0)[0] : AIC=969.948, Time=0.01 sec
ARIMA(1,1,2)(0,0,0)[0] intercept : AIC=972.919, Time=0.10 sec
ARIMA(2,1,1)(0,0,0)[0] intercept : AIC=972.738, Time=0.13 sec
ARIMA(3,1,2)(0,0,0)[0] intercept : AIC=969.483, Time=0.52 sec
ARIMA(2,1,3)(0,0,0)[0] intercept : AIC=971.491, Time=0.22 sec
ARIMA(1,1,1)(0,0,0)[0] intercept : AIC=971.366, Time=0.14 sec
ARIMA(1,1,3)(0,0,0)[0] intercept : AIC=972.667, Time=0.13 sec
ARIMA(3,1,1)(0,0,0)[0] intercept : AIC=973.267, Time=0.13 sec
ARIMA(3,1,3)(0,0,0)[0] intercept : AIC=966.431, Time=0.57 sec
ARIMA(2,1,2)(0,0,0)[0] : AIC=968.841, Time=0.20 sec
Best model: ARIMA(2,1,2)(0,0,0)[0] intercept
Total fit time: 2.567 seconds
'''
model2.summary()
model2.plot_diagnostics(figsize=(16,8))
plt.show()
test.shape
'''
(63,)
'''
# 예측 -> 리스트로 변환
pred = model2.predict(n_periods=len(test)).to_list()
# 데이터프레임 생성
test_pred = pd.DataFrame({'test':test, 'pred':pred}, index=test.index)
test_pred
plt.plot(train, label='Train')
plt.plot(test, label='Test')
plt.plot(test_pred.pred, label='Predicted')
plt.legend()
plt.show()
# one point forcast 함수 정의, 신뢰구간도 함께 담아보기
def forcast_one_step():
fc, conf = model2.predict(n_periods=1, return_conf_int=True)
return fc.tolist()[0], np.asarray(conf).tolist()[0]
# 값들을 담을 빈 리스트를 생성
y_pred = []
pred_upper = []
pred_lower = []
# for문을 통한 예측 및 모델 업데이트를 반복함
for new_ob in test:
fc, conf = forcast_one_step()
y_pred.append(fc)
pred_upper.append(conf[1])
pred_lower.append(conf[0])
model2.update(new_ob)
test_pred2 = pd.DataFrame({'test':test, 'pred':y_pred})
y_pred_df = test_pred2['pred'] # Series로 반환
y_pred_df
'''
Date
2017-12-28 170.843675
2017-12-29 171.682863
2018-01-02 169.646252
2018-01-03 172.295839
2018-01-04 172.064889
...
2018-03-23 169.268631
2018-03-26 165.184800
2018-03-27 172.805556
2018-03-28 168.247672
2018-03-29 166.840596
Name: pred, Length: 63, dtype: float64
'''
plt.plot(train, label='Train')
plt.plot(test, label='Test')
plt.plot(y_pred_df, label='Predicted')
plt.legend()
plt.show()
model2.summary()
# sklearn으로 MAPE 계산
from sklearn.metrics import mean_absolute_percentage_error
print(f"MAPE : {mean_absolute_percentage_error(test, y_pred):.3f}")
'''
MAPE : 0.013
'''
# numpy로 직접 계산
def MAPE(y_test, y_pred):
return np.mean(np.abs((test - y_pred) / y_test))
print(f"MAPE : {MAPE(test, y_pred):.3f}")
'''
MAPE : 0.013
'''
다음 글에서는 다변량 시계열 분석 기법에 대해 다루어보고, 이후엔 딥러닝 기법들도 다루어볼 예정입니다.
안녕하세요!
글 잘 보았습니다.
관련해서 질문이 있어 댓글 남깁니다.
test 데이터는 우리가 현재 예측 시점에서 알지 못하는 정보라고 생각하는데요
Ex) 향후 16주를 예측한다고 했을 때, 다음 주 ~ 16주 이후 모두 모르는 값.
아래의 모델 업데이트는 실제 test 데이터 값인 new_ob를 사용해서 이루어지는 것이
맞는 방법인지 의문이 들더군요.
오히려 모델의 예측 값으로 업데이트를 해야하는 것이 아닌가 라는 생각도 들구요
혹시 관련해서 제가 잘못 이해하고 있는 부분이 있거나, 작성자님의 의견 말씀해주시면 감사하겠습니다!
for문을 통한 예측 및 모델 업데이트를 반복함
for new_ob in test:
fc, conf = forcast_one_step()
y_pred.append(fc)
pred_upper.append(conf[1])
pred_lower.append(conf[0])
model2.update(new_ob)