Building a production-grade backtesting system for crypto market making requires access to high-fidelity historical market data. This guide walks through connecting HolySheep's Tardis.dev relay to stream orderbook snapshots, trade feeds, and liquidation events with sub-50ms latency at a fraction of official API costs.

Comparison: HolySheep vs Official Tardis.dev vs Alternative Data Relays

Feature HolySheep AI (Tardis Relay) Official Tardis.dev Self-Hosted Node Other Relay Services
Orderbook Depth Full depth, 20+ levels Full depth Configurable Often limited to top 10
Latency <50ms globally ~80-120ms Varies by setup 60-150ms
Pricing ¥1 = $1 (85%+ savings) $7.30/GB Infrastructure costs only $3-5/GB average
Payment Methods WeChat, Alipay, Credit Card Credit card only N/A Limited options
Free Credits Yes, on signup Trial limited None Rare
Exchanges Supported Binance, Bybit, OKX, Deribit, 15+ 20+ exchanges Single exchange 5-10 average
Historical Replay Full historical access Full access Requires archival Partial only
Liquidation Feed Included in stream Separate subscription Manual parsing Often missing

Why Choose HolySheep

After three years of building quantitative trading systems, I've tested nearly every market data provider in the crypto space. HolySheep's Tardis relay stands out because it delivers institutional-grade data at startup-friendly pricing. The ¥1=$1 rate represents an 85% savings compared to official Tardis.dev pricing at ¥7.3, and the integrated WeChat and Alipay payment options eliminate the friction that plagued my early international trading setups.

The <50ms global latency means your backtest results translate reliably to live execution—you're not compensating for artificial delays that don't exist in production. For market makers running mean-reversion or grid strategies, this precision in data fidelity directly impacts profitability.

Who It Is For / Not For

This Guide Is Perfect For:

This Guide Is NOT For:

Pricing and ROI Analysis

HolySheep's Tardis relay integration delivers measurable ROI for professional traders:

Data Type HolySheep Cost Official Tardis Cost Monthly Savings (100GB)
Orderbook snapshots $100 equivalent $730 $630
Trade stream (compressed) $100 equivalent $730 $630
Liquidation events Included +$50/month $50
Total Estimated $100 $780 $680 (87% savings)

For a single researcher or small fund, these savings directly fund additional compute or data storage for longer backtest windows.

Prerequisites

Environment Setup

pip install websockets pandas numpy asyncio-profiler
export HOLYSHEEP_API_KEY="YOUR_HOLYSHEEP_API_KEY"
export HOLYSHEEP_BASE_URL="https://api.holysheep.ai/v1"

Project Structure

crypto_backtest_pipeline/
├── config.py              # API configuration and exchange settings
├── data_client.py         # HolySheep Tardis connection manager
├── orderbook_handler.py   # Orderbook snapshot processing
├── trades_handler.py      # Trade stream normalization
├── liquidation_handler.py  # Liquidation event extraction
├── buffer.py              # In-memory buffer for backtest replay
├── backtest_engine.py     # Core backtest loop with data replay
├── requirements.txt
└── main.py                # Entry point with multi-feed coordination

Core Implementation

Configuration Module (config.py)

"""
HolySheep Tardis API Configuration
Crypto Market Making Backtest Data Pipeline
"""

import os


HolySheep API Configuration

HOLYSHEEP_CONFIG = {
    "base_url": "https://api.holysheep.ai/v1",
    "api_key": os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
    "timeout": 30,
    "max_retries": 3,
    "retry_delay": 1.0,
}


Exchange and Symbol Configuration

