시계열 데이터

시간의 흐름에 따라 관측된 데이터

수집된 데이터의 변수의 개수에 따라 단변량 시계열 데이터와 다변량 시계열 데이터로 나눈다

단변량 시계열 데이터 : 동일한 간격의 사간 증가에 대해 순차적으로 기록되며 한 개의 변수 관측치로 구성된 데이터를 의미
다변량 시계열 데이터 : 동일한 간격의 시간 증가에 대해 순차적으로 기록되지만 두 개 이상의 변수 관측치로 구성된 데이터를 의미

시계열 데이터는 계절성(S), 추세성(T), 순환성(C), 불규칙성(I) 이라는 특징이 존재

 

시계열 분석을 위해서는 시계열이 정상성을 만족해야 함
정상성 (stationary)
시간에 관계없이 데이터가 일정한 상태를 유지하는 것을 의미
시계열 데이터의 평균과 분산에 체계적인 변화 및 주기적 변동이 없다는 것
미래는 확률적으로 과거와 동일하다는 것

정상 시계열의 조건
평균은 모든 시점(시간 t)에 대해 일정하다
분산은 모든 시점(시간 t)에 대해 일정하다
자기 공분산은 시점(시간 t)에 의존하지 않고, 단지 시차에만 의존한다

 

차분: 현 시점 값 - 이전 시점 값

 

정상 시계열로 전환

평균이 일정하지 않은 경우: 원계열에 차분 사용
계절성을 갖는 비정상시계열: 계절 차분 사용
분산이 일정하지 않은 경우: 원계열에 자연로그(변환) 사용

 

pct_change(): 퍼센트 변화율

rooling(): 이동 평균선

df=pd.read_csv(CSV_PATH, parse_dates=['date'])
df['return'] = df['close'].pct_change() # pct_change() 는 이전 행 대비 퍼센트 변화율(%) 을 계산. (현재값 - 이전값) / 이전값 계산 결과를 반환
df['roll_mean_5'] = df['close'].rolling(5).mean() # 이동 평균선 이전 n개의 평균

