ในโลกของการเทรดคริปโตเชิงปริมาณ (Quantitative Trading) การเข้าถึงข้อมูล L2 Order Book คุณภาพสูงเป็นรากฐานสำคัญของระบบทุกตัว ไม่ว่าจะเป็น Market Making, Arbitrage Bot หรือ Alpha Research บทความนี้จะพาคุณสร้างระบบดึงข้อมูล Order Book จาก 3 Exchange ยอดนิยมอย่าง Binance, OKX และ Hyperliquid ผ่าน Tardis.dev API แบบ Unified โดยใช้ Python

ทำไมต้อง Tardis.dev?

Tardis.dev เป็น Data Aggregation Layer ที่รวม Historical Data จาก Exchange หลายตัวเข้าด้วยกันผ่าน API เดียว ข้อดีหลักคือ:

สถาปัตยกรรมระบบ

ก่อนเข้าสู่โค้ด มาดูภาพรวมสถาปัตยกรรมที่เราจะสร้างกัน:

+------------------+     +------------------+     +------------------+
|    Binance       |     |      OKX         |     |   Hyperliquid   |
|  L2 Order Book   |     |  L2 Order Book   |     |  L2 Order Book  |
+--------+---------+     +--------+---------+     +--------+---------+
         |                        |                        |
         v                        v                        v
+--------+---------+     +--------+---------+     +--------+---------+
|    Tardis.dev    |<----|    Tardis.dev    |<----|    Tardis.dev    |
|   Exchange API   |     |   Exchange API   |     |   Exchange API   |
+--------+---------+     +--------+---------+     +--------+---------+
         |                        |                        |
         +------------------------+------------------------+
                                  |
                                  v
                      +-------------------+
                      |  Python Consumer  |
                      |  - Async/Await    |
                      |  - Concurrency    |
                      |  - Buffer Queue   |
                      +-------------------+
                                  |
                                  v
                      +-------------------+
                      |  Data Processor   |
                      |  - Normalization  |
                      |  - Enrichment     |
                      |  - Storage        |
                      +-------------------+

การติดตั้งและ Setup

# ติดตั้ง dependencies
pip install tardis-dev aiohttp asyncpg redis-python pandas numpy


หรือใช้ requirements.txt


tardis-dev>=2.0.0


aiohttp>=3.9.0


pandas>=2.0.0


numpy>=1.24.0


asyncpg>=0.29.0


redis>=5.0.0


python-dotenv>=1.0.0

โครงสร้างโปรเจกต์

# project structure
tardis_unified_consumer/
├── config/
│   ├── __init__.py
│   └── settings.py
├── clients/
│   ├── __init__.py
│   ├── base_client.py
│   ├── binance_client.py
│   ├── okx_client.py
│   └── hyperliquid_client.py
├── models/
│   ├── __init__.py
│   └── orderbook.py
├── storage/
│   ├── __init__.py
│   └── postgres_writer.py
├── main.py
├── requirements.txt
└── .env

Configuration และ Environment

# config/settings.py
import os
from dataclasses import dataclass
from typing import Dict, List

@dataclass
class ExchangeConfig:
    exchange: str
    symbols: List[str]
    channels: List[str]
    api_key: str

@dataclass
class DatabaseConfig:
    host: str
    port: int
    database: str
    user: str
    password: str

@dataclass
class RedisConfig:
    host: str
    port: int
    db: int
    password: str = None


HolySheep AI - สำหรับ Real-time Analytics

@dataclass 
class LLMConfig:
    base_url: str = "https://api.holysheep.ai/v1"  # ประหยัด 85%+ vs OpenAI
    api_key: str = "YOUR_HOLYSHEEP_API_KEY"
    model: str = "gpt-4.1"  # $8/MTok vs $15 ที่อื่น
    max_tokens: int = 1000
    temperature: float = 0.3