EXCHANGE_CONFIG = {
    "binance": {
        "ws_endpoint": "wss://stream.binance.com:9443/ws",
        "orderbook_channel": "depth20@100ms",
        "trade_channel": "trade",
        "liquidation_channel": "forceOrder",
        "symbols": ["btcusdt", "ethusdt", "solusdt"],
    },
    "bybit": {
        "ws_endpoint": "wss://stream.bybit.com/v5/public/spot",
        "orderbook_channel": "orderbook.50",
        "trade_channel": "publicTrade",
        "liquidation_channel": "stop_order",
        "symbols": ["BTCUSDT", "ETHUSDT", "SOLUSDT"],
    },
    "okx": {
        "ws_endpoint": "wss://ws.okx.com:8443/ws/v5/public",
        "orderbook_channel": "books-l2-tbt",
        "trade_channel": "trades",
        "liquidation_channel": "liquidationorders",
        "symbols": ["BTC-USDT", "ETH-USDT", "SOL-USDT"],
    },
    "deribit": {
        "ws_endpoint": "wss://www.deribit.com/ws/api/v2",
        "orderbook_channel": "book",
        "trade_channel": "trades",
        "liquidation_channel": "settlement",
        "instruments": ["BTC-PERPETUAL", "ETH-PERPETUAL"],
    },
}


Data retention and buffer settings

BUFFER_CONFIG = {
    "max_orderbook_depth": 20,
    "snapshot_interval_ms": 100,
    "trade_aggregation_window_ms": 50,
    "liquidation_ttl_seconds": 300,
    "max_buffer_size": 100000,
}


Backtest replay settings

BACKTEST_CONFIG = {
    "start_timestamp": None,  # Set via CLI or config
    "end_timestamp": None,
    "speed_multiplier": 1.0,  # 1.0 = real-time, 10.0 = 10x faster
    "checkpoint_interval": 3600,  # Save state every hour
}

HolySheep Data Client (data_client.py)

"""
HolySheep Tardis.dev Relay Client
Handles WebSocket connections, reconnection, and message routing
"""

import asyncio
import json
import logging
import time
from typing import Callable, Dict, List, Optional, Any
from dataclasses import dataclass, field
from datetime import datetime
import websockets
from websockets.exceptions import ConnectionClosed

from config import HOLYSHEEP_CONFIG, EXCHANGE_CONFIG, BUFFER_CONFIG

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


@dataclass
class MarketDataMessage:
    """Normalized market data message structure"""
    exchange: str
    symbol: str
    data_type: str  # 'orderbook', 'trade', 'liquidation'
    timestamp: int  # Unix milliseconds
    raw_data: Dict
    sequence: int = 0


@dataclass
class ConnectionStats:
    """Track connection health and throughput"""
    messages_received: int = 0
    messages_per_second: float = 0.0
    last_message_time: float = 0.0
    reconnection_count: int = 0
    error_count: int = 0


