암호화폐 거래소에서 제공하는 L2 오더북(호가창) 데이터는 시세 변동, 유동성 분석, алгоритмическая торговля에 핵심적인 정보다. 하지만 실시간으로 쏟아지는 L2 데이터를 효율적으로 다운로드하고 파싱하는 것은 생각보다 복잡한 작업이다. 이 튜토리얼에서는 BTC/ETH 오더북 L2 데이터 다운로드와 파싱 성능을 최적화하는 실전 기법을 다룬다.
왜 L2 오더북 데이터인가?
거래소 API에서 제공하는 오더북 데이터는 크게 3단계로 나뉜다:
- L1 (Trade Data): 최종 체결가와 거래량만 제공
- L2 (Order Book): 각 가격대의 매수/매도 주문량 포함
- L3 (Full Order Book): 개별 주문 단위까지 상세히 제공 (기관 전용)
대부분의 개발자와 알고리즘 트레이더에게 L2 오더북 데이터가 가장 실용적이다. 실제 거래량 대비 미결제 주문을 분석하면 시장 심리 파악과 유동성 핍spots 발견이 가능하다.
사전 준비: 개발 환경 구성
먼저 필요한 라이브러리를 설치한다. 이 튜토리얼에서는 Binance, Coinbase, Kraken 등 주요 거래소의 WebSocket과 REST API를 활용한다.
# Python 환경 설정
pip install websockets asyncio aiohttp pandas numpy msgpack orjson ujson
고성능 JSON 파싱을 위한 orjson 설치 (표준 json 대비 3-5배 빠름)
msgpack: 바이너리 직렬화로 데이터 전송량 40% 절감
프로젝트 구조
mkdir orderbook_parser
cd orderbook_parser
mkdir src config data logs핵심 주제: L2 데이터 다운로드와 파싱 최적화
1. WebSocket 대 REST API: 어떤 방식을 선택해야 할까?
L2 오더북 데이터를 가져오는 방식은 크게 두 가지다. 각각의 장단점을 비교해보자.
| 기준 | WebSocket | REST API |
|---|---|---|
| 데이터 주기 | 실시간 (밀리초 단위) | 폴링 방식 (최소 1초) |
| 서버 부하 | 연결 유지 방식 (낮음) | 요청마다 새 연결 (높음) |
| 구현 복잡도 | 중간 (비동기 처리 필요) | 낮음 (동기 함수 호출) |
| 데이터 누락 | 없음 (실시간 스트림) | 폴링 간격 동안 누락 가능 |
| 적합한 용도 | 알고리즘 트레이딩, 실시간 모니터링 | 히스토리컬 데이터 다운로드, 배치 분석 |
| Rate Limit | 관대한 편 (연결 수 기준) | 엄격 (요청 수 기준, 분당 1200회 등) |
결론: 실시간 거래 시스템에는 WebSocket이 필수이고, 과거 데이터 분석에는 REST API가 적합하다. HolySheep AI를 활용하면 여러 거래소의 API를 단일 엔드포인트로 통합 관리할 수 있어 개발 효율성이 크게 향상된다.
2. Binance WebSocket으로 실시간 L2 데이터 수신하기
# binance_orderbook.py
import asyncio
import websockets
import json
import orjson # 고성능 JSON 파싱
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List, Optional
import time
@dataclass
class OrderBookEntry:
price: float
quantity: float
@dataclass
class OrderBook:
bids: Dict[float, float] = field(default_factory=dict) # price -> quantity
asks: Dict[float, float] = field(default_factory=dict)
last_update_id: int = 0
symbol: str = ""
def update_bids(self, updates: List[List[str]], force: bool = False):
"""매수 주문 실시간 업데이트"""
for price, qty in updates:
price, qty = float(price), float(qty)
if qty == 0:
self.bids.pop(price, None)
else:
self.bids[price] = qty
# 정리: 가격이 너무 떨어진 주문 제거
if len(self.bids) > 100:
sorted_bids = sorted(self.bids.items(), key=lambda x: -x[0])[:100]
self.bids = dict(sorted_bids)
def update_asks(self, updates: List[List[str]], force: bool = False):
"""매도 주문 실시간 업데이트"""
for price, qty in updates:
price, qty = float(price), float(qty)
if qty == 0:
self.asks.pop(price, None)
else:
self.asks[price] = qty
if len(self.asks) > 100:
sorted_asks = sorted(self.asks.items(), key=lambda x: x[0])[:100]
self.asks = dict(sorted_asks)
def get_spread(self) -> float:
"""스프레드 계산"""
if self.bids and self.asks:
best_bid = max(self.bids.keys())
best_ask = min(self.asks.keys())
return best_ask - best_bid
return 0.0
def get_mid_price(self) -> float:
"""중간가 계산"""
if self.bids and self.asks:
return (max(self.bids.keys()) + min(self.asks.keys())) / 2
return 0.0
class BinanceOrderBookClient:
def __init__(self, symbol: str = "btcusdt", depth: int = 20):
self.symbol = symbol.lower()
self.depth = depth
self.orderbook = OrderBook(symbol=symbol)
self.running = False
self.message_count = 0
self.start_time = None
self.last_stats_time = time.time()
def get_stream_url(self) -> str:
"""Binance WebSocket 스트림 URL 생성"""
return f"wss://stream.binance.com:9443/ws/{self.symbol}@depth{self.depth}@100ms"
async def connect(self):
"""WebSocket 연결 및 데이터 수신"""
self.running = True
self.start_time = time.time()
async with websockets.connect(self.get_stream_url()) as ws:
print(f"🔗 Binance {self.symbol.upper()} 오더북 연결됨")
print(f"📊 스트림: depth{self.depth}@100ms (100ms 갱신)")
while self.running:
try:
# orjson으로高速 JSON 파싱 (표준 json 대비 5배 빠름)
message = await asyncio.wait_for(ws.recv(), timeout=30.0)
data = orjson.loads(message)
self.message_count += 1
# 오더북 업데이트
if 'b' in data:
self.orderbook.update_bids(data['b'])
if 'a' in data:
self.orderbook.update_asks(data['a'])
if 'u' in data:
self.orderbook.last_update_id = data['u']
# 5초마다 통계 출력
current_time = time.time()
if current_time - self.last_stats_time >= 5:
elapsed = current_time - self.start_time
msg_per_sec = self.message_count / elapsed
print(f"\n⏱️ [{elapsed:.1f}s] 메시지: {self.message_count}개 "
f"({msg_per_sec:.1f}/sec)")
print(f"📈 Bid: {len(self.orderbook.bids)}개, "
f"Ask: {len(self.orderbook.asks)}개")
print(f"💰 스프레드: ${self.orderbook.get_spread():.2f}, "
f"중간가: ${self.orderbook.get_mid_price():,.2f}")
# 최고 매수/매도 3단계 표시
top_bids = sorted(self.orderbook.bids.items(),
key=lambda x: -x[0])[:3]
top_asks = sorted(self.orderbook.asks.items(),
key=lambda x: x[0])[:3]
print("매수(Bid)Top3:",
[(f"${p:.2f}", f"{q:.4f}") for p, q in top_bids])
print("매도(Ask)Top3:",
[(f"${p:.2f}", f"{q:.4f}") for p, q in top_asks])
self.last_stats_time = current_time
except asyncio.TimeoutError:
print("⏰ 연결 타임아웃 - 재연결 시도...")
except Exception as e:
print(f"❌ 오류 발생: {e}")
await asyncio.sleep(1)
def stop(self):
"""연결 종료"""
self.running = False
실행
async def main():
client = BinanceOrderBookClient(symbol="btcusdt", depth=20)
try:
await client.connect()
except KeyboardInterrupt:
print("\n🛑 연결 종료됨")
client.stop()
if __name__ == "__main__":
asyncio.run(main())3. HolySheep AI API로 L2 데이터 AI 분석하기
L2 오더북 데이터를 수집한 후, 이를 AI로 분석하면 시장 움직임을 예측하고 거래 전략을 세울 수 있다. HolySheep AI의 통합 API를 사용하면 여러 모델을 비교하며 최적의 분석 결과를 얻을 수 있다.
# orderbook_analysis.py
import aiohttp
import asyncio
import orjson
from typing import Dict, List, Optional
from dataclasses import dataclass
import time
@dataclass
class OrderBookAnalysis:
symbol: str
timestamp: int
spread_pct: float
mid_price: float
bid_volume: float
ask_volume: float
imbalance_ratio: float # 매수/매도 불균형 비율
depth_score: float # 유동성 깊이 점수
class HolySheepAIClient:
"""HolySheep AI API 클라이언트 - 단일 API 키로 다중 모델 지원"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
async def analyze_with_model(
self,
model: str,
prompt: str,
temperature: float = 0.3
) -> Dict:
"""지정된 모델로 분석 수행"""
async with aiohttp.ClientSession() as session:
payload = {
"model": model,
"messages": [
{"role": "system", "content": "당신은 암호화폐 시장 분석 전문가입니다."},
{"role": "user", "content": prompt}
],
"temperature": temperature,
"max_tokens": 500
}
start_time = time.time()
async with session.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json=payload
) as response:
elapsed_ms = (time.time() - start_time) * 1000
if response.status != 200:
error = await response.text()
raise Exception(f"API 오류: {response.status} - {error}")
result = await response.json()
usage = result.get("usage", {})
return {
"model": model,
"content": result["choices"][0]["message"]["content"],
"latency_ms": round(elapsed_ms, 2),
"tokens_used": usage.get("total_tokens", 0),
"cost_usd": self._calculate_cost(model, usage)
}
def _calculate_cost(self, model: str, usage: Dict) -> float:
"""토큰 사용량 기준 비용 계산 (HolySheep AI 가격)"""
prompt_tokens = usage.get("prompt_tokens", 0)
completion_tokens = usage.get("completion_tokens", 0)
pricing = {
"gpt-4.1": (8.0, 8.0), # $8/MTok
"claude-sonnet-4": (15.0, 15.0), # $15/MTok
"gemini-2.5-flash": (2.5, 2.5), # $2.50/MTok
"deepseek-v3.2": (0.42, 0.42), # $0.42/MTok
}
if model in pricing:
p_price, c_price = pricing[model]
return (prompt_tokens * p_price + completion_tokens * c_price) / 1_000_000
return 0.0
async def analyze_orderbook_multi_model(
self,
analysis: OrderBookAnalysis
) -> List[Dict]:
"""여러 모델로 동시 분석 (비용/성능 비교)"""
prompt = f"""BTC/USDT 오더북 분석 결과를 바탕으로 시장 해석을 제공해주세요.
현재 데이터:
-symbol: {analysis.symbol}
-스프레드: {analysis.spread_pct:.4f}%
-중간가: ${analysis.mid_price:,.2f}
-매수 총량: {analysis.bid_volume:.4f} BTC
-매도 총량: {analysis.ask_volume:.4f} BTC
-불균형 비율: {analysis.imbalance_ratio:.4f} (양수=매수 우세, 음수=매도 우세)
-유동성 점수: {analysis.depth_score:.2f}/100
단기로 3줄 분석과 거래 신호를 제공해주세요."""
models = ["deepseek-v3.2", "gemini-2.5-flash", "claude-sonnet-4"]
tasks = [self.analyze_with_model(model, prompt) for model in models]
results = await asyncio.gather(*tasks)
return results
async def main():
# HolySheep AI API 키 설정
api_key = "YOUR_HOLYSHEEP_API_KEY" # https://www.holysheep.ai/register에서 가입
client = HolySheepAIClient(api_key)
# 샘플 오더북 분석 데이터
sample_analysis = OrderBookAnalysis(
symbol="BTC/USDT",
timestamp=int(time.time()),
spread_pct=0.0234,
mid_price=67450.00,
bid_volume=125.45,
ask_volume=118.32,
imbalance_ratio=0.058,
depth_score=78.5
)
print("🚀 HolySheep AI 다중 모델 분석 시작...")
print(f"📊 분석 대상: {sample_analysis.symbol}\n")
results = await client.analyze_orderbook_multi_model(sample_analysis)
print("=" * 70)
print("📈 모델별 분석 결과 비교")
print("=" * 70)
for result in results:
print(f"\n🤖 모델: {result['model']}")
print(f"⏱️ 응답시간: {result['latency_ms']}ms")
print(f"💰 비용: ${result['cost_usd']:.6f}")
print(f"📝 결과:\n{result['content']}")
print("-" * 70)
# 총 비용 및 성능 요약
total_cost = sum(r['cost_usd'] for r in results)
avg_latency = sum(r['latency_ms'] for r in results) / len(results)
print(f"\n💵 총 비용: ${total_cost:.6f}")
print(f"⚡ 평균 응답시간: {avg_latency:.2f}ms")
print(f"💡 추천: {'gemini-2.5-flash' if avg_latency < 2000 else 'deepseek-v3.2'} (가성비最优)")
if __name__ == "__main__":
asyncio.run(main())L2 데이터 파싱 성능 최적화 기법
1. 바이너리 직렬화: MessagePack vs Protocol Buffers
대량의 오더북 데이터를 저장하거나 전송할 때, JSON 대신 바이너리 포맷을 사용하면 네트워크 대역폭과 디스크 공간을 크게 절약할 수 있다.
# binary_serialization.py
import msgpack
import json
import time
import random
from dataclasses import dataclass, asdict
@dataclass
class OrderBookSnapshot:
symbol: str
timestamp: int
bids: list # [[price, quantity], ...]
asks: list # [[price, quantity], ...]
def to_msgpack(self) -> bytes:
"""MessagePack 바이너리로 직렬화"""
data = asdict(self)
return msgpack.packb(data, use_bin_type=True)
@classmethod
def from_msgpack(cls, data: bytes) -> 'OrderBookSnapshot':
"""MessagePack 바이너리에서 역직렬화"""
decoded = msgpack.unpackb(data, raw=False)
return cls(**decoded)
def generate_sample_orderbook(symbol: str = "BTC/USDT") -> OrderBookSnapshot:
"""테스트용 샘플 오더북 생성"""
base_price = 67000
bids = [[base_price - i * 10 + random.uniform(-2, 2),
random.uniform(0.1, 5.0)] for i in range(50)]
asks = [[base_price + i * 10 + random.uniform(-2, 2),
random.uniform(0.1, 5.0)] for i in range(1, 51)]
return OrderBookSnapshot(
symbol=symbol,
timestamp=int(time.time() * 1000),
bids=bids,
asks=asks
)
def benchmark_serialization(iterations: int = 10000):
"""직렬화 성능 벤치마크"""
orderbook = generate_sample_orderbook()
# JSON 직렬화
json_data = json.dumps(asdict(orderbook))
json_encoded = json_data.encode('utf-8')
# MessagePack 직렬화
msgpack_data = orderbook.to_msgpack()
print(f"📊 직렬화 크기 비교 ({iterations}회 평균)")
print("=" * 50)
print(f"JSON 크기: {len(json_encoded):,} bytes")
print(f"MessagePack 크기: {len(msgpack_data):,} bytes")
print(f"크기 절감율: {(1 - len(msgpack_data)/len(json_encoded))*100:.1f}%")
print()
# 인코딩 속도 벤치마크
print("⚡ 인코딩 속도 벤치마크")
start = time.perf_counter()
for _ in range(iterations):
_ = json.dumps(asdict(orderbook))
json_time = time.perf_counter() - start
start = time.perf_counter()
for _ in range(iterations):
_ = orderbook.to_msgpack()
msgpack_time = time.perf_counter() - start
print(f"JSON 인코딩: {json_time*1000:.2f}ms ({iterations}회)")
print(f"MessagePack 인코딩: {msgpack_time*1000:.2f}ms ({iterations}회)")
print(f"속도 향상: {json_time/msgpack_time:.2f}x")
print()
# 디코딩 속도 벤치마크
print("🔄 디코딩 속도 벤치마크")
start = time.perf_counter()
for _ in range(iterations):
_ = json.loads(json_data)
json_decode_time = time.perf_counter() - start
start = time.perf_counter()
for _ in range(iterations):
_ = OrderBookSnapshot.from_msgpack(msgpack_data)
msgpack_decode_time = time.perf_counter() - start
print(f"JSON 디코딩: {json_decode_time*1000:.2f}ms ({iterations}회)")
print(f"MessagePack 디코딩: {msgpack_decode_time*1000:.2f}ms ({iterations}회)")
print(f"속도 향상: {json_decode_time/msgpack_decode_time:.2f}x")
# 저장 공간 절감 효과
daily_snapshots = 86400 # 1일치 (1초 간격)
json_size_gb = len(json_encoded) * daily_snapshots / (1024**3)
msgpack_size_gb = len(msgpack_data) * daily_snapshots / (1024**3)
print(f"\n💾 1일 데이터 저장 공간 ({daily_snapshots:,} 스냅샷)")
print(f"JSON: {json_size_gb:.2f} GB")
print(f"MessagePack: {msgpack_size_gb:.2f} GB")
print(f"절약: {json_size_gb - msgpack_size_gb:.2f} GB ({(1 - msgpack_size_gb/json_size_gb)*100:.1f}%)")
if __name__ == "__main__":
benchmark_serialization()2. 메모리 최적화: 슬라이딩 윈도우와 델타 업데이트
오더북 데이터는 실시간으로 변경되므로 전체 스냅샷 대신 변경분만 전송하면 네트워크 트래픽을劇적으로 줄일 수 있다.
# delta_update.py
from collections import deque
from dataclasses import dataclass, field
from typing import Dict, List, Tuple, Optional
from enum import Enum
import time
class UpdateType(Enum):
ADD = "add"
MODIFY = "modify"
REMOVE = "remove"
@dataclass
class OrderUpdate:
price: float
quantity: float
update_type: UpdateType
class OptimizedOrderBook:
"""
메모리 최적화된 오더북 구현
- 슬라이딩 윈도우: 최근 N개 업데이트만 메모리에 유지
- 델타 업데이트: 변경분만 기록
- 풀 스냅샷 주기적 저장으로 일관성 보장
"""
def __init__(self, max_depth: int = 100, window_size: int = 1000):
self.bids: Dict[float, float] = {} # price -> quantity
self.asks: Dict[float, float] = {}
self.max_depth = max_depth
self.window_size = window_size
# 델타 업데이트 히스토리 (슬라이딩 윈도우)
self.update_history: deque = deque(maxlen=window_size)
# 풀 스냅샷 카운터 (100회마다 전체 스냅샷 저장)
self.update_count = 0
self.full_snapshot_interval = 100
# 통계
self.total_updates = 0
self.delta_updates = 0
def apply_delta(self, side: str, price: float, quantity: float) -> Optional[OrderUpdate]:
"""델타 업데이트 적용"""
book = self.bids if side == "bid" else self.asks
update_type: UpdateType
if quantity == 0:
if price in book:
update_type = UpdateType.REMOVE
del book[price]
else:
return None
else:
if price in book:
update_type = UpdateType.MODIFY
else:
update_type = UpdateType.ADD
book[price] = quantity
# 델타 히스토리에 기록
update = OrderUpdate(price, quantity, update_type)
self.update_history.append({
"side": side,
"update": update,
"timestamp": time.time()
})
self.update_count += 1
self.delta_updates += 1
self.total_updates += 1
# 정리: 깊이 제한
self._prune_depth(side)
return update
def _prune_depth(self, side: str):
"""오더북 깊이 제한으로 메모리 절약"""
book = self.bids if side == "bid" else self.asks
if side == "bid":
# 매수: 높은 가격순으로 정렬 후 초과분 제거
sorted_prices = sorted(book.keys(), reverse=True)
else:
# 매도: 낮은 가격순으로 정렬 후 초과분 제거
sorted_prices = sorted(book.keys())
if len(sorted_prices) > self.max_depth:
remove_prices = sorted_prices[self.max_depth:]
for price in remove_prices:
del book[price]
def get_depth(self, levels: int = 10) -> Tuple[List, List]:
"""상위 N단계 가격대 반환"""
top_bids = sorted(self.bids.items(), key=lambda x: -x[0])[:levels]
top_asks = sorted(self.asks.items(), key=lambda x: x[0])[:levels]
return top_bids, top_asks
def calculate_metrics(self) -> Dict:
"""유동성 지표 계산"""
bid_prices = sorted(self.bids.keys(), reverse=True)
ask_prices = sorted(self.asks.keys())
if not bid_prices or not ask_prices:
return {"spread": 0, "mid_price": 0, "bid_depth": 0, "ask_depth": 0}
best_bid = bid_prices[0]
best_ask = ask_prices[0]
spread = best_ask - best_bid
mid_price = (best_bid + best_ask) / 2
spread_pct = (spread / mid_price) * 100 if mid_price > 0 else 0
# 총 유동성 (상위 10단계)
bid_depth = sum(self.bids.get(p, 0) for p in bid_prices[:10])
ask_depth = sum(self.asks.get(p, 0) for p in ask_prices[:10])
return {
"spread": spread,
"spread_pct": spread_pct,
"mid_price": mid_price,
"bid_depth": bid_depth,
"ask_depth": ask_depth,
"imbalance": (bid_depth - ask_depth) / (bid_depth + ask_depth) if (bid_depth + ask_depth) > 0 else 0
}
def get_memory_stats(self) -> Dict:
"""메모리 사용량 통계"""
import sys
# 딕셔너리 크기估算
bids_size = sys.getsizeof(self.bids) + sum(
sys.getsizeof(k) + sys.getsizeof(v) for k, v in self.bids.items()
)
asks_size = sys.getsizeof(self.asks) + sum(
sys.getsizeof(k) + sys.getsizeof(v) for k, v in self.asks.items()
)
# 델타 히스토리 크기
history_size = sys.getsizeof(self.update_history)
for item in self.update_history:
history_size += sys.getsizeof(item)
return {
"bids_entries": len(self.bids),
"asks_entries": len(self.asks),
"bids_memory_bytes": bids_size,
"asks_memory_bytes": asks_size,
"history_entries": len(self.update_history),
"history_memory_bytes": history_size,
"total_memory_bytes": bids_size + asks_size + history_size,
"total_updates": self.total_updates,
"delta_updates_pct": (self.delta_updates / self.total_updates * 100) if self.total_updates > 0 else 0
}
테스트
if __name__ == "__main__":
import random
book = OptimizedOrderBook(max_depth=100, window_size=500)
# 1000회 업데이트 시뮬레이션
print("📊 델타 업데이트 메모리 최적화 테스트")
print("=" * 50)
base_price = 67000
for i in range(1000):
side = random.choice(["bid", "ask"])
if side == "bid":
price = base_price - random.randint(1, 500)
else:
price = base_price + random.randint(1, 500)
quantity = random.uniform(0.1, 10.0)
book.apply_delta(side, price, quantity)
metrics = book.calculate_metrics()
stats = book.get_memory_stats()
print(f"✅ 총 업데이트: {stats['total_updates']:,}회")
print(f"📉 델타 업데이트: {stats['delta_updates_pct']:.1f}%")
print(f"💾 메모리 사용량: {stats['total_memory_bytes']:,} bytes")
print(f"📦 bids: {stats['bids_entries']}개 ({stats['bids_memory_bytes']:,} bytes)")
print(f"📦 asks: {stats['asks_entries']}개 ({stats['asks_memory_bytes']:,} bytes)")
print(f"📜 히스토리: {stats['history_entries']}개 ({stats['history_memory_bytes']:,} bytes)")
print()
print(f"💰 스프레드: ${metrics['spread']:.2f} ({metrics['spread_pct']:.4f}%)")
print(f"📈 중간가: ${metrics['mid_price']:.2f}")
print(f"📊 불균형 비율: {metrics['imbalance']:.4f}")실전 프로젝트: 완전한 L2 데이터 파이프라인
# complete_pipeline.py
"""
완전한 L2 오더북 데이터 파이프라인
- 다중 거래소 지원 (Binance, Coinbase, Kraken)
- 실시간 데이터 수집 → 메시지팩 변환 → 분석
- HolySheep AI API 연동
"""
import asyncio
import aiohttp
import websockets
import orjson
import msgpack
import time
import random
from dataclasses import dataclass, asdict
from typing import Dict, List, Optional
from collections import defaultdict
from enum import Enum
import json
class Exchange(Enum):
BINANCE = "binance"
COINBASE = "coinbase"
KRAKEN = "kraken"
@dataclass
class NormalizedOrderBook:
exchange: str
symbol: str
timestamp: int
bids: List[List[float]] # [[price, quantity], ...]
asks: List[List[float]] # [[price, quantity], ...]
def to_msgpack(self) -> bytes:
data = asdict(self)
return msgpack.packb(data, use_bin_type=True)
@classmethod
def from_msgpack(cls, data: bytes) -> 'NormalizedOrderBook':
decoded = msgpack.unpackb(data, raw=False)
return cls(**decoded)
class MultiExchangeCollector:
"""다중 거래소 오더북 수집기"""
def __init__(self):
self.orderbooks: Dict[str, Dict[str, NormalizedOrderBook]] = defaultdict(dict)
self.running = False
self.collected_count = 0
def normalize_binance(self, data: dict, symbol: str) -> NormalizedOrderBook:
"""Binance 데이터 정규화"""
return NormalizedOrderBook(
exchange="binance",
symbol=symbol,
timestamp=int(time.time() * 1000),
bids=[[float(p), float(q)] for p, q in data.get('bids', data.get('b', []))],
asks=[[float(p), float(q)] for p, q in data.get('asks', data.get('a', []))]
)
async def collect_binance(self, symbol: str = "btcusdt", duration_sec: int = 10):
"""Binance WebSocket에서 데이터 수집"""
url = f"wss://stream.binance.com:9443/ws/{symbol}@depth20@100ms"
print(f"📡 Binance {symbol} 수집 시작 ({duration_sec}초)")
async with websockets.connect(url) as ws:
start_time = time.time()
while time.time() - start_time < duration_sec:
try:
msg = await asyncio.wait_for(ws.recv(), timeout=5.0)
data = orjson.loads(msg)
ob = self.normalize_binance(data, symbol)
self.orderbooks["binance"][symbol] = ob
self.collected_count += 1
except asyncio.TimeoutError:
print("⏰ 타임아웃")
break
elapsed = time.time() - start_time
rate = self.collected_count / elapsed if elapsed > 0 else 0
print(f"✅ Binance 완료: {self.collected_count}개 수집 ({rate:.1f}/sec)")
async def collect_multiple(self, symbols: List[str], duration: int = 10):
"""여러 심볼 동시 수집"""
self.running = True
self.collected_count = 0
tasks = [self.collect_binance(sym, duration) for sym in symbols]
await asyncio.gather(*tasks)
print(f"\n📊 총 수집: {self.collected_count}개 오더북")
return self.orderbooks
async def analyze_with_holysheep(self, api_key: str):
"""HolySheep AI로 분석"""
# 샘플 분석 결과
sample = NormalizedOrderBook(
exchange="binance",
symbol="btcusdt",
timestamp=int(time.time() * 1000),
bids=[[67000, 1.5], [66999, 2.3], [66998, 0.8]],
asks=[[67001, 1.2], [67002, 3.1], [67003, 1.0]]
)
metrics = self._calculate_metrics(sample)
prompt = f"""BTC/USDT 오더북 분석:
스프레드: ${metrics['spread']:.2f}
중간가: ${metrics['mid']:.2f}
매수 총량: {metrics['bid_vol']:.4f}
매도 총량: {metrics['ask_vol']:.4f}
간단한 시장 해석을 2줄로 제공."""
async with aiohttp.ClientSession() as session:
payload = {
"model": "gemini-2.5-flash",
"messages":