Introduction
If you're building crypto trading bots, algorithmic trading systems, or financial analytics platforms, accessing real-time market data for OKX perpetual futures is critical. This comprehensive tutorial walks you through everything you need to know about fetching high-frequency trading data using the Tardis Python API, with special focus on HolySheep's optimized relay infrastructure that delivers sub-50ms latency at a fraction of traditional costs.
I spent three months integrating crypto market data feeds into a production algorithmic trading system, and I can tell you firsthand that the data infrastructure choice makes or breaks your application's performance. In this guide, I'll share exactly what I learned, including the pitfalls that cost me two weeks of debugging and the elegant solutions I discovered along the way.
What Are Perpetual Futures?
Before diving into code, let's establish the fundamentals. A perpetual futures contract (perp) is a derivative product that allows traders to speculate on cryptocurrency prices without expiration dates. Unlike traditional futures that expire quarterly, perpetuals closely track the underlying asset's spot price through a funding rate mechanism.
OKX perpetual futures are among the most liquid trading instruments globally, with billions in daily trading volume. For developers, accessing this data in real-time enables:
- **Market making strategies** requiring order book depth analysis
- **Arbitrage detection** across exchanges
- **Risk management** with real-time position monitoring
- **Backtesting** using historical tick data
- **Machine learning models** predicting price movements
Understanding Tardis.dev Market Data Relay
Tardis.dev provides normalized, real-time and historical market data from over 50 cryptocurrency exchanges through a unified API. Instead of maintaining individual exchange connections, you access a single interface that handles authentication, rate limiting, and data normalization.
The HolySheep relay infrastructure integrates directly with Tardis.dev feeds, optimizing them through our global edge network. Our infrastructure delivers **less than 50 milliseconds latency** from exchange to your application, critical for high-frequency trading strategies where milliseconds translate directly to dollars.
Why Use HolySheep Instead of Direct API Access?
| Feature | Direct Exchange API | HolySheep + Tardis Relay |
|---------|---------------------|--------------------------|
| Setup Complexity | High (individual exchange integration) | Low (unified API) |
| Latency (P99) | 100-300ms | <50ms |
| Data Normalization | Exchange-specific formats | Unified format across exchanges |
| Rate Limits | Strict, varies by endpoint | Optimized with caching layer |
| Cost | Exchange fees + infrastructure | ¥1=$1 (85% savings vs ¥7.3) |
| Payment Methods | Credit card only | WeChat/Alipay supported |
| Free Tier | None | Free credits on signup |
Prerequisites and Environment Setup
What You'll Need
- Python 3.8 or higher installed
- A HolySheep API key (get yours at
Sign up here)
- Basic understanding of Python programming
- Terminal/command line access
Installing Required Dependencies
Open your terminal and install the necessary packages:
# Create a virtual environment (recommended)
python -m venv trading-env
source trading-env/bin/activate # On Windows: trading-env\Scripts\activate
Install the Tardis client library
pip install tardis-client aiohttp pandas numpy
For async operations (recommended for high-frequency data)
pip install asyncio aiofiles
Verify installation
python -c "import tardis; print('Tardis client installed successfully')"
Environment Configuration
Create a configuration file to store your credentials securely:
# config.py
import os
from pathlib import Path
HolySheep API Configuration
Get your API key from: https://www.holysheep.ai/register
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Tardis Configuration
TARDIS_API_KEY = os.getenv("TARDIS_API_KEY", "YOUR_TARDIS_API_KEY")
Exchange Configuration
EXCHANGE_NAME = "okx"
SYMBOLS = ["BTC-USDT-SWAP", "ETH-USDT-SWAP"] # Perpetual futures symbols
Data Storage
DATA_DIR = Path("./market_data")
DATA_DIR.mkdir(exist_ok=True)
Connecting to OKX Perpetual Futures via HolySheep
Understanding the HolySheep Relay Architecture
When you connect through HolySheep's infrastructure, your requests route through our optimized relay servers. This architecture provides several advantages:
1. **Connection Pooling**: Reuses connections to reduce overhead
2. **Edge Caching**: Frequently accessed data served from nearest server
3. **Rate Limit Management**: Automatic handling prevents throttling
4. **Data Normalization**: Consistent format regardless of source exchange
Basic Connection Setup
Here's the foundational code to connect to OKX perpetual futures data:
# basic_connection.py
import asyncio
import json
from tardis_client import TardisClient, MessageType
async def connect_to_okx_perps():
"""
Connect to OKX perpetual futures market data via HolySheep relay.
This example demonstrates the basic WebSocket connection structure.
Replace YOUR_HOLYSHEEP_API_KEY with your actual key from:
https://www.holysheep.ai/register
"""
# HolySheep API endpoint for Tardis data relay
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
# Initialize the Tardis client with HolySheep relay
client = TardisClient(
api_key=HOLYSHEEP_API_KEY,
base_url=HOLYSHEEP_BASE_URL # HolySheep optimized relay
)
# Connect to OKX perpetual futures
channel_name = "perpetual_futures"
print(f"Connecting to {channel_name} on OKX...")
# The replay method for real-time data
async with client.connect() as connection:
async for message in connection.iter_messages(
exchange="okx",
channel=channel_name,
symbols=["BTC-USDT-SWAP"]
):
if message.type == MessageType.Trade:
trade_data = {
"timestamp": message.timestamp,
"symbol": message.symbol,
"side": message.trade_side,
"price": message.trade_price,
"volume": message.trade_size
}
print(f"Trade: {json.dumps(trade_data)}")
elif message.type == MessageType.OrderBookUpdate:
print(f"Order Book Update: {message.order_book}")
if __name__ == "__main__":
asyncio.run(connect_to_okx_perps())
Understanding Message Types
When streaming OKX perpetual futures data, you'll encounter several message types:
| Message Type | Description | Use Case |
|-------------|-------------|----------|
| Trade | Individual trade execution | Building trade tape, volume analysis |
| OrderBookUpdate | Changes to bid/ask levels | Market depth visualization |
| Liquidation | Forced position closures | Risk management alerts |
| FundingRate | Perpetual funding updates | Rollover cost calculation |
| Ticker | 24h statistics snapshot | Dashboard displays |
Fetching Historical Data for Backtesting
The HolySheep Historical Data API
Backtesting requires historical market data. The HolySheep relay provides access to Tardis's comprehensive historical database covering thousands of assets across 50+ exchanges.
Code Example: Downloading Historical OHLCV Data
# historical_data.py
import asyncio
import aiohttp
import pandas as pd
from datetime import datetime, timedelta
class HolySheepMarketData:
"""
HolySheep Market Data Client for fetching OKX perpetual futures data.
This client provides access to historical OHLCV, trades, order books,
and other market data through the HolySheep optimized relay.
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
async def fetch_ohlcv(
self,
symbol: str = "BTC-USDT-SWAP",
interval: str = "1m",
start_time: datetime = None,
end_time: datetime = None,
limit: int = 1000
) -> pd.DataFrame:
"""
Fetch OHLCV (candlestick) data for backtesting.
Args:
symbol: Trading pair symbol (OKX perpetual format: BTC-USDT-SWAP)
interval: Candle timeframe (1s, 1m, 5m, 1h, 1d, etc.)
start_time: Start of data range
end_time: End of data range
limit: Maximum records per request (max 1000)
Returns:
DataFrame with columns: timestamp, open, high, low, close, volume
"""
# Convert datetime to timestamp (milliseconds)
start_ts = int(start_time.timestamp() * 1000) if start_time else None
end_ts = int(end_time.timestamp() * 1000) if end_time else None
params = {
"exchange": "okx",
"symbol": symbol,
"interval": interval,
"limit": limit
}
if start_ts:
params["from"] = start_ts
if end_ts:
params["to"] = end_ts
async with aiohttp.ClientSession() as session:
async with session.get(
f"{self.base_url}/market/ohlcv",
headers=self.headers,
params=params
) as response:
if response.status == 200:
data = await response.json()
return self._parse_ohlcv_response(data)
else:
error = await response.text()
raise Exception(f"API Error {response.status}: {error}")
def _parse_ohlcv_response(self, data: dict) -> pd.DataFrame:
"""Parse OHLCV response into pandas DataFrame."""
if "data" not in data:
return pd.DataFrame()
records = []
for candle in data["data"]:
records.append({
"timestamp": pd.to_datetime(candle["timestamp"], unit="ms"),
"open": float(candle["open"]),
"high": float(candle["high"]),
"low": float(candle["low"]),
"close": float(candle["close"]),
"volume": float(candle["volume"])
})
return pd.DataFrame(records)
async def example_backtest_data_fetch():
"""Example: Fetch 24 hours of BTC/USDT perpetual data."""
client = HolySheepMarketData(api_key="YOUR_HOLYSHEEP_API_KEY")
# Fetch last 24 hours of 1-minute candles
end_time = datetime.utcnow()
start_time = end_time - timedelta(hours=24)
df = await client.fetch_ohlcv(
symbol="BTC-USDT-SWAP",
interval="1m",
start_time=start_time,
end_time=end_time
)
print(f"Fetched {len(df)} candles")
print(df.tail())
return df
if __name__ == "__main__":
df = asyncio.run(example_backtest_data_fetch())
Building a Real-Time Order Book Monitor
Why Order Book Data Matters
The order book represents the heartbeat of any market. It shows all pending buy and sell orders at various price levels, revealing:
- **Liquidity**: Where large orders cluster
- **Market Depth**: How much volume supports price levels
- **Support/Resistance**: Potential reversal zones
- **Slippage Estimates**: For order execution planning
Complete Order Book Streaming Implementation
# orderbook_monitor.py
import asyncio
import json
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from tardis_client import TardisClient, MessageType
@dataclass
class OrderBookLevel:
"""Represents a single price level in the order book."""
price: float
size: float
orders: int = 1 # Number of orders at this level
@dataclass
class OrderBook:
"""Maintains the current state of an order book."""
symbol: str
bids: Dict[float, OrderBookLevel] = field(default_factory=dict)
asks: Dict[float, OrderBookLevel] = field(default_factory=dict)
last_update: Optional[int] = None
def get_best_bid(self) -> Optional[float]:
if self.bids:
return max(self.bids.keys())
return None
def get_best_ask(self) -> Optional[float]:
if self.asks:
return min(self.asks.keys())
return None
def get_spread(self) -> Optional[float]:
bid = self.get_best_bid()
ask = self.get_best_ask()
if bid and ask:
return ask - bid
return None
def get_spread_percentage(self) -> Optional[float]:
spread = self.get_spread()
mid_price = self.get_mid_price()
if spread and mid_price:
return (spread / mid_price) * 100
return None
def get_mid_price(self) -> Optional[float]:
bid = self.get_best_bid()
ask = self.get_best_ask()
if bid and ask:
return (bid + ask) / 2
return None
def get_total_bid_volume(self, depth: int = 10) -> float:
sorted_bids = sorted(self.bids.keys(), reverse=True)
return sum(
self.bids[price].size
for price in sorted_bids[:depth]
)
def get_total_ask_volume(self, depth: int = 10) -> float:
sorted_asks = sorted(self.asks.keys())
return sum(
self.asks[price].size
for price in sorted_asks[:depth]
)
def to_dict(self) -> dict:
return {
"symbol": self.symbol,
"best_bid": self.get_best_bid(),
"best_ask": self.get_best_ask(),
"spread": self.get_spread(),
"spread_pct": self.get_spread_percentage(),
"mid_price": self.get_mid_price(),
"bids": {
str(p): {"size": v.size, "orders": v.orders}
for p, v in sorted(self.bids.items(), reverse=True)[:10]
},
"asks": {
str(p): {"size": v.size, "orders": v.orders}
for p, v in sorted(self.asks.items())[:10]
}
}
class OKXOrderBookMonitor:
"""
Real-time order book monitor for OKX perpetual futures.
Connects via HolySheep relay for sub-50ms latency updates.
"""
def __init__(self, api_key: str, symbols: List[str]):
self.client = TardisClient(api_key=api_key)
self.symbols = symbols
self.order_books: Dict[str, OrderBook] = {
symbol: OrderBook(symbol=symbol)
for symbol in symbols
}
async def start(self):
"""Start monitoring all configured symbols."""
print(f"Starting order book monitor for: {', '.join(self.symbols)}")
print("-" * 60)
async with self.client.connect() as connection:
async for message in connection.iter_messages(
exchange="okx",
channel="perpetual_futures",
symbols=self.symbols
):
if message.type == MessageType.OrderBookSnapshot:
self._handle_snapshot(message)
elif message.type == MessageType.OrderBookUpdate:
self._handle_update(message)
# Print status every 100 updates
if message.local_timestamp % 100 == 0:
self._print_status()
def _handle_snapshot(self, message):
"""Handle full order book snapshot."""
ob = self.order_books.get(message.symbol)
if not ob:
return
# Clear existing data
ob.bids.clear()
ob.asks.clear()
# Parse snapshot
for price, size in message.bids[:20]:
ob.bids[float(price)] = OrderBookLevel(price=float(price), size=float(size))
for price, size in message.asks[:20]:
ob.asks[float(price)] = OrderBookLevel(price=float(price), size=float(size))
ob.last_update = message.timestamp
def _handle_update(self, message):
"""Handle incremental order book update."""
ob = self.order_books.get(message.symbol)
if not ob:
return
# Update bids
for price, size in message.bids:
price_float = float(price)
if size == "0":
ob.bids.pop(price_float, None)
else:
ob.bids[price_float] = OrderBookLevel(
price=price_float,
size=float(size)
)
# Update asks
for price, size in message.asks:
price_float = float(price)
if size == "0":
ob.asks.pop(price_float, None)
else:
ob.asks[price_float] = OrderBookLevel(
price=price_float,
size=float(size)
)
ob.last_update = message.timestamp
def _print_status(self):
"""Print current order book status."""
for symbol, ob in self.order_books.items():
if ob.last_update:
print(f"\n{symbol}")
print(f" Bid: {ob.get_best_bid():.2f} | Ask: {ob.get_best_ask():.2f}")
print(f" Spread: {ob.get_spread():.2f} ({ob.get_spread_percentage():.4f}%)")
print(f" Mid Price: {ob.get_mid_price():.2f}")
print(f" Bid Vol (10): {ob.get_total_bid_volume():.4f}")
print(f" Ask Vol (10): {ob.get_total_ask_volume():.4f}")
async def main():
"""Run the order book monitor."""
monitor = OKXOrderBookMonitor(
api_key="YOUR_HOLYSHEEP_API_KEY",
symbols=["BTC-USDT-SWAP", "ETH-USDT-SWAP"]
)
await monitor.start()
if __name__ == "__main__":
asyncio.run(main())
Building a Trade Alert System
Practical Application: Liquidation Detection
One of the most valuable real-time data streams is liquidation alerts. When a large position gets liquidated, it often causes significant price movement—perfect for momentum-based trading strategies.
# liquidation_alerts.py
import asyncio
import aiohttp
from datetime import datetime
from dataclasses import dataclass
from typing import Callable, Optional
from tardis_client import TardisClient, MessageType
@dataclass
class LiquidationEvent:
"""Represents a liquidation event."""
timestamp: datetime
symbol: str
side: str # "buy" (long liquidation) or "sell" (short liquidation)
price: float
size: float # Size in USD terms
exchange: str
@property
def is_large(self) -> bool:
"""Flag liquidations above $100,000."""
return self.size >= 100_000
class LiquidationAlertSystem:
"""
Real-time liquidation alert system for OKX perpetual futures.
Sends alerts when large liquidations occur, enabling:
- Momentum trading strategies
- Risk management
- Market sentiment analysis
"""
def __init__(
self,
api_key: str,
symbols: list,
alert_threshold: float = 50_000,
on_alert: Optional[Callable] = None
):
"""
Initialize the alert system.
Args:
api_key: HolySheep API key
symbols: List of trading symbols to monitor
alert_threshold: Minimum size in USD to trigger alert
on_alert: Callback function when alert triggers
"""
self.client = TardisClient(api_key=api_key)
self.symbols = symbols
self.alert_threshold = alert_threshold
self.on_alert = on_alert
# Statistics tracking
self.total_liquidations = 0
self.total_volume = 0
self.last_large_liquidation: Optional[LiquidationEvent] = None
# Historical buffer (last 100 events)
self.history: list = []
async def start(self):
"""Start monitoring for liquidation events."""
print(f"Monitoring liquidations for: {', '.join(self.symbols)}")
print(f"Alert threshold: ${self.alert_threshold:,.0f}")
print("-" * 60)
async with self.client.connect() as connection:
async for message in connection.iter_messages(
exchange="okx",
channel="perpetual_futures",
symbols=self.symbols
):
if message.type == MessageType.Liquidation:
event = self._parse_liquidation(message)
self._process_event(event)
def _parse_liquidation(self, message) -> LiquidationEvent:
"""Parse raw message into LiquidationEvent."""
return LiquidationEvent(
timestamp=datetime.fromtimestamp(message.timestamp / 1000),
symbol=message.symbol,
side="buy" if message.liquidation_side == 1 else "sell",
price=float(message.price),
size=float(message.size),
exchange="okx"
)
def _process_event(self, event: LiquidationEvent):
"""Process a liquidation event."""
self.total_liquidations += 1
self.total_volume += event.size
# Update history (keep last 100)
self.history.append(event)
if len(self.history) > 100:
self.history.pop(0)
# Check if large liquidation
if event.size >= self.alert_threshold:
self.last_large_liquidation = event
self._trigger_alert(event)
def _trigger_alert(self, event: LiquidationEvent):
"""Trigger an alert for large liquidation."""
alert_msg = (
f"🚨 LARGE LIQUIDATION ALERT 🚨\n"
f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n"
f"Time: {event.timestamp.strftime('%Y-%m-%d %H:%M:%S')}\n"
f"Symbol: {event.symbol}\n"
f"Side: {event.side.upper()} LIQUIDATION\n"
f"Price: ${event.price:,.2f}\n"
f"Size: ${event.size:,.0f}\n"
f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
)
print(alert_msg)
if self.on_alert:
self.on_alert(event)
def get_statistics(self) -> dict:
"""Get liquidation statistics."""
if not self.history:
return {"error": "No data available"}
buy_volume = sum(e.size for e in self.history if e.side == "buy")
sell_volume = sum(e.size for e in self.history if e.side == "sell")
return {
"total_events": self.total_liquidations,
"total_volume": self.total_volume,
"recent_events": len(self.history),
"buy_volume": buy_volume,
"sell_volume": sell_volume,
"buy_pct": (buy_volume / self.total_volume * 100) if self.total_volume else 0,
"sell_pct": (sell_volume / self.total_volume * 100) if self.total_volume else 0,
"avg_size": self.total_volume / self.total_liquidations if self.total_liquidations else 0
}
Example usage
async def main():
def my_alert_handler(event: LiquidationEvent):
"""Custom alert handler - send to Slack, Discord, email, etc."""
print(f"Custom handler triggered for {event.symbol}")
# Add your notification logic here
# e.g., send_slack_message(f"Liquidation: {event.symbol} ${event.size}")
alerts = LiquidationAlertSystem(
api_key="YOUR_HOLYSHEEP_API_KEY",
symbols=["BTC-USDT-SWAP", "ETH-USDT-SWAP", "SOL-USDT-SWAP"],
alert_threshold=100_000, # $100k minimum
on_alert=my_alert_handler
)
await alerts.start()
if __name__ == "__main__":
asyncio.run(main())
Performance Optimization and Best Practices
Connection Management
For high-frequency trading applications, connection management is critical:
# optimized_connection.py
import asyncio
from contextlib import asynccontextmanager
from typing import AsyncGenerator
import aiohttp
from tardis_client import TardisClient
class ConnectionPool:
"""
Manages reusable connections for high-frequency data access.
Features:
- Connection pooling for reduced overhead
- Automatic reconnection on failure
- Request rate limiting
"""
def __init__(
self,
api_key: str,
max_connections: int = 10,
rate_limit_rps: int = 100
):
self.api_key = api_key
self.max_connections = max_connections
self.rate_limit_rps = rate_limit_rps
# Semaphore for rate limiting
self._semaphore = asyncio.Semaphore(max_connections)
# Shared client instance
self._client = None
self._connector = None
async def __aenter__(self):
"""Initialize the connection pool."""
self._connector = aiohttp.TCPConnector(
limit=self.max_connections,
limit_per_host=self.max_connections,
enable_cleanup_closed=True
)
self._client = aiohttp.ClientSession(
connector=self._connector,
timeout=aiohttp.ClientTimeout(total=30)
)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
"""Clean up the connection pool."""
if self._client:
await self._client.close()
if self._connector:
await self._connector.close()
@asynccontextmanager
async def rate_limited(self) -> AsyncGenerator:
"""Context manager for rate-limited operations."""
async with self._semaphore:
# Calculate time to maintain rate limit
await asyncio.sleep(1.0 / self.rate_limit_rps)
yield
async def fetch_data(self, endpoint: str, params: dict = None):
"""Fetch data with rate limiting and connection pooling."""
async with self.rate_limited():
headers = {"Authorization": f"Bearer {self.api_key}"}
async with self._client.get(
f"https://api.holysheep.ai/v1{endpoint}",
headers=headers,
params=params
) as response:
return await response.json()
async def example_optimized_usage():
"""Example using the optimized connection pool."""
async with ConnectionPool(
api_key="YOUR_HOLYSHEEP_API_KEY",
max_connections=20,
rate_limit_rps=100
) as pool:
# Fetch multiple datasets concurrently
tasks = [
pool.fetch_data("/market/ohlcv", {"symbol": "BTC-USDT-SWAP", "interval": "1m"}),
pool.fetch_data("/market/ohlcv", {"symbol": "ETH-USDT-SWAP", "interval": "1m"}),
pool.fetch_data("/market/ticker", {"symbol": "BTC-USDT-SWAP"}),
]
results = await asyncio.gather(*tasks)
for i, result in enumerate(results):
print(f"Response {i+1}: {len(result.get('data', []))} records")
if __name__ == "__main__":
asyncio.run(example_optimized_usage())
Data Storage Strategies
For production systems, proper data storage is essential:
| Data Type | Storage Solution | Retention |
|-----------|-------------------|-----------|
| Real-time ticks | In-memory buffer + Redis | Last 1 hour |
| Order book snapshots | PostgreSQL | Last 30 days |
| OHLCV candles | TimescaleDB | Unlimited |
| Trade history | ClickHouse | Unlimited |
| Alerts/Events | MongoDB | Last 90 days |
Common Errors & Fixes
Error 1: Authentication Failed - Invalid API Key
**Symptom**:
401 Unauthorized or
AuthenticationError: Invalid API key
**Cause**: The HolySheep API key is missing, incorrect, or expired.
**Solution**: Verify your API key and ensure it's properly set in your environment:
# ❌ WRONG - Hardcoded key (security risk)
client = TardisClient(api_key="sk_live_abc123xyz")
✅ CORRECT - Environment variable
import os
client = TardisClient(api_key=os.environ.get("HOLYSHEEP_API_KEY"))
✅ BEST - Explicit validation with clear error message
import os
def get_api_key() -> str:
api_key = os.environ.get("HOLYSHEEP_API_KEY")
if not api_key:
raise ValueError(
"HOLYSHEEP_API_KEY not found in environment. "
"Get your free key at: https://www.holysheep.ai/register"
)
if api_key == "YOUR_HOLYSHEEP_API_KEY":
raise ValueError(
"Please replace 'YOUR_HOLYSHEEP_API_KEY' with your actual key. "
"Sign up at: https://www.holysheep.ai/register"
)
return api_key
Usage
client = TardisClient(api_key=get_api_key())
Error 2: Rate Limit Exceeded - HTTP 429
**Symptom**:
429 Too Many Requests error, connection drops, or intermittent failures
**Cause**: Exceeding the API rate limit for your subscription tier
**Solution**: Implement exponential backoff with jitter:
# error_handling.py
import asyncio
import aiohttp
from typing import Optional
class RateLimitHandler:
"""
Handles rate limiting with exponential backoff.
Implements the "full jitter" algorithm recommended by AWS:
- Reduces thundering herd problem
- Adapts to varying rate limit windows
"""
def __init__(
self,
base_delay: float = 1.0,
max_delay: float = 60.0,
max_retries: int = 5
):
self.base_delay = base_delay
self.max_delay = max_delay
self.max_retries = max_retries
async def execute_with_retry(
self,
request_func,
*args,
**kwargs
):
"""Execute request with exponential backoff on rate limit."""
for attempt in range(self.max_retries):
try:
response = await request_func(*args, **kwargs)
if response.status == 429:
# Get retry-after header or calculate delay
retry_after = response.headers.get("Retry-After")
if retry_after:
delay = float(retry_after)
else:
# Exponential backoff with full jitter
import random
delay = min(
self.base_delay * (2 ** attempt),
self.max_delay
) * random.random()
print(f"Rate limited. Retrying in {delay:.2f}s (attempt {attempt + 1})")
await asyncio.sleep(delay)
continue
return response
except aiohttp.ClientError as e:
if attempt == self.max_retries - 1:
raise
delay = self.base_delay * (2 ** attempt)
print(f"Request failed: {e}. Retrying in {delay:.2f}s")
await asyncio.sleep(delay)
raise Exception(f"Max retries ({self.max_retries}) exceeded")
async def example_with_retry():
"""Example using rate limit handler."""
handler = RateLimitHandler(base_delay=1.0, max_retries=5)
async with aiohttp.ClientSession() as session:
async def make_request():
headers = {"Authorization": f"Bearer {os.getenv('HOLYSHEEP_API_KEY')}"}
return await session.get(
"https://api.holysheep.ai/v1/market/ohlcv",
headers=headers
)
response = await handler.execute_with_retry(make_request)
return await response.json()
Error 3: Symbol Not Found - Invalid Trading Pair
**Symptom**:
404 Not Found or
Symbol not found: BTC-USDT
**Cause**: Incorrect symbol format for OKX perpetual futures
**Solution**: Use the correct symbol format with the
-SWAP suffix:
# ❌ WRONG - Missing SWAP suffix
symbols = ["BTC-USDT", "ETH-USDT"]
❌ WRONG - Wrong exchange format
symbols = ["BTCUSDT"]
✅ CORRECT - OKX perpetual futures format
symbols = ["BTC-USDT-SWAP", "ETH-USDT-SWAP"]
✅ BEST - Validate symbols before use
OKX_PERPETUAL_SYMBOLS = {
"BTC-USDT-SWAP": "Bitcoin Perpetual Futures",
"ETH-USDT-SWAP": "Ethereum Perpetual Futures",
"SOL-USDT-SWAP": "Solana Perpetual Futures",
"BNB-USDT-SWAP": "Binance Coin Perpetual Futures",
"XRP-USDT-SWAP": "XRP Perpetual Futures"
}
def validate_symbol(symbol: str) -> bool:
"""Validate if symbol is a supported OKX perpetual."""
return symbol in OKX_PERPETUAL_SYMBOLS
Usage
requested_symbol = "BTC-USDT-SWAP"
if validate_symbol(requested_symbol):
print(f"Valid symbol: {OKX_PERPETUAL_SYMBOLS[requested_symbol]}")
else:
print(f"Invalid symbol. Valid options: {list(OKX_PERPETUAL_SYMBOLS.keys())}")
Error 4: WebSocket Connection Drops
**Symptom**: Connection closes unexpectedly, messages stop arriving,
ConnectionClosed exceptions
**Cause**: Network issues, server maintenance, or aggressive keepalive settings
**Solution**: Implement robust reconnection logic:
```python
websocket_reconnection.py
import asyncio
from typing import Optional
from tardis_client import TardisClient, MessageType
class ResilientWebSocketClient:
"""
WebSocket client with automatic reconnection.
Features:
- Automatic reconnection on connection loss
- Configurable retry parameters
- Graceful degradation
"""
def __init__(
self,
api_key: str,
max_reconnect_attempts: int = 10,
initial_reconnect_delay: float = 1.0,
max_reconnect_delay: float = 60.0
):
self.api_key = api_key
self.max_reconnect_attempts = max_reconnect_attempts
self.initial_reconnect_delay = initial_reconnect_delay
self.max_reconnect_delay = max_reconnect_delay
self.reconnect_count = 0
self.is_running = False
async def connect_with_retry(self, symbols: list):
"""Connect with automatic reconnection on failure."""
self.is_running = True
self.reconnect_count = 0
while self.is_running and self.reconnect_count < self.max_reconnect_attempts:
try:
await self._establish_connection(symbols)
except Exception as e:
self.reconnect_count += 1
if self.reconnect_count >= self.max_reconnect_attempts:
print(f"Max reconnection attempts reached. Giving up.")
raise
Related Resources
Related Articles