Build a Real-Time Gold Market Data Feed in 45 Minutes

import asyncio
import websockets
import json
from datetime import datetime

# Personal note: Learned this after my first connection kept timing out
# Always set ping_interval or exchanges drop you after 30 seconds

class GoldVenueConnector:
    def __init__(self, venue_name, ws_url, symbols):
        self.venue = venue_name
        self.url = ws_url
        self.symbols = symbols
        self.ws = None
        
    async def connect(self):
        """Maintain persistent connection with auto-reconnect"""
        while True:
            try:
                async with websockets.connect(
                    self.url,
                    ping_interval=20,  # Keep connection alive
                    ping_timeout=10
                ) as ws:
                    self.ws = ws
                    await self.subscribe()
                    await self.receive_quotes()
            except Exception as e:
                print(f"[{self.venue}] Connection lost: {e}. Reconnecting in 5s...")
                await asyncio.sleep(5)
    
    async def subscribe(self):
        """Send subscription message (format varies by venue)"""
        if self.venue == "COMEX":
            # CME uses channel-based subscriptions
            sub_msg = {
                "action": "subscribe",
                "channels": [f"quotes.{sym}" for sym in self.symbols]
            }
        elif self.venue == "NYMEX":
            # NYMEX uses symbol array
            sub_msg = {
                "type": "subscribe",
                "symbols": self.symbols
            }
        else:  # OTC/Spot
            sub_msg = {
                "cmd": "sub",
                "instruments": self.symbols
            }
        
        await self.ws.send(json.dumps(sub_msg))
        print(f"[{self.venue}] Subscribed to {len(self.symbols)} symbols")
    
    async def receive_quotes(self):
        """Stream quotes and normalize format"""
        async for message in self.ws:
            raw_quote = json.loads(message)
            normalized = self.normalize_quote(raw_quote)
            if normalized:
                await self.on_quote(normalized)
    
    def normalize_quote(self, raw):
        """Convert venue-specific format to standard quote"""
        # Watch out: Each venue has different field names
        if self.venue == "COMEX":
            return {
                "symbol": raw.get("s"),  # s = symbol on CME
                "bid": float(raw["b"]),
                "ask": float(raw["a"]),
                "bid_size": int(raw["bs"]),
                "ask_size": int(raw["as"]),
                "timestamp": raw["t"],  # Unix ms
                "venue": self.venue
            }
        elif self.venue == "NYMEX":
            return {
                "symbol": raw["sym"],
                "bid": float(raw["bid_price"]),
                "ask": float(raw["ask_price"]),
                "bid_size": int(raw["bid_qty"]),
                "ask_size": int(raw["ask_qty"]),
                "timestamp": raw["exchange_time"],
                "venue": self.venue
            }
        # Add other venue parsers...
        return None
    
    async def on_quote(self, quote):
        """Override this to handle normalized quotes"""
        pass

# Usage
comex = GoldVenueConnector(
    "COMEX",
    "wss://stream.cmegroup.com/live",
    ["GC.FUT"]  # Gold futures
)

import redis.asyncio as redis
from typing import Dict, Optional

class ConsolidatedTapeProvider:
    def __init__(self):
        self.redis = None
        self.quote_cache: Dict[str, list] = {}  # symbol -> [quotes]
        self.nbbo_callbacks = []
        
    async def initialize(self):
        """Connect to Redis for quote deduplication"""
        self.redis = await redis.from_url("redis://localhost:6379")
        print("✓ Connected to Redis")
        
    async def process_quote(self, quote: dict):
        """
        Main consolidation logic:
        1. Deduplicate (same price/size from venue = duplicate)
        2. Update quote cache
        3. Recalculate NBBO
        """
        # Personal note: Without dedup, I was processing the same quote
        # 3-5 times from different feeds. Killed my throughput.
        
        quote_key = f"{quote['symbol']}:{quote['venue']}:{quote['bid']}:{quote['ask']}"
        
        # Check if we've seen this exact quote in last 500ms
        is_duplicate = await self.redis.get(quote_key)
        if is_duplicate:
            return  # Skip duplicate
        
        # Store quote signature for 500ms
        await self.redis.setex(quote_key, 0.5, "1")
        
        # Update quote cache for this symbol
        symbol = quote['symbol']
        if symbol not in self.quote_cache:
            self.quote_cache[symbol] = []
        
        # Remove stale quotes from this venue (older than 2 seconds)
        current_time = quote['timestamp']
        self.quote_cache[symbol] = [
            q for q in self.quote_cache[symbol]
            if q['venue'] != quote['venue'] and (current_time - q['timestamp']) < 2000
        ]
        
        # Add new quote
        self.quote_cache[symbol].append(quote)
        
        # Calculate new NBBO
        nbbo = self.calculate_nbbo(symbol)
        
        # Publish to subscribers
        for callback in self.nbbo_callbacks:
            await callback(nbbo)
    
    def calculate_nbbo(self, symbol: str) -> Optional[dict]:
        """
        Calculate National Best Bid and Offer
        Best Bid = Highest bid across all venues
        Best Ask = Lowest ask across all venues
        """
        quotes = self.quote_cache.get(symbol, [])
        if not quotes:
            return None
        
        # Find best bid (highest)
        best_bid_quote = max(quotes, key=lambda q: q['bid'])
        # Find best ask (lowest)
        best_ask_quote = min(quotes, key=lambda q: q['ask'])
        
        nbbo = {
            "symbol": symbol,
            "bid": best_bid_quote['bid'],
            "ask": best_ask_quote['ask'],
            "bid_venue": best_bid_quote['venue'],
            "ask_venue": best_ask_quote['venue'],
            "bid_size": best_bid_quote['bid_size'],
            "ask_size": best_ask_quote['ask_size'],
            "spread": round(best_ask_quote['ask'] - best_bid_quote['bid'], 2),
            "timestamp": max(q['timestamp'] for q in quotes),
            "venue_count": len(set(q['venue'] for q in quotes))
        }
        
        return nbbo
    
    def subscribe_to_nbbo(self, callback):
        """Register callback for NBBO updates"""
        self.nbbo_callbacks.append(callback)