class Settings:
    # Tardis.dev API Token
    TARDIS_API_KEY: str = os.getenv("TARDIS_API_KEY", "your_tardis_api_key")
    
    # Exchange Configurations
    EXCHANGES: Dict[str, ExchangeConfig] = {
        "binance": ExchangeConfig(
            exchange="binance",
            symbols=["BTCUSDT", "ETHUSDT", "SOLUSDT"],
            channels=["orderbook"],
            api_key=os.getenv("TARDIS_API_KEY")
        ),
        "okx": ExchangeConfig(
            exchange="okx",
            symbols=["BTC-USDT", "ETH-USDT", "SOL-USDT"],
            channels=["books"],
            api_key=os.getenv("TARDIS_API_KEY")
        ),
        "hyperliquid": ExchangeConfig(
            exchange="hyperliquid",
            symbols=["BTC", "ETH", "SOL"],
            channels=["orderbook"],
            api_key=os.getenv("TARDIS_API_KEY")
        )
    }
    
    # Database
    DATABASE: DatabaseConfig = DatabaseConfig(
        host=os.getenv("DB_HOST", "localhost"),
        port=int(os.getenv("DB_PORT", "5432")),
        database=os.getenv("DB_NAME", "orderbook_db"),
        user=os.getenv("DB_USER", "postgres"),
        password=os.getenv("DB_PASSWORD", "password")
    )
    
    # Redis for caching
    REDIS: RedisConfig = RedisConfig(
        host=os.getenv("REDIS_HOST", "localhost"),
        port=int(os.getenv("REDIS_PORT", "6379")),
        db=int(os.getenv("REDIS_DB", "0"))
    )
    
    # LLM for analytics (HolySheep AI)
    LLM: LLMConfig = LLMConfig()
    
    # Performance settings
    BUFFER_SIZE: int = 1000
    FLUSH_INTERVAL_SEC: int = 5
    MAX_CONCURRENT_STREAMS: int = 10
    
    @classmethod
    def get_tardis_url(cls, exchange: str) -> str:
        """Generate Tardis.dev WebSocket URL for exchange"""
        return f"wss://tardis.dev/v1/stream/{exchange}?token={cls.TARDIS_API_KEY}"
    
    @classmethod
    def get_tardis_rest_url(cls, exchange: str, symbol: str, channel: str) -> str:
        """Generate Tardis.dev REST API URL"""
        return f"https://tardis.dev/v1/history/{exchange}/{symbol}?channel={channel}&token={cls.TARDIS_API_KEY}"

settings = Settings()

Unified Order Book Model

# models/orderbook.py
from dataclasses import dataclass, field
from typing import Dict, List, Tuple, Optional
from datetime import datetime
from decimal import Decimal
import json

@dataclass
class PriceLevel:
    """Single price level in order book"""
    price: float
    size: float
    order_count: int = 0
    
    def to_dict(self) -> dict:
        return {
            "price": self.price,
            "size": self.size,
            "order_count": self.order_count
        }

