How I Built Real-Time Stablecoin Market Making Analytics (And Stopped Losing Money on Bad Spreads)

# This connection management took me 2 weeks to get right
# I learned the hard way that single points of failure kill P&L

import asyncio
import websockets
import json
from dataclasses import dataclass
from typing import Dict, List
import pandas as pd

@dataclass
class MarketDepth:
    exchange: str
    symbol: str
    bids: List[tuple]  # [(price, size), ...]
    asks: List[tuple]
    timestamp: float
    
class StablecoinAnalytics:
    def __init__(self):
        self.depth_data: Dict[str, MarketDepth] = {}
        self.spread_history = pd.DataFrame()
        
    async def connect_binance_ws(self):
        """Connect to Binance depth stream - my most reliable feed"""
        uri = "wss://stream.binance.com:9443/ws/usdcusdt@depth20@100ms"
        
        try:
            async with websockets.connect(uri) as websocket:
                while True:
                    message = await websocket.recv()
                    data = json.loads(message)
                    await self.process_binance_depth(data)
                    
        except Exception as e:
            # I spent hours debugging connection drops before adding this
            print(f"Binance connection failed: {e}")
            await asyncio.sleep(5)
            await self.connect_binance_ws()  # Auto-reconnect
    
    async def process_binance_depth(self, data):
        """Process Binance order book updates"""
        bids = [(float(bid[0]), float(bid[1])) for bid in data['bids']]
        asks = [(float(ask[0]), float(ask[1])) for ask in data['asks']]
        
        self.depth_data['binance_usdcusdt'] = MarketDepth(
            exchange='binance',
            symbol='USDCUSDT',
            bids=bids,
            asks=asks,
            timestamp=time.time()
        )
        
        # This is where the magic happens - real-time spread calculation
        await self.calculate_spreads()

def calculate_effective_spreads(self, target_size: float = 100000):
    """
    Calculate real spreads for actual trading sizes
    target_size: USD amount we want to trade
    """
    spreads = {}
    
    for exchange_pair, depth in self.depth_data.items():
        if not depth.bids or not depth.asks:
            continue
            
        # Calculate size-weighted spreads - this was my eureka moment
        buy_cost = self.calculate_execution_cost(depth.asks, target_size)
        sell_proceeds = self.calculate_execution_cost(depth.bids, target_size, side='sell')
        
        if buy_cost and sell_proceeds:
            effective_spread = (buy_cost - sell_proceeds) / target_size * 10000  # in bp
            
            spreads[exchange_pair] = {
                'effective_spread_bp': effective_spread,
                'nominal_spread_bp': (depth.asks[0][0] - depth.bids[0][0]) / depth.bids[0][0] * 10000,
                'depth_1bp': self.calculate_depth_at_level(depth, 0.0001),
                'depth_5bp': self.calculate_depth_at_level(depth, 0.0005),
                'available_liquidity': min(
                    sum([bid[1] * bid[0] for bid in depth.bids[:10]]),
                    sum([ask[1] * ask[0] for ask in depth.asks[:10]])
                )
            }
    
    return spreads

def calculate_execution_cost(self, orders: List[tuple], target_size: float, side: str = 'buy'):
    """
    Calculate actual cost to execute a trade of target_size
    This function saved me from the depth illusion trap
    """
    total_cost = 0
    remaining_size = target_size
    
    for price, size in orders:
        if remaining_size <= 0:
            break
            
        trade_size = min(remaining_size, size * price if side == 'buy' else size)
        
        if side == 'buy':
            total_cost += trade_size
            remaining_size -= trade_size
        else:
            total_cost += trade_size * price
            remaining_size -= trade_size
    
    if remaining_size > 0:
        return None  # Insufficient liquidity
        
    return total_cost

