암호화폐 거래소에서 제공하는 Tick 데이터는 시장 microstructure를 이해하는 핵심原料입니다. 저는 3년 동안 고빈도 트레이딩 시스템을 구축하며 수십억 건의 Tick 데이터를 처리한 경험이 있습니다. 이 튜토리얼에서는 다양한 거래소 API에서 Tick 데이터를 수집하고, 이를 기반으로 주문서를 실시간으로 재구축하는 프로덕션 레벨의 시스템을 구현하겠습니다. HolySheep AI를 활용한 AI 기반 시장 이상 탐지까지 포함하여 완전한 데이터 파이프라인을 구축해 보겠습니다.
Tick 데이터의 본질과 시장 microstructure
Tick 데이터는 거래소에서 발생하는 모든 시장 활동을 캡처하는 가장 세밀한 단위입니다. 각 Tick은 새로운 거래(Trade), 가격 변동(Price Change), 또는 주문서 변동(Order Book Update)을 나타냅니다. 1초에 수백 개의 Tick이 발생할 수 있는 고流动性 시장에서는 효율적인 데이터 구조와 처리 방식이 필수적입니다.
Tick 데이터의 구성 요소
class Tick:
"""단일 Tick 데이터를 표현하는 기본 구조"""
timestamp: int # 마이크로초 단위 타임스탬프 (Unix epoch)
exchange: str # 거래소 식별자 (binance, bybit, okx 등)
symbol: str # 거래 페어 (BTC/USDT)
price: Decimal # 실행 가격
quantity: Decimal # 거래 수량
side: str # buy 또는 sell (누가 누구와 거래했는가)
trade_id: str # 고유 거래 ID
is_buyer_maker: bool # 메이커가 매수자인지 여부
def to_dict(self) -> dict:
return {
"timestamp": self.timestamp,
"exchange": self.exchange,
"symbol": self.symbol,
"price": float(self.price),
"quantity": float(self.quantity),
"side": self.side,
"trade_id": self.trade_id,
"is_buyer_maker": self.is_buyer_maker
}
class OrderBookLevel:
"""주문서의 단일 레벨 (Bid 또는 Ask)"""
price: Decimal
quantity: Decimal
def __repr__(self):
return f"{self.price} x {self.quantity}"
class OrderBookSnapshot:
"""주문서 스냅샷"""
timestamp: int
bids: List[OrderBookLevel] # 매수 주문 (가격 내림차순)
asks: List[OrderBookLevel] # 매도 주문 (가격 오름차순)
@property
def best_bid(self) -> Optional[Decimal]:
return self.bids[0].price if self.bids else None
@property
def best_ask(self) -> Optional[Decimal]:
return self.asks[0].price if self.asks else None
@property
def spread(self) -> Optional[Decimal]:
if self.best_bid and self.best_ask:
return self.best_ask - self.best_bid
return None
@property
def mid_price(self) -> Optional[Decimal]:
if self.best_bid and self.best_ask:
return (self.best_bid + self.best_ask) / 2
return None주요 거래소 API 비교 분석
암호화폐 Tick 데이터를 제공하는 주요 거래소의 API를 비교해 보겠습니다. 지연 시간, 데이터 품질, 가용성에 따라 최적의 선택이 달라집니다.
| 거래소 | WebSocket 지연 | REST 지연 | Rate Limit | 히스토리cal 데이터 | 주문서 깊이 | 월간 비용 |
|---|---|---|---|---|---|---|
| Binance Spot | ~5ms | ~50ms | 1200/min | 제한적 (7일) | 20 레벨 | 무료 ~ $100 |
| Binance Futures | ~3ms | ~45ms | 2400/min | 제한적 | 500 레벨 | 무료 ~ $200 |
| Bybit | ~2ms | ~40ms | 600/min | 제한적 | 200 레벨 | 무료 ~ $150 |
| OKX | ~4ms | ~55ms | 600/min | 제한적 | 400 레벨 | 무료 ~ $100 |
| CoinAPI | ~100ms | ~200ms | tier별 | 완전 (수년) | 변경 | $79 ~ $699 |
| Exocharts | ~50ms | ~80ms | 변경 | 완전 | 변경 | $199 ~ |
이런 팀에 적합
- 지연 시간 민감 시스템 운영: Bybit나 Binance Futures는 ~2-3ms 지연으로 마켓메이킹, 변동성 Arbitration, 자동 거래에 적합합니다.
- 역사 데이터 분석: 수년치 과거 데이터가 필요한 머신러닝 모델링, 백테스팅에는 CoinAPI나 Exocharts가 필수적입니다.
- 여러 거래소 통합: 크로스 거래소 전략을 위해 여러 출처의 데이터를 조합해야 하는 팀에게 이상적입니다.
- 리스크 관리 시스템: 실시간 주문서 분석을 통한 리스크 모니터링이 필요한 헤지펀드 및 거래소.
비적합한 경우
- 단순 가격 추적: 1분봉 이상으로 충분한 경우 거래소 무료 API로 충분합니다.
- 초소형 예산: 월 $50 이하 예산으로 고품질 데이터가 필요한 경우.
- 비트코인 전용: Binance 무료 티어만으로도 대부분의比特币 전략에 충분할 수 있습니다.
Binance WebSocket을 통한 Tick 데이터 수집
Binance는 암호화폐 시장에서 가장 높은流动性를 자랑하며, WebSocket을 통해 실시간 Tick 데이터를 무료로 제공합니다. 저는 Binance WebSocket을 기본 소스로 사용하되, 구조화된 코드를 통해 여러 거래소로 확장 가능하게 설계하겠습니다.
import asyncio
import json
import hmac
import hashlib
import time
from decimal import Decimal
from typing import Dict, List, Optional, Callable, Any
from dataclasses import dataclass, field
from enum import Enum
import aiohttp
from collections import defaultdict
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class Exchange(Enum):
BINANCE = "binance"
BYBIT = "bybit"
OKX = "okx"
@dataclass
class TickConfig:
"""거래소별 WebSocket 설정"""
exchange: Exchange
symbols: List[str]
streams: List[str] # 구독할 스트림 목록
base_url: str
api_key: Optional[str] = None
api_secret: Optional[str] = None
ping_interval: int = 20
ping_timeout: int = 10
class BinanceWebSocketClient:
"""
Binance WebSocket 클라이언트
3개 스트림 동시 구독 지원:
- trades: 실시간 거래
- ticker: 24시간 통계
- depth: 주문서 업데이트
"""
def __init__(self, config: TickConfig):
self.config = config
self._ws: Optional[aiohttp.ClientWebSocketResponse] = None
self._session: Optional[aiohttp.ClientSession] = None
self._running = False
self._reconnect_delay = 1
self._max_reconnect_delay = 60
# 콜백 핸들러
self._trade_handlers: List[Callable[[dict], None]] = []
self._orderbook_handlers: List[Callable[[dict], None]] = []
self._ticker_handlers: List[Callable[[dict], None]] = []
# 통계
self._stats = {
"messages_received": 0,
"messages_per_second": 0,
"last_message_time": 0,
"reconnects": 0
}
self._message_timestamps: List[float] = []
async def connect(self):
"""WebSocket 연결 수립"""
self._session = aiohttp.ClientSession()
# 다중 스트림 URL 생성
streams = "/".join(self.config.streams)
url = f"{self.config.base_url}/stream?streams={streams}"
logger.info(f"[Binance] WebSocket 연결 시도: {url}")
try:
self._ws = await self._session.ws_connect(
url,
heartbeat=self.config.ping_interval,
timeout=aiohttp.ClientTimeout(total=30)
)
self._running = True
self._reconnect_delay = 1
logger.info("[Binance] WebSocket 연결 성공")
except Exception as e:
logger.error(f"[Binance] WebSocket 연결 실패: {e}")
raise
async def disconnect(self):
"""WebSocket 연결 해제"""
self._running = False
if self._ws:
await self._ws.close()
if self._session:
await self._session.close()
logger.info("[Binance] WebSocket 연결 해제")
def on_trade(self, handler: Callable[[dict], None]):
"""거래 이벤트 핸들러 등록"""
self._trade_handlers.append(handler)
def on_orderbook(self, handler: Callable[[dict], None]):
"""주문서 업데이트 핸들러 등록"""
self._orderbook_handlers.append(handler)
def on_ticker(self, handler: Callable[[dict], None]):
"""티커 업데이트 핸들러 등록"""
self._ticker_handlers.append(handler)
async def _process_message(self, raw_message: dict):
"""수신된 메시지 처리 및 라우팅"""
try:
data = raw_message.get("data", {})
stream = raw_message.get("stream", "")
self._stats["messages_received"] += 1
current_time = time.time()
self._message_timestamps.append(current_time)
# 1초 이상된 타임스탬프 제거 (滑动窗口)
self._message_timestamps = [
t for t in self._message_timestamps
if current_time - t < 1
]
self._stats["messages_per_second"] = len(self._message_timestamps)
self._stats["last_message_time"] = current_time
# 스트림 타입별 라우팅
if "trade" in stream:
for handler in self._trade_handlers:
await self._safe_handler(handler, data)
elif "depth" in stream or "@depth" in stream:
for handler in self._orderbook_handlers:
await self._safe_handler(handler, data)
elif "ticker" in stream:
for handler in self._ticker_handlers:
await self._safe_handler(handler, data)
except Exception as e:
logger.error(f"[Binance] 메시지 처리 오류: {e}")
async def _safe_handler(self, handler: Callable, data: Any):
"""핸들러 실행 시 예외 처리"""
try:
if asyncio.iscoroutinefunction(handler):
await handler(data)
else:
handler(data)
except Exception as e:
logger.error(f"[Binance] 핸들러 실행 오류: {e}")
async def _send_ping(self):
"""주기적 Ping 전송 (Binance는 자동 ping 지원하지만 명시적发送)"""
while self._running:
await asyncio.sleep(self.config.ping_interval)
if self._ws and not self._ws.closed:
try:
await self._ws.ping()
except Exception as e:
logger.warning(f"[Binance] Ping 실패: {e}")
async def listen(self):
"""메시지 리스닝 루프"""
await self.connect()
# Ping 루틴 시작
ping_task = asyncio.create_task(self._send_ping())
try:
async for msg in self._ws:
if not self._running:
break
if msg.type == aiohttp.WSMsgType.TEXT:
try:
data = json.loads(msg.data)
await self._process_message(data)
except json.JSONDecodeError as e:
logger.warning(f"[Binance] JSON 파싱 오류: {e}")
elif msg.type == aiohttp.WSMsgType.ERROR:
logger.error(f"[Binance] WebSocket 오류: {msg.data}")
break
elif msg.type == aiohttp.WSMsgType.CLOSED:
logger.warning("[Binance] WebSocket 연결 종료")
break
except asyncio.CancelledError:
logger.info("[Binance] 리스닝 취소됨")
finally:
ping_task.cancel()
await self.disconnect()
async def reconnect(self):
"""재연결 로직 (지수 백오프)"""
self._stats["reconnects"] += 1
await self.disconnect()
await asyncio.sleep(self._reconnect_delay)
# 지수 백오프: 1초, 2초, 4초, 8초... 최대 60초
self._reconnect_delay = min(
self._reconnect_delay * 2,
self._max_reconnect_delay
)
await self.listen()
def get_stats(self) -> dict:
"""통계 정보 반환"""
return self._stats.copy()
사용 예제
async def main():
config = TickConfig(
exchange=Exchange.BINANCE,
symbols=["btcusdt", "ethusdt"],
streams=[
"btcusdt@trade",
"ethusdt@trade",
"btcusdt@depth20@100ms",
"ethusdt@depth20@100ms"
],
base_url="wss://stream.binance.com:9443"
)
client = BinanceWebSocketClient(config)
# 핸들러 등록
async def handle_trade(data):
print(f"Trade: {data['s']} @ {data['p']} x {data['q']}")
async def handle_orderbook(data):
print(f"OrderBook: Bids={len(data.get('b', []))}, Asks={len(data.get('a', []))}")
client.on_trade(handle_trade)
client.on_orderbook(handle_orderbook)
try:
await client.listen()
except KeyboardInterrupt:
print(f"Stats: {client.get_stats()}")
if __name__ == "__main__":
asyncio.run(main())주문서 재구축 엔진 구현
주문서를 정확하게 재구축하려면 incremental 업데이트(Delta Updates)와 전체 스냅샷(Snapshot)을 조합해야 합니다. 저는 Price-Time 우선순위(Prioirty) 방식으로 효율적인 주문서 구조를 구현하겠습니다.
from sortedcontainers import SortedDict
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass, field
from collections import defaultdict
from threading import RLock
import time
from decimal import Decimal
@dataclass
class OrderBookLevel:
"""주문서 레벨 (가격별 수량)"""
price: Decimal
quantity: Decimal
def __post_init__(self):
self.price = Decimal(str(self.price))
self.quantity = Decimal(str(self.quantity))
def __hash__(self):
return hash(str(self.price))
def is_zero(self) -> bool:
"""수량이 0인지 확인"""
return self.quantity == 0
class OrderBookReconstructor:
"""
주문서 재구축 엔진
- SortedDict를 사용한 O(log n) 삽입/삭제
- 스냅샷 + Delta 방식으로 네트워크 비용 절감
- 멀티 심볼 지원
"""
def __init__(self, max_levels: int = 100):
self.max_levels = max_levels
# 멀티 심볼 지원: symbol -> {bids: SortedDict, asks: SortedDict}
self._books: Dict[str, Dict[str, SortedDict]] = defaultdict(
lambda: {
"bids": SortedDict(), # price -> quantity
"asks": SortedDict(), # price -> quantity
"last_update_id": 0,
"last_seq_num": 0
}
)
# 메타데이터
self._metadata: Dict[str, dict] = defaultdict(dict)
# 잠금 (멀티스레딩 지원)
self._lock = RLock()
# 통계를 위한 카운터
self._update_count = 0
self._snapshot_count = 0
def apply_snapshot(
self,
symbol: str,
bids: List[Tuple[str, str]],
asks: List[Tuple[str, str]],
last_update_id: int
):
"""
전체 스냅샷 적용
Binance REST API에서 가져온 초기 주문서 설정에 사용
"""
with self._lock:
book = self._books[symbol]
# 기존 데이터 클리어
book["bids"].clear()
book["asks"].clear()
# Bid 업데이트 (가격 내림차순 정렬)
for price, qty in sorted(bids, key=lambda x: -float(x[0])):
qty_decimal = Decimal(qty)
if qty_decimal > 0:
book["bids"][Decimal(price)] = qty_decimal
# Ask 업데이트 (가격 오름차순 정렬)
for price, qty in sorted(asks, key=lambda x: float(x[0])):
qty_decimal = Decimal(qty)
if qty_decimal > 0:
book["asks"][Decimal(price)] = qty_decimal
book["last_update_id"] = last_update_id
self._snapshot_count += 1
# 최대 레벨 제한
self._trim_levels(symbol)
def apply_depth_update(
self,
symbol: str,
bids: List[Tuple[str, str, str]], # price, quantity, ignore
asks: List[Tuple[str, str, str]],
update_id: int,
is_final: bool = True
):
"""
Delta 업데이트 적용
bids/asks: [(price, quantity, is_book_update), ...]
is_book_update가 false인 경우 이 업데이트는 건너뜀
"""
with self._lock:
book = self._books[symbol]
# 순서 검증 (U==u > last_update_id 필수)
if update_id <= book["last_update_id"]:
# 오래된 업데이트는 무시
return False
# Bid 업데이트
for price_str, qty_str, _ in bids:
price = Decimal(price_str)
qty = Decimal(qty_str)
if qty == 0:
# 수량이 0이면 해당 가격 레벨 제거
if price in book["bids"]:
del book["bids"][price]
else:
book["bids"][price] = qty
# Ask 업데이트
for price_str, qty_str, _ in asks:
price = Decimal(price_str)
qty = Decimal(qty_str)
if qty == 0:
if price in book["asks"]:
del book["asks"][price]
else:
book["asks"][price] = qty
book["last_update_id"] = update_id
self._update_count += 1
# 최대 레벨 제한
self._trim_levels(symbol)
return True
def apply_depth_stream(
self,
symbol: str,
bids: List[List[str]], # [price, quantity]
asks: List[List[str]],
update_id: int,
event_time: int
):
"""
WebSocket Depth Stream 업데이트 적용
"""
with self._lock:
book = self._books[symbol]
# Bid 업데이트
for item in bids:
price = Decimal(item[0])
qty = Decimal(item[1])
if qty == 0:
book["bids"].pop(price, None)
else:
book["bids"][price] = qty
# Ask 업데이트
for item in asks:
price = Decimal(item[0])
qty = Decimal(item[1])
if qty == 0:
book["asks"].pop(price, None)
else:
book["asks"][price] = qty
book["last_update_id"] = update_id
self._metadata[symbol]["last_event_time"] = event_time
self._update_count += 1
self._trim_levels(symbol)
def _trim_levels(self, symbol: str):
"""최대 레벨 수 초과 시トリ밍"""
book = self._books[symbol]
# Bid: 높은 가격순으로 max_levels 유지
if len(book["bids"]) > self.max_levels:
keys_to_remove = list(book["bids"].keys())[self.max_levels:]
for key in keys_to_remove:
del book["bids"][key]
# Ask: 낮은 가격순으로 max_levels 유지
if len(book["asks"]) > self.max_levels:
keys_to_remove = list(book["asks"].keys())[self.max_levels:]
for key in keys_to_remove:
del book["asks"][key]
def get_snapshot(self, symbol: str) -> dict:
"""주문서 스냅샷 반환"""
with self._lock:
book = self._books.get(symbol)
if not book:
return {}
bids = [
{"price": str(p), "quantity": str(q)}
for p, q in list(book["bids"].items())[:self.max_levels]
]
asks = [
{"price": str(p), "quantity": str(q)}
for p, q in list(book["asks"].items())[:self.max_levels]
]
return {
"symbol": symbol,
"last_update_id": book["last_update_id"],
"bids": bids,
"asks": asks,
"best_bid": bids[0] if bids else None,
"best_ask": asks[0] if asks else None,
"spread": float(Decimal(asks[0]["price"]) - Decimal(bids[0]["price"])) if bids and asks else None,
"mid_price": float((Decimal(bids[0]["price"]) + Decimal(asks[0]["price"])) / 2) if bids and asks else None,
"timestamp": self._metadata[symbol].get("last_event_time", 0)
}
def get_level(self, symbol: str, level: int = 0) -> dict:
"""특정 레벨의 가격 가져오기"""
with self._lock:
book = self._books.get(symbol)
if not book:
return {}
bids = list(book["bids"].items())
asks = list(book["asks"].items())
return {
"bid_level": {
"price": str(bids[level][0]) if len(bids) > level else None,
"quantity": str(bids[level][1]) if len(bids) > level else None
},
"ask_level": {
"price": str(asks[level][0]) if len(asks) > level else None,
"quantity": str(asks[level][1]) if len(asks) > level else None
}
}
def get_vwap(self, symbol: str, depth: int = 10) -> Optional[Decimal]:
"""
Volume Weighted Average Price 계산
특정 깊이까지의 모든 주문 수량과 가격을 고려한 평균가
"""
with self._lock:
book = self._books.get(symbol)
if not book:
return None
total_volume = Decimal("0")
weighted_price = Decimal("0")
# Bid VWAP
for price, qty in list(book["bids"].items())[:depth]:
weighted_price += price * qty
total_volume += qty
# Ask VWAP
for price, qty in list(book["asks"].items())[:depth]:
weighted_price += price * qty
total_volume += qty
if total_volume == 0:
return None
return weighted_price / total_volume
def get_imbalance(self, symbol: str, levels: int = 10) -> Optional[Decimal]:
"""
주문서 불균형 (Order Book Imbalance) 계산
> 0: 매수 압력, < 0: 매도 압력
"""
with self._lock:
book = self._books.get(symbol)
if not book:
return None
bid_volume = sum(
qty for _, qty in list(book["bids"].items())[:levels]
)
ask_volume = sum(
qty for _, qty in list(book["asks"].items())[:levels]
)
if bid_volume + ask_volume == 0:
return None
return (bid_volume - ask_volume) / (bid_volume + ask_volume)
def get_depth(self, symbol: str, levels: int = 20) -> dict:
"""주문서 깊이 정보 반환"""
with self._lock:
book = self._books.get(symbol)
if not book:
return {}
bid_depth = 0
bid_volume = Decimal("0")
ask_depth = 0
ask_volume = Decimal("0")
for price, qty in list(book["bids"].items())[:levels]:
bid_depth += 1
bid_volume += qty
for price, qty in list(book["asks"].items())[:levels]:
ask_depth += 1
ask_volume += qty
return {
"bid_levels": bid_depth,
"bid_volume": float(bid_volume),
"ask_levels": ask_depth,
"ask_volume": float(ask_volume),
"total_volume": float(bid_volume + ask_volume),
"imbalance": float(self.get_imbalance(symbol, levels))
}
def get_stats(self) -> dict:
"""통계 정보 반환"""
return {
"total_updates": self._update_count,
"total_snapshots": self._snapshot_count,
"tracked_symbols": len(self._books)
}
성능 벤치마크
def benchmark():
"""주문서 재구축 엔진 성능 테스트"""
import random
reconstructor = OrderBookReconstructor(max_levels=100)
# 초기 스냅샷 설정
bids = [(str(50000 + i * 10), str(random.uniform(0.1, 10))) for i in range(50)]
asks = [(str(50010 + i * 10), str(random.uniform(0.1, 10))) for i in range(50)]
reconstructor.apply_snapshot("BTCUSDT", bids, asks, 1)
# 업데이트 성능 테스트
iterations = 100000
start = time.perf_counter()
for i in range(iterations):
# 랜덤 업데이트
update_bids = [(str(50000 + random.randint(0, 500)), str(random.uniform(0, 5))) for _ in range(5)]
update_asks = [(str(50010 + random.randint(0, 500)), str(random.uniform(0, 5))) for _ in range(5)]
reconstructor.apply_depth_stream("BTCUSDT", update_bids, update_asks, i + 2, 0)
elapsed = time.perf_counter() - start
print(f"=== 벤치마크 결과 ===")
print(f"반복 횟수: {iterations:,}")
print(f"총 소요 시간: {elapsed:.3f}s")
print(f"초당 업데이트: {iterations/elapsed:,.0f}")
print(f"평균 업데이트 시간: {elapsed/iterations*1000:.4f}ms")
print(f"현재 통계: {reconstructor.get_stats()}")
if __name__ == "__main__":
benchmark()완전한 Tick 수집 및 주문서 재구축 파이프라인
이제 앞서 구현한 컴포넌트를 통합하여 프로덕션 수준의 Tick 수집 및 주문서 재구축 파이프라인을 구축하겠습니다. HolySheep AI를 활용한 AI 기반 시장 이상 탐지까지 포함됩니다.
import asyncio
import aiohttp
import json
import time
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from collections import deque
from threading import Thread
import logging
from decimal import Decimal
HolySheep AI SDK - AI 기반 시장 분석
NOTE: HolySheep는 본래 AI API 게이트웨이이나, 시장 데이터와 결합하여
AI 기반 이상 거래 탐지 시스템 구축에 활용 가능
import openai
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
HolySheep AI 설정
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # 실제 API 키로 교체
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
openai.api_key = HOLYSHEEP_API_KEY
openai.api_base = HOLYSHEEP_BASE_URL
@dataclass
class MarketAlert:
"""시장 이상 거래 경고"""
timestamp: int
symbol: str
alert_type: str # 'price_spike', 'liquidity_shift', 'order_imbalance'
severity: str # 'low', 'medium', 'high', 'critical'
details: dict
ai_analysis: Optional[str] = None
class TickCollector:
"""
Tick 데이터 수집기 + 주문서 재구축 통합 시스템
"""
def __init__(self, symbols: List[str]):
self.symbols = symbols
self.orderbook_reconstructor = OrderBookReconstructor(max_levels=100)
# Tick 데이터 버퍼 (최근 N개 저장)
self.tick_buffer: Dict[str, deque] = {
symbol: deque(maxlen=1000) for symbol in symbols
}
# 경고 큐
self.alert_queue: asyncio.Queue[MarketAlert] = asyncio.Queue()
# 실행 중 플래그
self._running = False
# Binance REST API용 세션
self._session: Optional[aiohttp.ClientSession] = None
# HolySheep AI 클라이언트
self._ai_client = openai
# 통계
self._stats = {
"ticks_collected": 0,
"alerts_generated": 0,
"alerts_analyzed_by_ai": 0,
"start_time": 0
}
async def initialize_orderbooks(self):
"""
Binance REST API에서 초기 주문서 스냅샷 가져와서 설정
"""
self._session = aiohttp.ClientSession()
for symbol in self.symbols:
try:
url = f"https://api.binance.com/api/v3/depth"
params = {"symbol": symbol.upper(), "limit": 100}
async with self._session.get(url, params=params) as resp:
if resp.status == 200:
data = await resp.json()
self.orderbook_reconstructor.apply_snapshot(
symbol,
data["bids"],
data["asks"],
data["lastUpdateId"]
)
logger.info(f"[{symbol}] 주문서 초기화 완료: {data['lastUpdateId']}")
else:
logger.error(f"[{symbol}] 주문서 가져오기 실패: {resp.status}")
except Exception as e:
logger.error(f"[{symbol}] 주문서 초기화 오류: {e}")
async def start_websocket_streams(self):
"""WebSocket 스트림 시작"""
self._running = True
self._stats["start_time"] = time.time()
# 다중 스트림 URL 생성
streams = []
for symbol in self.symbols:
streams.append(f"{symbol}@trade")
streams.append(f"{symbol}@depth20@100ms")
url = f"wss://stream.binance.com:9443/stream?