@dataclass
class OrderBook:
    """Unified Order Book model across all exchanges"""
    exchange: str
    symbol: str
    timestamp: datetime
    local_timestamp: datetime = field(default_factory=datetime.utcnow)
    
    # Bid side (buy orders) - sorted high to low
    bids: List[PriceLevel] = field(default_factory=list)
    
    # Ask side (sell orders) - sorted low to high
    asks: List[PriceLevel] = field(default_factory=list)
    
    # Sequence number for ordering
    sequence: int = 0
    
    # Exchange-specific raw data
    raw_data: Optional[dict] = None
    
    # Derived metrics
    @property
    def best_bid(self) -> Optional[PriceLevel]:
        return self.bids[0] if self.bids else None
    
    @property
    def best_ask(self) -> Optional[PriceLevel]:
        return self.asks[0] if self.asks else None
    
    @property
    def spread(self) -> Optional[float]:
        if self.best_bid and self.best_ask:
            return self.best_ask.price - self.best_bid.price
        return None
    
    @property
    def spread_bps(self) -> Optional[float]:
        """Spread in basis points"""
        if self.spread and self.mid_price:
            return (self.spread / self.mid_price) * 10000
        return None
    
    @property
    def mid_price(self) -> Optional[float]:
        if self.best_bid and self.best_ask:
            return (self.best_bid.price + self.best_ask.price) / 2
        return None
    
    @property
    def bid_depth(self) -> float:
        """Total bid size"""
        return sum(level.size for level in self.bids)
    
    @property
    def ask_depth(self) -> float:
        """Total ask size"""
        return sum(level.size for level in self.asks)
    
    @property
    def imbalance(self) -> Optional[float]:
        """Order book imbalance: (bid - ask) / (bid + ask)"""
        total = self.bid_depth + self.ask_depth
        if total > 0:
            return (self.bid_depth - self.ask_depth) / total
        return None
    
    def to_dict(self) -> dict:
        return {
            "exchange": self.exchange,
            "symbol": self.symbol,
            "timestamp": self.timestamp.isoformat(),
            "local_timestamp": self.local_timestamp.isoformat(),
            "bids": [level.to_dict() for level in self.bids],
            "asks": [level.to_dict() for level in self.asks],
            "best_bid": self.best_bid.to_dict() if self.best_bid else None,
            "best_ask": self.best_ask.to_dict() if self.best_ask else None,
            "spread": self.spread,
            "spread_bps": self.spread_bps,
            "mid_price": self.mid_price,
            "bid_depth": self.bid_depth,
            "ask_depth": self.ask_depth,
            "imbalance": self.imbalance,
            "sequence": self.sequence
        }
    
    def to_sql_tuple(self) -> Tuple:
        return (
            self.exchange,
            self.symbol,
            self.timestamp,
            self.local_timestamp,
            json.dumps(self.bids[:10]),  # Top 10 levels
            json.dumps(self.asks[:10]),
            self.best_bid.price if self.best_bid else None,
            self.best_bid.size if self.best_bid else None,
            self.best_ask.price if self.best_ask else None,
            self.best_ask.size if self.best_ask else None,
            self.spread,
            self.spread_bps,
            self.mid_price,
            self.bid_depth,
            self.ask_depth,
            self.imbalance,
            self.sequence
        )
    
    @staticmethod
    def normalize_symbol(exchange: str, symbol: str) -> str:
        """Normalize symbol format across exchanges"""
        # Remove common separators and convert to uppercase
        normalized = symbol.replace("-", "").replace("_", "").replace("/", "").upper()
        
        # Exchange-specific mappings
        if exchange == "binance":
            if normalized.endswith("USDT"):
                return normalized
        elif exchange == "okx":
            if normalized.endswith("USDT"):
                return normalized.replace("USDT", "USDT")  # OKX uses BTC-USDT format
        elif exchange == "hyperliquid":
            # Hyperliquid uses BTC, ETH without suffix for perpetuals
            return normalized
        
        return normalized


class OrderBookNormalizer:
    """Normalize order book data from different exchanges to unified format"""
    
    @staticmethod
    def normalize_binance(data: dict) -> OrderBook:
        """Normalize Binance order book data"""
        # Binance timestamp is in milliseconds
        ts = datetime.utcfromtimestamp(data.get("timestamp", 0) / 1000)
        
        bids = [
            PriceLevel(price=float(bid[0]), size=float(bid[1]), order_count=int(bid[2]) if len(bid) > 2 else 0)
            for bid in data.get("bids", [])[:20]
        ]
        asks = [
            PriceLevel(price=float(ask[0]), size=float(ask[1]), order_count=int(ask[2]) if len(ask) > 2 else 0)
            for ask in data.get("asks", [])[:20]
        ]
        
        return OrderBook(
            exchange="binance",
            symbol=data.get("symbol", "").upper(),
            timestamp=ts,
            bids=bids,
            asks=asks,
            sequence=data.get("sequenceId", 0),
            raw_data=data
        )
    
    @staticmethod
    def normalize_okx(data: dict) -> OrderBook:
        """Normalize OKX order book data"""
        # OKX timestamp format: "2024-01-15T10:30:00.000Z"
        ts = datetime.fromisoformat(data.get("timestamp", "2024-01-01T00:00:00.000Z").replace("Z", "+00:00"))
        
        bids = [
            PriceLevel(price=float(bid[0]), size=float(bid[1]), order_count=int(bid[2]) if len(bid) > 2 else 0)
            for bid in data.get("bids", [])[:20]
        ]
        asks = [
            PriceLevel(price=float(ask[0]), size=float(ask[1]), order_count=int(ask[2]) if len(ask) > 2 else 0)
            for ask in data.get("asks", [])[:20]
        ]
        
        return OrderBook(
            exchange="okx",
            symbol=data.get("symbol", "").replace("-", "").upper(),
            timestamp=ts,
            bids=bids,
            asks=asks,
            sequence=data.get("sequenceId", 0),
            raw_data=data
        )
    
    @staticmethod
    def normalize_hyperliquid(data: dict) -> OrderBook:
        """Normalize Hyperliquid order book data"""
        # Hyperliquid uses different format
        ts = datetime.utcnow()
        
        # Extract levels from Hyperliquid format
        book = data.get("data", {}).get("orderbook", {})
        
        bids = [
            PriceLevel(price=float(bid[0]), size=float(bid[1]))
            for bid in book.get("bids", [])[:20]
        ]
        asks = [
            PriceLevel(price=float(ask[0]), size=float(ask[1]))
            for ask in book.get("asks", [])[:20]
        ]
        
        return OrderBook(
            exchange="hyperliquid",
            symbol=data.get("data", {}).get("symbol", "").upper(),
            timestamp=ts,
            bids=bids,
            asks=asks,
            sequence=data.get("data", {}).get("seqNum", 0),
            raw_data=data
        )

