호가창(LOB)이란 무엇인가
호가창(Limit Order Book, LOB)은 특정 자산에 대해 매수·매도 주문이 가격별로 쌓여 있는 데이터 구조입니다. 파이썬 호가창 분석 자동매매는 이 실시간 주문 데이터를 파싱하여 시장 수급 불균형을 감지하고, 자동으로 매매 신호를 생성하는 전략입니다. 단순 캔들 차트 기반 분석과 달리 호가창 데이터는 아직 체결되지 않은 대기 주문을 보여주므로, 가격 변동 전에 시장 참여자들의 의도를 파악할 수 있습니다.
호가창 데이터 구조 이해
호가창은 크게 매수 호가(Bid)와 매도 호가(Ask)로 나뉩니다. 각 호가 레벨에는 가격과 수량이 포함됩니다.
| 구성 요소 | 설명 | 활용 |
|---|---|---|
| Best Bid | 최고 매수 가격 | 즉시 매도 가능 가격 |
| Best Ask | 최저 매도 가격 | 즉시 매수 가능 가격 |
| Spread | Best Ask – Best Bid | 유동성 지표 |
| Depth | 각 가격 레벨의 누적 수량 | 지지/저항 강도 |
| Imbalance | 매수/매도 수량 비율 | 방향성 예측 |
파이썬으로 호가창 데이터 수집하기
바이낸스 거래소의 WebSocket API를 이용하면 실시간 호가창 데이터를 효율적으로 수집할 수 있습니다. ccxt 라이브러리 또는 직접 WebSocket 연결을 사용합니다.
import asyncio
import websockets
import json
from collections import defaultdict
class OrderBookCollector:
"""실시간 호가창 데이터 수집기"""
def __init__(self, symbol: str = "btcusdt", depth: int = 20):
self.symbol = symbol
self.depth = depth
self.orderbook = {"bids": {}, "asks": {}}
self.url = f"wss://stream.binance.com:9443/ws/{symbol}@depth{depth}@100ms"
async def connect(self):
async with websockets.connect(self.url) as ws:
while True:
msg = json.loads(await ws.recv())
self._update(msg)
def _update(self, data: dict):
for bid in data.get("bids", data.get("b", [])):
price, qty = float(bid[0]), float(bid[1])
if qty == 0:
self.orderbook["bids"].pop(price, None)
else:
self.orderbook["bids"][price] = qty
for ask in data.get("asks", data.get("a", [])):
price, qty = float(ask[0]), float(ask[1])
if qty == 0:
self.orderbook["asks"].pop(price, None)
else:
self.orderbook["asks"][price] = qty
def get_snapshot(self) -> dict:
bids = sorted(self.orderbook["bids"].items(), reverse=True)[:self.depth]
asks = sorted(self.orderbook["asks"].items())[:self.depth]
return {"bids": bids, "asks": asks}핵심 지표: 호가 불균형(Order Book Imbalance)
호가 불균형(OBI)은 매수와 매도 주문량의 차이를 수치화한 지표입니다. OBI가 양수이면 매수 압력이 강하고, 음수이면 매도 압력이 강합니다. 이 지표는 단기 가격 방향 예측에 매우 유용합니다.
import numpy as np
from dataclasses import dataclass
@dataclass
class ImbalanceSignal:
obi: float # Order Book Imbalance (-1 ~ 1)
voi: float # Volume Order Imbalance
weighted_mid: float # 가중 중간가
spread_bps: float # 스프레드 (basis points)
class OrderBookAnalyzer:
"""호가창 분석 및 시그널 생성"""
def __init__(self, levels: int = 10):
self.levels = levels
self.history = []
def compute_obi(self, bids: list, asks: list) -> float:
"""기본 호가 불균형 계산"""
bid_vol = sum(qty for _, qty in bids[:self.levels])
ask_vol = sum(qty for _, qty in asks[:self.levels])
if bid_vol + ask_vol == 0:
return 0.0
return (bid_vol - ask_vol) / (bid_vol + ask_vol)
def compute_weighted_obi(self, bids: list, asks: list) -> float:
"""거리 가중 호가 불균형 (가까운 호가에 더 높은 가중치)"""
if not bids or not asks:
return 0.0
mid = (bids[0][0] + asks[0][0]) / 2
bid_weighted = sum(
qty / (1 + abs(price - mid) / mid * 100)
for price, qty in bids[:self.levels]
)
ask_weighted = sum(
qty / (1 + abs(price - mid) / mid * 100)
for price, qty in asks[:self.levels]
)
total = bid_weighted + ask_weighted
if total == 0:
return 0.0
return (bid_weighted - ask_weighted) / total
def compute_voi(self, prev_bids: list, prev_asks: list,
curr_bids: list, curr_asks: list) -> float:
"""Volume Order Imbalance - 호가 변화량 기반"""
if not prev_bids or not curr_bids:
return 0.0
# 매수 측 변화
delta_bid = 0
if curr_bids[0][0] > prev_bids[0][0]:
delta_bid = curr_bids[0][1]
elif curr_bids[0][0] == prev_bids[0][0]:
delta_bid = curr_bids[0][1] - prev_bids[0][1]
# 매도 측 변화
delta_ask = 0
if curr_asks[0][0] < prev_asks[0][0]:
delta_ask = curr_asks[0][1]
elif curr_asks[0][0] == prev_asks[0][0]:
delta_ask = curr_asks[0][1] - prev_asks[0][1]
return delta_bid - delta_ask
def weighted_mid_price(self, bids: list, asks: list) -> float:
"""수량 가중 중간가격"""
if not bids or not asks:
return 0.0
best_bid, bid_qty = bids[0]
best_ask, ask_qty = asks[0]
return (best_bid * ask_qty + best_ask * bid_qty) / (bid_qty + ask_qty)
def analyze(self, snapshot: dict, prev_snapshot: dict = None) -> ImbalanceSignal:
bids, asks = snapshot["bids"], snapshot["asks"]
obi = self.compute_weighted_obi(bids, asks)
mid = self.weighted_mid_price(bids, asks)
spread = (asks[0][0] - bids[0][0]) / mid * 10000 if bids and asks else 0
voi = 0.0
if prev_snapshot:
voi = self.compute_voi(
prev_snapshot["bids"], prev_snapshot["asks"],
bids, asks
)
signal = ImbalanceSignal(obi=obi, voi=voi, weighted_mid=mid, spread_bps=spread)
self.history.append(signal)
return signal대형 주문벽 감지 알고리즘
호가창에서 비정상적으로 큰 주문(일명 “월(Wall)”)은 가격의 지지 또는 저항 역할을 합니다. 이를 자동으로 감지하면 매매 타이밍 결정에 도움이 됩니다. 관련하여 오더플로우 자동매매 분석 글도 참고하세요.
from typing import List, Tuple
@dataclass
class WallDetection:
price: float
quantity: float
side: str # "bid" or "ask"
strength: float # 평균 대비 배수
class WallDetector:
"""대형 주문벽 감지기"""
def __init__(self, threshold_multiplier: float = 3.0):
self.threshold = threshold_multiplier
def detect_walls(self, bids: list, asks: list) -> List[WallDetection]:
walls = []
# 매수벽 감지
bid_quantities = [qty for _, qty in bids]
if bid_quantities:
avg_bid = np.mean(bid_quantities)
for price, qty in bids:
if qty > avg_bid * self.threshold:
walls.append(WallDetection(
price=price, quantity=qty,
side="bid", strength=qty / avg_bid
))
# 매도벽 감지
ask_quantities = [qty for _, qty in asks]
if ask_quantities:
avg_ask = np.mean(ask_quantities)
for price, qty in asks:
if qty > avg_ask * self.threshold:
walls.append(WallDetection(
price=price, quantity=qty,
side="ask", strength=qty / avg_ask
))
return sorted(walls, key=lambda w: w.strength, reverse=True)
def is_spoofing_candidate(self, wall: WallDetection,
cancel_rate: float) -> bool:
"""스푸핑(허위 주문) 의심 판별"""
return wall.strength > 5.0 and cancel_rate > 0.8자동매매 전략 통합
호가창 분석 결과를 실제 자동매매 전략에 통합합니다. OBI와 VOI 시그널을 결합하여 진입/청산 판단을 수행합니다. 자동매매 레이턴시 최적화 기법을 함께 적용하면 체결 품질을 높일 수 있습니다.
from enum import Enum
class Signal(Enum):
LONG = "long"
SHORT = "short"
NEUTRAL = "neutral"
class LOBTradingStrategy:
"""호가창 기반 자동매매 전략"""
def __init__(self, obi_threshold: float = 0.3,
voi_threshold: float = 50.0,
spread_max_bps: float = 5.0):
self.obi_threshold = obi_threshold
self.voi_threshold = voi_threshold
self.spread_max_bps = spread_max_bps
self.analyzer = OrderBookAnalyzer(levels=10)
self.wall_detector = WallDetector(threshold_multiplier=3.0)
self.position = 0.0
def generate_signal(self, snapshot: dict,
prev_snapshot: dict) -> Signal:
signal = self.analyzer.analyze(snapshot, prev_snapshot)
# 스프레드가 너무 넓으면 거래 안 함
if signal.spread_bps > self.spread_max_bps:
return Signal.NEUTRAL
# 대형 벽 확인
walls = self.wall_detector.detect_walls(
snapshot["bids"], snapshot["asks"]
)
# OBI + VOI 결합 시그널
obi_signal = signal.obi
voi_signal = 1 if signal.voi > self.voi_threshold else (
-1 if signal.voi < -self.voi_threshold else 0
)
# 매수벽이 강하면 롱 시그널 강화
bid_walls = [w for w in walls if w.side == "bid"]
ask_walls = [w for w in walls if w.side == "ask"]
wall_bias = 0
if bid_walls and not ask_walls:
wall_bias = 0.1
elif ask_walls and not bid_walls:
wall_bias = -0.1
combined = obi_signal + wall_bias
if combined > self.obi_threshold and voi_signal >= 0:
return Signal.LONG
elif combined < -self.obi_threshold and voi_signal <= 0:
return Signal.SHORT
return Signal.NEUTRAL
async def run(self, collector: OrderBookCollector):
"""메인 실행 루프"""
prev = None
while True:
curr = collector.get_snapshot()
if not curr["bids"] or not curr["asks"]:
await asyncio.sleep(0.1)
continue
if prev:
sig = self.generate_signal(curr, prev)
if sig == Signal.LONG and self.position <= 0:
print(f"[LONG] mid={self.analyzer.history[-1].weighted_mid:.2f}")
self.position = 1.0
elif sig == Signal.SHORT and self.position >= 0:
print(f"[SHORT] mid={self.analyzer.history[-1].weighted_mid:.2f}")
self.position = -1.0
prev = curr
await asyncio.sleep(0.1)백테스트와 성능 평가
호가창 전략의 백테스트는 틱 레벨 데이터가 필요합니다. 바이낸스의 과거 호가 스냅샷을 저장하여 시뮬레이션하는 방법을 소개합니다.
| 성능 지표 | 목표 기준 | 설명 |
|---|---|---|
| 승률 | 55% 이상 | OBI 기반 단기 예측 정확도 |
| 평균 보유 시간 | 1~30초 | 초단타 전략 특성 |
| 손익비 | 1:1 이상 | 고빈도라 손익비보다 승률 중요 |
| 최대 낙폭 | 2% 이내 | 레버리지 미적용 기준 |
| 거래 비용 비율 | 수익의 30% 미만 | 수수료·슬리피지 포함 |
실전 적용 시 주의사항
호가창 기반 자동매매는 강력하지만 몇 가지 함정이 있습니다.
- 스푸핑(Spoofing): 대형 주문이 체결 직전에 취소되는 허위 주문 기법입니다. 주문 체류 시간을 모니터링하여 필터링해야 합니다.
- 히든 오더: 아이스버그 주문처럼 일부만 호가에 노출되는 주문이 있습니다. 체결 데이터와 교차 검증이 필요합니다.
- 레이턴시: 호가 데이터는 밀리초 단위로 변하므로 처리 속도가 핵심입니다. C++ 또는 Rust로 핵심 로직을 포팅하는 경우도 있습니다.
- 거래소 차이: 각 거래소마다 호가 데이터 형식과 업데이트 빈도가 다릅니다. 거래소별 어댑터 패턴을 적용하세요.
- 과적합 위험: 호가 패턴은 시장 환경에 따라 빠르게 변합니다. 정기적인 파라미터 재최적화가 필수입니다.
마치며
파이썬 호가창 분석 자동매매는 시장의 미시구조(Market Microstructure)를 활용하는 고급 전략입니다. OBI, VOI, 주문벽 감지 등의 지표를 조합하면 캔들 차트만으로는 불가능한 초단기 수급 변화를 포착할 수 있습니다. 다만 레이턴시, 거래 비용, 스푸핑 등 실전 리스크를 충분히 고려한 뒤 적용해야 합니다. 볼륨 프로파일 자동매매 전략과 결합하면 더욱 정교한 시스템을 구축할 수 있습니다.