One post tagged with "WebSocket"

# ❌ BAD: REST API polling (500ms+ latency)
import time
import requests

while True:
    response = requests.get("https://api.binance.com/api/v3/ticker/price?symbol=BTCUSDT")
    price = response.json()['price']
    print(f"BTC: ${price}")
    time.sleep(0.1)  # Poll every 100ms

# Problems:
# - 500ms+ total latency (network + processing)
# - Wasted bandwidth (99% unchanged data)
# - Rate limited after 1200 requests/minute
# - Missed price updates between polls
# - No guaranteed delivery

# ❌ COMPLEX: Manual WebSocket handling
import asyncio
import websockets
import json

async def binance_ticker():
    url = "wss://stream.binance.com:9443/ws/btcusdt@ticker"

    while True:  # Reconnection loop
        try:
            async with websockets.connect(url) as ws:
                while True:
                    message = await ws.recv()
                    data = json.loads(message)
                    print(f"Price: {data['c']}")

        except websockets.exceptions.ConnectionClosed:
            print("Connection closed, reconnecting...")
            await asyncio.sleep(1)
        except Exception as e:
            print(f"Error: {e}")
            await asyncio.sleep(5)

asyncio.run(binance_ticker())

# Problems:
# - Manual reconnection logic
# - No ping/pong handling
# - Missing error recovery
# - No message buffering
# - 50+ lines for production-ready code

# ✅ GOOD: Automatic WebSocket management
from stockapi import BinanceParser

parser = BinanceParser()

# Real-time ticker stream
for update in parser.stream_ticker("BTCUSDT"):
    print(f"Price: {update['price']}")
    print(f"Volume: {update['volume']}")
    print(f"Change: {update['change_24h']}%")

# Automatically handles:
# - WebSocket connection
# - Ping/pong keepalive
# - Automatic reconnection
# - Error recovery
# - Message parsing
# - 99.9% uptime guarantee

# ❌ BAD: Multiple WebSocket connections
import asyncio
import websockets

async def subscribe_symbol(symbol):
    url = f"wss://stream.binance.com:9443/ws/{symbol.lower()}@ticker"
    async with websockets.connect(url) as ws:
        async for message in ws:
            # Process message
            pass

# Subscribe to 100 symbols
symbols = ["BTCUSDT", "ETHUSDT", ...]  # 100 symbols
tasks = [subscribe_symbol(s) for s in symbols]
await asyncio.gather(*tasks)

# Problems:
# - 100 WebSocket connections (resource intensive)
# - Connection limit issues
# - Difficult to manage
# - High memory usage
# - Complex error handling

# ✅ GOOD: Single multiplexed stream
from stockapi import BinanceParser

parser = BinanceParser()

# Single WebSocket, multiple symbols
symbols = ["BTCUSDT", "ETHUSDT", "BNBUSDT", ...]  # 100+ symbols

for update in parser.stream_tickers(symbols):
    symbol = update['symbol']
    price = update['price']
    print(f"{symbol}: ${price}")

# Single WebSocket connection handles all symbols
# Automatic message routing
# Memory efficient
# Easy error recovery

# ❌ BAD: Repeated full snapshots
import requests

while True:
    # Fetch full order book (1000 levels)
    response = requests.get(
        "https://api.binance.com/api/v3/depth",
        params={"symbol": "BTCUSDT", "limit": 1000}
    )
    orderbook = response.json()

    # Process full orderbook every time
    analyze_orderbook(orderbook)
    time.sleep(0.1)

# Problems:
# - Massive bandwidth waste (full book every 100ms)
# - High latency (500ms+)
# - Rate limited
# - Inefficient processing

# ✅ GOOD: Incremental order book updates
from stockapi import BinanceParser

parser = BinanceParser()

# Real-time order book with incremental updates
orderbook = parser.stream_orderbook("BTCUSDT", depth=100)

for update in orderbook:
    if update['type'] == 'snapshot':
        # Initial full snapshot
        bids = update['bids']  # [[price, quantity], ...]
        asks = update['asks']
    else:
        # Incremental update (only changes)
        for bid in update['bids']:
            price, quantity = bid
            if quantity == 0:
                # Remove level
                remove_bid_level(price)
            else:
                # Update level
                update_bid_level(price, quantity)