Base Client และ Async Architecture

# clients/base_client.py
import asyncio
import json
import logging
from abc import ABC, abstractmethod
from typing import Dict, List, Callable, Optional, Any
from datetime import datetime
import aiohttp
from dataclasses import dataclass, field
from collections import deque
import signal
import sys

from models.orderbook import OrderBook, OrderBookNormalizer
from config.settings import settings

logger = logging.getLogger(__name__)

@dataclass
class StreamStats:
    """Statistics for a stream"""
    messages_received: int = 0
    messages_processed: int = 0
    errors: int = 0
    last_message_time: Optional[datetime] = None
    latency_ms: deque = field(default_factory=lambda: deque(maxlen=1000))
    
    @property
    def avg_latency_ms(self) -> float:
        return sum(self.latency_ms) / len(self.latency_ms) if self.latency_ms else 0
    
    @property
    def p99_latency_ms(self) -> float:
        if not self.latency_ms:
            return 0
        sorted_latencies = sorted(self.latency_ms)
        idx = int(len(sorted_latencies) * 0.99)
        return sorted_latencies[idx]

class BaseTardisClient(ABC):
    """Base class for Tardis.dev unified stream consumer"""
    
    def __init__(
        self,
        exchange: str,
        symbols: List[str],
        channels: List[str],
        on_message: Optional[Callable[[OrderBook], None]] = None,
        buffer_size: int = 1000,
        flush_interval: int = 5
    ):
        self.exchange = exchange
        self.symbols = symbols
        self.channels = channels
        self.on_message = on_message
        self.buffer_size = buffer_size
        self.flush_interval = flush_interval
        
        self.websocket_url = settings.get_tardis_url(exchange)
        self.ws: Optional[aiohttp.ClientWebSocketResponse] = None
        self.session: Optional[aiohttp.ClientSession] = None
        
        self.buffer: deque = deque(maxlen=buffer_size)
        self.stats = StreamStats()
        
        self._running = False
        self._subscribed = False
        
        # Setup graceful shutdown
        signal.signal(signal.SIGINT, self._signal_handler)
        signal.signal(signal.SIGTERM, self._signal_handler)
    
    def _signal_handler(self, signum, frame):
        """Handle shutdown signals gracefully"""
        logger.info(f"Received signal {signum}, initiating graceful shutdown...")
        self._running = False
    
    async def connect(self) -> bool:
        """Establish WebSocket connection to Tardis.dev"""
        try:
            self.session = aiohttp.ClientSession()
            
            # Build subscription message
            subscribe_msg = {
                "type": "subscribe",
                "exchange": self.exchange,
                "symbols": self.symbols,
                "channels": self.channels
            }
            
            # Connect with headers
            headers = {
                "Authorization": f"Bearer {settings.TARDIS_API_KEY}"
            }
            
            logger.info(f"Connecting to {self.websocket_url}")
            self.ws = await self.session.ws_connect(
                self.websocket_url,
                headers=headers,
                timeout=aiohttp.ClientTimeout(total=30)
            )
            
            # Send subscription
            await self.ws.send_json(subscribe_msg)
            logger.info(f"Subscribed to {self.exchange}: {self.symbols}")
            
            self._subscribed = True
            return True
            
        except aiohttp.ClientError as e:
            logger.error(f"Connection failed: {e}")
            return False
        except Exception as e:
            logger.error(f"Unexpected error during connection: {e}")
            return False
    
    async def disconnect(self):
        """Close WebSocket connection"""
        self._running = False
        
        if self.ws:
            await self.ws.close()
            logger.info(f"Disconnected from {self.exchange}")
        
        if self.session:
            await self.session.close()
    
    async def consume(self):
        """Main consume loop"""
        self._running = True
        
        while self._running:
            try:
                if not self.ws or self.ws.closed:
                    connected = await self.connect()
                    if not connected:
                        await asyncio.sleep(5)  # Retry after 5 seconds
                        continue
                
                msg = await self.ws.receive()
                
                if msg.type == aiohttp.WSMsgType.TEXT:
                    await self._handle_message(msg.data)
                elif msg.type == aiohttp.WSMsgType.ERROR:
                    logger.error(f"WebSocket error: {msg.data}")
                    self.stats.errors += 1
                elif msg.type == aiohttp.WSMsgType.CLOSED:
                    logger.warning("WebSocket closed by server")
                    break
                    
            except aiohttp.ClientError as e:
                logger.error(f"Client error: {e}")
                self.stats.errors += 1
                await asyncio.sleep(1)
            except Exception as e:
                logger.error(f"Error in consume loop: {e}")
                self.stats.errors += 1
    
    async def _handle_message(self, data: str):
        """Process incoming message"""
        try:
            self.stats.messages_received += 1
            
            # Parse JSON
            parsed = json.loads(data)
            
            # Calculate latency
            if "timestamp" in parsed:
                msg_time = datetime.fromisoformat(
                    parsed["timestamp"].replace("Z", "+00:00")
                ) if isinstance(parsed["timestamp"], str) else datetime.utcfromtimestamp(parsed["timestamp"] / 1000)
                
                latency = (datetime.utcnow() - msg_time).total_seconds() * 1000
                self.stats.latency_ms.append(latency)
            
            # Normalize order book
            orderbook = self._normalize_message(parsed)
            
            if orderbook:
                self.stats.messages_processed += 1
                self.stats.last_message_time = datetime.utcnow()
                
                # Add to buffer
                self.buffer.append(orderbook)
                
                # Call callback
                if self.on_message:
                    await self._safe_callback(orderbook)
                
                # Auto flush if buffer full
                if len(self.buffer) >= self.buffer_size:
                    await self._flush_buffer()
        
        except json.JSONDecodeError as e:
            logger.warning(f"Invalid JSON: {e}")
        except Exception as e:
            logger.error(f"Error processing message: {e}")
            self.stats.errors += 1
    
    async def _safe_callback(self, orderbook: OrderBook):
        """Safely execute callback"""
        try:
            if asyncio.iscoroutinefunction(self.on_message):
                await self.on_message(orderbook)
            else:
                self.on_message(orderbook)
        except Exception as e:
            logger.error(f"Callback error: {e}")
    
    @abstractmethod
    def _normalize_message(self, data: dict) -> Optional[OrderBook]:
        """Normalize exchange-specific message to OrderBook"""
        pass
    
    async def _flush_buffer(self):
        """Flush buffer to storage"""
        if self.buffer:
            logger.info(f"Flushing {len(self.buffer)} messages")
            self.buffer.clear()
    
    def get_stats(self) -> dict:
        """Get stream statistics"""
        return {
            "exchange": self.exchange,
            "messages_received": self.stats.messages_received,
            "messages_processed": self.stats.messages_processed,
            "errors": self.stats.errors,
            "last_message_time": self.stats.last_message_time.isoformat() if self.stats.last_message_time else None,
            "avg_latency_ms": round(self.stats.avg_latency_ms, 2),
            "p99_latency_ms": round(self.stats.p99_latency_ms, 2),
            "buffer_size": len(self.buffer)
        }