'''
rolling(window, min_periods=None, center=False)
window: 개수
min_periods: 최소 유효값 개수(기본=None -> window가 꽉차야 값 산출)
center=True면 윈도우를 현재 행을 중심으로 좌우에 펼침. 미래 예측함. 금지

feature_cols = [c for c in df.columns if c not in ["date", "close", "target_next"]]
cut = int(len(df) * 0.8)

train, valid = df.iloc[:cut].copy(), df.iloc[cut:].copy()

X_tr, y_tr = train[feature_cols], train["target_next"]
X_va, y_va = valid[feature_cols], valid["target_next"]


def smape(y_true, y_pred): # 대칭 평균 절대 백분율 오차
    y_true = np.array(y_true)
    y_pred = np.array(y_pred)
    denom = (np.abs(y_true) + np.abs(y_pred))
    denom = np.where(denom == 0, 1e-9, denom)  # 0으로 나눔 방지
    return 100 * np.mean(np.abs(y_true - y_pred) / denom)


yhat_naive = valid["close"].values  # 오늘의 종가를 내일의 종가 예측으로 사용


mae_naive = mean_absolute_error(y_va, yhat_naive)
rmse_naive = np.sqrt(mean_squared_error(y_va, yhat_naive))
smape_naive = smape(y_va, yhat_naive)

print(f"BaseLine - Naive: MAE={mae_naive:.4f}, RMSE={rmse_naive:.4f}, sMAPE={smape_naive:.4f}")

shift_k = 5
yhat_seas = valid["close"].shift(1 - shift_k)  # 4일 후의 종가를 내일의 종가로 예측
# print(yhat_seas.isna().sum())
mask = ~yhat_seas.isna()

mae_naive = mean_absolute_error(y_va[mask], yhat_seas[mask])
rmse_naive = np.sqrt(mean_squared_error(y_va[mask], yhat_seas[mask]))
smape_naive = smape(y_va[mask], yhat_seas[mask])

print(f"BaseLine - Naive: MAE={mae_naive:.4f}, RMSE={rmse_naive:.4f}, sMAPE={smape_naive:.4f}")

 

 

TimeSeriesSplit

▪ 시계열 전용 교차검증
▪ 과거 → 현재 순서를 유지한 채, 점점 늘어나는(또는 고정 길이의) 학습 구간 + 미래 테스트 구간으로 나눠서 5번 평가
▪ 랜덤 셔플이 없고, 과거 정보로 미래를 예측하는 구조라 누수 방지에 적합

분할 방식

• 총 샘플 수가 N인 테스트 구간 크기는 대략 N // (n_splits + 1)
• Fold 1: train = [0 ... i1-1], test = [i1 ... i2-1]
• Fold 2: train = [0 ... i2-1], test = [i2 ... i3-1]
• …
• Fold 5: train = [0 ... i5-1], test = [i5 ... i6-1]

▪ 과거 데이터가 누적되면서 테스트 블록을 앞으로 한 칸씩 밀어가며 평가

 

지수 평활법(Exponential Smoothing)
▪ 이동 평균법의 종류로, 최근관측치에 더 높은 가중치를 부여하는
방법으로 최신의 변동을 민감하게 반영
▪ 최근 시점에 큰 가중치, 과거 시점으로 갈수록 가중치를 지수적으
로 줄여 나감
◼ 지수 평활법을 사용하는 모델
▪ SimpleExpSmoothing (SES)
▪ Holt (Holt’s Linear Trend)
▪ ExponentialSmoothing (Holt–Winters)

 

Simple Exponential Smoothing (SES, 단순지수평활)

개념

• 하나의 시계열 데이터(단변량) 만을 사용한다.
• 추세(trend) 나 계절성(seasonality) 이 없다고 가정한다.
• 최근 관측값일수록 더 큰 가중치(α, 알파)를 주어 평균 수준(level) 을 추정하는 방식이다.
• 즉, 오래된 데이터보다는 최신 데이터에 더 민감하게 반응하는 형태의 이동평균이라고 보면 된다.

사용 목적 및 적합한 상황

• 데이터가 크게 변하지 않고 비교적 안정적일 때, 즉 변화가 완만한 경우.
• 추세나 계절성 패턴이 뚜렷하지 않은 시계열에서 사용한다.
• “내일의 값은 오늘과 거의 비슷할 것”이라고 보는 나이브 예측보다,
  조금 더 부드럽게 변화 추세를 반영하고 싶을 때 적합하다.
• 단기적인 노이즈를 줄이고 최근 데이터의 방향성을 반영한 예측을 원할 때 사용한다.

핵심 파라미터

initialization_method

• 초기값(level의 시작점)을 정하는 방법.
• 보통 "estimated" 옵션을 사용한다. (초기값도 함께 최적화됨)
• 즉, 모델이 학습 과정에서 가장 오차가 작아지도록 초기값을 스스로 찾는다.

smoothing_level (α, 알파, 0~1 사이 값)

• “현재 관측값 yty_tyt​을 얼마나 강하게 반영할 것인가”를 결정하는 계수.
• 값의 의미
  • α가 1에 가까울수록 → 최근 값에 매우 민감하게 반응 (변화에 빠르게 따라감)
  • α가 0에 가까울수록 → 이전 평균을 더 많이 신뢰 (예측이 부드럽고 느리게 반응)
• 지정하지 않으면 보통 내부에서 최적값을 자동 추정한다.

# 이동평균
k = 5
hist = pd.concat([train['close'], valid['close']])
hist_ma = hist.rolling(k).mean()
hist_roll_aligned = hist_ma.loc[valid.index]

mae = mean_absolute_error(y_va, hist_roll_aligned)
rmse = np.sqrt(mean_squared_error(y_va, hist_roll_aligned))
smape_rf = smape(y_va, hist_roll_aligned)

print(f"roll 5 mean: MAE={mae:.4f}, RMSE={rmse:.4f}, sMAPE={smape_rf:.4f}")

# 단순 지수 평활
ses_fit = SimpleExpSmoothing(train['close'], initialization_method="estimated").fit()
print(valid.index[0], valid.index[-1])
ses_pred = ses_fit.predict(start=valid.index[0], end=valid.index[-1])
ses_pred_aligned = ses_pred.shift(-1)

mask = ~ses_pred_aligned.isna()
mae = mean_absolute_error(y_va[mask], hist_roll_aligned[mask])
rmse = np.sqrt(mean_squared_error(y_va[mask], hist_roll_aligned[mask]))
smape_rf = smape(y_va[mask], hist_roll_aligned[mask])

print(f"roll 5 mean : MAE={mae:.4f}, RMSE={rmse:.4f}, sMAPE={smape_rf:.4f}")

 

Holt (Holt’s Linear Trend)

개념 요약

• **Simple Exponential Smoothing(SES)**에서 한 단계 발전된 형태.
• 추세(Trend) 를 함께 고려하는 지수평활법.
• “레벨(Level)” + “추세(Trend)” 두 가지 상태값을 동시에 갱신하면서 예측을 수행.
• 간단히 말해,
  “최근 데이터의 평균적인 수준 + 상승·하락하는 추세”를 모두 반영해서 미래를 예측함.
• 가끔 Holt-Winters의 추세만 쓰는 형태라고도 부름
  (Holt-Winters는 추세 + 계절성까지 포함하지만, Holt는 계절성은 없음).

---

사용 예시

• 추세가 뚜렷한 시계열 데이터 (예: 지속적으로 오르거나 내리는 주가, 판매량 등)에 적합.
• 계절성(주기적인 패턴)은 없거나 약할 때 사용.
• 단순히 “최근값을 평활화(SES)” 하는 것보다,
  시간에 따라 증가·감소하는 경향까지 예측에 반영하고 싶을 때 좋음.
• 하지만 추세를 너무 민감하게 반영하면 과적합이 될 수 있어서→ damped=True 옵션으로 완만하게 조정하는 경우가 많음.
  (이걸 “댐핑(damping)”이라고 함 — 추세가 장기적으로 폭주하지 않게 완화시킴.)

핵심 파라미터

damped_trend=True	추세를 완만하게 만들어 장기 폭주 방지. 예를 들어 상승세가 끝없이 커지지 않게 조절함.
smoothing_level (0–1)	관측값(실제값)을 얼마나 강하게 반영할지 조절. 1에 가까울수록 최근값 반영이 강함, 0에 가까울수록 과거 평균 반영이 강함.
smoothing_trend (0–1)	추세(상승/하락)를 얼마나 강하게 반영할지 조절. 1이면 추세 변화에 즉각 반응, 0이면 추세를 거의 고정.

 

Exponential Smoothing (Holt–Winters)

개념 요약

• Holt-Winters는 지수평활법(Exponential Smoothing)의 가장 일반적인 형태야.
• 하나의 시계열을 구성하는 세 가지 요소를 모두 다룸:
  레벨(Level) + 추세(Trend) + 계절성(Seasonality)
• 즉, 단순한 평균 변화뿐 아니라 시간적 패턴과 주기적 반복까지 반영함.
• 계절성(Seasonality) 표현 방식은 두 가지 중 선택:
  • additive(덧셈형) → 변화 폭이 일정할 때
  • multiplicative(곱셈형) → 변화 폭이 비율로 커질 때

---

사용 예시

• 분기별, 월별, 주별 등 주기적으로 반복되는 패턴(계절성) 이 뚜렷할 때 적합.
• 예시
  • 월별 수요 변화
  • 주중/주말에 따라 다른 방문자 수
  • 분기별 매출 패턴 등

즉, 단순한 추세(상승·하락)뿐 아니라
반복적인 계절 패턴까지 고려해야 할 때 사용.

 핵심 파라미터

파라미터설명

trend	추세 반영 여부. "add"는 선형추세, None은 추세 없음.
seasonal	계절성 반영 방식. "add"(덧셈형), "mul"(곱셈형), 또는 None.
seasonal_periods	계절 주기의 길이. 예: 월별 데이터에서 1년 주기면 12. 주간 데이터에서 1개월 주기면 4 정도.
damped_trend=True	추세를 완만하게 조정해서 장기 폭주를 막음.
initialization_method="estimated"	초기값(level, trend, seasonality)을 자동으로 최적화해서 학습 시작.

 

실습

DAILY_SALES_PATH = '/content/drive/MyDrive/multicampus/시각화/data/daily_sales.csv'

# 이동 평균선으로 이전 n개 평균 구하기 
df=pd.read_csv(DAILY_SALES_PATH, parse_dates=['date'])
df["revenue"] = df["sales"] * df["price"]
df
df['return'] = df['revenue'].pct_change() # pct_change() 는 이전 행 대비 퍼센트 변화율(%) 을 계산. (현재값 - 이전값) / 이전값 계산 결과를 반환

df['roll_mean_5'] = df['revenue'].rolling(5).mean() # 이동 평균선 이전 n개의 평균
df['roll_mean_20'] = df['revenue'].rolling(20).mean() # 이동 평균선 이전 n개의 평균
print(df.head(10))
# 추세 시각화
plt.figure(figsize=(9, 5))
plt.plot(df["date"], df["revenue"], label="revenue Price")
plt.plot(df["date"], df["roll_mean_5"], label="5-day Mean")
plt.plot(df["date"], df["roll_mean_20"], label="20-day Mean")
plt.legend()
plt.tight_layout()
plt.show()

 

 

# 
df["dow"] = df["date"].dt.dayofweek
df["month"] = df["date"].dt.month

df["ret_1"] = df["revenue"].pct_change(1)
df["ret_5"] = df["revenue"].pct_change(5) # 5일전 대비 퍼센트

LAGS = [1, 2, 3, 5, 7, 9]
for i in LAGS:
    df[f"revenue_lag{i}"] = df["revenue"].shift(i) # 이전 값 추가. 시간 지연

df["roll_mean_5"] = df["revenue"].rolling(5).mean().shift(1) # 직전 5일 평균 revenue. shift(1)은 미래 데이터 누출 방지
df["roll_std_5"] = df["revenue"].rolling(5).std().shift(1) # 직전 5일 revenue의 표준편차
df["roll_mean_20"] = df["revenue"].rolling(20).mean().shift(1)

df["target_next"] = df["revenue"].shift(-1)


df = df.dropna().reset_index(drop=True)
df.head()

 

 

# 대칭 평균 절대 백분율 오차로 BaseLine 2개 만들어보기

feature_cols = [c for c in df.columns if c not in ["date", "revenue", "target_next"]]
cut = int(len(df) * 0.8)

train, valid = df.iloc[:cut].copy(), df.iloc[cut:].copy() # train이 80퍼, valid가 20퍼

X_tr, y_tr = train[feature_cols], train["target_next"]
X_va, y_va = valid[feature_cols], valid["target_next"]


def smape(y_true, y_pred): # 대칭 평균 절대 백분율 오차
    y_true = np.array(y_true)
    y_pred = np.array(y_pred)
    denom = (np.abs(y_true) + np.abs(y_pred))
    denom = np.where(denom == 0, 1e-9, denom)  # 0으로 나눔 방지. np.where(조건, 참일때, 거짓일때)
    return 100 * np.mean(np.abs(y_true - y_pred) / denom)


yhat_naive = valid["revenue"].values  # 오늘의 종가를 내일의 종가 예측으로 사용


mae_naive = mean_absolute_error(y_va, yhat_naive)
rmse_naive = np.sqrt(mean_squared_error(y_va, yhat_naive))
smape_naive = smape(y_va, yhat_naive)

print(f"BaseLine - Naive: MAE={mae_naive:.4f}, RMSE={rmse_naive:.4f}, sMAPE={smape_naive:.4f}")

shift_k = 5
yhat_seas = valid["revenue"].shift(1 - shift_k)  # 4일 후의 종가를 오늘의 종가 예측으로 본다
# print(yhat_seas.isna().sum())
mask = ~yhat_seas.isna()

mae_naive = mean_absolute_error(y_va[mask], yhat_seas[mask])
rmse_naive = np.sqrt(mean_squared_error(y_va[mask], yhat_seas[mask]))
smape_naive = smape(y_va[mask], yhat_seas[mask])

print(f"BaseLine - Naive: MAE={mae_naive:.4f}, RMSE={rmse_naive:.4f}, sMAPE={smape_naive:.4f}")

 

 

# 랜덤포레스트
rf = RandomForestRegressor(
    n_estimators=400,
    bootstrap=False,
    max_features='sqrt',
    random_state=RANDOM_STATE,
    n_jobs=-1
)
rf.fit(X_tr, y_tr)

pred_va = rf.predict(X_va)
mae = mean_absolute_error(y_va, pred_va)
rmse = np.sqrt(mean_squared_error(y_va, pred_va))
smape_rf = smape(y_va, pred_va)

print(f"RandomForestRegressor : MAE={mae:.4f}, RMSE={rmse:.4f}, sMAPE={smape_rf:.4f}")

plt.figure(figsize=(13, 5))
plt.plot(valid["date"], y_va, label="True")
plt.plot(valid["date"], pred_va, label="Pred")
plt.legend()
plt.tight_layout()
plt.show()

 

 

# 시계열 데이터를 여러 구간으로 나눠 랜덤포레스트로 예측 성능을 평가
def evaluate_window(df_in, feature_cols, date_col, y_col):
    X = df_in[feature_cols]
    y = df_in[y_col]

    tscv = TimeSeriesSplit(n_splits=5)
    fold_metrics = []

    for k, (tr, te) in enumerate(tscv.split(X)):
        X_tr, X_te = X.iloc[tr], X.iloc[te]
        y_tr, y_te = y.iloc[tr], y.iloc[te]

        rf = RandomForestRegressor(
            n_estimators=400,
            bootstrap=False,
            max_features='sqrt',
            random_state=RANDOM_STATE,
            n_jobs=-1
        )
        rf.fit(X_tr, y_tr)
        pred = rf.predict(X_te)
        mae = mean_absolute_error(y_te, pred)
        rmse = np.sqrt(mean_squared_error(y_te, pred))
        smape_rf = smape(y_te, pred)
        fold_metrics.append({"fold": k + 1, "MAE": mae, "RMSE": rmse, "sMAPE": smape_rf})

    return pd.DataFrame(fold_metrics)

bt = evaluate_window(df, feature_cols, "date", "target_next")
print("MAE mean+-std:", bt["MAE"].mean())

 

 

# 이동평균
k = 5
hist = pd.concat([train['revenue'], valid['revenue']])
hist_ma = hist.rolling(k).mean()
hist_roll_aligned = hist_ma.loc[valid.index]

mae = mean_absolute_error(y_va, hist_roll_aligned)
rmse = np.sqrt(mean_squared_error(y_va, hist_roll_aligned))
smape_rf = smape(y_va, hist_roll_aligned)

print(f"roll 5 mean: MAE={mae:.4f}, RMSE={rmse:.4f}, sMAPE={smape_rf:.4f}")

 

 

# 단순 지수 평활
ses_fit = SimpleExpSmoothing(train["revenue"], initialization_method="estimated").fit()
print(valid.index[0], valid.index[-1])
ses_pred = ses_fit.predict(start=valid.index[0], end=valid.index[-1])
ses_pred_aligned = ses_pred.shift(-1)

mask = ~ses_pred_aligned.isna()
mae = mean_absolute_error(y_va[mask], ses_pred_aligned[mask])
rmse = np.sqrt(mean_squared_error(y_va[mask], ses_pred_aligned[mask]))
smape_rf = smape(y_va[mask], ses_pred_aligned[mask])

print(f"ses : MAE={mae:.4f}, RMSE={rmse:.4f}, sMAPE={smape_rf:.4f}")

 

# Exponential Smoothing (Holt–Winters)
s = 90
hw_fit = ExponentialSmoothing(
    train['revenue'],
    trend='add',
    seasonal='add',
    seasonal_periods=s,
    initialization_method='estimated'
).fit()

hw_pred = hw_fit.forecast(steps=len(valid))
hw_pred.index = y_va.index

mask = ~hw_pred.isna()
mae = mean_absolute_error(y_va[mask], hw_pred[mask])
rmse = np.sqrt(mean_squared_error(y_va[mask], hw_pred[mask]))
smape_rf = smape(y_va[mask], hw_pred[mask])

print(f"Holt_Winters : MAE={mae:.4f}, RMSE={rmse:.4f}, sMAPE={smape_rf:.4f}")