# Minimal bandwidth (only changes)
# Sub-100ms updates
# Automatic snapshot recovery
# Guaranteed consistency

# ❌ BAD: Manual multi-exchange WebSockets
import asyncio

async def binance_stream():
    # Binance-specific WebSocket logic
    pass

async def coinbase_stream():
    # Coinbase-specific WebSocket logic
    pass

async def kraken_stream():
    # Kraken-specific WebSocket logic
    pass

# Each exchange has different:
# - WebSocket URL format
# - Authentication method
# - Message format
# - Reconnection logic
# - Rate limits

# Result: 500+ lines of integration code per exchange

# ✅ GOOD: Unified multi-exchange streaming
from stockapi import BinanceParser, CoinbaseParser, KrakenParser

# Same interface across all exchanges
parsers = {
    'binance': BinanceParser(),
    'coinbase': CoinbaseParser(),
    'kraken': KrakenParser(),
}

async def aggregate_streams(symbol):
    streams = [
        parser.stream_ticker(symbol)
        for parser in parsers.values()
    ]

    async for exchange, update in combine_streams(streams):
        print(f"{exchange}: ${update['price']}")

# Unified interface
# Same data format
# Automatic normalization
# Built-in arbitrage detection

# ✅ Production-ready stream with resilience
from stockapi import BinanceParser

parser = BinanceParser(
    reconnect_attempts=float('inf'),  # Never give up
    reconnect_delay=1.0,               # 1s between attempts
    ping_interval=20,                  # Keepalive every 20s
    ping_timeout=10,                   # 10s ping timeout
)

# Handles all failure scenarios:
# - Network interruptions
# - Exchange disconnections
# - API rate limits
# - Message corruption
# - Timeout errors

for update in parser.stream_ticker("BTCUSDT"):
    # Will automatically recover from any error
    process_update(update)

# ✅ Handle burst traffic without data loss
from stockapi import BinanceParser

parser = BinanceParser(
    buffer_size=10000,     # Buffer up to 10k messages
    buffer_strategy='drop_oldest',  # Drop old on overflow
)

# During high volatility:
# - Messages buffered during processing
# - No data loss up to buffer limit
# - Configurable overflow strategy
# - Memory-safe operation

# ✅ Track end-to-end latency
from stockapi import BinanceParser
import time

parser = BinanceParser()

for update in parser.stream_ticker("BTCUSDT"):
    # Exchange timestamp
    exchange_time = update['timestamp']

    # Local receipt time
    local_time = time.time() * 1000

    # Calculate latency
    latency = local_time - exchange_time

    print(f"Latency: {latency:.2f}ms")

    # Typical results:
    # - Binance: 20-50ms
    # - Coinbase: 30-60ms
    # - NYSE: 50-100ms
    # StockAPI adds <10ms overhead

# ✅ Production-ready trading bot in 30 lines
from stockapi import BinanceParser, CoinbaseParser

class ArbitrageBot:
    def __init__(self):
        self.binance = BinanceParser()
        self.coinbase = CoinbaseParser()

    def run(self, symbol):
        # Stream from both exchanges simultaneously
        binance_stream = self.binance.stream_ticker(symbol)
        coinbase_stream = self.coinbase.stream_ticker(symbol)

        binance_price = None
        coinbase_price = None

        while True:
            # Get latest from both (non-blocking)
            binance_price = next(binance_stream, binance_price)
            coinbase_price = next(coinbase_stream, coinbase_price)

            if binance_price and coinbase_price:
                spread = abs(
                    binance_price['price'] - coinbase_price['price']
                )

                if spread > 10:  # $10 arbitrage opportunity
                    self.execute_arbitrage(
                        binance_price,
                        coinbase_price
                    )

bot = ArbitrageBot()
bot.run("BTCUSDT")

# Real-time arbitrage detection
# Sub-100ms latency
# 99.9% uptime
# Production-ready