class BinanceClient(BaseTardisClient):
    """Binance-specific Tardis.dev client"""
    
    def __init__(self, symbols: List[str], **kwargs):
        super().__init__(exchange="binance", symbols=symbols, channels=["orderbook"], **kwargs)
    
    def _normalize_message(self, data: dict) -> Optional[OrderBook]:
        if data.get("channel") == "orderbook":
            return OrderBookNormalizer.normalize_binance(data)
        return None


class OKXClient(BaseTardisClient):
    """OKX-specific Tardis.dev client"""
    
    def __init__(self, symbols: List[str], **kwargs):
        super().__init__(exchange="okx", symbols=symbols, channels=["books"], **kwargs)
    
    def _normalize_message(self, data: dict) -> Optional[OrderBook]:
        if data.get("channel") == "books" or data.get("channel") == "books-l2-tbt":
            return OrderBookNormalizer.normalize_okx(data)
        return None


class HyperliquidClient(BaseTardisClient):
    """Hyperliquid-specific Tardis.dev client"""
    
    def __init__(self, symbols: List[str], **kwargs):
        super().__init__(exchange="hyperliquid", symbols=symbols, channels=["orderbook"], **kwargs)
    
    def _normalize_message(self, data: dict) -> Optional[OrderBook]:
        if data.get("type") == "orderbook":
            return OrderBookNormalizer.normalize_hyperliquid(data)
        return None

