암호화폐 거래소를 대상으로 한 실시간 시세 데이터 수집 시스템은 금융 데이터를 다루는 가장 도전적인 시나리오 중 하나입니다. millisecond 단위의 지연과 99.9% 이상의 가용성이 요구되며, 동시에 여러 거래소의 WebSocket 스트림을 안정적으로 구독해야 합니다. 이 튜토리얼에서는 Python asyncio를 활용한 프로덕션 레벨 다중 거래소 WebSocket 연결 관리 아키텍처를 소개하겠습니다.
문제 정의 및 시스템 요구사항
암호화폐 거래소의 WebSocket API는 REST API와 달리 persistent connection을 유지하며 실시간 체결/호가창 데이터를 push 방식으로 전달합니다. 다중 거래소 통합 시 다음과 같은 과제가 존재합니다:
- 연결 관리 복잡성: Binance, Bybit, OKX, CoinEx 등 각 거래소별 고유 프로토콜과 재연결 메커니즘
- 메시지 처리 병목: 초당 수천 건의 메시지를 비동기적으로 처리해야 하는throughput 문제
- 데이터 정합성: 서로 다른 거래소 간 시간 동기화 및 순서 보장
- 장애 복구: 네트워크 단절, 거래소 서버 장애 시 자동 재연결
- 리소스 효율성: 수백 개의 trading pair를 동시에 구독할 때의 메모리/CPU 관리
아키텍처 설계
전체 시스템 구조
┌─────────────────────────────────────────────────────────────────┐
│ Application Layer │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────────┐ │
│ │ Price Alert │ │ Portfolio │ │ Arbitrage Detector │ │
│ │ Service │ │ Aggregator │ │ │ │
│ └──────┬──────┘ └──────┬──────┘ └────────────┬────────────┘ │
└─────────┼────────────────┼──────────────────────┼───────────────┘
│ │ │
▼ ▼ ▼
┌─────────────────────────────────────────────────────────────────┐
│ Aggregator Layer ( asyncio.Queue ) │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ UnifiedMarketData: 거래소별 포맷 → 표준 포맷 변환 │ │
│ └─────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────┐
│ Connection Manager Layer │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ Binance │ │ Bybit │ │ OKX │ │ CoinEx │ │
│ │ Connector │ │ Connector │ │ Connector │ │ Connector │ │
│ └────────────┘ └────────────┘ └────────────┘ └────────────┘ │
└─────────────────────────────────────────────────────────────────┘
│ │ │ │
▼ ▼ ▼ ▼
wss://stream wss://stream wss://ws.okx wss://stream
핵심 컴포넌트 설계
# exchange_aggregator/core/connection_manager.py
import asyncio
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Dict, List, Optional, Set
from collections import defaultdict
import json
logger = logging.getLogger(__name__)
class ExchangeStatus(Enum):
DISCONNECTED = "disconnected"
CONNECTING = "connecting"
CONNECTED = "connected"
RECONNECTING = "reconnecting"
ERROR = "error"
@dataclass
class WebSocketMessage:
exchange: str
symbol: str
message_type: str
data: dict
timestamp: datetime = field(default_factory=datetime.utcnow)
raw_data: Optional[str] = None
@dataclass
class UnifiedTicker:
symbol: str
exchanges: Dict[str, dict] # exchange -> ticker_data
best_bid_price: float
best_ask_price: float
best_bid_exchange: str
best_ask_exchange: str
spread_bps: float
timestamp: datetime = field(default_factory=datetime.utcnow)
class BaseExchangeConnector(ABC):
"""거래소 WebSocket 커넥터 기본 클래스"""
def __init__(
self,
exchange_name: str,
symbols: List[str],
message_queue: asyncio.Queue,
ping_interval: int = 20,
ping_timeout: int = 10,
max_reconnect_attempts: int = 10,
reconnect_delay_base: float = 1.0,
):
self.exchange_name = exchange_name
self.symbols = set(symbols)
self.queue = message_queue
self.ping_interval = ping_interval
self.ping_timeout = ping_timeout
self.max_reconnect_attempts = max_reconnect_attempts
self.reconnect_delay_base = reconnect_delay_base
self._websocket: Optional[asyncio.WebSocketClientProtocol] = None
self._connection_lock = asyncio.Lock()
self._status = ExchangeStatus.DISCONNECTED
self._reconnect_attempts = 0
self._should_run = False
self._subscriptions: Set[str] = set()
self._last_pong_time: Optional[datetime] = None
self._processing_stats = {"messages": 0, "errors": 0, "reconnects": 0}
@property
def status(self) -> ExchangeStatus:
return self._status
@abstractmethod
def get_websocket_url(self) -> str:
"""거래소별 WebSocket URL 반환"""
pass
@abstractmethod
def format_subscription_message(self, symbols: List[str]) -> str:
"""구독 요청 메시지 포맷 생성"""
pass
@abstractmethod
def format_unsubscription_message(self, symbols: List[str]) -> str:
"""구독 취소 메시지 포맷 생성"""
pass
@abstractmethod
async def parse_message(self, raw_message: str) -> List[WebSocketMessage]:
"""수신 메시지를 표준 포맷으로 변환"""
pass
def _create_ping_task(self) -> asyncio.Task:
"""Ping/Pong 유지工作任务创建"""
async def ping_loop():
while self._should_run and self._status == ExchangeStatus.CONNECTED:
try:
await asyncio.sleep(self.ping_interval)
if self._websocket:
await self._websocket.ping()
logger.debug(f"{self.exchange_name}: Ping sent")
except Exception as e:
logger.error(f"{self.exchange_name}: Ping error - {e}")
break
return asyncio.create_task(ping_loop())
async def connect(self) -> bool:
"""WebSocket 연결 수립"""
async with self._connection_lock:
if self._status == ExchangeStatus.CONNECTED:
return True
self._status = ExchangeStatus.CONNECTING
self._should_run = True
try:
url = self.get_websocket_url()
logger.info(f"{self.exchange_name}: Connecting to {url}")
self._websocket = await asyncio.wait_for(
asyncio.get_event_loop().create_connection(
asyncio.WebSocketClientProtocol,
url
)
)[0].__aenter__()
self._status = ExchangeStatus.CONNECTED
self._reconnect_attempts = 0
self._last_pong_time = datetime.utcnow()
# 구독 메시지 전송
await self._send_subscription()
logger.info(f"{self.exchange_name}: Connected successfully")
return True
except asyncio.TimeoutError:
logger.error(f"{self.exchange_name}: Connection timeout")
self._status = ExchangeStatus.ERROR
return False
except Exception as e:
logger.error(f"{self.exchange_name}: Connection failed - {e}")
self._status = ExchangeStatus.ERROR
return False
async def _send_subscription(self):
"""구독 요청 전송"""
if self.symbols:
message = self.format_subscription_message(list(self.symbols))
await self._websocket.send(message)
self._subscriptions.update(self.symbols)
logger.info(f"{self.exchange_name}: Subscribed to {len(self.symbols)} symbols")
async def disconnect(self):
"""연결 해제"""
self._should_run = False
async with self._connection_lock:
if self._websocket:
try:
await self._websocket.close()
except Exception:
pass
self._websocket = None
self._status = ExchangeStatus.DISCONNECTED
self._subscriptions.clear()
logger.info(f"{self.exchange_name}: Disconnected")
async def _handle_reconnect(self):
"""자동 재연결 처리"""
if self._reconnect_attempts >= self.max_reconnect_attempts:
logger.critical(
f"{self.exchange_name}: Max reconnect attempts reached"
)
self._status = ExchangeStatus.ERROR
return
self._status = ExchangeStatus.RECONNECTING
self._reconnect_attempts += 1
self._processing_stats["reconnects"] += 1
# 지수 백오프 딜레이
delay = min(
self.reconnect_delay_base * (2 ** (self._reconnect_attempts - 1)),
60.0
)
logger.warning(
f"{self.exchange_name}: Reconnecting in {delay:.1f}s "
f"(attempt {self._reconnect_attempts}/{self.max_reconnect_attempts})"
)
await asyncio.sleep(delay)
await self.connect()
async def run(self):
"""메시지 리스닝 루프"""
await self.connect()
ping_task = None
if self._status == ExchangeStatus.CONNECTED:
ping_task = self._create_ping_task()
while self._should_run:
try:
if not self._websocket or self._status != ExchangeStatus.CONNECTED:
await self._handle_reconnect()
continue
try:
raw_message = await asyncio.wait_for(
self._websocket.recv(),
timeout=self.ping_timeout + 5
)
except asyncio.TimeoutError:
logger.warning(f"{self.exchange_name}: No message received, checking connection")
continue
# 바이너리 메시지 처리
if isinstance(raw_message, bytes):
raw_message = raw_message.decode('utf-8')
# 메시지 파싱 및 큐 전달
parsed_messages = await self.parse_message(raw_message)
for msg in parsed_messages:
await self.queue.put(msg)
self._processing_stats["messages"] += 1
except asyncio.CancelledError:
logger.info(f"{self.exchange_name}: Task cancelled")
break
except Exception as e:
self._processing_stats["errors"] += 1
logger.error(f"{self.exchange_name}: Message handling error - {e}")
await self._handle_reconnect()
await asyncio.sleep(1)
finally:
if ping_task and ping_task.done():
ping_task = self._create_ping_task()
if ping_task:
ping_task.cancel()
def get_stats(self) -> dict:
"""처리 통계 반환"""
return {
"exchange": self.exchange_name,
"status": self._status.value,
"subscriptions": len(self._subscriptions),
**self._processing_stats
}
# exchange_aggregator/exchanges/binance_connector.py
import asyncio
import json
import zlib
from typing import List
from .connection_manager import BaseExchangeConnector, WebSocketMessage
class BinanceConnector(BaseExchangeConnector):
"""Binance WebSocket 커넥터"""
COMPRESSED_STREAM_URL = "wss://stream.binance.com:9443/stream"
STREAMS_URL = "wss://stream.binance.com:9443/stream"
def __init__(self, symbols: List[str], message_queue: asyncio.Queue, **kwargs):
# Binance는 심볼을 소문자로 변환
normalized = [s.lower().replace('/', '') for s in symbols]
super().__init__("binance", normalized, message_queue, **kwargs)
self._decompressor = zlib.decompressobj(31)
def get_websocket_url(self) -> str:
# Combined stream URL
streams = [f"{s}@ticker" for s in self.symbols]
stream_param = '/'.join(streams)
return f"{self.STREAMS_URL}?streams={stream_param}"
def format_subscription_message(self, symbols: List[str]) -> str:
"""Binance는 초기 URL에 스트림 포함, 별도 구독 불필요"""
return ""
def format_unsubscription_message(self, symbols: List[str]) -> str:
return json.dumps({
"method": "UNSUBSCRIBE",
"params": [f"{s}@ticker" for s in symbols],
"id": 1
})
async def parse_message(self, raw_message: str) -> List[WebSocketMessage]:
"""Binance 24hr Ticker 메시지 파싱"""
messages = []
try:
# 압축 해제 처리
try:
data = json.loads(raw_message)
except json.JSONDecodeError:
# 압축된 메시지 처리
decompressed = self._decompress(raw_message)
if decompressed:
data = json.loads(decompressed)
else:
return messages
# Combined stream 포맷: {"stream": "...", "data": {...}}
if "stream" in data and "data" in data:
stream_data = data["data"]
symbol = stream_data.get("s", "").lower()
if symbol in self.symbols:
msg = WebSocketMessage(
exchange=self.exchange_name,
symbol=symbol,
message_type="ticker",
data={
"bid_price": float(stream_data.get("b", 0)),
"ask_price": float(stream_data.get("a", 0)),
"last_price": float(stream_data.get("c", 0)),
"volume_24h": float(stream_data.get("v", 0)),
"quote_volume_24h": float(stream_data.get("q", 0)),
"bid_qty": float(stream_data.get("B", 0)),
"ask_qty": float(stream_data.get("A", 0)),
},
raw_data=raw_message
)
messages.append(msg)
# 개별 스트림 포맷
elif "e" in data:
symbol = data.get("s", "").lower()
if symbol in self.symbols:
msg = WebSocketMessage(
exchange=self.exchange_name,
symbol=symbol,
message_type=data.get("e", "unknown"),
data=data,
raw_data=raw_message
)
messages.append(msg)
except Exception as e:
import logging
logging.error(f"Binance parse error: {e}")
return messages
def _decompress(self, raw: bytes) -> str:
"""zlib 압축 해제"""
try:
return self._decompressor.decompress(raw).decode('utf-8')
except Exception:
return ""
class BybitConnector(BaseExchangeConnector):
"""Bybit WebSocket 커넥터"""
WS_URL = "wss://stream.bybit.com/v5/public/spot"
def __init__(self, symbols: List[str], message_queue: asyncio.Queue, **kwargs):
# Bybit는 심볼 형식 유지
super().__init__("bybit", symbols, message_queue, **kwargs)
def get_websocket_url(self) -> str:
return self.WS_URL
def format_subscription_message(self, symbols: List[str]) -> str:
import json
return json.dumps({
"op": "subscribe",
"args": [f"tickers.{s}" for s in symbols]
})
def format_unsubscription_message(self, symbols: List[str]) -> str:
import json
return json.dumps({
"op": "unsubscribe",
"args": [f"tickers.{s}" for s in symbols]
})
async def parse_message(self, raw_message: str) -> List[WebSocketMessage]:
import json
messages = []
try:
data = json.loads(raw_message)
# 구독 확인 응답
if data.get("op") == "subscribe":
return messages
# 티커 데이터
topic = data.get("topic", "")
if topic.startswith("tickers."):
symbol = topic.split(".")[1]
if symbol in self.symbols:
ticker_data = data.get("data", {})
msg = WebSocketMessage(
exchange=self.exchange_name,
symbol=symbol,
message_type="ticker",
data={
"bid_price": float(ticker_data.get("bp1", 0)),
"ask_price": float(ticker_data.get("ap1", 0)),
"last_price": float(ticker_data.get("lastPrice", 0)),
"volume_24h": float(ticker_data.get("volume24h", 0)),
"quote_volume_24h": float(ticker_data.get("turnover24h", 0)),
},
raw_data=raw_message
)
messages.append(msg)
except Exception as e:
import logging
logging.error(f"Bybit parse error: {e}")
return messages
class OKXConnector(BaseExchangeConnector):
"""OKX WebSocket 커넥터"""
WS_URL = "wss://ws.okx.com:8443/ws/v5/public"
def __init__(self, symbols: List[str], message_queue: asyncio.Queue, **kwargs):
# OKX는 BTC-USDT 형식
normalized = [s.replace('/', '-') for s in symbols]
super().__init__("okx", normalized, message_queue, **kwargs)
def get_websocket_url(self) -> str:
return self.WS_URL
def format_subscription_message(self, symbols: List[str]) -> str:
import json
return json.dumps({
"op": "subscribe",
"args": [{
"channel": "tickers",
"instId": s
} for s in symbols]
})
def format_unsubscription_message(self, symbols: List[str]) -> str:
import json
return json.dumps({
"op": "unsubscribe",
"args": [{
"channel": "tickers",
"instId": s
} for s in symbols]
})
async def parse_message(self, raw_message: str) -> List[WebSocketMessage]:
import json
messages = []
try:
data = json.loads(raw_message)
# 핑 응답
if data.get("event") == "pong":
self._last_pong_time = datetime.utcnow()
return messages
# 구독 확인
if data.get("event") == "subscribe":
return messages
# 데이터 메시지
if "data" in data:
for ticker_data in data["data"]:
symbol = ticker_data.get("instId", "")
if symbol in self.symbols:
msg = WebSocketMessage(
exchange=self.exchange_name,
symbol=symbol,
message_type="ticker",
data={
"bid_price": float(ticker_data.get("bidPx", 0)),
"ask_price": float(ticker_data.get("askPx", 0)),
"last_price": float(ticker_data.get("last", 0)),
"volume_24h": float(ticker_data.get("vol24h", 0)),
"quote_volume_24h": float(ticker_data.get("volCcy24h", 0)),
},
raw_data=raw_message
)
messages.append(msg)
except Exception as e:
import logging
logging.error(f"OKX parse error: {e}")
return messages
데이터 집계 및 큐 처리 레이어
# exchange_aggregator/core/aggregator.py
import asyncio
import logging
from dataclasses import dataclass, field
from datetime import datetime
from typing import Dict, List, Optional
from collections import defaultdict
import time
from .connection_manager import WebSocketMessage, UnifiedTicker
logger = logging.getLogger(__name__)
@dataclass
class AggregatorConfig:
"""집계기 설정"""
queue_maxsize: int = 10000
batch_size: int = 100
batch_timeout: float = 0.1 # 100ms
stats_interval: float = 60.0 # 통계 리포트 간격
# 심볼별 거래소 가중치 (arbitrage detection용)
exchange_priority: Dict[str, int] = field(default_factory=lambda: {
"binance": 1,
"bybit": 2,
"okx": 3,
"coinex": 4,
})
# 가격 스프레드 임계값 (basis points)
arbitrage_threshold_bps: float = 10.0 # 10bps 이상 차이가 나면 arbitrage 기회
class MarketDataAggregator:
"""다중 거래소 마켓 데이터 집계기"""
def __init__(self, config: AggregatorConfig = None):
self.config = config or AggregatorConfig()
# 심볼별 최신 티커 데이터: symbol -> {exchange -> ticker_data}
self._ticker_cache: Dict[str, Dict[str, dict]] = defaultdict(dict)
# Unified ticker 캐시
self._unified_tickers: Dict[str, UnifiedTicker] = {}
# 콜백 핸들러
self._callbacks: List[callable] = []
# 통계
self._stats = {
"total_messages": 0,
"messages_per_exchange": defaultdict(int),
"arbitrage_opportunities": 0,
"callback_latency_ms": 0,
}
# 내부 큐
self._internal_queue: asyncio.Queue = asyncio.Queue(
maxsize=self.config.queue_maxsize
)
self._running = False
self._start_time: Optional[datetime] = None
def register_callback(self, callback: callable):
"""데이터 콜백 등록"""
self._callbacks.append(callback)
async def enqueue(self, message: WebSocketMessage):
"""메시지 큐에 추가"""
try:
self._internal_queue.put_nowait(message)
self._stats["total_messages"] += 1
self._stats["messages_per_exchange"][message.exchange] += 1
except asyncio.QueueFull:
logger.warning(f"Queue full, dropping message: {message.symbol}")
async def _process_message(self, message: WebSocketMessage):
"""단일 메시지 처리"""
symbol = message.symbol
# 캐시 업데이트
self._ticker_cache[symbol][message.exchange] = {
**message.data,
"timestamp": message.timestamp,
"raw": message.raw_data,
}
# 통합 티커 생성
unified = self._compute_unified_ticker(symbol)
if unified:
self._unified_tickers[symbol] = unified
# Arbitrage 기회 감지
if unified.spread_bps >= self.config.arbitrage_threshold_bps:
self._stats["arbitrage_opportunities"] += 1
logger.info(
f"Arbitrage opportunity: {symbol} - "
f"Buy {unified.best_ask_exchange}@{unified.best_ask_price}, "
f"Sell {unified.best_bid_exchange}@{unified.best_bid_price}, "
f"Spread: {unified.spread_bps:.2f}bps"
)
def _compute_unified_ticker(self, symbol: str) -> Optional[UnifiedTicker]:
"""심볼별 최적 Bid/Ask 계산"""
exchange_tickers = self._ticker_cache.get(symbol, {})
if not exchange_tickers:
return None
# 최고 Bid 찾기 (구매 의사 표현)
best_bid_exchange = None
best_bid_price = 0.0
best_bid_qty = 0.0
# 최저 Ask 찾기 (판매 의사 표현)
best_ask_exchange = None
best_ask_price = float('inf')
best_ask_qty = 0.0
for exchange, ticker in exchange_tickers.items():
bid_price = ticker.get('bid_price', 0)
ask_price = ticker.get('ask_price', float('inf'))
if bid_price > best_bid_price:
best_bid_price = bid_price
best_bid_exchange = exchange
best_bid_qty = ticker.get('bid_qty', 0)
if ask_price < best_ask_price:
best_ask_price = ask_price
best_ask_exchange = exchange
best_ask_qty = ticker.get('ask_qty', 0)
if best_bid_exchange is None or best_ask_exchange is None:
return None
# 스프레드 계산 (bps)
if best_ask_price > 0:
spread_bps = ((best_bid_price - best_ask_price) / best_ask_price) * 10000
else:
spread_bps = 0.0
return UnifiedTicker(
symbol=symbol,
exchanges={ex: ticker for ex, ticker in exchange_tickers.items()},
best_bid_price=best_bid_price,
best_ask_price=best_ask_price,
best_bid_exchange=best_bid_exchange,
best_ask_exchange=best_ask_exchange,
spread_bps=spread_bps,
timestamp=datetime.utcnow()
)
async def _batch_processor(self):
"""배치 처리 루프"""
batch: List[WebSocketMessage] = []
last_batch_time = time.monotonic()
while self._running:
try:
# 배칭 로직
timeout = self.config.batch_timeout - (time.monotonic() - last_batch_time)
try:
if timeout > 0:
message = await asyncio.wait_for(
self._internal_queue.get(),
timeout=timeout
)
batch.append(message)
except asyncio.TimeoutError:
pass
# 배치 처리 조건
should_process = (
len(batch) >= self.config.batch_size or
(len(batch) > 0 and time.monotonic() - last_batch_time >= self.config.batch_timeout)
)
if should_process and batch:
for msg in batch:
await self._process_message(msg)
batch.clear()
last_batch_time = time.monotonic()
except Exception as e:
logger.error(f"Batch processor error: {e}")
await asyncio.sleep(0.1)
async def _callback_dispatcher(self):
"""콜백 디스패처 루프"""
while self._running:
try:
if self._unified_tickers:
start = time.monotonic()
# 모든 콜백에 unified tickers 전달
tasks = [
callback(self._unified_tickers.copy())
for callback in self._callbacks
]
if tasks:
await asyncio.gather(*tasks, return_exceptions=True)
self._stats["callback_latency_ms"] = (
(time.monotonic() - start) * 1000
)
await asyncio.sleep(0.05) # ~20Hz 콜백 주기
except Exception as e:
logger.error(f"Callback dispatcher error: {e}")
await asyncio.sleep(1)
async def _stats_reporter(self):
"""통계 리포터"""
while self._running:
await asyncio.sleep(self.config.stats_interval)
uptime = (datetime.utcnow() - self._start_time).total_seconds()
msg_rate = self._stats["total_messages"] / uptime if uptime > 0 else 0
logger.info(
f"=== Aggregator Stats (uptime: {uptime:.0f}s) ===\n"
f"Total messages: {self._stats['total_messages']}\n"
f"Msg rate: {msg_rate:.1f}/s\n"
f"Arbitrage opportunities: {self._stats['arbitrage_opportunities']}\n"
f"Callback latency: {self._stats['callback_latency_ms']:.2f}ms\n"
f"Messages per exchange: {dict(self._stats['messages_per_exchange'])}\n"
f"Queue size: {self._internal_queue.qsize()}"
)
async def start(self):
"""집계기 시작"""
self._running = True
self._start_time = datetime.utcnow()
# 백그라운드 태스크 시작
self._batch_task = asyncio.create_task(self._batch_processor())
self._callback_task = asyncio.create_task(self._callback_dispatcher())
self._stats_task = asyncio.create_task(self._stats_reporter())
logger.info("MarketDataAggregator started")
async def stop(self):
"""집계기 중지"""
self._running = False
await asyncio.gather(
self._batch_task,
self._callback_task,
self._stats_task,
return_exceptions=True
)
logger.info("MarketDataAggregator stopped")
def get_stats(self) -> dict:
"""통계 반환"""
return {
**self._stats,
"queue_size": self._internal_queue.qsize(),
"cached_symbols": len(self._ticker_cache),
"uptime_seconds": (
(datetime.utcnow() - self._start_time).total_seconds()
if self._start_time else 0
)
}
프로덕션 실행 관리자
# exchange_aggregator/runner.py
import asyncio
import logging
import signal
from typing import List, Dict
from datetime import datetime
from .core.connection_manager import BaseExchangeConnector
from .core.aggregator import MarketDataAggregator, AggregatorConfig
from .exchanges.binance_connector import BinanceConnector, BybitConnector, OKXConnector
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s | %(levelname)-8s | %(name)-20s | %(message)s'
)
logger = logging.getLogger(__name__)
class MultiExchangeRunner:
"""다중 거래소 WebSocket 구독 런너"""
def __init__(self, config: AggregatorConfig = None):
self.config = config or AggregatorConfig()
# 메시지 큐 생성
self._message_queue: asyncio.Queue = asyncio.Queue(
maxsize=config.queue_maxsize if config else 10000
)
# 커넥터 저장소
self._connectors: Dict[str, BaseExchangeConnector] = {}
# 집계기
self._aggregator = MarketDataAggregator(config)
# 실행 상태
self._running = False
self._shutdown_event = asyncio.Event()
def add_exchange(
self,
exchange_name: str,
symbols: List[str],
connector_class=None,
**connector_kwargs
):
"""거래소 커넥터 추가"""
if connector_class is None:
# 기본 커넥터 매핑
connector_map = {
"binance": BinanceConnector,
"bybit": BybitConnector,
"okx": OKXConnector,
}
connector_class = connector_map.get(exchange_name.lower())
if connector_class is None:
raise ValueError(f"Unknown exchange: {exchange_name}")
connector = connector_class(
symbols=symbols,
message_queue=self._message_queue,
**connector_kwargs
)
self._connectors[exchange_name] = connector
logger.info(
f"Added exchange: {exchange_name} with {len(symbols)} symbols"
)
def register_callback(self, callback):
"""데이터 콜백 등록"""
self._aggregator.register_callback(callback)
async def _signal_handler(self):
"""시그널 핸들링"""
loop = asyncio.get_event_loop()
for sig in (signal.SIGTERM, signal.SIGINT):
try:
loop.add_signal_handler(
sig,
lambda s=sig: asyncio.create_task(self._handle_shutdown(s))
)
except NotImplementedError:
# Windows에서는 시그널 핸들링 제한적
pass
async def _handle_shutdown(self, sig):
""" graceful shutdown 처리"""
logger.info(f"Received signal {sig.name}, initiating graceful shutdown...")
await self.shutdown()
async def _connector_manager(self):
"""연결 상태 모니터링 및 재연결 관리"""
while self._running:
await asyncio.sleep(10) # 10초마다 상태 체크
for name, connector in self._connectors.items():
stats = connector.get_stats()
# 연결 실패 시 알림
if stats["status"] == "error":
logger.error(
f"Exchange {name} in ERROR state - "
f"manual intervention may be required"
)
# 높은 에러율 감지
if stats["messages"] > 0:
error_rate = stats["errors"] / stats["messages"]
if error_rate > 0.05: # 5% 이상 에러율
logger.warning(
f"Exchange {name} error rate: {error_rate*100:.1f}%"
)
async def run(self):
"""런너 실행"""
logger.info("Starting Multi-Exchange Runner...")
self._running = True
# 시그널 핸들러 등록
await self._signal_handler()
# 집계기 시작
await self._aggregator.start()
# 커넥터 태스크 생성
connector_tasks = []
for connector in self._connectors.values():
task = asyncio.create_task(connector.run())
connector_tasks.append(task)
# 연결 관리자 태스크
manager_task = asyncio.create_task(self._connector_manager())
try:
# 모든 태스크 완료 대기 ( graceful shutdown 전까지)
await asyncio.gather(
*connector_tasks,
manager_task,
return_exceptions=True
)
except asyncio.CancelledError:
logger.info("Tasks cancelled")
finally:
await self._cleanup()