은닉 마르코프 모델이란?
은닉 마르코프 모델(Hidden Markov Model, HMM)은 관측할 수 없는 숨겨진 상태가 시간에 따라 전이되며, 각 상태에서 관측 가능한 데이터를 생성한다고 가정하는 확률 모형입니다. 금융 시장에서는 상승장·횡보장·하락장 같은 보이지 않는 시장 레짐을 숨겨진 상태로 모델링하고, 수익률·변동성 등 관측 데이터로부터 현재 레짐을 추론합니다.
기존 이동평균이나 RSI 같은 기술적 지표가 과거 가격의 단순 변환인 반면, HMM은 시장의 구조적 상태 변화를 확률적으로 포착하여 더 근본적인 매매 신호를 제공합니다.
HMM의 3가지 핵심 구성요소
| 구성요소 | 수학적 표현 | 매매 전략 해석 |
|---|---|---|
| 초기 확률 (π) | 각 상태의 시작 확률 | 시장이 특정 레짐에서 시작할 확률 |
| 전이 확률 (A) | 상태 간 전환 확률 행렬 | 상승장→하락장 전환 확률 등 |
| 방출 확률 (B) | 각 상태에서 관측값 분포 | 각 레짐의 수익률·변동성 분포 |
파이썬 HMM 레짐 감지 구현
hmmlearn 라이브러리를 활용하여 가격 데이터에서 시장 레짐을 자동 분류하는 코드입니다.
import numpy as np
import pandas as pd
from hmmlearn.hmm import GaussianHMM
from sklearn.preprocessing import StandardScaler
class HMMRegimeDetector:
def __init__(self, n_regimes=3, n_iter=200):
"""
:param n_regimes: 레짐 수 (기본 3: 상승/횡보/하락)
:param n_iter: EM 알고리즘 반복 횟수
"""
self.n_regimes = n_regimes
self.model = GaussianHMM(
n_components=n_regimes,
covariance_type='full',
n_iter=n_iter,
random_state=42
)
self.scaler = StandardScaler()
self.regime_map = {}
def prepare_features(self, prices):
"""가격 데이터에서 특성 추출"""
df = pd.DataFrame({'price': prices})
df['returns'] = df['price'].pct_change()
df['volatility'] = df['returns'].rolling(20).std()
df['momentum'] = df['price'].pct_change(10)
df = df.dropna()
features = df[['returns', 'volatility', 'momentum']].values
return features, df.index
def fit(self, prices):
"""HMM 학습"""
features, idx = self.prepare_features(prices)
scaled = self.scaler.fit_transform(features)
self.model.fit(scaled)
# 레짐 라벨링 (평균 수익률 기준 정렬)
means = self.model.means_[:, 0] # 수익률 평균
sorted_idx = np.argsort(means)
labels = ['하락', '횡보', '상승']
for i, si in enumerate(sorted_idx):
self.regime_map[si] = labels[i]
return self
def predict_regime(self, prices):
"""현재 레짐 예측"""
features, idx = self.prepare_features(prices)
scaled = self.scaler.transform(features)
states = self.model.predict(scaled)
probs = self.model.predict_proba(scaled)
current_state = states[-1]
return {
'regime': self.regime_map.get(
current_state, f'state_{current_state}'
),
'state_id': int(current_state),
'probabilities': {
self.regime_map.get(i, f'state_{i}'): float(p)
for i, p in enumerate(probs[-1])
},
'transition_matrix': self.model.transmat_.tolist()
}
def regime_history(self, prices):
"""전체 기간 레짐 히스토리"""
features, idx = self.prepare_features(prices)
scaled = self.scaler.transform(features)
states = self.model.predict(scaled)
return [
self.regime_map.get(s, f'state_{s}')
for s in states
]HMM 기반 자동매매 봇
레짐 감지 결과를 실시간 매매에 적용하는 전략 봇입니다. 각 레짐별로 다른 매매 규칙을 적용합니다.
import ccxt
import time
from datetime import datetime
class HMMTradingBot:
def __init__(self, exchange, symbol, capital,
lookback=500):
self.exchange = exchange
self.symbol = symbol
self.capital = capital
self.lookback = lookback
self.detector = HMMRegimeDetector(n_regimes=3)
self.position = 0
self.trades = []
self.trained = False
# 레짐별 전략 설정
self.regime_strategy = {
'상승': {
'action': 'long',
'size_pct': 0.3,
'stop_loss': 0.03
},
'횡보': {
'action': 'neutral',
'size_pct': 0,
'stop_loss': 0.01
},
'하락': {
'action': 'short',
'size_pct': 0.2,
'stop_loss': 0.02
}
}
def fetch_prices(self):
"""과거 가격 데이터 조회"""
ohlcv = self.exchange.fetch_ohlcv(
self.symbol, '1h',
limit=self.lookback
)
return [c[4] for c in ohlcv] # 종가
def train_model(self):
"""HMM 모델 학습"""
prices = self.fetch_prices()
self.detector.fit(prices)
self.trained = True
print("[HMM] 모델 학습 완료")
def get_signal(self):
"""현재 레짐 기반 매매 신호"""
prices = self.fetch_prices()
if not self.trained:
self.detector.fit(prices)
self.trained = True
result = self.detector.predict_regime(prices)
regime = result['regime']
confidence = result['probabilities'].get(regime, 0)
strategy = self.regime_strategy.get(regime)
return {
'regime': regime,
'confidence': confidence,
'action': strategy['action'],
'size_pct': strategy['size_pct'],
'stop_loss': strategy['stop_loss'],
'transition_probs': result['probabilities']
}
def execute(self, signal, current_price):
"""매매 실행"""
action = signal['action']
confidence = signal['confidence']
size = signal['size_pct'] * confidence
# 신뢰도 60% 미만이면 매매 안 함
if confidence < 0.6:
return
target_qty = (self.capital * size) / current_price
if action == 'long' and self.position <= 0:
# 숏 청산 후 롱 진입
if self.position < 0:
self.exchange.create_market_order(
self.symbol, 'buy', abs(self.position)
)
if target_qty > 0:
self.exchange.create_market_order(
self.symbol, 'buy', target_qty
)
self.position = target_qty
self._log('LONG', target_qty, current_price,
signal)
elif action == 'short' and self.position >= 0:
if self.position > 0:
self.exchange.create_market_order(
self.symbol, 'sell', self.position
)
if target_qty > 0:
self.exchange.create_market_order(
self.symbol, 'sell', target_qty
)
self.position = -target_qty
self._log('SHORT', target_qty, current_price,
signal)
elif action == 'neutral' and self.position != 0:
side = 'sell' if self.position > 0 else 'buy'
self.exchange.create_market_order(
self.symbol, side, abs(self.position)
)
self._log('FLAT', abs(self.position),
current_price, signal)
self.position = 0
def run(self, retrain_hours=24, check_interval=300):
"""
메인 루프
:param retrain_hours: 모델 재학습 주기(시간)
:param check_interval: 신호 체크 간격(초)
"""
last_train = 0
print(f"[HMM 봇 시작] {self.symbol}")
while True:
try:
# 주기적 재학습
now = time.time()
if now - last_train > retrain_hours * 3600:
self.train_model()
last_train = now
signal = self.get_signal()
ticker = self.exchange.fetch_ticker(self.symbol)
price = ticker['last']
print(f"[{datetime.now().strftime('%H:%M')}] "
f"레짐={signal['regime']} "
f"({signal['confidence']:.1%}) "
f"| 포지션={self.position:.4f}")
self.execute(signal, price)
except Exception as e:
print(f"[오류] {e}")
time.sleep(check_interval)
def _log(self, action, qty, price, signal):
self.trades.append({
'time': datetime.now().isoformat(),
'action': action,
'qty': qty,
'price': price,
'regime': signal['regime'],
'confidence': signal['confidence']
})최적 레짐 수 결정
HMM의 성능은 레짐 수(n_components)에 크게 좌우됩니다. BIC(베이지안 정보 기준)와 AIC(아카이케 정보 기준)로 최적 레짐 수를 결정합니다.
def find_optimal_regimes(prices, max_regimes=6):
"""BIC 기반 최적 레짐 수 탐색"""
detector = HMMRegimeDetector()
features, _ = detector.prepare_features(prices)
scaler = StandardScaler()
scaled = scaler.fit_transform(features)
results = []
for n in range(2, max_regimes + 1):
model = GaussianHMM(
n_components=n,
covariance_type='full',
n_iter=200,
random_state=42
)
model.fit(scaled)
bic = -2 * model.score(scaled) * len(scaled)
+ n * (n + 2 * features.shape[1]) * np.log(len(scaled))
results.append({'n_regimes': n, 'bic': bic,
'log_likelihood': model.score(scaled)})
print(f" n={n} | BIC={bic:.1f}")
best = min(results, key=lambda x: x['bic'])
print(f"n최적 레짐 수: {best['n_regimes']}")
return best['n_regimes']HMM 매매 전략 장단점
| 장점 | 단점 |
|---|---|
| 시장 구조 변화를 확률적으로 포착 | 학습 데이터에 과적합 위험 |
| 레짐별 차별화된 전략 적용 가능 | 레짐 전환 감지에 지연 발생 |
| 전이 확률로 레짐 지속 기간 예측 | 가우시안 가정이 꼬리 위험 과소평가 |
| 다른 전략과 앙상블 가능 | 하이퍼파라미터 민감도 높음 |
실전 운영 팁
- 롤링 윈도우 재학습 — 고정 데이터가 아닌 최근 N일 데이터로 주기적 재학습하여 시장 변화 반영
- 다중 타임프레임 — 1시간·4시간·일봉 각각에 HMM을 적용하고, 다수결로 최종 레짐 결정
- 전이 확률 활용 — 현재 레짐의 자기 전이 확률이 높으면 포지션 유지, 낮으면 축소
- 신뢰도 기반 사이징 — 레짐 확률이 높을수록 포지션 크기 확대
- 워크포워드 검증 — 학습·검증 기간을 분리한 워크포워드 백테스트로 과적합 방지
관련 글
- 시장 레짐 감지 퀀트 전략 — 레짐 기반 투자의 기초 개념과 다양한 접근법
- 멀티 전략 앙상블 자동매매 — HMM을 포함한 복합 전략 포트폴리오 구성
- XGBoost 퀀트 매매 신호 예측 — ML 기반 매매 신호 생성의 다른 접근