암호화폐 거래에서 호가창(Order Book)은 매수·매도 대기 주문을 실시간으로 보여주는 핵심 데이터입니다. 이 튜토리얼에서는 OKX 거래소의 WebSocket API를 사용하여 깊이 있는 호가창 데이터를 수집하고, 파싱하며, 효과적으로 저장하는 방법을 초보자부터 차근차근 알려드리겠습니다.
1. 호가창 데이터란 무엇인가?
호가창은 특정 거래쌍(예: BTC/USDT)의 현재 대기 중인 매수 주문(Bid)과 매도 주문(Ask)을 가격 순으로 정렬한 표입니다.
호가창 구조 이해하기
- Bid (매수): 다른 사람들이 사고 싶어하는 가격과 수량
- Ask (Sell, 매도): 다른 사람들이 팔고 싶어하는 가격과 수량
- Spread: 가장 낮은 매도가와 가장 높은 매수가의 차이
- Depth: 호가창의 깊이, 즉 몇 단계까지 주문이 있는지
[스크린샷 힌트] 실제 호가창 화면
화면 왼쪽에 초록색 Bid 주문이, 오른쪽에 빨간색 Ask 주문이 테이블 형태로 표시됩니다. 중앙의 파란색 영역이 현재 스프레드입니다.
2. WebSocket이 필요한 이유
tradicionais HTTP API는 클라이언트가 요청을 보내야만 서버가 응답합니다. 하지만 호가창은毫秒(밀리초) 단위로 계속 변하기 때문에:
- HTTP 폴링: 1초마다 요청 → 데이터 지연 발생, 서버 부하 증가
- WebSocket: 서버가 데이터 변경 시 즉시推送 → 실시간, 효율적
3. 사전 준비 작업
3.1 Python 설치
# Python 3.8 이상 권장
Terminal 또는 명령 프롬프트에서 확인
python --version
또는
python3 --version아직 Python이 없다면 python.org에서 다운로드하세요.
3.2 필요한 라이브러리 설치
# WebSocket 클라이언트 라이브러리
pip install websockets
데이터 저장 및 분석용
pip install pandas
실시간 차트 시각화 (선택사항)
pip install matplotlib4. OKX WebSocket API 연결 기본
4.1 연결 주소 확인
OKX는 공식 WebSocket 서버를 제공합니다:
- 실시간 거래:
wss://ws.okx.com:8443/ws/v5/public - 개인 거래:
wss://ws.okx.com:8443/ws/v5/private(인증 필요)
4.2 첫 번째 WebSocket 연결 테스트
import asyncio
import websockets
import json
async def test_connection():
"""OKX WebSocket 서버 연결 테스트"""
url = "wss://ws.okx.com:8443/ws/v5/public"
try:
async with websockets.connect(url) as ws:
print("✅ OKX 서버 연결 성공!")
# 연결 유지 확인용 ping 메시지
ping_msg = {"op": "ping"}
await ws.send(json.dumps(ping_msg))
# 서버 응답 대기
response = await asyncio.wait_for(ws.recv(), timeout=10)
print(f"📥 서버 응답: {response}")
except Exception as e:
print(f"❌ 연결 실패: {e}")
실행
asyncio.run(test_connection())[실행 결과 힌트] 터미널에 "✅ OKX 서버 연결 성공!"과 함께 서버에서 pong 응답이 표시되면 정상입니다.
5. 호가창 데이터 구독하기
5.1 구독 메시지 구조
OKX WebSocket에서 데이터를 받으려면 구독(subscribe) 메시지를 보내야 합니다.
# 호가창 구독 요청 형식
{
"op": "subscribe",
"args": [
{
"channel": "books5", # 5단계 호가창
"instId": "BTC-USDT" # 거래쌍
}
]
}channel 옵션 설명:
books5: 5단계 호가창 (빠른 업데이트용)books50: 50단계 깊이 호가창books-l2-tbt: Tick-by-Tick 전체 호가창
5.2 완전한 호가창 데이터 수신 코드
import asyncio
import websockets
import json
import pandas as pd
from datetime import datetime
class OKXOrderBook:
"""OKX 호가창 데이터 관리 클래스"""
def __init__(self, symbol="BTC-USDT"):
self.symbol = symbol
self.url = "wss://ws.okx.com:8443/ws/v5/public"
self.bids = [] # 매수 주문
self.asks = [] # 매도 주문
self.last_update = None
async def subscribe(self):
"""호가창 데이터 구독"""
async with websockets.connect(self.url) as ws:
# 구독 요청 보내기
subscribe_msg = {
"op": "subscribe",
"args": [{
"channel": "books50",
"instId": self.symbol
}]
}
await ws.send(json.dumps(subscribe_msg))
print(f"📡 {self.symbol} 호가창 구독 시작...")
# 메시지 수신 루프
while True:
try:
message = await ws.recv()
data = json.loads(message)
# 구독 확인 메시지 처리
if "event" in data:
if data["event"] == "subscribe":
print("✅ 구독 성공!")
continue
# 호가창 데이터 파싱
await self.parse_orderbook(data)
except websockets.exceptions.ConnectionClosed:
print("❌ 연결이 종료되었습니다.")
break
async def parse_orderbook(self, data):
"""호가창 데이터 파싱 및 저장"""
try:
# 데이터 구조 확인
if "data" not in data:
return
for item in data["data"]:
# bids: 매수 주문 [가격, 수량, 정밀도]
self.bids = [[float(p), float(q)] for p, q in item.get("bids", [])]
# asks: 매도 주문
self.asks = [[float(p), float(q)] for p, q in item.get("asks", [])]
self.last_update = datetime.now()
# 상위 5개만 출력 (테스트용)
self.print_top_levels()
except Exception as e:
print(f"파싱 오류: {e}")
def print_top_levels(self):
"""상위 호가창 출력"""
print(f"\n⏰ {self.last_update.strftime('%H:%M:%S.%f')[:-3]}")
print("=" * 50)
print("📈 매도 (Ask) 📉 매수 (Bid)")
print("-" * 50)
for i in range(min(5, len(self.asks), len(self.bids))):
ask_price, ask_qty = self.asks[i]
bid_price, bid_qty = self.bids[i]
print(f"{ask_price:>12.2f} | {ask_qty:>10.4f} {bid_price:>12.2f} | {bid_qty:>10.4f}")
# 스프레드 계산
if self.asks and self.bids:
spread = self.asks[0][0] - self.bids[0][0]
spread_pct = (spread / self.asks[0][0]) * 100
print("-" * 50)
print(f"스프레드: {spread:.2f} ({spread_pct:.4f}%)")
실행
async def main():
orderbook = OKXOrderBook("BTC-USDT")
await orderbook.subscribe()
30초간 테스트 후 종료
async def timed_test():
orderbook = OKXOrderBook("BTC-USDT")
await asyncio.wait_for(orderbook.subscribe(), timeout=30)
기본 실행 (무한 루프)
asyncio.run(main())
테스트 실행 (30초)
asyncio.run(timed_test())[실행 결과 힌트] 실행하면 매초마다 BTC-USDT 호가창이 업데이트되며, 매도·매수 가격이 실시간으로 변하는 것을 확인할 수 있습니다.
6. 데이터베이스 저장하기
6.1 SQLite로 간단히 저장
처음에는 가볍고 설치 불필요한 SQLite 데이터베이스를 사용합니다.
import sqlite3
import json
from datetime import datetime
class OrderBookDB:
"""호가창 데이터를 SQLite로 저장하는 클래스"""
def __init__(self, db_name="orderbook.db"):
self.conn = sqlite3.connect(db_name, check_same_thread=False)
self.create_tables()
def create_tables(self):
"""테이블 생성"""
cursor = self.conn.cursor()
# 호가창 스냅샷 테이블
cursor.execute('''
CREATE TABLE IF NOT EXISTS orderbook_snapshots (
id INTEGER PRIMARY KEY AUTOINCREMENT,
symbol TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
best_bid REAL,
best_ask REAL,
spread REAL,
bid_volume REAL,
ask_volume REAL,
raw_data TEXT
)
''')
# 개별 주문 테이블 (세밀한 분석용)
cursor.execute('''
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
symbol TEXT NOT NULL,
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
side TEXT, -- 'bid' 또는 'ask'
price REAL,
quantity REAL,
level INTEGER -- 호가창 깊이 레벨
)
''')
self.conn.commit()
print("📦 데이터베이스 테이블 생성 완료!")
def save_snapshot(self, symbol, bids, asks):
"""호가창 스냅샷 저장"""
if not bids or not asks:
return
cursor = self.conn.cursor()
# 최우선 가격과 거래량 계산
best_bid = bids[0][0]
best_ask = asks[0][0]
spread = best_ask - best_bid
bid_volume = sum(q for _, q in bids)
ask_volume = sum(q for _, q in asks)
cursor.execute('''
INSERT INTO orderbook_snapshots
(symbol, best_bid, best_ask, spread, bid_volume, ask_volume, raw_data)
VALUES (?, ?, ?, ?, ?, ?, ?)
''', (
symbol,
best_bid,
best_ask,
spread,
bid_volume,
ask_volume,
json.dumps({"bids": bids, "asks": asks})
))
self.conn.commit()
return cursor.lastrowid
def save_orders(self, symbol, bids, asks):
"""개별 주문 데이터 저장"""
cursor = self.conn.cursor()
# 매수 주문 저장
for level, (price, qty) in enumerate(bids[:50], 1):
cursor.execute('''
INSERT INTO orders (symbol, side, price, quantity, level)
VALUES (?, 'bid', ?, ?, ?)
''', (symbol, price, qty, level))
# 매도 주문 저장
for level, (price, qty) in enumerate(asks[:50], 1):
cursor.execute('''
INSERT INTO orders (symbol, side, price, quantity, level)
VALUES (?, 'ask', ?, ?, ?)
''', (symbol, price, qty, level))
self.conn.commit()
def get_recent_spreads(self, symbol, limit=100):
"""최근 스프레드 히스토리 조회"""
cursor = self.conn.cursor()
cursor.execute('''
SELECT timestamp, spread, spread * 100.0 / best_ask as spread_pct
FROM orderbook_snapshots
WHERE symbol = ?
ORDER BY timestamp DESC
LIMIT ?
''', (symbol, limit))
return cursor.fetchall()
def close(self):
"""데이터베이스 연결 종료"""
self.conn.close()
print("🔒 데이터베이스 연결 종료")
사용 예시
db = OrderBookDB("btc_orderbook.db")
db.save_snapshot("BTC-USDT", [[50000, 1.5], [49900, 2.0]], [[50010, 1.2], [50020, 3.0]])
print("✅ 데이터 저장 완료!")
최근 스프레드 확인
recent = db.get_recent_spreads("BTC-USDT", 10)
for ts, spread, spread_pct in recent:
print(f"{ts}: 스프레드 {spread:.2f} ({spread_pct:.4f}%)")
db.close()6.2 실시간 저장과 조회 통합
import asyncio
import websockets
import json
from datetime import datetime
위에서 정의한 OrderBookDB 클래스 사용
class OrderBookDB:
# ... 이전 코드와 동일 ...
class LiveOrderBookTracker:
"""실시간 호가창 추적 및 저장"""
def __init__(self, symbol="BTC-USDT", save_interval=60):
self.symbol = symbol
self.url = "wss://ws.okx.com:8443/ws/v5/public"
self.db = OrderBookDB(f"{symbol.replace('-', '_')}.db")
self.save_interval = save_interval # 저장 간격(초)
self.save_count = 0
async def start(self):
"""추적 시작"""
print(f"🚀 {self.symbol} 실시간 호가창 추적 시작")
print(f" 저장 간격: {self.save_interval}초")
async with websockets.connect(self.url) as ws:
# 구독
subscribe_msg = {
"op": "subscribe",
"args": [{
"channel": "books50",
"instId": self.symbol
}]
}
await ws.send(json.dumps(subscribe_msg))
# 데이터 수신 루프
while True:
try:
message = await ws.recv()
data = json.loads(message)
if "data" in data:
for item in data["data"]:
bids = [[float(p), float(q)] for p, q in item.get("bids", [])]
asks = [[float(p), float(q)] for p, q in item.get("asks", [])]
# 실시간 표시
self.display_summary(bids, asks)
# 주기적 저장
self.save_count += 1
if self.save_count >= self.save_interval:
self.db.save_snapshot(self.symbol, bids, asks)
self.db.save_orders(self.symbol, bids, asks)
self.save_count = 0
except Exception as e:
print(f"오류: {e}")
break
def display_summary(self, bids, asks):
"""요약 정보 표시"""
if bids and asks:
spread = asks[0][0] - bids[0][0]
total_bid_vol = sum(q for _, q in bids)
total_ask_vol = sum(q for _, q in asks)
imbalance = (total_bid_vol - total_ask_vol) / (total_bid_vol + total_ask_vol)
imbalance_indicator = "📈 매수 우세" if imbalance > 0 else "📉 매도 우세"
print(f"{datetime.now().strftime('%H:%M:%S')} | "
f"스프레드: {spread:.2f} | "
f"VWAP: {imbalance:+.2%} | "
f"{imbalance_indicator}")
실행
tracker = LiveOrderBookTracker("BTC-USDT", save_interval=30)
asyncio.run(tracker.start())7. 고급 분석 기능
7.1 VWAP (Volume Weighted Average Price) 계산
import pandas as pd
def calculate_vwap(bids, asks):
"""
VWAP 계산: 호가창 깊이별 미결제 거래량 가중 평균 가격
Args:
bids: [[price, quantity], ...]
asks: [[price, quantity], ...]
"""
all_orders = []
# 모든 주문 합치기
for price, qty in bids:
all_orders.append({"price": price, "qty": qty, "side": "bid"})
for price, qty in asks:
all_orders.append({"price": price, "qty": qty, "side": "ask"})
df = pd.DataFrame(all_orders)
# 누적 거래량
df["cumulative_qty"] = df["qty"].cumsum()
df["cumulative_value"] = df["price"] * df["qty"].cumsum()
# VWAP = 총 거래대금 / 총 거래량
total_value = (df["price"] * df["qty"]).sum()
total_qty = df["qty"].sum()
return total_value / total_qty if total_qty > 0 else 0
테스트
bids = [[50000, 1.5], [49900, 2.0], [49800, 3.0]]
asks = [[50010, 1.2], [50020, 3.0], [50030, 2.5]]
vwap = calculate_vwap(bids, asks)
print(f"VWAP: {vwap:.2f} USDT")7.2 시장 미세거래 분석
import pandas as pd
class MarketMicrostructure:
"""시장 미세구조 분석 클래스"""
def __init__(self, symbol):
self.symbol = symbol
def analyze_imbalance(self, bids, asks):
"""호가창 불균형 분석"""
bid_volume = sum(q for _, q in bids)
ask_volume = sum(q for _, q in asks)
total_volume = bid_volume + ask_volume
# 불균형 지표 (-1 ~ +1)
imbalance = (bid_volume - ask_volume) / total_volume if total_volume > 0 else 0
# 해석
if imbalance > 0.3:
interpretation = "강한 매수 압박"
elif imbalance > 0.1:
interpretation = "약한 매수 우세"
elif imbalance < -0.3:
interpretation = "강한 매도 압박"
elif imbalance < -0.1:
interpretation = "약한 매도 우세"
else:
interpretation = "중립"
return {
"bid_volume": bid_volume,
"ask_volume": ask_volume,
"imbalance": imbalance,
"interpretation": interpretation
}
def calculate_depth_metrics(self, bids, asks, levels=10):
"""호가창 깊이 지표 계산"""
bid_levels = bids[:levels]
ask_levels = asks[:levels]
# 특정 가격까지의 누적 거래량
bid_depth = sum(q for _, q in bid_levels)
ask_depth = sum(q for _, q in ask_levels)
# 미결제 거래량 비율
order_ratio = bid_depth / ask_depth if ask_depth > 0 else 0
# VWAP 기반 가격 변동 예상
mid_price = (bids[0][0] + asks[0][0]) / 2 if bids and asks else 0
return {
"bid_depth": bid_depth,
"ask_depth": ask_depth,
"depth_ratio": order_ratio,
"mid_price": mid_price,
"potential_support": bids[levels-1][0] if len(bids) >= levels else bids[-1][0],
"potential_resistance": asks[levels-1][0] if len(asks) >= levels else asks[-1][0]
}
사용 예시
analyzer = MarketMicrostructure("BTC-USDT")
result = analyzer.analyze_imbalance(bids, asks)
print(f"분석 결과: {result['interpretation']} (불균형: {result['imbalance']:.2%})")
depth = analyzer.calculate_depth_metrics(bids, asks)
print(f"지지선 예상: {depth['potential_support']}")
print(f"저항선 예상: {depth['potential_resistance']}")8. 데이터 시각화
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches
def visualize_orderbook(bids, asks, title="호가창 시각화"):
"""호가창을 막대 그래프로 시각화"""
fig, ax = plt.subplots(figsize=(12, 6))
# 가격과 수량 분리
bid_prices = [b[0] for b in bids[:20]]
bid_qtys = [b[1] for b in bids[:20]]
ask_prices = [a[0] for a in asks[:20]]
ask_qtys = [a[1] for a in asks[:20]]
# 매수(초록) vs 매도(빨강)
ax.barh(range(len(bid_prices)), bid_qtys, color='#26a69a', label='Bid (매수)')
ax.barh(range(len(ask_prices)), ask_qtys, color='#ef5350', label='Ask (매도)')
ax.set_yticks(range(max(len(bid_prices), len(ask_prices))))
ax.set_yticklabels(bid_prices[::-1] + ask_prices, fontsize=8)
ax.set_xlabel('수량 (Quantity)')
ax.set_title(title)
ax.legend()
plt.tight_layout()
plt.show()
시각화 실행 (테스트용 더미 데이터)
test_bids = [[50000 - i*10, 1.5 - i*0.05] for i in range(20)]
test_asks = [[50000 + i*10, 1.2 + i*0.05] for i in range(20)]
visualize_orderbook(test_bids, test_asks, "BTC-USDT 호가창")자주 발생하는 오류와 해결책
오류 1: ConnectionRefusedError - WebSocket 연결 거부
# ❌ 오류 메시지:
websockets.exceptions.ConnectionRefusedError: [Errno 111] Connection refused
원인:
1. 잘못된 WebSocket URL
2. 네트워크 연결 문제
3. 방화벽 차단
✅ 해결 방법:
1. URL 확인 (특히 포트 번호)
CORRECT_URL = "wss://ws.okx.com:8443/ws/v5/public" # 포트 8443
WRONG_URL = "wss://ws.okx.com/ws/v5/public" # 포트 누락 ❌
2. 연결 시간 초과 설정 추가
async def connect_with_timeout():
url = "wss://ws.okx.com:8443/ws/v5/public"
try:
async with websockets.connect(url, ping_interval=30, ping_timeout=10) as ws:
# 연결 성공 시 코드
print("연결 성공!")
except websockets.exceptions.ConnectionRefused:
print("⚠️ 연결이 거부되었습니다. URL과 네트워크를 확인하세요.")
except asyncio.TimeoutError:
print("⚠️ 연결 시간이 초과되었습니다.")
3. 프록시 환경에서는:
import os
os.environ['HTTPS_PROXY'] = 'http://your-proxy:port'
os.environ['HTTP_PROXY'] = 'http://your-proxy:port'오류 2: JSONDecodeError - 데이터 파싱 실패
# ❌ 오류 메시지:
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)
원인:
1. 빈 메시지 수신
2. 바이너리 데이터 수신
3. 서버 에러 메시지
✅ 해결 방법:
async def safe_recv(ws):
"""안전한 메시지 수신 및 파싱"""
try:
message = await asyncio.wait_for(ws.recv(), timeout=30)
# 빈 메시지 체크
if not message or message.strip() == '':
return None
# 바이너리 데이터 체크
if isinstance(message, bytes):
# gzip 압축 해제
import gzip
message = gzip.decompress(message).decode('utf-8')
# JSON 파싱
data = json.loads(message)
return data
except json.JSONDecodeError as e:
print(f"⚠️ JSON 파싱 실패: {e}")
return None
except asyncio.TimeoutError:
print("⚠️ 메시지 수신 시간 초과")
return None
except Exception as e:
print(f"⚠️ 예상치 못한 오류: {type(e).__name__}: {e}")
return None
사용법
while True:
data = await safe_recv(ws)
if data is not None:
# 정상 데이터 처리
await process_data(data)오류 3: 중복 구독导致的메시지 폭증
# ❌ 오류 메시지:
같은 구독 채널에서 중복된 데이터 수신
또는 메모리 사용량이 계속 증가
원인:
1. 구독 메시지 여러 번 전송
2. 재연결 시 구독 중복
3. 구독 해제 없이 재구독
✅ 해결 방법:
class SubscriptionManager:
"""구독 상태 관리 클래스"""
def __init__(self):
self.subscriptions = set() # 현재 구독 목록
self.ws = None
async def subscribe(self, ws, channel, instId):
"""멱등성 구독 - 중복 방지"""
key = f"{channel}:{instId}"
# 이미 구독된 경우 무시
if key in self.subscriptions:
print(f"ℹ️ 이미 구독됨: {key}")
return
# 구독 요청
subscribe_msg = {
"op": "subscribe",
"args": [{
"channel": channel,
"instId": instId
}]
}
await ws.send(json.dumps(subscribe_msg))
self.subscriptions.add(key)
print(f"✅ 구독 성공: {key}")
async def unsubscribe(self, ws, channel, instId):
"""구독 취소"""
key = f"{channel}:{instId}"
if key not in self.subscriptions:
return
unsubscribe_msg = {
"op": "unsubscribe",
"args": [{
"channel": channel,
"instId": instId
}]
}
await ws.send(json.dumps(unsubscribe_msg))
self.subscriptions.discard(key)
print(f"🔕 구독 취소: {key}")
def clear_subscriptions(self):
"""모든 구독 초기화 (재연결 시)"""
self.subscriptions.clear()
print("🧹 구독 목록 초기화")
사용법
manager = SubscriptionManager()
재연결 시 구독 목록만 클리어 (새로운 연결에서 재구독)
async def reconnect():
await manager.unsubscribe_all(ws) # 이전 연결 정리
manager.clear_subscriptions()
# 새 연결 수립 후 다시 구독오류 4: Rate Limiting - 요청 제한 초과
# ❌ 오류 메시지:
{"event":"error","msg":"Illegal request","code":"30039"}
원인:
1. 너무 많은 구독 요청
2.短时间内多次重连
✅ 해결 방법:
import asyncio
class RateLimiter:
"""레이트 리미터 - 요청 간격 제어"""
def __init__(self, max_requests=10, time_window=1):
self.max_requests = max_requests
self.time_window = time_window
self.requests = []
async def acquire(self):
"""요청 가능 여부 확인 및 대기"""
now = asyncio.get_event_loop().time()
# 오래된 요청 기록 삭제
self.requests = [t for t in self.requests if now - t < self.time_window]
if len(self.requests) >= self.max_requests:
# 제한 초과 시 대기
wait_time = self.time_window - (now - self.requests[0])
print(f"⏳ Rate limit 대기: {wait_time:.2f}초")
await asyncio.sleep(wait_time)
self.requests.append(now)
async def subscribe_with_limit(self, ws, channel, instId):
"""레이트 리밋 적용 구독"""
await self.acquire()
msg = {"op": "subscribe", "args": [{"channel": channel, "instId": instId}]}
await ws.send(json.dumps(msg))
사용
limiter = RateLimiter(max_requests=5, time_window=1)
여러 구독 시
for symbol in ["BTC-USDT", "ETH-USDT", "SOL-USDT"]:
await limiter.subscribe_with_limit(ws, "books50", symbol)
await asyncio.sleep(0.2) # 각 구독 간 0.2초 간격오류 5: 데이터 순서 꼬임 (Race Condition)
# ❌ 오류 메시지:
호가창 데이터가 비순차적으로 도착하여 상태 불일치
원인:
1. 네트워크 지연 차이
2. 비동기 처리 순서 보장 불가
3. seq不一致
✅ 해결 방법:
import asyncio
from dataclasses import dataclass
from typing import List
@dataclass
class OrderBookSnapshot:
"""호가창 스냅샷"""
seq: int
bids: List[List[float]]
asks: List[List[float]]
timestamp: float
class OrderedOrderBook:
"""순서 보장 호가창 관리"""
def __init__(self):
self.bids = {} # price -> quantity
self.asks = {}
self.last_seq = 0
self.buffer = [] # 순서 맞지 않는 데이터 버퍼
def update(self, data):
"""데이터 업데이트 (순서 보장)"""
seq = int(data.get("seq", 0))
bids = data.get("bids", [])
asks = data.get("asks", [])
# 순서 확인
if seq <= self.last_seq:
print(f"⚠️ 순서 꼬임: seq {seq} < last_seq {self.last_seq}")
return False
# 순서가 맞으면 직접 업데이트
for price, qty in bids:
if qty == 0:
self.bids.pop(float(price), None)
else:
self.bids[float(price)] = float(qty)
for price, qty in asks:
if qty == 0:
self.asks.pop(float(price), None)
else:
self.asks[float(price)] = float(qty)
self.last_seq = seq
return True
def get_sorted(self, side="bid", limit=20):
"""정렬된 호가 반환"""
if side == "bid":
return sorted(self.bids.items(), key=lambda x: -x[0])[:limit]
else:
return sorted(self.asks.items(), key=lambda x: x[0])[:limit]
사용
orderbook = OrderedOrderBook()
데이터 수신 시
orderbook.update({"seq": 1001, "bids": bids, "asks": asks})9. 실제 애플리케이션 예시
9.1 시장気配分析기
import asyncio
import websockets
import json
from datetime import datetime
class MarketSentimentAnalyzer:
"""시장 분위기 분석기"""
def __init__(self, symbols=["BTC-USDT", "ETH-USDT"]):
self.symbols = symbols
self.orderbooks = {s: {"bids": [], "asks": []} for s in symbols}
self.url = "wss://ws.okx.com:8443/ws/v5/public"
async def start(self):
"""분석 시작"""
async with websockets.connect(self.url) as ws:
# 모든 심볼 구독
for symbol in self.symbols:
msg = {
"op": "subscribe",
"args": [{"channel": "books5", "instId": symbol}]
}
await ws.send(json.dumps(msg))
print("