GaussianHMM으로 시장 국면을 분류한 방법

2026. 5. 3. 18:23ㆍ카테고리 없음

Passive

미국 지수 투자자를 위한 금융 데이터 분석 사이트

dinsightlab.com

GaussianHMM으로 시장 국면을 분류한 방법

AI 기반 미국 주식 분석 플랫폼 Passive의 핵심 기능인 Noise vs Signal 국면 분류기 구현 과정을 기록한다.

왜 시장 국면 분류가 필요했는가

ETF 투자자에게 가장 위험한 순간은 "지금 시장이 이성적으로 움직이고 있는가, 아니면 감정에 의해 움직이고 있는가"를 구분하지 못할 때다.

주가가 오르고 있어도 펀더멘털 대비 고평가된 상태에서 감정이 이끄는 상승이라면 매수 타이밍이 아니다. 반대로 주가가 급락해도 펀더멘털이 견고하다면 공포에 의한 과매도 구간일 수 있다.

이 판단을 숫자로 만들고 싶었다. "지금 시장이 이성적인가 감정적인가"를 정량화하는 것. 그게 Passive 펀더멘털 탭의 출발점이었다.

왜 HMM인가

처음 시도한 방법은 단순한 임계값 기반 분류였다. CAPE가 30 이상이면 고평가, VIX가 25 이상이면 공포 구간이라는 식이었다. 문제는 각 지표가 독립적으로 신호를 내고, 지표끼리 충돌할 때 어떤 걸 우선해야 하는지 기준이 없었다.

여러 지표를 동시에 고려해서 하나의 상태를 추정하는 방법이 필요했다. Hidden Markov Model이 딱 맞았다. 시장의 실제 국면("숨겨진 상태")은 직접 관측할 수 없고, 우리가 보는 건 그 상태에서 나온 관측값(지표들)뿐이라는 구조가 시장을 표현하기에 자연스러웠다.

GaussianHMM을 선택한 이유는 연속형 관측값을 다루기 때문이다. CAPE, HY 스프레드, 실현 변동성 같은 지표들은 범주형이 아닌 연속형이고, GaussianHMM은 각 상태에서 관측값이 정규분포를 따른다고 가정한다.

8개 피처 설계

모델 입력으로 8개 월별 피처를 사용한다. 18년치 월별 데이터(약 219개월)가 학습 데이터다.

FEATURE_NAMES = [
    'fundamental_gap',  # CAPE 기반 가격-이익 괴리
    'erp_zscore',       # 주식 위험 프리미엄 표준화
    'residual_corr',    # 섹터 잔차 상관관계
    'dispersion',       # 종목간 수익률 분산
    'amihud',           # Amihud 비유동성 지표
    'vix_term',         # VIX 기간 구조 (VIX/VIX3M)
    'hy_spread',        # HY 신용 스프레드
    'realized_vol',     # 실현 변동성
]

각 피처는 시장의 다른 측면을 포착한다.

fundamental_gap은 Shiller 데이터에서 계산한다. 12개월 주가 누적 수익률에서 12개월 EPS 누적 성장률을 뺀 값이다. 양수면 가격이 이익보다 빠르게 올랐다는 뜻, 즉 비싸졌다는 신호다.

shiller['log_P'] = np.log(shiller['P'])
shiller['log_E'] = np.log(shiller['E'].clip(lower=0.01))
fundamental_gap = (shiller['log_P'].diff(12) - shiller['log_E'].diff(12)).dropna()

erp_zscore는 주식 위험 프리미엄의 역사적 표준화값이다. ERP = 주식 수익률(1/CAPE) - TIPS 실질금리. 이 값의 rolling z-score를 사용한다.

erp = erp_df['ey'] - erp_df['tips']
erp_rm = erp.rolling(120, min_periods=60).mean()
erp_rs = erp.rolling(120, min_periods=60).std()
erp_zscore = ((erp - erp_rm) / erp_rs).abs().dropna()

residual_corr은 시장 전체 움직임(SPY 베타)을 제거한 잔차 수익률 간 상관관계다. 감정 지배 구간에서는 개별 종목들이 펀더멘털과 무관하게 같은 방향으로 움직이기 때문에 잔차 상관이 높아진다.