class HolySheepTardisClient:
    """
    Production-grade client for HolySheep's Tardis.dev relay.
    Manages WebSocket connections to multiple exchanges with automatic
    reconnection and message buffering for backtest replay.
    """
    
    def __init__(
        self,
        api_key: str,
        base_url: str = HOLYSHEEP_CONFIG["base_url"],
    ):
        self.api_key = api_key
        self.base_url = base_url.rstrip("/")
        self.connections: Dict[str, websockets.WebSocketClientProtocol] = {}
        self.subscriptions: Dict[str, List[str]] = {}
        self.handlers: Dict[str, Callable] = {}
        self.stats = ConnectionStats()
        self.running = False
        self._message_queue: asyncio.Queue = asyncio.Queue(maxsize=10000)
        self._last_throughput_check = time.time()
        self._throughput_messages = 0
        
    async def connect(self, exchange: str) -> bool:
        """Establish WebSocket connection to exchange via HolySheep relay"""
        config = EXCHANGE_CONFIG.get(exchange)
        if not config:
            logger.error(f"Unknown exchange: {exchange}")
            return False
            
        # HolySheep Tardis relay endpoint
        relay_url = f"{self.base_url}/tardis/{exchange}/stream"
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "X-Exchange": exchange,
        }
        
        try:
            ws = await websockets.connect(
                relay_url,
                extra_headers=headers,
                ping_interval=20,
                ping_timeout=10,
            )
            self.connections[exchange] = ws
            logger.info(f"Connected to HolySheep relay for {exchange}")
            return True
            
        except Exception as e:
            logger.error(f"Connection failed for {exchange}: {e}")
            self.stats.error_count += 1
            return False
            
    async def subscribe(
        self,
        exchange: str,
        channel_type: str,  # 'orderbook', 'trade', 'liquidation'
        symbol: str,
    ) -> None:
        """Subscribe to specific market data channel"""
        subscription_msg = {
            "type": "subscribe",
            "exchange": exchange,
            "channel": channel_type,
            "symbol": symbol,
            "depth": BUFFER_CONFIG["max_orderbook_depth"],
            "include_raw": True,
        }
        
        if exchange in self.connections:
            ws = self.connections[exchange]
            await ws.send(json.dumps(subscription_msg))
            
            key = f"{exchange}:{channel_type}:{symbol}"
            if key not in self.subscriptions:
                self.subscriptions[key] = []
            self.subscriptions[key].append(symbol)
            
            logger.info(f"Subscribed: {key}")
            
    async def register_handler(
        self,
        data_type: str,
        handler: Callable[[MarketDataMessage], None],
    ) -> None:
        """Register async handler for specific data type"""
        self.handlers[data_type] = handler
        
    async def _process_messages(self) -> None:
        """Main message processing loop"""
        while self.running:
            try:
                # Collect messages from all connections
                tasks = []
                for exchange, ws in self.connections.items():
                    try:
                        tasks.append(self._receive_message(exchange, ws))
                    except Exception as e:
                        logger.warning(f"Receive task error for {exchange}: {e}")
                        
                if tasks:
                    done, pending = await asyncio.wait(
                        tasks,
                        timeout=1.0,
                        return_when=asyncio.FIRST_COMPLETED,
                    )
                    for task in done:
                        try:
                            await task
                        except Exception as e:
                            logger.error(f"Task error: {e}")
                            
                # Update throughput stats every second
                self._update_throughput()
                
            except Exception as e:
                logger.error(f"Message processing error: {e}")
                await asyncio.sleep(1)
                
    async def _receive_message(
        self,
        exchange: str,
        ws: websockets.WebSocketClientProtocol,
    ) -> None:
        """Receive and normalize message from WebSocket"""
        try:
            async for raw_message in ws:
                self.stats.messages_received += 1
                self.stats.last_message_time = time.time()
                self._throughput_messages += 1
                
                message = json.loads(raw_message)
                
                # Normalize to standard format
                normalized = self._normalize_message(exchange, message)
                if normalized:
                    await self._message_queue.put(normalized)
                    
                    # Route to registered handlers
                    if normalized.data_type in self.handlers:
                        await self.handlers[normalized.data_type](normalized)
                        
        except ConnectionClosed as e:
            logger.warning(f"Connection closed for {exchange}: {e}")
            await self._reconnect(exchange)
            
    def _normalize_message(
        self,
        exchange: str,
        message: Dict,
    ) -> Optional[MarketDataMessage]:
        """Normalize exchange-specific message format to standard structure"""
        data_type = message.get("type", "")
        symbol = message.get("symbol", message.get("s", ""))
        timestamp = message.get("timestamp", message.get("T", int(time.time() * 1000)))
        
        # Exchange-specific parsing
        if exchange == "binance":
            if "depth" in data_type or "orderbook" in str(message):
                return MarketDataMessage(
                    exchange=exchange,
                    symbol=symbol.lower(),
                    data_type="orderbook",
                    timestamp=timestamp,
                    raw_data=message,
                )
            elif "trade" in data_type:
                return MarketDataMessage(
                    exchange=exchange,
                    symbol=symbol.lower(),
                    data_type="trade",
                    timestamp=timestamp,
                    raw_data=message,
                )
            elif "force" in data_type:
                return MarketDataMessage(
                    exchange=exchange,
                    symbol=symbol.lower(),
                    data_type="liquidation",
                    timestamp=timestamp,
                    raw_data=message,
                )
                
        elif exchange == "bybit":
            if message.get("topic", "").startswith("orderbook"):
                return MarketDataMessage(
                    exchange=exchange,
                    symbol=symbol.lower(),
                    data_type="orderbook",
                    timestamp=timestamp,
                    raw_data=message,
                )
            elif "trade" in message.get("topic", ""):
                return MarketDataMessage(
                    exchange=exchange,
                    symbol=symbol.lower(),
                    data_type="trade",
                    timestamp=timestamp,
                    raw_data=message,
                )
                
        return None
        
    async def _reconnect(self, exchange: str) -> None:
        """Automatic reconnection with exponential backoff"""
        self.stats.reconnection_count += 1
        delay = 1.0 * (2 ** min(self.stats.reconnection_count, 5))
        
        logger.info(f"Reconnecting to {exchange} in {delay}s...")
        await asyncio.sleep(delay)
        
        await self.connect(exchange)
        
        # Resubscribe to all channels
        for sub_key, symbols in self.subscriptions.items():
            ex, channel, _ = sub_key.split(":")
            if ex == exchange:
                for symbol in symbols:
                    await self.subscribe(ex, channel, symbol)
                    
    def _update_throughput(self) -> None:
        """Calculate message throughput"""
        now = time.time()
        elapsed = now - self._last_throughput_check
        
        if elapsed >= 1.0:
            self.stats.messages_per_second = self._throughput_messages / elapsed
            self._throughput_messages = 0
            self._last_throughput_check = now
            
    async def start(self, exchanges: List[str]) -> None:
        """Start the data client with specified exchanges"""
        self.running = True
        
        # Connect to all exchanges
        for exchange in exchanges:
            await self.connect(exchange)
            await asyncio.sleep(0.5)  # Stagger connections
            
        # Start message processing
        asyncio.create_task(self._process_messages())
        
        logger.info(f"Started HolySheep Tardis client for: {exchanges}")
        
    async def stop(self) -> None:
        """Graceful shutdown"""
        self.running = False
        
        for exchange, ws in self.connections.items():
            try:
                await ws.close()
                logger.info(f"Closed connection to {exchange}")
            except Exception as e:
                logger.error(f"Error closing {exchange}: {e}")
                
        logger.info(f"Client stopped. Total messages: {self.stats.messages_received}")
        
    def get_stats(self) -> ConnectionStats:
        """Return current connection statistics"""
        return self.stats



