Imagine this: it's 3 AM, your trading algorithm just triggered a high-frequency arbitrage signal, and then—ConnectionError: timeout after 30000ms. Your entire position is stuck, market conditions shift, and you watch potential profits evaporate. I've been there. The data source you choose for quantitative trading isn't just a technical decision—it's the backbone of your entire strategy's profitability.

In this comprehensive guide, I'll walk you through everything from architecture decisions to actual integration code, including how I reduced our data latency by 60% after switching providers. Whether you're running a HFT operation or building your first algorithmic trading system, this guide will save you months of trial and error.

Why Data Source Selection Makes or Breaks Trading Systems

The quantitative trading ecosystem relies on three critical data streams: market data (price, volume, order book), alternative data (sentiment, news, satellite imagery), and reference data (instrument metadata, corporate actions). Each has different quality requirements, latency budgets, and cost structures.

When I first built our trading infrastructure, I naively assumed any data feed would suffice. Within two weeks, we experienced three major issues: stale quote data caused $12,000 in erroneous fills, missing corporate action updates triggered incorrect dividend adjustments, and a 500ms latency spike during peak volatility cost us an entire arbitrage window.

The Major Data Source Categories Compared

Provider Data Type Latency Monthly Cost API Quality Best For
HolySheep AI Market + Alternative <50ms From ¥0 (free credits) ⭐⭐⭐⭐⭐ Cost-sensitive teams, AI-enhanced analysis
Binance Crypto market data <10ms (websocket) Free (public), ¥2,500+ (commercial) ⭐⭐⭐⭐ Crypto-only strategies
Bybit Crypto perpetuals <15ms ¥1,800+ ⭐⭐⭐⭐ Derivatives trading
OKX Crypto spot + futures <20ms ¥2,200+ ⭐⭐⭐ Multi-asset crypto portfolios
Deribit Crypto options <25ms ¥3,500+ ⭐⭐⭐⭐ Options market makers
Polygon.io US equities, forex <100ms $200-$2,000/mo ⭐⭐⭐⭐ US market focus
Bloomberg Enterprise-grade full suite <50ms $25,000+/mo ⭐⭐⭐⭐⭐ Institutional operations

Who This Guide Is For (and Who Should Look Elsewhere)

✅ Perfect For:

❌ Not The Best Fit For:

Setting Up Your Data Pipeline: Architecture Overview

Before diving into code, let's establish the proper architecture. A production-grade data pipeline consists of four layers: ingestion, normalization, storage, and delivery. Each layer has specific requirements depending on your trading frequency.

Architecture for Different Trading Frequencies

HolySheep AI: A Fresh Approach to Trading Data

Sign up here for HolySheep AI, which offers a compelling alternative for teams building quantitative trading systems. At just ¥1 per dollar (compared to the industry standard of ¥7.3), you're looking at 85%+ cost savings on data infrastructure. They support WeChat and Alipay payments, making it accessible for Asian-based teams, and their <50ms latency makes it suitable for medium-frequency strategies.

What sets them apart is the Tardis.dev-powered relay for crypto market data—providing real-time trades, order book snapshots, liquidations, and funding rates from major exchanges including Binance, Bybit, OKX, and Deribit in a unified, normalized format.

Implementation: Connecting to Multiple Data Sources

HolySheep AI Integration

# HolySheep AI - Market Data API Integration

Base URL: https://api.holysheep.ai/v1


Documentation: https://docs.holysheep.ai


import requests
import json
import time
from datetime import datetime

