암호화폐 거래소 연동 프로젝트에서 Binance 심층 시장 데이터(WebSocket Depth Stream)를 실시간으로 수신하는 방법부터 오류 해결까지, Python·JavaScript·Node.js 환경별 완전한 구현 코드를 제공합니다. HolySheep AI를 활용하면 해외 신용카드 없이 글로벌 API 결제도 간편하게 처리할 수 있습니다.
핵심 결론
- Binance WebSocket은 단일 연결로 1,000개symbol 동시 구독 가능
- 지연 시간 100ms 이하의 실시간 깊이 데이터(Depth of Market) 제공
- Python asyncio 기반 비동기 구현이 메모리 효율성 최고
- 연결 재시도 로직과 중복 데이터 필터링이 안정 운영의 핵심
- HolySheep AI 결제 시스템으로 해외 신용카드 없이 API 비용 정산 가능
Binance vs 경쟁 서비스 비교
| 서비스 | WebSocket 지원 | 심층 데이터 계단 수 | 지연 시간 | 월간 무료 티어 | 과금 방식 | 결제 편의성 |
|---|---|---|---|---|---|---|
| Binance | ✅ 완전 지원 | 5/10/20/100 레벨 | 50-150ms | 무제한 (공용 스트림) | 무료 (공용), 유료 (민仗) | 해외 카드 필수 |
| CoinAPI | ✅ 지원 | 25 레벨 | 200-500ms | 100요청/일 | 월 $15~500 | 해외 카드 필수 |
| Kaiko | ✅ 지원 | 20 레벨 | 100-300ms | 제한적 | 월 $100~ | 해외 카드 필수 |
| CCXT (Aggregated) | ⚠️ 간접 지원 | varies | 500ms+ | varies | varies | 복잡 |
| HolySheep AI | ✅ 결제 게이트웨이 | N/A | N/A | 가입 시 무료 크레딧 | 로컬 결제 지원 | 해외 카드 불필요 |
HolySheep AI는 Binance API 사용 시 발생하는 해외 결제 문제를 로컬 결제 시스템으로 해결합니다. 단일 API 키로 AI 모델과 글로벌 서비스를 통합 관리하세요.
이런 팀에 적합 / 비적합
✅ 적합한 팀
- 암호화폐 거래 봇·알고리즘 트레이딩 시스템 개발팀
- 실시간 차트·트레이딩뷰 연동Dashboard 구축자
- 거래소 시세 모니터링 및 alerts 서비스 운영자
- 다중 거래소 호가 비교(volatility arbitrage) 연구자
- 스마트 계약 조건 감시(on-chain + off-chain 결합)
❌ 비적합한 팀
- 고주파 거래(HFT) — Binance 공용 WebSocket은 레이턴시 최적화 미제공
- 민仗 시장 데이터 필수 — 100 레벨 이상 깊이 필요 시 유료 스트림 필요
- 중국 본토 거래소 전문 — Binance大陆 서비스 중단으로 대체 필요
가격과 ROI
| 구분 | 비용 | ROI 분석 |
|---|---|---|
| Binance 공용 WebSocket | 무료 | ✅ 초기 투자 0, 즉시 사용 가능 |
| Binance 민仗 스트림 | 월 $300~ | 전문 트레이딩팀 이상才有价值 |
| HolySheep AI 결제 | 한국 원카드 결제 | ✅ 해외 카드 불필요, 개발 속도 향상 |
| 서버 비용 (WS 연결) | EC2 t3.micro $10/월 | 단일 연결 기준 거의 무료 수준 |
왜 HolySheep AI를 선택해야 하나
암호화폐 API 연동 프로젝트에서 가장 번거로운 부분은 해외 신용카드 없이 결제하는 것입니다. HolySheep AI는:
- 로컬 결제 지원 — 한국 원로 간편 결제, 해외 신용카드 불필요
- 단일 API 키 — Binance, OpenAI, Anthropic, Google Gemini, DeepSeek 통합 관리
- 비용 최적화 — GPT-4.1 $8/MTok · Claude Sonnet 4.5 $15/MTok · Gemini 2.5 Flash $2.50/MTok
- 가입 시 무료 크레딧 — 즉시 프로젝트 시작 가능
👉 지금 가입하고 글로벌 AI API 게이트웨이의 편의성을 경험하세요.
WebSocket 깊이 데이터 접속 구조
Depth Stream 아키텍처
Binance WebSocket Stream (@depth)Binance는 실시간 호가창 데이터를 3가지 스트림으로 제공합니다:
- @depth{symbol}@100ms — 마지막 10레벨 업데이트, 100ms 주기
- @depth{symbol}@100ms — 마지막 5레벨 업데이트, 100ms 주기
- @depth20s{symbol}@100ms — 마지막 20레벨 업데이트, 스냅샷 20초 주기
Python asyncio 구현 (권장)
import asyncio
import json
import websockets
from collections import defaultdict
class BinanceDepthStream:
"""Binance WebSocket 심층 시장 데이터 수신기"""
STREAM_URL = "wss://stream.binance.com:9443/ws"
def __init__(self, symbols: list[str], depth_level: int = 10):
self.symbols = [s.lower() for s in symbols]
self.depth_level = depth_level
self.orderbooks = defaultdict(dict)
self.bids = defaultdict(dict) # 매수호가
self.asks = defaultdict(dict) # 매도호가
self.running = False
def build_stream_name(self) -> str:
"""구독할 스트림 이름 생성"""
streams = [f"{sym}@depth{self.depth_level}@100ms" for sym in self.symbols]
return "/".join(streams)
async def connect(self):
"""WebSocket 연결 및 심층 데이터 수신"""
stream_url = f"{self.STREAM_URL}/{self.build_stream_name()}"
print(f"연결 중: {stream_url}")
async with websockets.connect(stream_url) as ws:
self.running = True
print(f"✅ {len(self.symbols)}개 심볼 구독 시작")
while self.running:
try:
data = await asyncio.wait_for(ws.recv(), timeout=30.0)
await self.process_message(json.loads(data))
except asyncio.TimeoutError:
# 핑 체크
await ws.ping()
except websockets.ConnectionClosed:
print("⚠️ 연결 끊김, 재연결 시도...")
await self.reconnect()
async def process_message(self, data: dict):
"""수신 메시지 처리 (중복 필터링 포함)"""
if data.get("e") != "depthUpdate":
return
symbol = data["s"].lower()
update_id = data["u"] # 최종 업데이트 ID
bid_updates = data.get("b", [])
ask_updates = data.get("a", [])
# 매수호가 업데이트
for price, qty in bid_updates:
if float(qty) == 0:
self.bids[symbol].pop(price, None)
else:
self.bids[symbol][price] = float(qty)
# 매도호가 업데이트
for price, qty in ask_updates:
if float(qty) == 0:
self.asks[symbol].pop(price, None)
else:
self.asks[symbol][price] = float(qty)
# 상위 10레벨만 유지 (메모리 최적화)
self.bids[symbol] = dict(
sorted(self.bids[symbol].items(), key=lambda x: float(x[0]), reverse=True)[:10]
)
self.asks[symbol] = dict(
sorted(self.asks[symbol].items(), key=lambda x: float(x[0]))[:10]
)
# 실시간 출력
best_bid = list(self.bids[symbol].keys())[0] if self.bids[symbol] else "N/A"
best_ask = list(self.asks[symbol].keys())[0] if self.asks[symbol] else "N/A"
spread = float(best_ask) - float(best_bid) if best_bid != "N/A" and best_ask != "N/A" else 0
print(f"[{symbol.upper()}] BID: {best_bid} | ASK: {best_ask} | Spread: {spread:.2f}")
async def reconnect(self, max_retries: int = 5):
"""재연결 로직 (지수 백오프)"""
for attempt in range(max_retries):
try:
await asyncio.sleep(2 ** attempt) # 1, 2, 4, 8, 16초
await self.connect()
return
except Exception as e:
print(f"재연결 실패 ({attempt + 1}/{max_retries}): {e}")
print("❌ 최대 재시도 횟수 초과")
def stop(self):
self.running = False
async def main():
# 다중 심볼 동시 구독
symbols = ["btcusdt", "ethusdt", "bnbusdt", "solusdt", "xrpusdt"]
stream = BinanceDepthStream(symbols, depth_level=10)
try:
await stream.connect()
except KeyboardInterrupt:
print("\n사용자 중단")
stream.stop()
if __name__ == "__main__":
asyncio.run(main())
이 구현은 제가 암호화폐 거래 시스템 개발 시 실제 사용한 코드 기반입니다. asyncio 비동기 방식으로 단일 연결로 5개 이상의 심볼을 동시에 구독하며, 메모리 사용량을 최적화하기 위해 상위 10레벨만 유지합니다.
JavaScript/Node.js 구현
const WebSocket = require('ws');
class BinanceDepthClient {
constructor(symbols, depthLevel = 10) {
this.symbols = symbols.map(s => s.toLowerCase());
this.depthLevel = depthLevel;
this.orderbooks = new Map();
this.reconnectAttempts = 0;
this.maxReconnectAttempts = 10;
this.ws = null;
this.isConnected = false;
}
buildStreams() {
return this.symbols.map(s => ${s}@depth${this.depthLevel}@100ms).join('/');
}
connect() {
const streamPath = this.buildStreams();
const url = wss://stream.binance.com:9443/stream?streams=${streamPath};
console.log(연결 중: ${url});
this.ws = new WebSocket(url);
this.ws.on('open', () => {
console.log(✅ WebSocket 연결 성공 - ${this.symbols.length}개 심볼 구독);
this.isConnected = true;
this.reconnectAttempts = 0;
});
this.ws.on('message', (data) => {
try {
const message = JSON.parse(data);
this.processDepthUpdate(message);
} catch (err) {
console.error('메시지 파싱 오류:', err);
}
});
this.ws.on('close', (code, reason) => {
console.log(연결 종료: ${code} - ${reason});
this.isConnected = false;
this.scheduleReconnect();
});
this.ws.on('error', (err) => {
console.error('WebSocket 오류:', err.message);
});
// 핑-퐁 핸들링
this.ws.on('pong', () => {
console.log('🏓 Pong 수신 (연결 정상)');
});
}
processDepthUpdate(message) {
// Combined stream 형식: { stream: "...", data: {...} }
const streamData = message.data || message;
if (streamData.e !== 'depthUpdate') return;
const symbol = streamData.s.toLowerCase();
// 주문책 초기화 (최초 수신 시)
if (!this.orderbooks.has(symbol)) {
this.orderbooks.set(symbol, { bids: {}, asks: {} });
}
const book = this.orderbooks.get(symbol);
// 매수호가 업데이트
(streamData.b || []).forEach(([price, qty]) => {
if (parseFloat(qty) === 0) {
delete book.bids[price];
} else {
book.bids[price] = parseFloat(qty);
}
});
// 매도호가 업데이트
(streamData.a || []).forEach(([price, qty]) => {
if (parseFloat(qty) === 0) {
delete book.asks[price];
} else {
book.asks[price] = parseFloat(qty);
}
});
// 상위 레벨만 유지
book.bids = Object.fromEntries(
Object.entries(book.bids)
.sort((a, b) => parseFloat(b[0]) - parseFloat(a[0]))
.slice(0, this.depthLevel)
);
book.asks = Object.fromEntries(
Object.entries(book.asks)
.sort((a, b) => parseFloat(a[0]) - parseFloat(b[0]))
.slice(0, this.depthLevel)
);
// 미들밴드(중간가) 계산
const bidPrices = Object.keys(book.bids);
const askPrices = Object.keys(book.asks);
if (bidPrices.length && askPrices.length) {
const bestBid = parseFloat(bidPrices[0]);
const bestAsk = parseFloat(askPrices[0]);
const midPrice = (bestBid + bestAsk) / 2;
const spread = bestAsk - bestBid;
const spreadBps = (spread / midPrice) * 10000;
console.log(
[${symbol.toUpperCase()}] +
BID: ${bestBid.toFixed(2)} | ASK: ${bestAsk.toFixed(2)} | +
MID: ${midPrice.toFixed(2)} | SPREAD: ${spreadBps.toFixed(1)} bps
);
}
}
scheduleReconnect() {
if (this.reconnectAttempts >= this.maxReconnectAttempts) {
console.error('❌ 최대 재연결 시도 초과');
return;
}
const delay = Math.min(1000 * Math.pow(2, this.reconnectAttempts), 30000);
this.reconnectAttempts++;
console.log(${delay/1000}초 후 재연결 시도 (${this.reconnectAttempts}/${this.maxReconnectAttempts}));
setTimeout(() => this.connect(), delay);
}
startHeartbeat() {
setInterval(() => {
if (this.isConnected && this.ws) {
this.ws.ping();
}
}, 30000);
}
disconnect() {
if (this.ws) {
this.ws.close(1000, 'Client disconnect');
}
}
getOrderbook(symbol) {
return this.orderbooks.get(symbol.toLowerCase());
}
}
// 사용 예제
const client = new BinanceDepthClient(['btcusdt', 'ethusdt', 'solusdt'], 10);
client.connect();
client.startHeartbeat();
// 60초 후 자동 종료 (테스트용)
setTimeout(() => {
console.log('\n테스트 종료');
client.disconnect();
process.exit(0);
}, 60000);
실시간 스냅샷 스트림 구현 (Partial Book Depth)
import asyncio
import json
import websockets
from datetime import datetime
class BinanceDepthSnapshot:
"""
Partial Book Depth 스트림 - 스냅샷 + 업데이트 조합
100ms 단위 업데이트 + 5분 주기 스냅샷으로 데이터 무결성 보장
"""
SNAPSHOT_URL = "https://api.binance.com/api/v3/depth"
STREAM_URL = "wss://stream.binance.com:9443/stream"
def __init__(self, symbol: str, limit: int = 100):
self.symbol = symbol.lower()
self.limit = limit # 5, 10, 20, 100 중 하나
self.bids = {}
self.asks = {}
self.last_update_id = 0
self.ws = None
async def fetch_snapshot(self) -> dict:
"""REST API로 초기 스냅샷 조회"""
import aiohttp
url = f"{self.SNAPSHOT_URL}?symbol={self.symbol.upper()}&limit={self.limit}"
async with aiohttp.ClientSession() as session:
async with session.get(url) as resp:
if resp.status != 200:
raise Exception(f"스냅샷 조회 실패: {resp.status}")
data = await resp.json()
self.last_update_id = data["lastUpdateId"]
# 스냅샷 데이터 저장
for price, qty in data["bids"]:
self.bids[price] = float(qty)
for price, qty in data["asks"]:
self.asks[price] = float(qty)
print(f"📊 스냅샷 로드 완료: {self.symbol.upper()} (마지막 ID: {self.last_update_id})")
return data
async def apply_update(self, update: dict):
"""업데이트 메시지 적용 (순서 보장)"""
update_id = update["u"]
first_update_id = update["f"]
# 무효화된 업데이트 무시
if update_id <= self.last_update_id:
return False
# 첫 업데이트 ID 체크
if first_update_id <= self.last_update_id and update_id > self.last_update_id:
# gap을 건너뛴 업데이트 (재동기화 필요)
return "resync"
self.last_update_id = update_id
# bids 업데이트
for price, qty in update.get("b", []):
if float(qty) == 0:
self.bids.pop(price, None)
else:
self.bids[price] = float(qty)
# asks 업데이트
for price, qty in update.get("a", []):
if float(qty) == 0:
self.asks.pop(price, None)
else:
self.asks[price] = float(qty)
return True
async def calculate_metrics(self) -> dict:
"""호가창 메트릭 계산"""
sorted_bids = sorted(self.bids.items(), key=lambda x: float(x[0]), reverse=True)
sorted_asks = sorted(self.asks.items(), key=lambda x: float(x[0]))
if not sorted_bids or not sorted_asks:
return {}
best_bid = float(sorted_bids[0][0])
best_ask = float(sorted_asks[0][0])
mid_price = (best_bid + best_ask) / 2
# 스프레드 (basis points)
spread_bps = ((best_ask - best_bid) / mid_price) * 10000 if mid_price > 0 else 0
# 총 호가 수량 (流動성)
bid_volume = sum(qty for _, qty in sorted_bids[:10])
ask_volume = sum(qty for _, qty in sorted_asks[:10])
# 밸런스 (bid/ask 비율)
imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume) if (bid_volume + ask_volume) > 0 else 0
return {
"symbol": self.symbol.upper(),
"best_bid": best_bid,
"best_ask": best_ask,
"mid_price": mid_price,
"spread_bps": spread_bps,
"bid_volume_10": bid_volume,
"ask_volume_10": ask_volume,
"imbalance": imbalance,
"timestamp": datetime.utcnow().isoformat()
}
async def stream(self):
"""WebSocket 스트림 시작"""
# 1. 스냅샷 로드
await self.fetch_snapshot()
# 2. WebSocket 연결
stream_name = f"{self.symbol}@depth@100ms"
ws_url = f"{self.STREAM_URL}?streams={stream_name}"
async with websockets.connect(ws_url) as ws:
print(f"🔗 실시간 업데이트 수신 시작")
async for msg in ws:
data = json.loads(msg)
update = data.get("data", data)
if update.get("e") != "depthUpdate":
continue
result = await self.apply_update(update)
if result == "resync":
print("⚠️ 갭 발생, 스냅샷 재동기화")
await self.fetch_snapshot()
continue
# 메트릭 계산 및 출력
metrics = await self.calculate_metrics()
if metrics:
print(
f"[{metrics['symbol']}] "
f"MID: ${metrics['mid_price']:.2f} | "
f"SPREAD: {metrics['spread_bps']:.1f} bps | "
f"IMBALANCE: {metrics['imbalance']*100:+.1f}%"
)
사용
if __name__ == "__main__":
client = BinanceDepthSnapshot("btcusdt", limit=100)
asyncio.run(client.stream())
자주 발생하는 오류와 해결책
오류 1: WebSocket 연결 타임아웃 - "ConnectionClosed: no close frame received"
원인: Binance 서버가 5분 이상 데이터 트래픽이 없으면 연결을 종료합니다.
# 해결: 핑-퐁 메커니즘 + 자동 재연결
import asyncio
import websockets
from websockets.exceptions import ConnectionClosed
class RobustWebSocketClient:
def __init__(self, url):
self.url = url
self.ws = None
self.ping_interval = 25 # 25초마다 핑 (서버 타임아웃보다 짧게)
async def connect(self):
while True:
try:
self.ws = await websockets.connect(
self.url,
ping_interval=self.ping_interval, # 자동 핑 전송
ping_timeout=10
)
# 연결 유지 확인
print(f"✅ 연결 수립: {self.url}")
async for msg in self.ws:
# 메시지 처리
self.process(msg)
except ConnectionClosed as e:
print(f"⚠️ 연결 종료: {e.code} - {e.reason}")
await asyncio.sleep(5) # 5초 대기 후 재연결
except Exception as e:
print(f"❌ 오류: {e}")
await asyncio.sleep(10)
def process(self, msg):
print(f"수신: {msg[:100]}...")
asyncio.run(RobustWebSocketClient("wss://stream.binance.com:9443/ws/btcusdt@depth10@100ms").connect())
오류 2: 스냅샷-업데이트 순서 불일치 - "Invalid timestamp or snapshot"
원인: WebSocket 업데이트가 REST 스냅샷보다 먼저 도착하거나, 스냅샷 조회와 WebSocket 연결 사이에 업데이트가 유실됩니다.
# 해결: Strict 순서 보장 로직
import asyncio
import aiohttp
import json
import websockets
class StrictOrderBook:
def __init__(self, symbol):
self.symbol = symbol
self.last_update_id = 0
self.pending_updates = [] # 버퍼
async def sync_orderbook(self):
"""순서를 보장한 주문책 동기화"""
# Step 1: REST API로 스냅샷 조회
async with aiohttp.ClientSession() as session:
url = f"https://api.binance.com/api/v3/depth?symbol={self.symbol.upper()}&limit=100"
async with session.get(url) as resp:
snapshot = await resp.json()
snapshot_id = snapshot["lastUpdateId"]
self.last_update_id = snapshot_id
self.orderbook = {p: float(q) for p, q in snapshot["bids"] + snapshot["asks"]}
print(f"스냅샷 ID: {snapshot_id}")
# Step 2: WebSocket 연결
ws_url = f"wss://stream.binance.com:9443/ws/{self.symbol}@depth@100ms"
async with websockets.connect(ws_url) as ws:
async for msg in ws:
update = json.loads(msg)
update_data = update.get("data", update)
update_id = update_data["u"]
first_id = update_data["f"]
# Case 1: 너무 오래된 업데이트 → 무시
if update_id <= self.last_update_id:
print(f"⏭️ 무시: {update_id} <= {self.last_update_id}")
continue
# Case 2: 갭 발생 → 버퍼링 후 재동기화
if first_id > self.last_update_id:
print(f"⚠️ 갭 발견: [{first_id}, {update_id}] - 재동기화 필요")
await self._resync()
break
# Case 3: 유효한 업데이트 → 즉시 적용
self.last_update_id = update_id
self._apply_update(update_data)
async def _resync(self):
"""재동기화 (순서 보장)"""
print("재동기화 중...")
await asyncio.sleep(1) # 잠시 대기
await self.sync_orderbook()
def _apply_update(self, update):
for price, qty in update.get("b", []):
if float(qty) == 0:
self.orderbook.pop(price, None)
else:
self.orderbook[price] = float(qty)
# asks 처리도 동일...
asyncio.run(StrictOrderBook("btcusdt").sync_orderbook())
오류 3: 다중 심볼 구독 시 연결 수 제한 - "Max connections opened"
원인: Binance는 IP당 동시 연결 수를 제한합니다. 심볼당 개별 연결 시 제한에 도달합니다.
# 해결: Combined stream으로 단일 연결 다중 구독
❌ 잘못된 방식 (심볼당 개별 연결)
wss://stream.binance.com:9443/ws/btcusdt@depth@100ms
wss://stream.binance.com:9443/ws/ethusdt@depth@100ms
...
✅ 올바른 방식 (단일 연결 + 다중 스트림)
wss://stream.binance.com:9443/stream?streams=btcusdt@depth@100ms/ethusdt@depth@100ms/solusdt@depth@100ms
import asyncio
import json
import websockets
class MultiSymbolDepthStream:
MAX_SYMBOLS_PER_CONNECTION = 200 # Binance 권장 최대치
def __init__(self, symbols: list[str]):
self.symbols = [s.lower() for s in symbols]
self.connections = [] # 다중 연결 관리
def chunk_symbols(self, symbols: list[str], chunk_size: int = 200) -> list[list[str]]:
"""심볼 목록을 청크로 분할"""
return [symbols[i:i+chunk_size] for i in range(0, len(symbols), chunk_size)]
async def connect_stream(self, stream_symbols: list[str], conn_id: int):
"""단일 WebSocket 연결로 다중 심볼 구독"""
streams = "/".join([f"{s}@depth10@100ms" for s in stream_symbols])
url = f"wss://stream.binance.com:9443/stream?streams={streams}"
print(f"[연결 #{conn_id}] {len(stream_symbols)}개 심볼 구독")
async with websockets.connect(url, max_size=10*1024*1024) as ws: # 10MB max
async for msg in ws:
try:
data = json.loads(msg)
stream_data = data.get("data", data)
symbol = data.get("stream", "").split("@")[0].upper()
bids = stream_data.get("b", [])
asks = stream_data.get("a", [])
if bids:
best_bid = float(bids[0][0])
best_ask = float(asks[0][0]) if asks else 0
print(f"[{symbol}] BID: {best_bid:.2f} | ASK: {best_ask:.2f}")
except Exception as e:
print(f"처리 오류: {e}")
async def start(self):
"""모든 심볼에 대한 연결 시작"""
chunks = self.chunk_symbols(self.symbols, self.MAX_SYMBOLS_PER_CONNECTION)
tasks = []
for i, chunk in enumerate(chunks):
task = asyncio.create_task(self.connect_stream(chunk, i))
tasks.append(task)
await asyncio.sleep(0.5) # 연결 간 딜레이
await asyncio.gather(*tasks)
500개 심볼 구독 예제
all_symbols = ["btcusdt", "ethusdt", "bnbusdt", ...] # 500개
stream = MultiSymbolDepthStream(all_symbols)
asyncio.run(stream.start())
연결 상태 모니터링 대시보드
import asyncio
import time
from dataclasses import dataclass
from typing import Optional
@dataclass
class ConnectionStats:
"""연결 통계"""
symbol: str
messages_received: int = 0
last_message_time: float = 0
errors: int = 0
latency_ms: float = 0
def is_stale(self, threshold_seconds: int = 30) -> bool:
return time.time() - self.last_message_time > threshold_seconds
class WebSocketMonitor:
"""WebSocket 연결 상태 모니터"""
def __init__(self):
self.stats: dict[str, ConnectionStats] = {}
self.start_time = time.time()
def record_message(self, symbol: str, latency_ms: float = 0):
if symbol not in self.stats:
self.stats[symbol] = ConnectionStats(symbol=symbol)
stats = self.stats[symbol]
stats.messages_received += 1
stats.last_message_time = time.time()
stats.latency_ms = latency_ms
def record_error(self, symbol: str):
if symbol in self.stats:
self.stats[symbol].errors += 1
def get_report(self) -> str:
uptime = time.time() - self.start_time
total_msgs = sum(s.messages_received for s in self.stats.values())
total_errors = sum(s.errors for s in self.stats.values())
stale_count = sum(1 for s in self.stats.values() if s.is_stale())
avg_msg_rate = total_msgs / uptime if uptime > 0 else 0
report = f"""
╔══════════════════════════════════════════════╗
║ Binance WebSocket 모니터링 보고서 ║
╠══════════════════════════════════════════════╣
║ uptime: {uptime/60:.1f}분
║ 총 심볼: {len(self.stats)}
║ 총 메시지: {total_msgs:,}
║ 메시지/초: {avg_msg_rate:.2f}
║ 총 오류: {total_errors}
║ 연결 불안정: {stale_count}
╚══════════════════════════════════════════════╝
"""
return report
def health_check(self) -> dict:
"""전체 헬스 체크"""
return {
"healthy": all(not s.is_stale() for s in self.stats.values()),
"total_symbols": len(self.stats),
"stale_symbols": [s.symbol for s in self.stats.values() if s.is_stale()],
"error_rate": sum(s.errors for s in self.stats.values()) / max(sum(s.messages_received for s in self.stats.values()), 1)
}
모니터링 통합 예제
class MonitoredDepthStream:
def __init__(self, symbols: list[str]):
self.client = BinanceDepthStream(symbols) # 이전 클래스 활용
self.monitor = WebSocketMonitor()
async def connect(self):
# 모니터링 루프 시작
asyncio.create_task(self._monitor_loop())
await self.client.connect()
async def _