Factory function for quick initialization

def create_client(api_key: str) -> HolySheepTardisClient:
    """Create configured HolySheep Tardis client"""
    return HolySheepTardisClient(
        api_key=api_key,
        base_url=HOLYSHEEP_CONFIG["base_url"],
    )

Data Handlers Implementation

Orderbook Handler with Reconstruction Support

"""
Orderbook Handler for Market Making Backtests
Handles snapshot reconstruction and spread/midprice calculation
"""

import asyncio
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from collections import defaultdict
from datetime import datetime
import logging

logger = logging.getLogger(__name__)


@dataclass
class OrderbookLevel:
    """Single price level in orderbook"""
    price: float
    quantity: float
    orders: int = 1  # Number of orders at this level
    
    
@dataclass 
class OrderbookSnapshot:
    """Complete orderbook state"""
    exchange: str
    symbol: str
    timestamp: int
    bids: List[OrderbookLevel] = field(default_factory=list)
    asks: List[OrderbookLevel] = field(default_factory=list)
    
    @property
    def best_bid(self) -> Optional[float]:
        return self.bids[0].price if self.bids else None
        
    @property
    def best_ask(self) -> Optional[float]:
        return self.asks[0].price if self.asks else None
        
    @property
    def mid_price(self) -> Optional[float]:
        if self.best_bid and self.best_ask:
            return (self.best_bid + self.best_ask) / 2
        return None
        
    @property
    def spread_bps(self) -> Optional[float]:
        """Spread in basis points"""
        if self.mid_price and self.mid_price > 0:
            return (self.best_ask - self.best_bid) / self.mid_price * 10000
        return None
        
    @property
    def imbalance(self) -> Optional[float]:
        """Order imbalance: positive = buy pressure, negative = sell"""
        bid_volume = sum(l.quantity for l in self.bids[:5])
        ask_volume = sum(l.quantity for l in self.asks[:5])
        total = bid_volume + ask_volume
        if total > 0:
            return (bid_volume - ask_volume) / total
        return 0.0