# Wire everything together
async def main():
    ctp = ConsolidatedTapeProvider()
    await ctp.initialize()
    
    # Create venue connectors
    venues = [
        GoldVenueConnector("COMEX", "wss://stream.cmegroup.com/live", ["GC.FUT"]),
        GoldVenueConnector("NYMEX", "wss://nymex-stream.com/quotes", ["MGC.FUT"]),
        # Add more venues...
    ]
    
    # Hook up quote flow: Venue -> CTP
    for venue in venues:
        venue.on_quote = ctp.process_quote
    
    # Subscribe to NBBO updates
    async def print_nbbo(nbbo):
        print(f"NBBO Update: {nbbo['symbol']} "
              f"Bid: ${nbbo['bid']} ({nbbo['bid_venue']}) "
              f"Ask: ${nbbo['ask']} ({nbbo['ask_venue']}) "
              f"Spread: ${nbbo['spread']} | "
              f"Venues: {nbbo['venue_count']}")
    
    ctp.subscribe_to_nbbo(print_nbbo)
    
    # Start all venue connections
    await asyncio.gather(*[v.connect() for v in venues])

if __name__ == "__main__":
    asyncio.run(main())

from collections import defaultdict, deque
from dataclasses import dataclass
import time

@dataclass
class VenueMetrics:
    quotes_received: int = 0
    stale_quotes: int = 0
    crossed_markets: int = 0
    avg_latency_ms: float = 0.0
    last_quote_time: float = 0.0

class MarketDataMonitor:
    def __init__(self):
        self.metrics = defaultdict(VenueMetrics)
        self.latency_samples = defaultdict(lambda: deque(maxlen=100))
        
    def record_quote(self, quote: dict):
        """Track quality metrics for each quote"""
        venue = quote['venue']
        m = self.metrics[venue]
        
        m.quotes_received += 1
        m.last_quote_time = time.time()
        
        # Check for crossed market (bid > ask = bad data)
        if quote['bid'] >= quote['ask']:
            m.crossed_markets += 1
            print(f"⚠️  Crossed market from {venue}: "
                  f"Bid ${quote['bid']} >= Ask ${quote['ask']}")
        
        # Calculate latency (exchange time -> our time)
        receive_time = time.time() * 1000  # Convert to ms
        exchange_time = quote['timestamp']
        latency = receive_time - exchange_time
        
        if latency > 1000:  # Stale if > 1 second old
            m.stale_quotes += 1
        
        # Track rolling average latency
        self.latency_samples[venue].append(latency)
        m.avg_latency_ms = sum(self.latency_samples[venue]) / len(self.latency_samples[venue])
    
    def get_health_report(self) -> dict:
        """Generate monitoring report"""
        report = {}
        for venue, m in self.metrics.items():
            stale_pct = (m.stale_quotes / m.quotes_received * 100) if m.quotes_received > 0 else 0
            crossed_pct = (m.crossed_markets / m.quotes_received * 100) if m.quotes_received > 0 else 0
            
            report[venue] = {
                "total_quotes": m.quotes_received,
                "avg_latency_ms": round(m.avg_latency_ms, 1),
                "stale_percentage": round(stale_pct, 2),
                "crossed_percentage": round(crossed_pct, 2),
                "last_quote_age_sec": round(time.time() - m.last_quote_time, 1)
            }
        
        return report

# Add to CTP
class ConsolidatedTapeProvider:
    def __init__(self):
        # ... existing code ...
        self.monitor = MarketDataMonitor()
        
    async def process_quote(self, quote: dict):
        self.monitor.record_quote(quote)  # Track metrics
        # ... rest of existing code ...

# Print health report every 30 seconds
async def print_health_report(ctp):
    while True:
        await asyncio.sleep(30)
        report = ctp.monitor.get_health_report()
        print("\n=== Market Data Health Report ===")
        for venue, metrics in report.items():
            print(f"{venue}:")
            print(f"  Quotes: {metrics['total_quotes']}")
            print(f"  Latency: {metrics['avg_latency_ms']}ms")
            print(f"  Stale: {metrics['stale_percentage']}%")
            print(f"  Crossed: {metrics['crossed_percentage']}%")
            print(f"  Last quote: {metrics['last_quote_age_sec']}s ago")