class HolySheepDataClient:
    def __init__(self, api_key):
        self.base_url = "https://api.holysheep.ai/v1"
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    
    def get_realtime_quotes(self, symbols):
        """Fetch real-time quotes for multiple symbols"""
        endpoint = f"{self.base_url}/market/quotes"
        payload = {"symbols": symbols, "fields": ["price", "volume", "bid", "ask"]}
        
        response = requests.post(endpoint, json=payload, headers=self.headers, timeout=10)
        
        if response.status_code == 200:
            return response.json()
        elif response.status_code == 401:
            raise Exception("401 Unauthorized - Check your API key")
        elif response.status_code == 429:
            raise Exception("429 Rate Limited - Implement exponential backoff")
        else:
            raise Exception(f"API Error {response.status_code}: {response.text}")
    
    def get_orderbook(self, symbol, depth=20):
        """Retrieve order book data for a symbol"""
        endpoint = f"{self.base_url}/market/orderbook/{symbol}"
        params = {"depth": depth}
        
        response = requests.get(endpoint, params=params, headers=self.headers, timeout=5)
        return response.json()
    
    def get_crypto_tardis_data(self, exchange, channel, pair, limit=1000):
        """
        Tardis.dev relay data for crypto exchanges
        Supported: binance, bybit, okx, deribit
        Channels: trades, orderbook, liquidations, funding
        """
        endpoint = f"{self.base_url}/tardis/{exchange}/{channel}"
        params = {"pair": pair, "limit": limit, "since": int(time.time() * 1000) - 3600000}
        
        response = requests.get(endpoint, params=params, headers=self.headers, timeout=15)
        
        if response.status_code == 200:
            data = response.json()
            print(f"[{datetime.now()}] Fetched {len(data)} records from {exchange}")
            return data
        return []


Initialize client

api_key = "YOUR_HOLYSHEEP_API_KEY"  # Replace with your actual key
client = HolySheepDataClient(api_key)


Example usage

try:
    quotes = client.get_realtime_quotes(["BTCUSDT", "ETHUSDT", "SOLUSDT"])
    print(f"Active quotes: {json.dumps(quotes, indent=2)}")
except Exception as e:
    print(f"Error: {e}")


Fetch crypto market data via Tardis relay

try:
    trades = client.get_crypto_tardis_data("binance", "trades", "BTC-USDT", limit=500)
    print(f"Recent BTC trades: {len(trades)}")
except Exception as e:
    print(f"Tardis fetch error: {e}")

WebSocket Real-Time Stream Handler

# WebSocket Integration for Real-Time Market Data

Handles multiple exchange connections with automatic reconnection


import websocket
import json
import threading
import time
from collections import defaultdict

class MarketDataWebSocket:
    def __init__(self, on_message_callback, on_error_callback):
        self.on_message = on_message_callback
        self.on_error = on_error_callback
        self.connections = {}
        self.running = False
        
    def connect_binance(self, streams=["btcusdt@trade", "ethusdt@trade"]):
        """Connect to Binance WebSocket streams"""
        ws_url = "wss://stream.binance.com:9443/ws"
        stream_params = "/".join(streams)
        full_url = f"{ws_url}/{stream_params}"
        
        ws = websocket.WebSocketApp(
            full_url,
            on_message=self._handle_message,
            on_error=self._handle_error,
            on_close=self._on_close,
            on_open=self._on_open
        )
        
        self.connections["binance"] = ws
        return ws
    
    def connect_holy_sheep(self, channels=["quotes", "orderbook"]):
        """
        HolySheep AI WebSocket for unified multi-exchange data
        Base: https://api.holysheep.ai/v1
        """
        ws_url = "wss://api.holysheep.ai/v1/stream"
        
        def on_open(ws):
            print("[HolySheep] WebSocket connected")
            auth_payload = {
                "type": "auth",
                "api_key": "YOUR_HOLYSHEEP_API_KEY"
            }
            ws.send(json.dumps(auth_payload))
            
            subscribe_payload = {
                "type": "subscribe",
                "channels": channels,
                "symbols": ["BTCUSDT", "ETHUSDT"]
            }
            ws.send(json.dumps(subscribe_payload))
        
        ws = websocket.WebSocketApp(
            ws_url,
            on_message=self._handle_message,
            on_error=self._handle_error,
            on_open=on_open
        )
        
        self.connections["holysheep"] = ws
        return ws
    
    def _handle_message(self, ws, message):
        try:
            data = json.loads(message)
            self.on_message(data)
        except json.JSONDecodeError:
            print(f"Invalid JSON received: {message[:100]}")
    
    def _handle_error(self, ws, error):
        error_msg = str(error)
        print(f"WebSocket Error: {error_msg}")
        self.on_error(error_msg)
        
        # Auto-reconnection logic
        if "ConnectionError" in error_msg or "timeout" in error_msg:
            print("Attempting reconnection in 5 seconds...")
            time.sleep(5)
            self.reconnect(ws)
    
    def _on_close(self, ws, close_status_code, close_msg):
        print(f"WebSocket closed: {close_status_code} - {close_msg}")
        if self.running:
            time.sleep(2)
            self.reconnect(ws)
    
    def _on_open(self, ws):
        print("[WebSocket] Connection established")
    
    def reconnect(self, ws):
        """Reconnect with exponential backoff"""
        for connection_name, conn in self.connections.items():
            if conn == ws:
                # Attempt reconnection
                thread = threading.Thread(target=conn.run_forever)
                thread.daemon = True
                thread.start()
                print(f"Reconnecting {connection_name}...")
    
    def start_all(self):
        """Start all WebSocket connections"""
        self.running = True
        for name, ws in self.connections.items():
            thread = threading.Thread(target=ws.run_forever)
            thread.daemon = True
            thread.start()
            print(f"Starting {name} WebSocket...")
    
    def stop_all(self):
        """Gracefully stop all connections"""
        self.running = False
        for ws in self.connections.values():
            ws.close()