class OrderbookManager:
    """
    Manages orderbook state for multiple symbols across exchanges.
    Supports incremental updates and full snapshot reconstruction.
    """
    
    def __init__(self, max_depth: int = 20):
        self.max_depth = max_depth
        self.orderbooks: Dict[Tuple[str, str], OrderbookSnapshot] = {}
        self.update_count = 0
        self.snapshot_history: List[OrderbookSnapshot] = []
        
    def update_from_binance(self, symbol: str, data: Dict) -> OrderbookSnapshot:
        """Parse Binance depth update message"""
        timestamp = data.get("E", data.get("updateId", 0))  # Event time or update ID
        bids = [
            OrderbookLevel(price=float(p), quantity=float(q))
            for p, q in data.get("bids", [])[:self.max_depth]
        ]
        asks = [
            OrderbookLevel(price=float(p), quantity=float(q))
            for p, q in data.get("asks", [])[:self.max_depth]
        ]
        
        snapshot = OrderbookSnapshot(
            exchange="binance",
            symbol=symbol,
            timestamp=timestamp,
            bids=bids,
            asks=asks,
        )
        
        key = (snapshot.exchange, symbol)
        self.orderbooks[key] = snapshot
        self.update_count += 1
        
        return snapshot
        
    def update_from_bybit(self, symbol: str, data: Dict) -> OrderbookSnapshot:
        """Parse Bybit orderbook message"""
        # Bybit format: {'type': 'snapshot'|'update', 'data': {...}}
        inner_data = data.get("data", data)
        timestamp = inner_data.get("ts", 0)
        
        bids = [
            OrderbookLevel(price=float(p), quantity=float(s))
            for p, s in inner_data.get("b", inner_data.get("bids", []))[:self.max_depth]
        ]
        asks = [
            OrderbookLevel(price=float(p), quantity=float(s))
            for p, s in inner_data.get("a", inner_data.get("asks", []))[:self.max_depth]
        ]
        
        snapshot = OrderbookSnapshot(
            exchange="bybit",
            symbol=symbol,
            timestamp=timestamp,
            bids=bids,
            asks=asks,
        )
        
        key = (snapshot.exchange, symbol)
        self.orderbooks[key] = snapshot
        self.update_count += 1
        
        return snapshot
        
    def get_snapshot(self, exchange: str, symbol: str) -> Optional[OrderbookSnapshot]:
        """Retrieve current orderbook snapshot"""
        return self.orderbooks.get((exchange, symbol))
        
    def calculate_maker_fee(self, exchange: str) -> float:
        """Return maker fee rate for exchange"""
        fees = {
            "binance": 0.001,  # 0.1%
            "bybit": 0.001,
            "okx": 0.0015,
            "deribit": 0.0005,
        }
        return fees.get(exchange, 0.001)
        
    def calculate_rebate(
        self,
        exchange: str,
        volume_30d: float,
        tier: str = "default",
    ) -> float:
        """Calculate maker rebate based on volume tier"""
        base_rebate = 0.0001  # 0.01% default
        tier_multipliers = {
            "default": 1.0,
            "silver": 1.2,
            "gold": 1.5,
            "platinum": 2.0,
            "diamond": 2.5,
        }
        return base_rebate * tier_multipliers.get(tier, 1.0)
        
    def estimate_slippage(
        self,
        exchange: str,
        symbol: str,
        side: str,  # 'buy' or 'sell'
        quantity: float,
    ) -> Tuple[float, float]:
        """
        Estimate execution slippage for a given order size.
        Returns (expected_price, slippage_bps).
        """
        snapshot = self.get_snapshot(exchange, symbol)
        if not snapshot:
            return 0.0, 0.0
            
        levels = snapshot.asks if side == "buy" else snapshot.bids
        remaining_qty = quantity
        total_cost = 0.0
        
        for level in levels:
            fill_qty = min(remaining_qty, level.quantity)
            total_cost += fill_qty * level.price
            remaining_qty -= fill_qty
            if remaining_qty <= 0:
                break
                
        if quantity > 0:
            avg_price = total_cost / quantity
            mid = snapshot.mid_price
            if mid:
                slippage_bps = abs(avg_price - mid) / mid * 10000
                return avg_price, slippage_bps
                
        return 0.0, 0.0