cov_spy = ret.rolling(60, min_periods=30).cov(spy_ret)
spy_var = spy_ret.rolling(60, min_periods=30).var()
beta = cov_spy / spy_var
residuals[ticker] = ret - beta * spy_ret

vix_term은 VIX를 VIX3M으로 나눈 값이다. 1보다 크면 단기 공포가 장기보다 크다는 뜻(역전 상태). 감정적 시장에서 이 값이 1을 넘는 경우가 많다.

모델 학습

RobustScaler로 전처리한 뒤 GaussianHMM을 학습한다. full covariance를 먼저 시도하고 실패하면 diag로 폴백한다.

scaler = RobustScaler()
X_scaled = scaler.fit_transform(X)

for cov_type in ('full', 'diag'):
    try:
        model = GaussianHMM(
            n_components=4,
            covariance_type=cov_type,
            n_iter=200,
            random_state=42,
        )
        model.fit(X_scaled)
        break
    except (ValueError, np.linalg.LinAlgError):
        if cov_type == 'diag':
            raise

RobustScaler를 선택한 이유는 금융 데이터의 이상치 때문이다. 2008년 금융위기나 2020년 코로나 급락 같은 극단적 사건이 StandardScaler의 평균과 분산을 왜곡한다. RobustScaler는 중앙값과 IQR을 기준으로 스케일링하기 때문에 이상치에 강건하다.

4개 상태를 선택한 이유는 실험 기반이다. 2개는 너무 단순하고 6개 이상은 해석 불가능한 상태가 생겼다. 4개가 "이성 / 약한 이성 / 약한 감정 / 강한 감정" 정도로 해석 가능한 최적점이었다.

noise_score: 국면을 하나의 숫자로

HMM이 4개 상태를 분류하지만, 사용자에게 "현재 상태 2번"이라고 보여주는 건 의미가 없다. 직관적인 점수로 변환해야 했다.

noise_score는 8개 피처의 가중합으로 계산한다.

def compute_noise_score(means: np.ndarray) -> np.ndarray:
    return (
        0.5 * np.abs(means[:, 0])   # fundamental_gap (절대값)
      + 0.3 * np.abs(means[:, 1])   # erp_zscore (절대값)
      + 1.0 * means[:, 2]           # residual_corr
      + 0.5 * means[:, 4]           # amihud
      + 2.0 * means[:, 5]           # vix_term (가장 높은 가중치)
      + 1.5 * means[:, 6]           # hy_spread
      + 2.0 * means[:, 7]           # realized_vol (가장 높은 가중치)
    )

가중치 선택 기준은 두 가지다. 첫째, 시장 감정을 직접적으로 반영하는 지표에 높은 가중치를 준다. vix_term과 realized_vol이 2.0인 이유다. 둘째, 절대값을 취하는 지표와 그대로 쓰는 지표를 구분한다. fundamental_gap과 erp_zscore는 방향보다 괴리의 크기가 중요하기 때문에 절대값을 취한다.

DB 저장 시 양수가 "감정적 시장"을 의미하도록 부호 컨벤션을 정했고, API 응답 단계에서 부호를 반전해서 사용자에게는 "양수 = 이성적 시장"으로 보여준다.

def _flip_noise_record(record):
    """DB 저장 부호(양수=감정) → 표시 부호(양수=이성) 변환."""
    if record.get('noise_score') is not None:
        record['noise_score'] = round(-float(record['noise_score']), 4)
    fc = record.get('feature_contributions')
    if isinstance(fc, list):
        for c in fc:
            if c.get('contribution') is not None:
                c['contribution'] = round(-float(c['contribution']), 4)
    return record

경량 예측: 10분마다 실시간 업데이트

전체 파이프라인(18년치 데이터 수집 + 모델 학습)은 3시간마다 실행된다. 그 사이에도 오늘의 국면은 업데이트해야 한다.

8개 피처 중 월별로만 업데이트되는 피처(fundamental_gap, erp_zscore, vix_term, hy_spread)는 모델 번들에 캐싱해두고, 실시간으로 계산 가능한 피처(residual_corr, dispersion, amihud, realized_vol)는 최근 60일 데이터로 재계산한다.