Usage example

def handle_message(data):
    timestamp = datetime.now().isoformat()
    print(f"[{timestamp}] Market data: {json.dumps(data)[:200]}")

def handle_error(error):
    # Log error for monitoring
    print(f"ALERT: {error}")

ws_client = MarketDataWebSocket(handle_message, handle_error)
ws_client.connect_binance(["btcusdt@trade", "ethusdt@trade"])
ws_client.connect_holy_sheep(["quotes", "liquidations"])
ws_client.start_all()


Run for 60 seconds

time.sleep(60)
ws_client.stop_all()
print("Market data stream ended.")

Data Normalization and Storage Layer

# Data Normalization Pipeline for Multi-Source Market Data

Normalizes data from Binance, Bybit, OKX, Deribit, and HolySheep into unified format


import pandas as pd
from typing import Dict, List, Any
from datetime import datetime
import numpy as np

class MarketDataNormalizer:
    """Converts exchange-specific data formats to unified schema"""
    
    @staticmethod
    def normalize_trade(trade: Dict, source: str) -> Dict:
        """Normalize trade data to unified format"""
        
        # HolySheep already provides normalized format
        if source == "holysheep":
            return {
                "timestamp": pd.to_datetime(trade["timestamp"], unit="ms"),
                "symbol": trade["symbol"],
                "price": float(trade["price"]),
                "volume": float(trade["volume"]),
                "side": trade.get("side", "buy"),
                "source": "holysheep"
            }
        
        # Binance format
        if source == "binance":
            return {
                "timestamp": pd.to_datetime(trade["T"], unit="ms"),
                "symbol": trade["s"],
                "price": float(trade["p"]),
                "volume": float(trade["q"]),
                "side": "buy" if trade["m"] else "sell",  # m = buyer is maker
                "source": "binance"
            }
        
        # Bybit format
        if source == "bybit":
            return {
                "timestamp": pd.to_datetime(trade["trade_time_ms"], unit="ms"),
                "symbol": trade["symbol"],
                "price": float(trade["price"]),
                "volume": float(trade["size"]),
                "side": trade["side"].lower(),
                "source": "bybit"
            }
        
        # OKX format
        if source == "okx":
            return {
                "timestamp": pd.to_datetime(int(trade[3]), unit="ms"),
                "symbol": trade[3],  # instId
                "price": float(trade[4]),
                "volume": float(trade[5]),
                "side": trade[6].lower(),
                "source": "okx"
            }
        
        raise ValueError(f"Unknown source: {source}")
    
    @staticmethod
    def normalize_orderbook(book: Dict, source: str, symbol: str) -> Dict:
        """Normalize order book data"""
        
        if source == "holysheep":
            return {
                "timestamp": datetime.now(),
                "symbol": symbol,
                "bids": [(float(b[0]), float(b[1])) for b in book.get("bids", [])],
                "asks": [(float(a[0]), float(a[1])) for a in book.get("asks", [])],
                "source": "holysheep"
            }
        
        # Generic normalization
        return {
            "timestamp": datetime.now(),
            "symbol": symbol,
            "bids": book.get("bids", [])[:20],
            "asks": book.get("asks", [])[:20],
            "source": source
        }