Global orderbook manager instance

ob_manager = OrderbookManager(max_depth=20)


async def orderbook_handler(message) -> None:
    """Async handler for orderbook updates from HolySheep relay"""
    exchange = message.exchange
    symbol = message.symbol
    raw = message.raw_data
    
    if exchange == "binance":
        snapshot = ob_manager.update_from_binance(symbol, raw)
    elif exchange == "bybit":
        snapshot = ob_manager.update_from_bybit(symbol, raw)
    else:
        return
        
    # Log spread metrics for monitoring
    if snapshot.spread_bps:
        logger.debug(
            f"{exchange}:{symbol} | "
            f"spread: {snapshot.spread_bps:.1f}bps | "
            f"imbalance: {snapshot.imbalance:.3f}"
        )

Trade and Liquidation Handlers

"""
Trade Stream and Liquidation Event Handlers
Integrates with HolySheep Tardis relay for complete market data coverage
"""

import asyncio
from dataclasses import dataclass
from typing import Dict, List, Optional
from datetime import datetime
from collections import deque
import logging
import json

logger = logging.getLogger(__name__)


@dataclass
class NormalizedTrade:
    """Standardized trade event"""
    exchange: str
    symbol: str
    timestamp: int
    price: float
    quantity: float
    side: str  # 'buy' or 'sell' (taker side)
    trade_id: str
    is_liquidation: bool = False
    is_market_maker_trade: bool = False
    
    
@dataclass
class LiquidationEvent:
    """Liquidation event with full context"""
    exchange: str
    symbol: str
    timestamp: int
    side: str  # 'long_liquidated' or 'short_liquidated'
    price: float
    quantity: float
    filled_value: float
    bankruptcy_price: float
    leverage: int
    order_type: str  # 'market' or 'limit'