Unified Consumer พร้อม Concurrency

# clients/unified_consumer.py
import asyncio
import logging
from typing import List, Dict, Optional, Callable
from datetime import datetime
import json
from collections import defaultdict

from clients.base_client import BaseTardisClient, BinanceClient, OKXClient, HyperliquidClient
from clients.binance_client import BinanceClient
from clients.okx_client import OKXClient
from clients.hyperliquid_client import HyperliquidClient
from models.orderbook import OrderBook
from config.settings import settings

logger = logging.getLogger(__name__)

class UnifiedOrderBookConsumer:
    """
    Unified consumer for multiple exchange order books.
    Handles concurrent connections with proper resource management.
    """
    
    def __init__(
        self,
        exchanges: Dict[str, List[str]],
        on_orderbook: Optional[Callable[[OrderBook], None]] = None,
        on_error: Optional[Callable[[str, Exception], None]] = None,
        max_concurrent: int = 10
    ):
        """
        Initialize unified consumer.
        
        Args:
            exchanges: Dict mapping exchange names to list of symbols
                      {"binance": ["BTCUSDT", "ETHUSDT"], "okx": ["BTC-USDT"]}
            on_orderbook: Callback for each order book update
            on_error: Callback for errors
            max_concurrent: Max concurrent stream connections
        """
        self.exchanges = exchanges
        self.on_orderbook = on_orderbook
        self.on_error = on_error
        self.max_concurrent = max_concurrent
        
        self.clients: Dict[str, BaseTardisClient] = {}
        self.tasks: List[asyncio.Task] = []
        self._running = False
        
        # Aggregate statistics
        self.aggregate_stats = {
            "total_messages": 0,
            "total_errors": 0,
            "by_exchange": defaultdict(lambda: {"messages": 0, "errors": 0})
        }
        
        self._init_clients()
    
    def _init_clients(self):
        """Initialize clients for each exchange"""
        exchange_configs = settings.EXCHANGES
        
        for exchange_name, symbols in self.exchanges.items():
            if exchange_name not in exchange_configs:
                logger.warning(f"Unknown exchange: {exchange_name}")
                continue
            
            config = exchange_configs[exchange_name]
            
            if exchange_name == "binance":
                client = BinanceClient(
                    symbols=symbols,
                    on_message=self._wrap_callback(exchange_name),
                    buffer_size=settings.BUFFER_SIZE
                )
            elif exchange_name == "okx":
                client = OKXClient(
                    symbols=symbols,
                    on_message=self._wrap_callback(exchange_name),
                    buffer_size=settings.BUFFER_SIZE
                )
            elif exchange_name == "hyperliquid":
                client = HyperliquidClient(
                    symbols=symbols,
                    on_message=self._wrap_callback(exchange_name),
                    buffer_size=settings.BUFFER_SIZE
                )
            else:
                continue
            
            self.clients[exchange_name] = client
            logger.info(f"Initialized client for {exchange_name} with symbols: {symbols}")
    
    def _wrap_callback(self, exchange: str) -> Callable:
        """Wrap callback with error handling and stats tracking"""
        async def wrapped_callback(orderbook: OrderBook):
            try:
                self.aggregate_stats["total_messages"] += 1
                self.aggregate_stats["by_exchange"][exchange]["messages"] += 1
                
                if self.on_orderbook:
                    await self._safe_callback(self.on_orderbook, orderbook)
                    
            except Exception as e:
                logger.error(f"Callback error for {exchange}: {e}")
                self.aggregate_stats["total_errors"] += 1
                self.aggregate_stats["by_exchange"][exchange]["errors"] += 1
                
                if self.on_error:
                    self.on_error(exchange, e)
        
        return wrapped_callback
    
    async def _safe_callback(self, callback: Callable, *args, **kwargs):
        """Safely execute callback"""
        try:
            if asyncio.iscoroutinefunction(callback):
                await callback(*args, **kwargs)
            else:
                callback(*args, **kwargs)
        except Exception as e:
            logger.error(f"Safe callback error: {e}")
    
    async def start(self):
        """Start all client streams concurrently"""
        self._running = True
        
        # Use semaphore to limit concurrent connections
        semaphore = asyncio.Semaphore(self.max_concurrent)
        
        async def bounded_consume(client: BaseTardisClient, exchange: str):
            async with semaphore:
                logger.info(f"Starting consumer for {exchange}")
                try:
                    await client.consume()
                except asyncio.CancelledError:
                    logger.info(f"Consumer for {exchange} cancelled")
                except Exception as e:
                    logger.error(f"Consumer error for {exchange}: {e}")
                    if self.on_error:
                        self.on_error(exchange, e)
                finally:
                    await client.disconnect()
        
        # Create tasks for all clients
        for exchange, client in self.clients.items():
            task = asyncio.create_task(bounded_consume(client, exchange))
            self.tasks.append(task)
        
        # Wait for all tasks
        await asyncio.gather(*self.tasks, return_exceptions=True)
    
    async def stop(self):
        """Stop all streams gracefully"""
        logger.info("Stopping unified consumer...")
        self._running = False
        
        # Cancel all tasks
        for task in self.tasks:
            if not task.done():
                task.cancel()
        
        # Wait for tasks to finish
        await asyncio.gather(*self.tasks, return_exceptions=True)
        
        # Disconnect all clients
        for exchange, client in self.clients.items():
            await client.disconnect()
        
        logger.info("All consumers stopped")
    
    def get_all_stats(self) -> dict:
        """Get statistics from all clients"""
        return {
            "aggregate": dict(self.aggregate_stats),
            "clients": {
                exchange: client.get_stats()
                for exchange, client in self.clients.items()
            }
        }


class CrossExchangeArbitrageDetector:
    """
    Detect arbitrage opportunities across exchanges using
    unified order book data.
    """
    
    def __init__(self, min_spread_bps: float = 5.0, min_size: float = 0.1):
        self.min_spread_bps = min_spread_bps
        self.min_size = min_size
        
        # Store latest order books by symbol
        self.order_books: Dict[str, Dict[str, OrderBook]] = defaultdict(dict)
    
    async def on_orderbook(self, orderbook: OrderBook):
        """Process incoming order book"""
        key = f"{orderbook.exchange}:{orderbook.symbol}"
        self.order_books[orderbook.symbol][orderbook.exchange] = orderbook
        
        # Check for arbitrage
        await self._check_arbitrage(orderbook.symbol)
    
    async def _check_arbitrage(self, symbol: str):
        """Check for cross-exchange arbitrage opportunities"""
        if symbol not in self.order_books:
            return
        
        books = self.order_books[symbol]
        if len(books) < 2:
            return
        
        opportunities = []
        
        exchanges = list(books.keys())
        for i, buy_exchange in enumerate(exchanges):
            for sell_exchange in exchanges[i+1:]:

แหล่งข้อมูลที่เกี่ยวข้อง
บทความที่เกี่ยวข้อง