class TradingDataStore:
    """In-memory store with rolling window for recent market data"""
    
    def __init__(self, window_size=10000):
        self.window_size = window_size
        self.trades = defaultdict(list)
        self.orderbooks = defaultdict(list)
        self.normalizer = MarketDataNormalizer()
    
    def add_trade(self, trade: Dict, source: str):
        """Add normalized trade to rolling window"""
        normalized = self.normalizer.normalize_trade(trade, source)
        self.trades[normalized["symbol"]].append(normalized)
        
        # Maintain window size
        if len(self.trades[normalized["symbol"]]) > self.window_size:
            self.trades[normalized["symbol"]] = self.trades[normalized["symbol"]][-self.window_size:]
    
    def get_recent_trades(self, symbol: str, n: int = 100) -> pd.DataFrame:
        """Get recent N trades as DataFrame"""
        trades = self.trades.get(symbol, [])[-n:]
        if not trades:
            return pd.DataFrame()
        return pd.DataFrame(trades)
    
    def calculate_vwap(self, symbol: str, window_minutes: int = 5) -> float:
        """Calculate Volume-Weighted Average Price"""
        df = self.get_recent_trades(symbol, n=10000)
        if df.empty:
            return 0.0
        
        cutoff = datetime.now() - pd.Timedelta(minutes=window_minutes)
        df = df[df["timestamp"] >= cutoff]
        
        if df.empty:
            return 0.0
        
        return np.average(df["price"], weights=df["volume"])
    
    def get_mid_price(self, symbol: str) -> float:
        """Get current mid price from latest order book"""
        books = self.orderbooks.get(symbol, [])
        if not books:
            return 0.0
        
        latest = books[-1]
        if not latest["bids"] or not latest["asks"]:
            return 0.0
        
        best_bid = latest["bids"][0][0]
        best_ask = latest["asks"][0][0]
        return (best_bid + best_ask) / 2


Usage example

store = TradingDataStore(window_size=50000)


Simulate adding data from multiple sources

sample_trade_holysheep = {
    "timestamp": 1704067200000,
    "symbol": "BTCUSDT",
    "price": 42500.50,
    "volume": 0.5432,
    "side": "buy"
}

store.add_trade(sample_trade_holysheep, "holysheep")
print(f"Stored trades count: {len(store.trades['BTCUSDT'])}")
print(f"VWAP (5min): ${store.calculate_vwap('BTCUSDT', 5):.2f}")
print(f"Mid price: ${store.get_mid_price('BTCUSDT'):.2f}")

Building a Simple Alpha Signal with Multi-Source Data

# Alpha Signal Generation using Multi-Source Market Data

Combines data from multiple exchanges to generate trading signals


import pandas as pd
import numpy as np
from typing import Tuple, Optional
import requests
import time

