개요: 호가창 불균형도(Order Book Imbalance)란?
호가창 불균형도는 특정 순간에 매수호가와 매도호가의 잔량을 비교하여 시장 심리를 정량화하는 지표입니다. 저는 3년간 암호화폐 단타 거래를 해왔는데, 이 지표가 단기 수익을 예측하는 데 놀라울 정도로 효과적이라는 것을 발견했습니다.
핵심 공식:
OBI = (매수호가 총잔량 - 매도호가 총잔량) / (매수호가 총잔량 + 매도호가 총잔량)
- OBI가 +0.5 이상이면 매수 우위 (Bullish 시그널)
- OBI가 -0.5 이하이면 매도 우위 (Bearish 시그널)
- OBI가 0에 가까우면 중립 구간
왜 이 지표가 단타에 유용한가?
저는 여러 단타 지표를 비교 분석했었고, 호가창 불균형도는 다음과 같은 이유로有效性(유효성)이 높았습니다:
- 초단타 대응: 온체인 데이터보다 실시간 호가창 데이터가 더 빠르게 반응
- 노이즈 필터링: 단순 가격 차트보다 매수/매도 압력의 방향성이 명확
- 백테스트 적합: 명확한 수치 기준(閾値(threshold)) 설정 가능
필수 라이브러리 설치
pip install requests pandas numpy scipy ccxt websocket-client
# 백테스트에 필요한 기본 설정
import requests
import pandas as pd
import numpy as np
import time
import json
from datetime import datetime
HolySheep AI API 설정
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
def holy_sheep_chat(prompt, model="gpt-4.1"):
"""HolySheep AI를 통한 시장 분석"""
headers = {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3 # 분석이므로 낮은 temperature 사용
}
response = requests.post(f"{BASE_URL}/chat/completions",
headers=headers, json=payload)
return response.json()
print("HolySheep AI 연결 완료!")
호가창 데이터 수집 클래스 구현
import ccxt
import asyncio
from collections import deque
class OrderBookAnalyzer:
def __init__(self, exchange_id='binance', symbol='BTC/USDT'):
self.exchange = getattr(ccxt, exchange_id)()
self.symbol = symbol
self.orderbook_history = deque(maxlen=1000)
self.obi_threshold_long = 0.4 # 롱 진입 임계값
self.obi_threshold_short = -0.4 # 숏 진입 임계값
def calculate_obi(self, orderbook):
"""호가창 불균형도 계산"""
bids_total = sum([float(b[1]) for b in orderbook['bids'][:10]])
asks_total = sum([float(a[1]) for a in orderbook['asks'][:10]])
if bids_total + asks_total == 0:
return 0
return (bids_total - asks_total) / (bids_total + asks_total)
def get_current_obi(self):
"""현재 호가창 불균형도 조회"""
orderbook = self.exchange.fetch_order_book(self.symbol)
obi = self.calculate_obi(orderbook)
return obi, orderbook
def generate_signal(self, obi):
"""매매 시그널 생성"""
if obi > self.obi_threshold_long:
return "LONG"
elif obi < self.obi_threshold_short:
return "SHORT"
return "HOLD"
테스트 실행
analyzer = OrderBookAnalyzer('binance', 'BTC/USDT')
current_obi, orderbook = analyzer.get_current_obi()
signal = analyzer.generate_signal(current_obi)
print(f"현재 OBI: {current_obi:.4f}")
print(f"생성된 시그널: {signal}")
백테스트 엔진 구현
import numpy as np
from dataclasses import dataclass
from typing import List, Tuple
@dataclass
class Trade:
entry_time: str
entry_price: float
direction: str # "LONG" or "SHORT"
exit_time: str
exit_price: float
pnl_percent: float
class BacktestEngine:
def __init__(self, initial_capital=10000, fee=0.001):
self.initial_capital = initial_capital
self.fee = fee # 거래 수수료 0.1%
self.trades: List[Trade] = []
self.capital_history = [initial_capital]
def run_backtest(self, price_data, obi_data,
holding_period=5, stop_loss=0.02, take_profit=0.03):
"""
백테스트 실행
Args:
price_data: 가격 데이터 리스트
obi_data: OBI 데이터 리스트
holding_period:持仓 기간 (분)
stop_loss:止损 비율
take_profit:利確 비율
"""
position = None
entry_price = 0
entry_time = 0
for i in range(len(price_data) - holding_period):
current_price = price_data[i]
current_obi = obi_data[i]
# 포지션 없음 → 시그널 확인
if position is None:
if current_obi > 0.4: # 롱 시그널
position = "LONG"
entry_price = current_price * (1 + self.fee)
entry_time = i
elif current_obi < -0.4: # 숏 시그널
position = "SHORT"
entry_price = current_price * (1 + self.fee)
entry_time = i
# 포지션 있음 → 종료 조건 확인
elif position == "LONG":
price_change = (current_price - entry_price) / entry_price
if price_change <= -stop_loss:
self._close_trade("STOP_LOSS", current_price, entry_time, i, position)
position = None
elif price_change >= take_profit:
self._close_trade("TAKE_PROFIT", current_price, entry_time, i, position)
position = None
elif i - entry_time >= holding_period:
self._close_trade("TIME_EXIT", current_price, entry_time, i, position)
position = None
elif position == "SHORT":
price_change = (entry_price - current_price) / entry_price
if