class TradeAggregator:
    """
    Aggregates trades into configurable windows for backtesting.
    Useful for VWAP calculations and volume analysis.
    """
    
    def __init__(self, window_ms: int = 1000):
        self.window_ms = window_ms
        self.current_window_start = 0
        self.window_trades: deque = deque(maxlen=10000)
        
    def add_trade(self, trade: NormalizedTrade) -> List[NormalizedTrade]:
        """Add trade and return completed window if applicable"""
        window_start = (trade.timestamp // self.window_ms) * self.window_ms
        
        if window_start > self.current_window_start:
            # New window - return completed trades
            completed = list(self.window_trades)
            self.window_trades.clear()
            self.current_window_start = window_start
            
        self.window_trades.append(trade)
        return []
        
    def get_vwap(self) -> Optional[float]:
        """Calculate volume-weighted average price for current window"""
        if not self.window_trades:
            return None
            
        total_volume = sum(t.quantity for t in self.window_trades)
        if total_volume == 0:
            return None
            
        vwap = sum(t.price * t.quantity for t in self.window_trades) / total_volume
        return vwap
        
    def get_volume_stats(self) -> Dict:
        """Return volume statistics for current window"""
        if not self.window_trades:
            return {"buy_volume": 0, "sell_volume": 0, "total_volume": 0}
            
        buy_vol = sum(t.quantity for t in self.window_trades if t.side == "buy")
        sell_vol = sum(t.quantity for t in self.window_trades if t.side == "sell")
        
        return {
            "buy_volume": buy_vol,
            "sell_volume": sell_vol,
            "total_volume": buy_vol + sell_vol,
            "buy_ratio": buy_vol / (buy_vol + sell_vol) if (buy_vol + sell_vol) > 0 else 0.5,
        }


class LiquidationTracker:
    """
    Tracks liquidation events for market impact analysis.
    HolySheep relay includes liquidation feeds that many providers charge extra for.
    """
    
    def __init__(self, ttl_seconds: int = 300):
        self.ttl_seconds = ttl_seconds
        self.liquidations: deque = deque(maxlen=10000)
        self.liquidation_value_24h: float = 0.0
        
    def add_liquidation(self, event: LiquidationEvent) -> None:
        """Record liquidation and update running statistics"""
        self.liquidations.append(event)
        
        # Update 24h stats (simplified - production should use actual time window)
        self.liquidation_value_24h += event.filled_value
        
    def get_recent_liquidations(
        self,
        exchange: Optional[str] = None,
        symbol: Optional[str] = None,
        min_value: float = 0,
    ) -> List[LiquidationEvent]:
        """Query recent liquidations with filters"""
        filtered = []
        
        for liq in self.liquidations:
            if exchange and liq.exchange != exchange:
                continue
            if symbol and liq.symbol != symbol:
                continue
            if liq.filled_value < min_value:
                continue
            filtered.append(liq)
            
        return filtered
        
    def calculate_market_impact(
        self,
        symbol: str,
        time_window_ms: int = 5000,
    ) -> float:
        """
        Estimate market impact from liquidations.
        Returns average price movement (in bps) following liquidations.
        """
        relevant = self.get_recent_liquidations(symbol=symbol)
        
        if len(relevant) < 3:
            return 0.0
            
        # In production, compare actual price movement post-liquidation
        # This is a simplified placeholder
        avg_liquidation_size = sum(l.filled_value for l in relevant) / len(relevant)
        
        # Rough estimation: larger liquidations = more impact
        return min(avg_liquidation_size / 1_000_000 * 5, 50)  # Cap at 50bps



Global instances

trade_aggregator = TradeAggregator(window_ms=1000)
liq_tracker = LiquidationTracker(ttl_seconds=300)


def parse_binance_trade(symbol: str, data: Dict) -> Optional[NormalizedTrade]:
    """Parse Binance trade message"""
    return NormalizedTrade(
        exchange="binance",
        symbol=symbol,
        timestamp=data.get("T", data.get("E", 0)),  # Trade time
        price=float(data.get("p", data.get("price", 0))),
        quantity=float(data.get("q", data.get("qty", 0))),
        side="buy" if data.get("m", True) else "sell",  # m=true means buyer is maker
        trade_id=str(data.get("t", data.get("tradeId", ""))),
    )


def parse_bybit_trade(symbol: str, data: Dict) -> Optional[NormalizedTrade]:
    """Parse Bybit trade message"""
    trades_data = data.get("data", [data])
    if not trades_data:
        return None
        
    trade = trades_data[0] if isinstance(trades_data, list) else trades_data
    
    return NormalizedTrade(
        exchange="bybit",
        symbol=symbol,
        timestamp=int(trade.get("T", trade.get("tradeTime", 0))),
        price=float(trade.get("p", trade.get("price", 0))),
        quantity=float(trade.get("v", trade.get("size", trade.get("qty", 0)))),
        side="buy" if trade.get("S") == "Buy" else "sell",
        trade_id=str(trade.get("i", trade.get("tradeId", ""))),
    )


def parse_binance_liquidation(symbol: str, data: Dict) -> Optional[LiquidationEvent]:
    """Parse Binance force liquidation event"""
    o = data.get("o", {})
    
    return LiquidationEvent(
        exchange="binance",
        symbol=symbol,
        timestamp=data.get("E", 0),
        side="long_liquidated" if o.get("s", "").endswith("USDT") else "short_liquidated",
        price=float(o.get("p", 0)),
        quantity=float(o.get("q", 0)),
        filled_value=float(o.get("q", 0)) * float(o.get("p", 0)),
        bankruptcy_price=float(o.get("ap", 0)),  # Auto-price or bankruptcy price
        leverage=int(o.get("l", 1)),
        order_type="market",
    )


async def trade_handler(message) -> None:
    """Handle incoming trade messages from HolySheep relay"""
    exchange = message.exchange
    symbol = message.symbol
    raw = message.raw_data
    
    trade = None
    
    if exchange == "binance":
        trade = parse_binance_trade(symbol, raw)
    elif exchange == "bybit":
        trade = parse_bybit_trade(symbol, raw)
        
    if trade:
        # Add to aggregator and

Related Resources
Related Articles