class AlphaSignalGenerator:
    """
    Generates trading signals by combining:
    - HolySheep AI unified market data
    - Direct exchange feeds
    - On-chain metrics (via HolySheep alternative data)
    """
    
    def __init__(self, holysheep_api_key: str):
        self.api_key = holysheep_api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.position = 0  # 1 = long, -1 = short, 0 = neutral
        self.signal_strength = 0.0
    
    def fetch_features(self, symbol: str) -> Dict:
        """Fetch all features needed for signal generation"""
        
        headers = {"Authorization": f"Bearer {self.api_key}"}
        
        # HolySheep AI unified data endpoint
        response = requests.post(
            f"{self.base_url}/alpha/features",
            json={"symbol": symbol, "include_onchain": True},
            headers=headers,
            timeout=10
        )
        
        if response.status_code != 200:
            return {}
        
        return response.json()
    
    def calculate_momentum(self, prices: pd.Series, period: int = 20) -> float:
        """Calculate momentum indicator"""
        if len(prices) < period:
            return 0.0
        return (prices.iloc[-1] - prices.iloc[-period]) / prices.iloc[-period]
    
    def calculate_volatility(self, prices: pd.Series, period: int = 20) -> float:
        """Calculate realized volatility (annualized)"""
        if len(prices) < period:
            return 0.0
        returns = prices.pct_change().dropna()
        return returns.std() * np.sqrt(365 * 24 * 60)  # Annualized
    
    def calculate_spread_zscore(self, pair_a: str, pair_b: str) -> float:
        """
        Calculate z-score of spread between two correlated pairs
        Used for pairs trading strategies
        """
        # Fetch data for both pairs
        headers = {"Authorization": f"Bearer {self.api_key}"}
        
        response = requests.post(
            f"{self.base_url}/market/quotes",
            json={"symbols": [pair_a, pair_b], "history": 100},
            headers=headers,
            timeout=10
        )
        
        if response.status_code != 200:
            return 0.0
        
        data = response.json()
        prices_a = pd.Series(data[pair_a]["prices"])
        prices_b = pd.Series(data[pair_b]["prices"])
        
        # Calculate spread
        spread = prices_a - prices_b * (prices_a.mean() / prices_b.mean())
        spread_mean = spread.mean()
        spread_std = spread.std()
        
        if spread_std == 0:
            return 0.0
        
        zscore = (spread.iloc[-1] - spread_mean) / spread_std
        return zscore
    
    def generate_signal(self, symbol: str) -> Tuple[str, float]:
        """
        Generate trading signal for a symbol
        Returns: (signal_type, confidence)
        signal_type: "BUY", "SELL", or "HOLD"
        confidence: 0.0 to 1.0
        """
        
        features = self.fetch_features(symbol)
        
        if not features:
            return "HOLD", 0.0
        
        # Momentum signal
        momentum = self.calculate_momentum(
            pd.Series(features.get("close_prices", []))
        )
        
        # Volatility signal
        volatility = self.calculate_volatility(
            pd.Series(features.get("close_prices", []))
        )
        
        # Sentiment score (from alternative data)
        sentiment = features.get("sentiment_score", 0.5)
        
        # Funding rate differential (for crypto perpetual)
        funding_diff = features.get("funding_rate_diff", 0.0)
        
        # Combine signals
        score = 0.0
        score += momentum * 0.4
        score += (sentiment - 0.5) * 2 * 0.3  # Normalize sentiment
        score += funding_diff * 0.3
        
        # Volatility adjustment
        if volatility > 0.5:  # High volatility
            score *= 0.7
        
        # Determine signal
        if score > 0.3:
            return "BUY", min(abs(score), 1.0)
        elif score < -0.3:
            return "SELL", min(abs(score), 1.0)
        else:
            return "HOLD", abs(score)
    
    def execute_signal(self, symbol: str, signal: str, confidence: float):
        """Execute signal based on confidence threshold"""
        
        if confidence < 0.5:
            print(f"[{symbol}] Signal {signal} confidence {confidence:.2f} below threshold")
            return
        
        print(f"[{symbol}] Executing {signal} with confidence {confidence:.2f}")
        
        # Here you would integrate with your brokerage/exchange API
        # Example: binance_client.place_order(symbol, signal)
        
        self.position = 1 if signal == "BUY" else (-1 if signal == "SELL" else 0)
        self.signal_strength = confidence