# 모델 번들에서 월별 피처 캐시값 로드
monthly = model_bundle['last_monthly_values']
fg_val = monthly['fundamental_gap']
ez_val = monthly['erp_zscore']

# vix_term은 Yahoo에서 실시간 업데이트
vix_hist = yf.Ticker('^VIX').history(period='5d', auto_adjust=True)
vix3m_hist = yf.Ticker('^VIX3M').history(period='5d', auto_adjust=True)
vt_val = float(vix_hist['Close'].dropna().iloc[-1]) / float(vix3m_hist['Close'].dropna().iloc[-1])

이 방식으로 10분마다 실행되는 경량 파이프라인에서도 최신 국면을 반영한다.

50일 백필

모델을 새로 학습하면 과거 50일치 국면도 새 모델로 재채점한다. 오래된 모델로 채점된 과거 데이터와 새 모델 사이의 불일치를 최소화하기 위해서다.

for date in spy_dates:
    # 해당 날짜 기준 20일 윈도우로 실시간 피처 재계산
    recent_resid = residuals[residuals.index <= date].iloc[-20:]

    pair_corrs = []
    for sector, stocks in SECTOR_STOCKS.items():
        avail = [s for s in stocks if s in recent_resid.columns]
        if len(avail) < 2:
            continue
        for s1, s2 in combinations(avail, 2):
            c = recent_resid[s1].corr(recent_resid[s2])
            pair_corrs.append(c)
    rc_val = float(np.nanmean(pair_corrs)) if pair_corrs else 0.0

    feat_vec = np.array([[fg_val, ez_val, rc_val, disp_val, ami_val, vt_val, hy_val, rv_val]])
    feat_scaled = scaler.transform(feat_vec)
    ns = float(compute_noise_score(feat_scaled)[0])
    pred_phase, pred_name = score_to_regime_name(ns)

배포 후 발견한 문제

가장 큰 문제는 FRED API 불안정성이었다. 간헐적으로 타임아웃이 나거나 빈 응답을 반환한다. FRED 없이는 hy_spread와 tips_rate 피처를 계산할 수 없어서 전체 파이프라인이 멈춘다.

해결책은 두 단계로 구성했다. 첫째, 재시도 로직을 지수 백오프로 구현했다.

def _fetch_fred(series_id, col_name, retries=3, timeout=(10, 30)):
    for attempt in range(retries):
        try:
            resp = session.get(url, timeout=timeout)
            resp.raise_for_status()
            return df
        except Exception as e:
            if attempt < retries - 1:
                wait = 3 * (2 ** attempt)  # 3초, 6초 대기
                time.sleep(wait)
            else:
                raise

둘째, 전체 파이프라인에서 FRED 데이터를 수집하면 fred_cache.pkl로 저장하고, 경량 파이프라인은 이 캐시를 우선 사용한다.

또 하나는 부호 컨벤션 혼선이었다. 개발 초기에 "양수 = 이성, 음수 = 감정"으로 정의했다가 중간에 반대로 뒤집었다. 이미 DB에 저장된 데이터와 새 데이터가 다른 컨벤션을 가지게 됐고, 결국 전체 DB를 일괄 반전하는 마이그레이션 스크립트를 써야 했다. 컨벤션 변경이 필요할 때는 처음부터 API 레이어에서 변환하는 방식을 잡는 게 맞다.

이 기능을 만들면서 배운 것

HMM은 학습 자체보다 피처 설계가 결과를 좌우한다는 걸 배웠다. fundamental_gap 없이 VIX만 넣으면 변동성 높낮이로만 국면이 분류되고, ERP를 빼면 밸류에이션 신호가 사라진다. 어떤 피처를 넣느냐가 곧 "시장을 어떻게 정의하느냐"의 문제다.

noise_score라는 단일 숫자로 압축하는 결정도 중요했다. HMM이 출력하는 4개 상태 확률보다 단일 점수가 사용자에게 훨씬 직관적으로 전달된다. 복잡한 모델의 결과를 단순화하는 것도 제품을 만드는 일의 일부다.

👉 Passive 바로가기: https://dinsightlab.com