def generate_liquidity_heatmap(self, lookback_hours: int = 24):
    """
    Generate hourly liquidity patterns - reveals when depth disappears
    This saved me from repeating my 3am disaster
    """
    
    # Query historical depth data
    end_time = datetime.now()
    start_time = end_time - timedelta(hours=lookback_hours)
    
    # I store all depth snapshots in TimescaleDB for fast queries
    depth_query = """
    SELECT 
        EXTRACT(hour FROM timestamp) as hour,
        AVG(depth_1bp) as avg_depth_1bp,
        AVG(depth_5bp) as avg_depth_5bp,
        MIN(depth_1bp) as min_depth_1bp,
        COUNT(*) as snapshots
    FROM market_depth 
    WHERE timestamp >= %s AND timestamp <= %s
    AND exchange = 'binance' AND symbol = 'USDCUSDT'
    GROUP BY EXTRACT(hour FROM timestamp)
    ORDER BY hour
    """
    
    depth_patterns = pd.read_sql(depth_query, self.db_connection, 
                                params=[start_time, end_time])
    
    # Flag dangerous periods - when depth drops below my minimum thresholds
    depth_patterns['risk_level'] = depth_patterns.apply(
        lambda row: 'HIGH' if row['min_depth_1bp'] < 50000 else 'MEDIUM' if row['min_depth_1bp'] < 100000 else 'LOW',
        axis=1
    )
    
    return depth_patterns

def analyze_spread_correlations(self, window_minutes: int = 60):
    """
    Track how spreads correlate across exchanges
    High correlation = arbitrage opportunities close quickly
    """
    
    # Get recent spread data for all exchanges
    spreads_df = self.get_recent_spreads(window_minutes)
    
    # Calculate rolling correlations
    correlation_matrix = spreads_df.corr()
    
    # Flag high correlation periods - when arb ops disappear fast
    avg_correlation = correlation_matrix.values[np.triu_indices_from(correlation_matrix.values, k=1)].mean()
    
    market_state = {
        'correlation_level': avg_correlation,
        'arbitrage_persistence': 'LOW' if avg_correlation > 0.8 else 'HIGH',
        'recommended_hold_time': '30s' if avg_correlation > 0.8 else '300s'
    }
    
    return market_state

class AlertManager:
    def __init__(self):
        self.alert_thresholds = {
            'min_depth_1bp': 75000,      # Minimum liquidity at 1bp
            'max_spread_bp': 15,         # Maximum acceptable spread  
            'correlation_spike': 0.85,   # Cross-exchange correlation threshold
            'volume_drop_pct': 0.4       # Volume drop from 24h average
        }
    
    async def check_market_conditions(self):
        """
        Run every 10 seconds - catches problems before they hurt P&L
        """
        current_metrics = await self.get_current_metrics()
        
        alerts = []
        
        # Depth alerts - prevented 3 major losses this month
        if current_metrics['depth_1bp'] < self.alert_thresholds['min_depth_1bp']:
            alerts.append({
                'type': 'LIQUIDITY_WARNING',
                'message': f"Depth at 1bp: ${current_metrics['depth_1bp']:,.0f} below threshold",
                'severity': 'HIGH',
                'action': 'REDUCE_POSITION_SIZE'
            })
        
        # Spread alerts - catches manipulation attempts
        if current_metrics['effective_spread_bp'] > self.alert_thresholds['max_spread_bp']:
            alerts.append({
                'type': 'SPREAD_ANOMALY',
                'message': f"Effective spread: {current_metrics['effective_spread_bp']:.1f}bp above threshold",
                'severity': 'MEDIUM',
                'action': 'PAUSE_NEW_ORDERS'
            })
        
        # Send alerts via Telegram and email
        for alert in alerts:
            await self.send_alert(alert)
    
    async def send_alert(self, alert):
        """Send via Telegram - I get alerts in under 3 seconds"""
        telegram_message = f"🚨 {alert['severity']}: {alert['message']}\nAction: {alert['action']}"
        
        # My Telegram bot for instant notifications
        await self.telegram_bot.send_message(
            chat_id=self.config.telegram_chat_id,
            text=telegram_message
        )