Initialize and run signal generation

api_key = "YOUR_HOLYSHEEP_API_KEY"
generator = AlphaSignalGenerator(api_key)


Generate signals for multiple symbols

symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT"]

for symbol in symbols:
    try:
        signal, confidence = generator.generate_signal(symbol)
        print(f"\n{symbol}: {signal} (confidence: {confidence:.2f})")
        
        # Auto-execute if confidence is high enough
        generator.execute_signal(symbol, signal, confidence)
        
    except Exception as e:
        print(f"Error generating signal for {symbol}: {e}")
    
    # Rate limiting - HolySheep free tier allows reasonable usage
    time.sleep(0.5)

Common Errors and Fixes

After implementing data pipelines for dozens of trading systems, I've encountered virtually every error. Here's how to resolve them quickly.

Error 1: "401 Unauthorized" on API Calls

Symptom: All API requests return 401 status with {"error": "Unauthorized", "message": "Invalid API key"}

Cause: The API key is missing, expired, or incorrectly formatted

Fix:

# CORRECT API Key Usage
import os


Method 1: Environment variable (recommended for production)

api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
    raise ValueError("HOLYSHEEP_API_KEY environment variable not set")


Method 2: Direct assignment (for testing)

api_key = "YOUR_HOLYSHEEP_API_KEY"  # Must match exactly as provided


Verify key format - HolySheep keys are typically 32+ characters

if len(api_key) < 32:
    print("WARNING: API key seems too short, verify it from dashboard")


Correct headers format

headers = {
    "Authorization": f"Bearer {api_key}",  # Note the "Bearer " prefix
    "Content-Type": "application/json"
}


Test authentication

import requests
response = requests.get(
    "https://api.holysheep.ai/v1/auth/verify",
    headers=headers,
    timeout=10
)

if response.status_code == 200:
    print("Authentication successful!")
else:
    print(f"Auth failed: {response.status_code} - {response.text}")

Error 2: "ConnectionError: timeout after 30000ms" on WebSocket

Symptom: WebSocket connections timeout, especially during high-volatility periods

Cause: Network congestion, exchange rate limiting, or server overload

Fix:

# WebSocket Timeout and Reconnection Handler
import websocket
import threading
import time
from datetime import datetime

class ResilientWebSocket:
    def __init__(self, url, api_key):
        self.url = url
        self.api_key = api_key
        self.ws = None
        self.reconnect_attempts = 0
        self.max_attempts = 10
        self.base_delay = 1
        self.max_delay = 60
        
    def connect(self):
        """Connect with timeout handling"""
        headers = [f"Authorization: Bearer {self.api_key}"]
        
        self.ws = websocket.WebSocketApp(
            self.url,
            header=headers,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close,
            on_open=self.on_open
        )
        
        # Run in thread with timeout
        thread = threading.Thread(target=self._run_with_timeout)
        thread.daemon = True
        thread.start()
        
    def _run_with_timeout(self):
        try:
            self.ws.run_forever(
                ping_timeout=30,           # Ping/pong timeout
                ping_interval=20,          # Send ping every 20 seconds
                max_queue=1024,             # Message queue size
                sslopt={"cert_reqs": False}  # SSL settings
            )
        except Exception as e:
            print(f"WebSocket error: {e}")
            self._attempt_reconnect()
    
    def _attempt_reconnect(self):
        """Exponential backoff reconnection"""
        if self.reconnect_attempts >= self.max_attempts:
            print("Max reconnection attempts reached")
            return
        
        delay = min(
            self.base_delay * (2 ** self.reconnect_attempts),
            self.max_delay
        )
        
        print(f"Reconnecting in {delay} seconds (attempt {self.reconnect_attempts + 1})")
        time.sleep(delay)
        
        self.reconnect_attempts += 1
        self.ws.close()
        self.connect()
    
    def on_open(self, ws):
        print(f"[{datetime.now()}] WebSocket connected")
        self.reconnect_attempts = 0  # Reset on success
        
        # Subscribe to channels
        subscribe_msg = {
            "type": "subscribe",
            "channels": ["trades", "orderbook"],
            "symbols": ["BTCUSDT", "ETHUSDT"]
        }
        ws.send(json.dumps(subscribe_msg))
    
    def on_message(self, ws, message):
        print(f"Received: {message[:100]}...")
    
    def on_error(self, ws, error):
        print(f"Error: {error}")
        if "timeout" in str(error).lower():
            self._attempt_reconnect()
    
    def on_close(self, ws, close_status_code, close_msg):
        print(f"Closed: {close_status_code} - {close_msg}")


