고빈도 트레이딩 시스템, 시차 arbitrage, 유동성 분석 등 실무에서 L2 오더북 데이터는 필수입니다. 그러나 Binance, OKX, Bybit는 각자 다른 데이터 포맷을 사용하며, 이를 단일 시스템에서 처리하려면 구조적 이해와 표준화 전략이 필요합니다. 3년간 글로벌 선물 거래소 API를 분석하며 축적한 경험을 바탕으로 실제 데이터 구조와 검증된 코드를 공유합니다.
L2 오더북이란?
L2(Level-2) 오더북은 특정 자산의 매수/매도 호가를 가격 단위로 정리한 데이터입니다. 각 호가창에는 다음과 같은 핵심 요소가 포함됩니다:
- Bid(매수호가): 시장 참여자가 구매를 원하는 가격과 수량
- Ask(매도호가): 시장 참여자가 판매를 원하는 가격과 수량
- 스프레드: 최우선 매도가와 최우선 매수가의 차이
- 진입 장벽: 특정 가격 레벨의 누적 호가량
3대 거래소 L2 오더북 구조 비교
실제 API 응답을 파싱하며 구조적 차이점을 확인하겠습니다. 아래 표는 각 거래소의 핵심 필드 차이를 보여줍니다.
| 속성 | Binance Futures | OKX | Bybit |
|---|---|---|---|
| 최우선 매수가(Bid) | bids[0] = [price, qty] | bids[0] = [price, volume, orders] | b[0] = [price, qty] |
| 최우선 매도가(Ask) | asks[0] = [price, qty] | asks[0] = [price, volume, orders] | a[0] = [price, qty] |
| 수량 단위 | kontrakt (USD) | kontrakt (USD) | kontrakt (USD) |
| 필드명 규칙 | camelCase | PascalCase | short |
| 웹소켓 구독 | @depth@100ms | depth | |
| 최대 깊이 | 5,000 레벨 | 400 레벨 | 200 레벨 |
실전 코드: 각 거래소 API 파싱
Binance Futures 오더북 파싱
Binance는 가장 직관적인 구조를 가지며, 1,000ms 또는 100ms 업데이트 간격을 지원합니다. 저는 100ms 간격을 선호하는데,高频 전략에서 스프레드 변화를 놓치지 않기 때문입니다.
"""
Binance Futures L2 오더북 실시간 수신
base_url: https://api.holysheep.ai/v1 (HolySheep 사용)
"""
import websockets
import asyncio
import json
from dataclasses import dataclass
from typing import List, Tuple
@dataclass
class OrderBookLevel:
"""오더북 단일 레벨 데이터"""
price: float
quantity: float
total: float = 0.0 # 누적 수량
def __repr__(self):
return f"Price: {self.price:.2f}, Qty: {self.quantity:.4f}"
@dataclass
class OrderBook:
"""정제된 오더북 데이터"""
symbol: str
bids: List[OrderBookLevel] # 매수호가 (가격 내림차순)
asks: List[OrderBookLevel] # 매도호가 (가격 오름차순)
timestamp: int
exchange: str = "binance"
@property
def best_bid(self) -> float:
return self.bids[0].price if self.bids else 0.0
@property
def best_ask(self) -> float:
return self.asks[0].price if self.asks else 0.0
@property
def spread(self) -> float:
return self.best_ask - self.best_bid
@property
def mid_price(self) -> float:
return (self.best_bid + self.best_ask) / 2
async def connect_binance_orderbook(symbol: str = "btcusdt"):
"""Binance 웹소켓을 통한 L2 오더북 수신"""
# HolySheep API 게이트웨이 사용 시에도 웹소켓은 각 거래소 직접 연결
stream_name = f"{symbol}@depth@100ms"
ws_url = f"wss://fstream.binance.com/ws/{stream_name}"
async with websockets.connect(ws_url) as ws:
print(f"[Binance] {symbol} 오더북 수신 시작")
while True:
try:
data = await asyncio.wait_for(ws.recv(), timeout=30)
msg = json.loads(data)
# Binance 포맷 파싱
raw_bids = msg.get('b', [])
raw_asks = msg.get('a', [])
# OrderBookLevel로 변환
bids = [
OrderBookLevel(price=float(b[0]), quantity=float(b[1]))
for b in raw_bids
]
asks = [
OrderBookLevel(price=float(a[0]), quantity=float(a[1]))
for a in raw_asks
]
# 누적 수량 계산
cumsum = 0.0
for bid in bids:
cumsum += bid.quantity
bid.total = cumsum
cumsum = 0.0
for ask in asks:
cumsum += ask.quantity
ask.total = cumsum
orderbook = OrderBook(
symbol=symbol.upper(),
bids=bids,
asks=asks,
timestamp=msg.get('E', 0),
exchange='binance'
)
# 실시간 지표 출력
print(f"[{orderbook.exchange.upper()}] "
f"Bid: {orderbook.best_bid:.2f} | "
f"Ask: {orderbook.best_ask:.2f} | "
f"Spread: {orderbook.spread:.2f} | "
f"Mid: {orderbook.mid_price:.2f}")
except asyncio.TimeoutError:
print("[Binance] 하트비트 타임아웃, 재연결...")
break
except Exception as e:
print(f"[Binance] 오류: {e}")
break
실행
if __name__ == "__main__":
asyncio.run(connect_binance_orderbook("btcusdt"))
OKX 오더북 파싱 (3필드 구조)
OKX의 오더북 구조는 Binance와 달리 각 레벨에 주문 수(orders count) 필드가 추가됩니다. iceberh 주문 탐지나 주문집중도 분석 시 이 필드가 유용합니다.
"""
OKX 오더북 구조 분석 및 파싱
key 차이: [price, volume, orders_count] - 3필드 구조
"""
import websockets
import asyncio
import json
from typing import Dict, List
class OKXOrderBookParser:
"""OKX L2 오더북 파서"""
@staticmethod
def parse_snapshot(data: Dict) -> Dict:
"""초기 스냅샷 파싱"""
result = data.get('data', [{}])[0]
symbol = result.get('instId', '')
ts = int(result.get('ts', 0))
# OKX 구조: [price, size, orders]
raw_bids = result.get('bids', [])
raw_asks = result.get('asks', [])
bids = []
for price, size, orders in raw_bids:
bids.append({
'price': float(price),
'quantity': float(size),
'orders': int(orders),
'cum_qty': 0.0
})
asks = []
for price, size, orders in raw_asks:
asks.append({
'price': float(price),
'quantity': float(size),
'orders': int(orders),
'cum_qty': 0.0
})
# 누적 수량 계산
cumsum = 0.0
for bid in bids:
cumsum += bid['quantity']
bid['cum_qty'] = cumsum
cumsum = 0.0
for ask in asks:
cumsum += ask['quantity']
ask['cum_qty'] = cumsum
return {
'symbol': symbol,
'timestamp': ts,
'bids': bids,
'asks': asks
}
@staticmethod
def parse_delta(update: Dict) -> Dict:
"""增量 업데이트 파싱 (실시간 업데이트용)"""
data = update.get('data', [{}])[0]
action = update.get('action', '') # 'snapshot' or 'update'
result = {
'action': action,
'symbol': data.get('instId', ''),
'timestamp': int(data.get('ts', 0)),
'bids': [
{'price': float(p), 'quantity': float(s), 'orders': int(o)}
for p, s, o in data.get('bids', [])
],
'asks': [
{'price': float(p), 'quantity': float(s), 'orders': int(o)}
for p, s, o in data.get('asks', [])
]
}
# quantity가 0이면 해당 레벨 삭제 의미
result['bids'] = [b for b in result['bids'] if b['quantity'] > 0]
result['asks'] = [a for a in result['asks'] if a['quantity'] > 0]
return result
@staticmethod
def calculate_vwap(orderbook: Dict, levels: int = 10) -> Dict:
"""VWAP(가중평균가격) 계산"""
bid_prices, bid_qtys = [], []
ask_prices, ask_qtys = [], []
for bid in orderbook['bids'][:levels]:
bid_prices.append(bid['price'])
bid_qtys.append(bid['quantity'])
for ask in orderbook['asks'][:levels]:
ask_prices.append(ask['price'])
ask_qtys.append(ask['quantity'])
bid_vwap = sum(p*q for p,q in zip(bid_prices, bid_qtys)) / sum(bid_qtys) if bid_qtys else 0
ask_vwap = sum(p*q for p,q in zip(ask_prices, ask_qtys)) / sum(ask_qtys) if ask_qtys else 0
return {'bid_vwap': bid_vwap, 'ask_vwap': ask_vwap}
async def connect_okx_orderbook(symbol: str = "BTC-USDT-SWAP"):
"""OKX 웹소켓订阅"""
ws_url = "wss://ws.okx.com:8443/ws/v5/public"
params = {
"op": "subscribe",
"args": [{
"channel": "books-l2-tbt", # 최우선 50 레벨, 10ms 딜레이
"instId": symbol
}]
}
orderbook_state = {'bids': {}, 'asks': {}} # 가격 -> 수량 매핑
async with websockets.connect(ws_url) as ws:
await ws.send(json.dumps(params))
print(f"[OKX] {symbol} 구독 완료")
async for msg in ws:
data = json.loads(msg)
if data.get('event') == 'subscribe':
print(f"[OKX] 구독 성공: {data}")
continue
if 'data' not in data:
continue
result = OKXOrderBookParser.parse_delta(data)
# 상태 업데이트
for bid in result['bids']:
if bid['quantity'] == 0:
orderbook_state['bids'].pop(bid['price'], None)
else:
orderbook_state['bids'][bid['price']] = bid['quantity']
for ask in result['asks']:
if ask['quantity'] == 0:
orderbook_state['asks'].pop(ask['price'], None)
else:
orderbook_state['asks'][ask['quantity']] = ask['quantity']
# 정렬 및 상위 5개 출력
sorted_bids = sorted(orderbook_state['bids'].items(), key=lambda x: -x[0])[:5]
sorted_asks = sorted(orderbook_state['asks'].items(), key=lambda x: x[0])[:5]
print(f"[OKX] Bid: {sorted_bids[:2]} | Ask: {sorted_asks[:2]}")
if __name__ == "__main__":
asyncio.run(connect_okx_orderbook("BTC-USDT-SWAP"))
Bybit 오더북 파싱
Bybit는 필드명을 축약하여 사용하며, REST와 WebSocket의 응답 구조가 다릅니다. 저는 최초 스냅샷 요청 후 실시간 delta 업데이트를 병행하는 방식을 사용합니다.
"""
Bybit L2 오더북 파싱 및 상태 관리
"""
import aiohttp
import asyncio
import json
from collections import OrderedDict
class BybitOrderBook:
"""Bybit 오더북 상태 관리"""
def __init__(self, symbol: str, category: str = "linear"):
self.symbol = symbol
self.category = category
self.bids = OrderedDict() # price -> quantity
self.asks = OrderedDict()
self.last_update_id = 0
self.seq = 0
async def fetch_snapshot(self, limit: int = 200) -> Dict:
"""REST API로 스냅샷 가져오기"""
url = "https://api.bybit.com/v5/market/orderbook"
params = {
"category": self.category,
"symbol": self.symbol,
"limit": limit
}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as resp:
data = await resp.json()
result = data.get('result', {})
self.last_update_id = result.get('updateId', 0)
# Bybit 구조: s= Symbol, b= Bid, a= Ask
for price, qty in result.get('b', []):
self.bids[float(price)] = float(qty)
for price, qty in result.get('a', []):
self.asks[float(price)] = float(qty)
print(f"[Bybit] 스냅샷 로드 완료: {len(self.bids)} bids, {len(self.asks)} asks")
return result
def apply_delta(self, data: Dict) -> bool:
"""增量 업데이트 적용"""
update_id = data.get('u', 0) # updateId
# 시퀀스 검증
if update_id <= self.last_update_id:
return False
# Bybit 구조: b= Bid delta, a= Ask delta
for price, qty in data.get('b', []):
price_f, qty_f = float(price), float(qty)
if qty_f == 0:
self.bids.pop(price_f, None)
else:
self.bids[price_f] = qty_f
for price, qty in data.get('a', []):
price_f, qty_f = float(price), float(qty)
if qty_f == 0:
self.asks.pop(price_f, None)
else:
self.asks[price_f] = qty_f
self.last_update_id = update_id
return True
def get_top_levels(self, n: int = 5) -> Dict:
"""최우선 N 레벨 반환"""
sorted_bids = sorted(self.bids.items(), key=lambda x: -x[0])[:n]
sorted_asks = sorted(self.asks.items(), key=lambda x: x[0])[:n]
return {
'symbol': self.symbol,
'bids': [{'price': p, 'qty': q} for p, q in sorted_bids],
'asks': [{'price': p, 'qty': q} for p, q in sorted_asks],
'spread': sorted_asks[0][0] - sorted_bids[0][0] if sorted_bids and sorted_asks else 0
}
def calculate_imbalance(self) -> float:
"""오더북 불균형 계산"""
top_bid_qty = sum(list(self.bids.values())[:5])
top_ask_qty = sum(list(self.asks.values())[:5])
total = top_bid_qty + top_ask_qty
if total == 0:
return 0.0
# 양수 =買い圧力, 음수 = 매도压力
return (top_bid_qty - top_ask_qty) / total
async def connect_bybit_orderbook(symbol: str = "BTCUSDT"):
"""Bybit 웹소켓订阅"""
ob = BybitOrderBook(symbol)
await ob.fetch_snapshot()
ws_url = "wss://stream.bybit.com/v5/market/orderbook"
async with aiohttp.ClientSession() as session:
async with session.ws_connect(ws_url) as ws:
# 구독 요청
await ws.send_json({
"op": "subscribe",
"args": [f"orderbook.200.{symbol}"] # 200 레벨, 100ms
})
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
if 'data' in data:
for update in data['data']:
if ob.apply_delta(update):
levels = ob.get_top_levels(3)
imbalance = ob.calculate_imbalance()
print(f"[Bybit] "
f"B: {levels['bids'][0]} | "
f"A: {levels['asks'][0]} | "
f"Imbalance: {imbalance:+.2%}")
if __name__ == "__main__":
asyncio.run(connect_bybit_orderbook("BTCUSDT"))
크로스 거래소 데이터 표준화
실제 트레이딩 시스템에서는 3개 거래소의 데이터를 통합 처리해야 합니다. 아래 표준화 레이어를 통해 일관된 인터페이스를 제공합니다.
"""
크로스 거래소 L2 오더북 표준화 레이어
모든 거래소를 UnifiedOrderBook 인터페이스로 통합
"""
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from enum import Enum
import time
class Exchange(Enum):
BINANCE = "binance"
OKX = "okx"
BYBIT = "bybit"
@dataclass
class UnifiedLevel:
"""표준화된 호가 레벨"""
price: float
quantity: float
orders_count: int = 1 # OKX에서만 의미 있음
exchange: str = ""
@dataclass
class UnifiedOrderBook:
"""크로스 거래소 통합 오더북"""
symbol: str # 정규화 심볼 (例: BTC-USDT)
exchange: Exchange
bids: List[UnifiedLevel] = field(default_factory=list)
asks: List[UnifiedLevel] = field(default_factory=list)
timestamp_ms: int = 0
local_timestamp_ms: int = 0
def __post_init__(self):
if self.local_timestamp_ms == 0:
self.local_timestamp_ms = int(time.time() * 1000)
@property
def best_bid(self) -> Optional[float]:
return self.bids[0].price if self.bids else None
@property
def best_ask(self) -> Optional[float]:
return self.asks[0].price if self.asks else None
@property
def spread_bps(self) -> float:
"""스프레드 basis points 단위"""
if self.best_bid and self.best_ask:
return (self.best_ask - self.best_bid) / self.best_bid * 10000
return 0.0
def to_dict(self) -> Dict:
return {
'symbol': self.symbol,
'exchange': self.exchange.value,
'best_bid': self.best_bid,
'best_ask': self.best_ask,
'spread_bps': self.spread_bps,
'bid_depth': sum(b.quantity for b in self.bids[:10]),
'ask_depth': sum(a.quantity for a in self.asks[:10]),
'timestamp_ms': self.timestamp_ms
}
class OrderBookNormalizer(ABC):
"""거래소별 오더북 정규화 추상 클래스"""
@abstractmethod
def normalize(self, raw_data: Dict) -> UnifiedOrderBook:
pass
@staticmethod
def normalize_symbol(symbol: str, exchange: Exchange) -> str:
"""심볼 정규화: BTCUSDT -> BTC-USDT"""
# 거래소별 심볼 매핑
if exchange == Exchange.OKX:
return symbol.replace('-', '').replace('_USDT', '-USDT')
return symbol
class BinanceNormalizer(OrderBookNormalizer):
"""Binance 오더북 정규화"""
def normalize(self, raw_data: Dict) -> UnifiedOrderBook:
symbol = raw_data.get('s', raw_data.get('symbol', ''))
bids = [
UnifiedLevel(
price=float(b[0]),
quantity=float(b[1]),
exchange='binance'
)
for b in raw_data.get('b', raw_data.get('bids', []))
]
asks = [
UnifiedLevel(
price=float(a[0]),
quantity=float(a[1]),
exchange='binance'
)
for a in raw_data.get('a', raw_data.get('asks', []))
]
return UnifiedOrderBook(
symbol=self.normalize_symbol(symbol, Exchange.BINANCE),
exchange=Exchange.BINANCE,
bids=bids,
asks=asks,
timestamp_ms=raw_data.get('E', raw_data.get('updateId', 0))
)
class OKXNormalizer(OrderBookNormalizer):
"""OKX 오더북 정규화"""
def normalize(self, raw_data: Dict) -> UnifiedOrderBook:
data = raw_data.get('data', [{}])[0]
symbol = data.get('instId', '')
bids = [
UnifiedLevel(
price=float(b[0]),
quantity=float(b[1]),
orders_count=int(b[2]) if len(b) > 2 else 1,
exchange='okx'
)
for b in data.get('bids', [])
]
asks = [
UnifiedLevel(
price=float(a[0]),
quantity=float(a[1]),
orders_count=int(a[2]) if len(a) > 2 else 1,
exchange='okx'
)
for a in data.get('asks', [])
]
return UnifiedOrderBook(
symbol=self.normalize_symbol(symbol, Exchange.OKX),
exchange=Exchange.OKX,
bids=bids,
asks=asks,
timestamp_ms=int(data.get('ts', 0))
)
class BybitNormalizer(OrderBookNormalizer):
"""Bybit 오더북 정규화"""
def normalize(self, raw_data: Dict) -> UnifiedOrderBook:
result = raw_data.get('result', raw_data)
symbol = result.get('s', result.get('symbol', ''))
bids = [
UnifiedLevel(
price=float(b[0]),
quantity=float(b[1]),
exchange='bybit'
)
for b in result.get('b', result.get('bids', []))
]
asks = [
UnifiedLevel(
price=float(a[0]),
quantity=float(a[1]),
exchange='bybit'
)
for a in result.get('a', result.get('asks', []))
]
return UnifiedOrderBook(
symbol=self.normalize_symbol(symbol, Exchange.BYBIT),
exchange=Exchange.BYBIT,
bids=bids,
asks=asks,
timestamp_ms=result.get('u', result.get('updateId', 0))
)
class OrderBookAggregator:
"""크로스 거래소 오더북 집계"""
def __init__(self):
self.normalizers = {
Exchange.BINANCE: BinanceNormalizer(),
Exchange.OKX: OKXNormalizer(),
Exchange.BYBIT: BybitNormalizer()
}
self.orderbooks: Dict[Exchange, UnifiedOrderBook] = {}
def update(self, exchange: Exchange, raw_data: Dict):
"""오더북 업데이트"""
normalizer = self.normalizers.get(exchange)
if normalizer:
self.orderbooks[exchange] = normalizer.normalize(raw_data)
def get_cross_exchange_spread(self, symbol: str) -> List[Dict]:
"""거래소 간 스프레드 비교"""
results = []
for exchange, ob in self.orderbooks.items():
if ob.symbol == symbol and ob.best_bid and ob.best_ask:
results.append({
'exchange': exchange.value,
'bid': ob.best_bid,
'ask': ob.best_ask,
'spread': ob.spread_bps,
'mid': (ob.best_bid + ob.best_ask) / 2
})
return sorted(results, key=lambda x: x['mid'])
def get_best_bid_offer(self, symbol: str) -> Dict:
"""최선 매수/매도 호가 (크로스 거래소)"""
spread_data = self.get_cross_exchange_spread(symbol)
if not spread_data:
return {}
return {
'best_bid_exchange': spread_data[0]['exchange'],
'best_bid': spread_data[0]['bid'],
'best_ask_exchange': spread_data[-1]['exchange'],
'best_ask': spread_data[-1]['ask'],
'max_spread_bps': spread_data[-1]['ask'] / spread_data[0]['bid'] - 1 * 10000
}
AI 기반 오더북 분석: HolySheep 통합
오더북 패턴을 AI로 분석하면 이상 거래 탐지,流动性 붕괴 예측, arbitrage 기회가 자동으로 감지됩니다. HolySheep AI를 사용하면 여러 모델을 단일 API 키로 조합하여 비용을 최적화할 수 있습니다.
"""
HolySheep AI API를 사용한 L2 오더북 패턴 분석
base_url: https://api.holysheep.ai/v1
"""
import aiohttp
import json
from typing import Dict, List
from dataclasses import dataclass
HolySheep API 설정
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
@dataclass
class OrderBookAnalysis:
"""AI 분석 결과"""
pattern: str
confidence: float
recommendation: str
risk_level: str
class HolySheepAIAnalyzer:
"""HolySheep AI API를 활용한 오더북 분석기"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
async def analyze_orderbook_pattern(self, orderbook_data: Dict) -> OrderBookAnalysis:
"""
GPT-4.1로 오더북 패턴 분석
비용: $8/MTok (HolySheep 기준)
"""
prompt = self._build_analysis_prompt(orderbook_data)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "당신은 고빈도 트레이딩 전문가입니다. L2 오더북 데이터를 분석하여 패턴을 식별하고 위험도를 평가합니다."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 500
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as resp:
result = await resp.json()
return self._parse_analysis_response(result)
async def detect_anomaly_fast(self, orderbook_data: Dict) -> Dict:
"""
Gemini 2.5 Flash로 빠른 이상 거래 탐지
비용: $2.50/MTok (HolySheep 기준) - 빠른 분석에 적합
"""
features = self._extract_features(orderbook_data)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "gemini-2.5-flash",
"messages": [
{"role": "user", "content": f"오더북 이상 패턴 탐지: {features}\n\n이상 있으면 'ALERT: [이유]' 형식으로 응답"}
],
"temperature": 0.1,
"max_tokens": 100
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as resp:
result = await resp.json()
return {'anomaly': result.get('choices', [{}])[0].get('message', {}).get('content', '')}
async def generate_trading_signal(self, historical_data: List[Dict]) -> str:
"""
DeepSeek V3.2로 트레이딩 시그널 생성
비용: $0.42/MTok (HolySheep 기준) - 대규모 분석에 경제적
"""
historical_summary = self._summarize_history(historical_data)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "system", "content": "암호화폐 트레이딩 어드바이저"},
{"role": "user", "content": f"히스토리 기반 시그널:\n{historical_summary}\n\n단기 트레이딩 시그널을 BUY/SELL/HOLD 중 하나로만 응답"}
],
"temperature": 0.5,
"max_tokens": 10
}
async with aiohttp.ClientSession() as session:
async with session.post(
f"{self.base_url}/chat/completions",
headers=headers,
json=payload
) as resp:
result = await resp.json()
return result.get('choices', [{}])[0].get('message', {}).get('content', 'HOLD')
def _build_analysis_prompt(self, data: Dict) -> str:
"""분석용 프롬프트 구성"""
return f"""
L2 오더북 데이터를 분석해주세요:
심볼: {data.get('symbol')}
거래소: {data.get('exchange')}
매수호가 (Top 5):
{json.dumps(data.get('bids', [])[:5], indent=2)}
매도호가 (Top 5):
{json.dumps(data.get('asks', [])[:5], indent=2)}
분석 항목:
1. 주문 밀도 패턴 (분산/집중)
2. 스프레드 상태 (정상/축소/확장)
3. 잠재적 arbitrage 기회
4.流动性 위험도
5. 투자 권고
"""
def _extract_features(self, data: Dict) -> str:
"""이상 탐지용 특징 추출"""
bids = data.get('bids', [])
asks = data.get('asks', [])
bid_qty = sum(b.get('quantity', 0) for b in bids[:10])
ask_qty = sum(a.get('quantity', 0) for a in asks[:10])
imbalance = (bid_qty - ask_qty) / (bid_qty + ask_qty) if (bid_qty + ask_qty) > 0 else 0
return f"bid_qty={bid_qty:.2f}, ask_qty={ask_qty:.2f}, imbalance={imbalance:+.2f}"
def _summarize_history(self, data: List[Dict]) -> str:
"""히스토리 데이터 요약"""
if not data:
return "데이터 없음"
spreads = [d.get('spread_bps', 0) for d in data]
imbalances = [d.get('imbalance', 0) for d in data]
return f"평균 스프레드: {sum(spreads)/len(spreads):.1f} bps, 평균 불균형: {sum(imbalances)/len(imbalances):+.2f}"
def _parse_analysis_response(self, response: Dict) -> OrderBookAnalysis:
"""API 응답 파싱"""
content = response.get('choices', [{}])[0].get('message', {}).get('content', '')
# 단순 파싱 (실제로는 더 정교한 파서 사용)
return OrderBookAnalysis(
pattern="분석 완료",
confidence=0.85,
recommendation=content[:200],
risk_level="MEDIUM"
)
사용 예시
async def main():
analyzer = HolySheepAIAnalyzer(HOLYSHEEP_API_KEY)
sample_orderbook = {
'symbol': 'BTC-USDT',
'exchange': 'binance',
'bids': [
{'price': 67450.00, 'quantity': 2.5},
{'price': 67448.50, 'quantity': 1.8},
{'price': 67447.00, 'quantity': 3.2},
],
'asks