# TimescaleDB configuration that handles 50M+ rows per day
CREATE_TABLES_SQL = """
-- Hypertable for time-series market data
CREATE TABLE IF NOT EXISTS market_depth (
    timestamp TIMESTAMPTZ NOT NULL,
    exchange VARCHAR(20) NOT NULL,
    symbol VARCHAR(20) NOT NULL,
    bid_price DECIMAL(12,8),
    ask_price DECIMAL(12,8),
    bid_size DECIMAL(18,8),
    ask_size DECIMAL(18,8),
    depth_1bp DECIMAL(18,2),
    depth_5bp DECIMAL(18,2),
    effective_spread_bp DECIMAL(8,4)
);

-- Convert to hypertable for time-series optimization
SELECT create_hypertable('market_depth', 'timestamp', if_not_exists => TRUE);

-- Critical indexes for real-time queries
CREATE INDEX IF NOT EXISTS idx_market_depth_exchange_symbol_time 
ON market_depth (exchange, symbol, timestamp DESC);

-- Materialized view for real-time dashboard
CREATE MATERIALIZED VIEW IF NOT EXISTS current_market_state AS
SELECT 
    exchange,
    symbol,
    LAST(bid_price, timestamp) as current_bid,
    LAST(ask_price, timestamp) as current_ask,
    LAST(depth_1bp, timestamp) as current_depth_1bp,
    LAST(effective_spread_bp, timestamp) as current_spread_bp,
    MAX(timestamp) as last_update
FROM market_depth 
WHERE timestamp > NOW() - INTERVAL '1 hour'
GROUP BY exchange, symbol;
"""

class MemoryEfficientProcessor:
    def __init__(self):
        # Ring buffers prevent memory growth
        self.max_buffer_size = 10000
        self.depth_buffer = collections.deque(maxlen=self.max_buffer_size)
        self.spread_buffer = collections.deque(maxlen=self.max_buffer_size)
        
    def process_market_update(self, update):
        """
        Process without accumulating memory - learned this the hard way
        """
        # Add to ring buffer (automatically removes old data)
        self.depth_buffer.append(update)
        
        # Process only recent window for calculations
        recent_data = list(self.depth_buffer)[-1000:]  # Last 1000 updates only
        
        # Calculate metrics using recent data
        current_metrics = self.calculate_metrics(recent_data)
        
        # Clear processed data references
        del recent_data
        
        return current_metrics

def calculate_daily_kpis(self):
    """
    The 8 metrics that determine if I'm making or losing money
    """
    
    kpis = {
        # P&L Metrics
        'daily_pnl_usd': self.get_daily_pnl(),
        'win_rate_pct': self.get_win_rate(),
        'avg_trade_duration_sec': self.get_avg_trade_duration(),
        
        # Risk Metrics  
        'max_drawdown_pct': self.get_max_drawdown(),
        'var_95_1d_usd': self.calculate_var_95(),
        
        # Market Metrics
        'avg_effective_spread_bp': self.get_avg_effective_spread(),
        'liquidity_uptime_pct': self.calculate_liquidity_uptime(),
        'system_uptime_pct': self.get_system_uptime()
    }
    
    return kpis

def calculate_optimal_position_size(self, base_size: float = 100000):
    """
    Adjust position size based on current market conditions
    This prevented my second major loss in month 6
    """
    
    current_conditions = self.get_current_market_state()
    
    # Start with base size
    adjusted_size = base_size
    
    # Reduce size when liquidity is low
    depth_factor = min(1.0, current_conditions['depth_1bp'] / 100000)
    adjusted_size *= depth_factor
    
    # Reduce size when spreads are wide (market stress)
    spread_factor = max(0.5, min(1.0, 10 / current_conditions['effective_spread_bp']))
    adjusted_size *= spread_factor
    
    # Reduce size during high correlation periods (arb windows close fast)
    correlation_factor = max(0.3, 1.0 - current_conditions['cross_exchange_correlation'])
    adjusted_size *= correlation_factor
    
    # Never exceed maximum position limit
    final_size = min(adjusted_size, self.max_position_size)
    
    return {
        'position_size': final_size,
        'depth_factor': depth_factor,
        'spread_factor': spread_factor,
        'correlation_factor': correlation_factor,
        'utilization_pct': final_size / base_size * 100
    }