Usage

ws = ResilientWebSocket(
    "wss://api.holysheep.ai/v1/stream",
    "YOUR_HOLYSHEEP_API_KEY"
)
ws.connect()


Keep running

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    ws.ws.close()
    print("Disconnected")

Error 3: "429 Rate Limit Exceeded" During High-Frequency Polling

Symptom: API returns 429 errors, data requests fail intermittently

Cause: Exceeding API rate limits (usually requests per minute)

Fix:

# Rate Limiting Implementation with Exponential Backoff
import time
import threading
from collections import deque
from datetime import datetime, timedelta
import requests

class RateLimitedClient:
    def __init__(self, api_key, requests_per_minute=60):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.rpm_limit = requests_per_minute
        self.request_times = deque(maxlen=requests_per_minute)
        self.lock = threading.Lock()
        
    def _wait_if_needed(self):
        """Ensure we don't exceed rate limit"""
        now = datetime.now()
        cutoff = now - timedelta(minutes=1)
        
        with self.lock:
            # Remove old requests from tracking
            while self.request_times and self.request_times[0] < cutoff:
                self.request_times.popleft()
            
            # If at limit, wait
            if len(self.request_times) >= self.rpm_limit:
                sleep_time = (self.request_times[0] - cutoff).total_seconds() + 0.1
                print(f"Rate limit reached, sleeping {sleep_time:.2f}s")
                time.sleep(sleep_time)
                self._wait_if_needed()  # Recursively check again
            
            self.request_times.append(now)
    
    def get_with_retry(self, endpoint, max_retries=5):
        """GET request with rate limiting and retry logic"""
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        for attempt in range(max_retries):
            self._wait_if_needed()
            
            try:
                response = requests.get(
                    f"{self.base_url}{endpoint}",
                    headers=headers,
                    timeout=10
                )
                
                if response.status_code == 200:
                    return response.json()
                elif response.status_code == 429:
                    # Hit rate limit, exponential backoff
                    wait_time = 2 ** attempt
                    print(f"429 Rate Limited, waiting {wait_time}s...")
                    time.sleep(wait_time)
                elif response.status_code == 401:
                    raise Exception("401 Unauthorized - check API key")
                else:
                    print(f"Error {response.status_code}: {response.text}")
                    
            except requests.exceptions.Timeout:
                print(f"Timeout on attempt {attempt + 1}, retrying...")
                time.sleep(2 ** attempt)
            except requests.exceptions.ConnectionError as e:
                print(f"Connection error: {e}, retrying...")
                time.sleep(5)
        
        raise Exception(f"Failed after {max_retries} attempts")
    
    def post_with_retry(self, endpoint, payload, max_retries=5):
        """POST request with rate limiting and retry logic"""
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        for attempt in range(max_retries):
            self._wait_if_needed()
            
            try:
                response = requests.post(
                    f"{self.base_url}{endpoint}",
                    json=payload,
                    headers=headers,
                    timeout=15
                )
                
                if response.status_code == 200:
                    return response.json()
                elif response.status_code == 429:
                    wait_time = 2 ** attempt
                    print(f"Rate limited, backing off {wait_time}s...")
                    time.sleep(w

Related Resources
Related Articles