Last Tuesday at 3:47 AM UTC, I ran a backtest that failed spectacularly. My Python script threw a 401 Unauthorized error when pulling 90 days of OHLCV data from OKX, and worse — the fallback to cached data introduced a 72-hour gap that corrupted my entire ML model's training set. After debugging through OKX's rate limiting headers and verifying my signature algorithm, I discovered the root cause: OKX's REST API requires a fresh timestamp alignment within 30 seconds of request time. This tutorial is the comprehensive guide I wish existed then — covering OKX historical data retrieval, clean data pipelines, and real AI model integration using HolySheep AI's sub-50ms latency endpoints.
Why Fetch Historical Data from OKX?
OKX ranks among the top five cryptocurrency exchanges by adjusted trading volume, offering deep liquidity across 400+ trading pairs. For algorithmic traders and quantitative researchers, OKX's public REST endpoints expose historical candlestick (OHLCV) data, order book snapshots, funding rates, and liquidations — all essential inputs for machine learning models predicting price movements, volatility regimes, or liquidation cascades.
When integrated with AI analysis pipelines, this historical data enables:
- Training supervised learning models on labeled price action patterns
- Fine-tuning large language models on crypto-specific terminology and market regimes
- Generating natural language market summaries via LLM inference
- Backtesting signal generation strategies with real exchange-grade data
Prerequisites
Before diving into code, ensure you have:
- Python 3.9+ installed (
python --version) - An OKX account with API key pair (public endpoints require no authentication for basic OHLCV data)
pip install requests pandas python-dotenv- Optional: HolySheep AI account for LLM inference (¥1=$1 pricing, WeChat/Alipay supported)
Step 1: Fetching Historical OHLCV Data from OKX REST API
OKX provides a public endpoint for candlestick data that requires no API key authentication. However, be aware of rate limits: 20 requests per 2 seconds for unauthenticated endpoints. I learned the hard way that batch requests with asyncio without respecting this limit results in HTTP 429 responses.
# okx_historical_fetcher.py
import requests
import pandas as pd
from datetime import datetime, timedelta
import time
OKX public API base URL
OKX_BASE_URL = "https://www.okx.com"
def fetch_ohlcv_data(
inst_id: str = "BTC-USDT",
bar: str = "1H", # 1m, 5m, 15m, 1H, 4H, 1D, etc.
start: str = None,
end: str = None,
limit: int = 100 # max 100 per request
) -> pd.DataFrame:
"""
Fetch historical OHLCV candlestick data from OKX public API.
Args:
inst_id: Instrument ID (e.g., "BTC-USDT", "ETH-USDT-SWAP")
bar: Timeframe granularity
start: Start time in ISO 8601 format (UTC)
end: End time in ISO 8601 format (UTC)
limit: Number of candles per request (max 100)
Returns:
DataFrame with columns: timestamp, open, high, low, close, volume
"""
endpoint = f"{OKX_BASE_URL}/api/v5/market/history-candles"
params = {
"instId": inst_id,
"bar": bar,
"limit": limit
}
if start:
# OKX requires RFC 3339 format with milliseconds
start_ms = int(datetime.fromisoformat(start.replace('Z', '+00:00')).timestamp() * 1000)
params["after"] = start_ms
if end:
end_ms = int(datetime.fromisoformat(end.replace('Z', '+00:00')).timestamp() * 1000)
params["before"] = end_ms
headers = {
"Content-Type": "application/json",
"User-Agent": "HolySheep-Tutorial/1.0" # Good practice for rate limit debugging
}
response = requests.get(endpoint, params=params, headers=headers, timeout=10)
# CRITICAL: Check rate limit headers
if response.status_code == 429:
retry_after = int(response.headers.get("X-RateLimit-Reset", 5))
print(f"Rate limited. Waiting {retry_after} seconds...")
time.sleep(retry_after)
response = requests.get(endpoint, params=params, headers=headers, timeout=10)
if response.status_code != 200:
raise ConnectionError(
f"OKX API returned {response.status_code}: {response.text}. "
f"Check OKX API status page for outages."
)
data = response.json()
if data.get("code") != "0":
raise ConnectionError(
f"OKX API error code {data.get('code')}: {data.get('msg')}. "
f"This often indicates timestamp drift — ensure server time is synced."
)
candles = data.get("data", [])
if not candles:
return pd.DataFrame(columns=["timestamp", "open", "high", "low", "close", "volume"])
# OKX returns array: [timestamp, open, high, low, close, volume, ...]
df = pd.DataFrame(candles, columns=[
"timestamp", "open", "high", "low", "close", "volume",
"quote_volume", "confirm", "bid_vol", "ask_vol"
])
# Convert timestamp (milliseconds) to datetime
df["timestamp"] = pd.to_datetime(df["timestamp"].astype(float) / 1000, unit="ms", utc=True)
# Convert numeric columns
for col in ["open", "high", "low", "close", "volume"]:
df[col] = pd.to_numeric(df[col], errors="coerce")
df = df.sort_values("timestamp").reset_index(drop=True)
return df[["timestamp", "open", "high", "low", "close", "volume"]]
Example usage: fetch 7 days of BTC-USDT hourly data
if __name__ == "__main__":
end_time = datetime.utcnow()
start_time = end_time - timedelta(days=7)
print(f"Fetching {start_time.isoformat()} to {end_time.isoformat()}...")
df = fetch_ohlcv_data(
inst_id="BTC-USDT",
bar="1H",
start=start_time.isoformat(),
end=end_time.isoformat(),
limit=100
)
print(f"Retrieved {len(df)} candles")
print(df.tail())
# Save to CSV for AI model training
df.to_csv("btc_usdt_1h_7d.csv", index=False)
print("Saved to btc_usdt_1h_7d.csv")
Step 2: Fetching Order Book and Liquidation Data
Beyond OHLCV candles, I often need order book depth snapshots and liquidation heatmaps for my volatility prediction models. OKX exposes these via separate endpoints.
# okx_orderbook_liquidations.py
import requests
import pandas as pd
from datetime import datetime
import time
OKX_BASE_URL = "https://www.okx.com"
def fetch_order_book(inst_id: str = "BTC-USDT", depth: int = 400) -> dict:
"""
Fetch current order book snapshot.
Args:
inst_id: Instrument ID
depth: Order book depth (max 400)
Returns:
Dict with 'bids' and 'asks' lists
"""
endpoint = f"{OKX_BASE_URL}/api/v5/market/books"
params = {
"instId": inst_id,
"sz": depth # Order book depth (max 400)
}
response = requests.get(endpoint, params=params, timeout=10)
if response.status_code != 200:
raise ConnectionError(f"Order book fetch failed: {response.status_code}")
data = response.json()
if data.get("code") != "0":
raise ConnectionError(f"OKX API error: {data.get('msg')}")
books = data.get("data", [{}])[0]
return {
"timestamp": datetime.utcnow().isoformat(),
"bids": [(float(b[0]), float(b[1])) for b in books.get("bids", [])],
"asks": [(float(a[0]), float(a[1])) for a in books.get("asks", [])],
"spread": float(books.get("asks", [[0]])[0][0]) - float(books.get("bids", [[0]])[0][0])
}
def fetch_liquidation_history(
inst_id: str = "BTC-USDT",
inst_type: str = "SWAP",
limit: int = 100
) -> pd.DataFrame:
"""
Fetch recent liquidation history (public endpoint).
Note: Requires authenticated endpoint for real-time liquidation stream.
Args:
inst_id: Instrument ID
inst_type: Instrument type (SPOT, SWAP, FUTURES, etc.)
limit: Number of records (max 100)
"""
endpoint = f"{OKX_BASE_URL}/api/v5/market/liquidation-history"
params = {
"instId": inst_id,
"instType": inst_type,
"limit": limit
}
response = requests.get(endpoint, params=params, timeout=10)
if response.status_code == 429:
print("Rate limited. Waiting 5 seconds...")
time.sleep(5)
response = requests.get(endpoint, params=params, timeout=10)
if response.status_code != 200:
raise ConnectionError(f"Liquidation history fetch failed: {response.status_code}")
data = response.json()
if data.get("code") != "0":
# Handle specific error codes
if data.get("code") == "58001":
raise ValueError(
"Invalid instrument type. Use SPOT, SWAP, FUTURES, or OPTION."
)
raise ConnectionError(f"OKX API error: {data.get('msg')}")
liquidations = data.get("data", [])
if not liquidations:
return pd.DataFrame(columns=["timestamp", "side", "size", "price", "trade_price"])
df = pd.DataFrame(liquidations, columns=[
"inst_id", "inst_type", "margin_currency", "size",
"price", "trade_price", "side", "force_order_type",
"created_at"
])
df["timestamp"] = pd.to_datetime(df["created_at"].astype(float) / 1000, unit="ms", utc=True)
return df[["timestamp", "side", "size", "price", "trade_price"]]
Example usage
if __name__ == "__main__":
# Get order book
book = fetch_order_book("BTC-USDT", depth=50)
print(f"Order book spread: ${book['spread']:.2f}")
print(f"Top 3 bids: {book['bids'][:3]}")
print(f"Top 3 asks: {book['asks'][:3]}")
# Get recent liquidations
print("\nFetching recent liquidations...")
liq_df = fetch_liquidation_history("BTC-USDT-SWAP", limit=20)
print(liq_df.head())
Step 3: Integrating with AI Models via HolySheep
Once you have cleaned historical data, the next step is analysis. I use HolySheep AI for AI inference because their ¥1=$1 pricing model (saving 85%+ versus domestic alternatives at ¥7.3) combined with WeChat/Alipay support makes it ideal for crypto traders. Their API delivers <50ms latency on inference calls — critical for time-sensitive market analysis.
Below is a complete pipeline that fetches OKX data, generates technical analysis prompts, and sends them to HolySheep's LLM endpoints for natural language insights.
# ai_market_analyzer.py
import requests
import pandas as pd
import os
from datetime import datetime, timedelta
HolySheep AI API Configuration
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
2026 Model Pricing (per 1M tokens input/output in USD)
MODEL_PRICING = {
"gpt-4.1": {"input": 8.00, "output": 8.00},
"claude-sonnet-4.5": {"input": 15.00, "output": 15.00},
"gemini-2.5-flash": {"input": 2.50, "output": 2.50},
"deepseek-v3.2": {"input": 0.42, "output": 0.42}, # Most cost-effective
}
def calculate_price_estimate(model: str, input_tokens: int, output_tokens: int) -> float:
"""Calculate cost in USD based on token counts."""
pricing = MODEL_PRICING.get(model, {"input": 0, "output": 0})
cost = (input_tokens / 1_000_000 * pricing["input"] +
output_tokens / 1_000_000 * pricing["output"])
return round(cost, 4)
def analyze_market_data(df: pd.DataFrame, model: str = "deepseek-v3.2") -> dict:
"""
Send OHLCV data to HolySheep AI for market analysis.
Args:
df: DataFrame with columns [timestamp, open, high, low, close, volume]
model: Model to use for analysis
Returns:
Dict with analysis results and cost
"""
# Calculate technical indicators
df = df.copy()
df["returns"] = df["close"].pct_change()
df["volatility_24h"] = df["returns"].rolling(24).std() * 100
df["volume_ma_24"] = df["volume"].rolling(24).mean()
recent = df.tail(48) # Last 48 candles (e.g., 48 hours of hourly data)
# Calculate summary statistics
price_change = ((recent["close"].iloc[-1] / recent["open"].iloc[0]) - 1) * 100
avg_volatility = recent["volatility_24h"].mean()
volume_ratio = recent["volume"].iloc[-1] / recent["volume_ma_24"].iloc[-1] if recent["volume_ma_24"].iloc[-1] > 0 else 0
# Construct analysis prompt
prompt = f"""You are a professional crypto market analyst. Based on the following BTC-USDT hourly data from the past 48 hours:
Recent Price: ${recent['close'].iloc[-1]:,.2f}
48h Price Change: {price_change:+.2f}%
Average Volatility: {avg_volatility:.2f}%
Volume vs 24h MA: {volume_ratio:.2f}x
24h High: ${recent['high'].max():,.2f}
24h Low: ${recent['low'].min():,.2f}
Please provide:
1. A brief market sentiment assessment (bullish/bearish/neutral)
2. Key support and resistance levels
3. Notable patterns or signals
4. Risk factors to watch
Keep the response concise (under 200 words)."""
# Estimate input tokens (rough: 4 chars ≈ 1 token)
input_tokens_est = len(prompt) // 4
output_tokens_est = 500 # Estimated response length
# Call HolySheep AI API
endpoint = f"{HOLYSHEEP_BASE_URL}/chat/completions"
headers = {
"Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
"Content-Type": "application/json"
}
payload = {
"model": model,
"messages": [
{"role": "system", "content": "You are a professional crypto market analyst."},
{"role": "user", "content": prompt}
],
"max_tokens": 500,
"temperature": 0.7
}
start_time = datetime.utcnow()
response = requests.post(endpoint, json=payload, headers=headers, timeout=30)
latency_ms = (datetime.utcnow() - start_time).total_seconds() * 1000
if response.status_code != 200:
raise ConnectionError(
f"HolySheep API error {response.status_code}: {response.text}. "
f"Verify your API key and check for rate limits."
)
result = response.json()
# Extract response
analysis_text = result["choices"][0]["message"]["content"]
usage = result.get("usage", {})
# Calculate actual cost
input_tokens_actual = usage.get("prompt_tokens", input_tokens_est)
output_tokens_actual = usage.get("completion_tokens", output_tokens_est)
cost_usd = calculate_price_estimate(model, input_tokens_actual, output_tokens_actual)
return {
"analysis": analysis_text,
"model_used": model,
"latency_ms": round(latency_ms, 2),
"input_tokens": input_tokens_actual,
"output_tokens": output_tokens_actual,
"cost_usd": cost_usd,
"timestamp": datetime.utcnow().isoformat(),
"price_data": {
"current_price": float(recent["close"].iloc[-1]),
"price_change_48h": round(price_change, 2),
"volatility": round(avg_volatility, 2),
"volume_ratio": round(volume_ratio, 2)
}
}
def batch_analyze_multiple_pairs(pairs: list, df_dict: dict, model: str = "deepseek-v3.2") -> list:
"""
Analyze multiple trading pairs efficiently.
Args:
pairs: List of instrument IDs (e.g., ["BTC-USDT", "ETH-USDT"])
df_dict: Dict mapping pair names to DataFrames
model: Model to use
Returns:
List of analysis results
"""
results = []
for pair in pairs:
if pair not in df_dict or df_dict[pair].empty:
print(f"Skipping {pair}: No data available")
continue
try:
result = analyze_market_data(df_dict[pair], model=model)
result["pair"] = pair
results.append(result)
print(f"✓ {pair}: {result['cost_usd']} USD ({result['latency_ms']}ms latency)")
except Exception as e:
print(f"✗ {pair}: {str(e)}")
return results
Example usage
if __name__ == "__main__":
# This would normally fetch real data, here using mock for demonstration
import numpy as np
# Mock OHLCV data for demonstration
dates = pd.date_range(end=datetime.utcnow(), periods=48, freq="H")
mock_df = pd.DataFrame({
"timestamp": dates,
"open": 67000 + np.random.randn(48).cumsum() * 100,
"high": 67200 + np.random.randn(48).cumsum() * 100,
"low": 66800 + np.random.randn(48).cumsum() * 100,
"close": 67000 + np.random.randn(48).cumsum() * 100,
"volume": np.random.randint(100, 500, 48) * 1e6
})
print("Analyzing market data with HolySheep AI...")
print(f"Using model: deepseek-v3.2 (most cost-effective at $0.42/MTok)\n")
result = analyze_market_data(mock_df, model="deepseek-v3.2")
print(f"=== Analysis Results ===")
print(f"Current Price: ${result['price_data']['current_price']:,.2f}")
print(f"48h Change: {result['price_data']['price_change_48h']:+.2f}%")
print(f"Volatility: {result['price_data']['volatility']:.2f}%")
print(f"\nAI Analysis:\n{result['analysis']}")
print(f"\nCost: ${result['cost_usd']} USD")
print(f"Latency: {result['latency_ms']}ms")
Step 4: Building a Complete Data Pipeline
For production use, I recommend wrapping everything in a clean data pipeline class that handles retries, caching, and error recovery.
# okx_pipeline.py
import requests
import pandas as pd
import json
import os
from datetime import datetime, timedelta
from pathlib import Path
import time
from typing import Optional, Generator
class OKXDataPipeline:
"""
Robust data pipeline for fetching and caching OKX market data.
Includes retry logic, rate limit handling, and local caching.
"""
def __init__(
self,
cache_dir: str = "./data_cache",
rate_limit_delay: float = 0.2, # 200ms between requests
max_retries: int = 3
):
self.base_url = "https://www.okx.com/api/v5"
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(parents=True, exist_ok=True)
self.rate_limit_delay = rate_limit_delay
self.max_retries = max_retries
def _get_cache_path(self, endpoint: str, params: dict) -> Path:
"""Generate cache file path from request parameters."""
param_str = "_".join(f"{k}={v}" for k, v in sorted(params.items()))
safe_name = f"{endpoint}_{param_str}.parquet".replace("/", "_")
return self.cache_dir / safe_name
def _is_cache_valid(self, cache_path: Path, max_age_hours: int = 1) -> bool:
"""Check if cached data is still valid."""
if not cache_path.exists():
return False
age = datetime.now() - datetime.fromtimestamp(cache_path.stat().st_mtime)
return age.total_seconds() < (max_age_hours * 3600)
def fetch_with_retry(
self,
endpoint: str,
params: dict = None,
use_cache: bool = True,
cache_max_age_hours: int = 1
) -> dict:
"""
Fetch data with automatic retry and caching.
Args:
endpoint: API endpoint (e.g., "/market/history-candles")
params: Query parameters
use_cache: Whether to use cached data
cache_max_age_hours: Maximum age of cached data in hours
Returns:
JSON response from OKX API
"""
params = params or {}
cache_path = self._get_cache_path(endpoint, params)
# Check cache first
if use_cache and self._is_cache_valid(cache_path, cache_max_age_hours):
print(f"Using cached data from {cache_path.name}")
return pd.read_parquet(cache_path).to_dict(orient="records")
# Make request with retries
url = f"{self.base_url}{endpoint}"
headers = {"Content-Type": "application/json"}
for attempt in range(self.max_retries):
try:
time.sleep(self.rate_limit_delay) # Respect rate limits
response = requests.get(
url,
params=params,
headers=headers,
timeout=15
)
# Handle rate limiting
if response.status_code == 429:
retry_after = int(response.headers.get("X-RateLimit-Reset", 60))
print(f"Rate limited. Retrying after {retry_after}s...")
time.sleep(retry_after)
continue
# Handle server errors
if response.status_code >= 500:
print(f"Server error {response.status_code}, attempt {attempt + 1}/{self.max_retries}")
time.sleep(2 ** attempt) # Exponential backoff
continue
# Handle client errors
if response.status_code >= 400:
error_data = response.json()
raise ConnectionError(
f"API error {response.status_code}: "
f"Code={error_data.get('code')}, Msg={error_data.get('msg')}"
)
# Success
data = response.json()
if data.get("code") != "0":
raise ConnectionError(
f"API error code {data.get('code')}: {data.get('msg')}"
)
# Cache the result
if use_cache:
records = data.get("data", [])
df = pd.DataFrame(records)
df.to_parquet(cache_path)
print(f"Cached {len(records)} records to {cache_path.name}")
return data
except requests.exceptions.Timeout:
print(f"Request timeout, attempt {attempt + 1}/{self.max_retries}")
if attempt == self.max_retries - 1:
raise ConnectionError("Request timed out after max retries")
def fetch_historical_ohlcv(
self,
inst_id: str,
bar: str = "1H",
start_time: datetime = None,
end_time: datetime = None,
use_cache: bool = True
) -> pd.DataFrame:
"""
Fetch historical OHLCV data with automatic pagination.
Args:
inst_id: Instrument ID
bar: Timeframe (1m, 5m, 15m, 1H, 4H, 1D)
start_time: Start time (defaults to 7 days ago)
end_time: End time (defaults to now)
use_cache: Use cached data if available
Returns:
DataFrame with OHLCV data
"""
start_time = start_time or (datetime.utcnow() - timedelta(days=7))
end_time = end_time or datetime.utcnow()
all_candles = []
current_after = int(start_time.timestamp() * 1000)
end_before = int(end_time.timestamp() * 1000)
max_requests = 100 # Safety limit
request_count = 0
print(f"Fetching {inst_id} {bar} data from {start_time.date()} to {end_time.date()}...")
while current_after < end_before and request_count < max_requests:
params = {
"instId": inst_id,
"bar": bar,
"after": current_after,
"limit": 100
}
data = self.fetch_with_retry("/market/history-candles", params, use_cache)
candles = data.get("data", [])
if not candles:
break
all_candles.extend(candles)
# Set pagination marker (next request's 'after' parameter)
current_after = int(candles[-1][0]) + 1
request_count += 1
print(f" Batch {request_count}: {len(candles)} candles (total: {len(all_candles)})")
if not all_candles:
return pd.DataFrame(columns=["timestamp", "open", "high", "low", "close", "volume"])
# Convert to DataFrame
df = pd.DataFrame(all_candles, columns=[
"timestamp", "open", "high", "low", "close", "volume",
"quote_volume", "confirm", "bid_vol", "ask_vol"
])
df["timestamp"] = pd.to_datetime(df["timestamp"].astype(float) / 1000, unit="ms", utc=True)
for col in ["open", "high", "low", "close", "volume"]:
df[col] = pd.to_numeric(df[col], errors="coerce")
return df[["timestamp", "open", "high", "low", "close", "volume"]].sort_values("timestamp")
Example usage
if __name__ == "__main__":
pipeline = OKXDataPipeline(
cache_dir="./my_crypto_data",
rate_limit_delay=0.25
)
# Fetch 30 days of BTC-USDT hourly data
df = pipeline.fetch_historical_ohlcv(
inst_id="BTC-USDT",
bar="1H",
start_time=datetime.utcnow() - timedelta(days=30),
use_cache=True
)
print(f"\nRetrieved {len(df)} candles")
print(f"Date range: {df['timestamp'].min()} to {df['timestamp'].max()}")
print(f"Total volume: {df['volume'].sum():,.0f}")
# Save for ML training
df.to_parquet("btc_usdt_1h_30d.parquet")
print("Saved to btc_usdt_1h_30d.parquet")
Common Errors and Fixes
Error 1: 401 Unauthorized
Error: {"code":"1","msg":"Authentication failed","data":null}
Cause: This error occurs when using OKX's private endpoints (like account data or order placement) without valid API credentials, or when your timestamp drift exceeds 30 seconds from OKX server time.
Fix: For public endpoints (OHLCV, order book), you don't need authentication. For private endpoints, ensure you:
# Correct API key authentication for OKX private endpoints
import hmac
import base64
import hashlib
from datetime import datetime
def generate_okx_signature(
timestamp: str,
method: str,
request_path: str,
body: str = ""
) -> str:
"""
Generate OKX API signature for authenticated endpoints.
Args:
timestamp: RFC 3339 format (e.g., "2024-01-15T10:30:00.000Z")
method: HTTP method (GET, POST, etc.)
request_path: API endpoint path (e.g., "/api/v5/account/balance")
body: Request body as string (empty string for GET)
Returns:
HMAC-SHA256 signature in base64
"""
message = timestamp + method + request_path + body
signature = hmac.new(
SECRET_KEY.encode('utf-8'),
message.encode('utf-8'),
hashlib.sha256
).digest()
return base64.b64encode(signature).decode('utf-8')
Verify server time before making authenticated requests
def verify_server_time():
"""Check if local time is within 30 seconds of OKX server time."""
response = requests.get("https://www.okx.com/api/v5/public/time")
server_time_ms = int(response.json()["data"][0]["ts"])
server_time = datetime.fromtimestamp(server_time_ms / 1000, tz=datetime.timezone.utc)
local_time = datetime.now(tz=datetime.timezone.utc)
drift = abs((server_time - local_time).total_seconds())
if drift > 30:
print(f"WARNING: Time drift of {drift:.1f}s detected. Sync your system clock!")
print("Run: sudo ntpdate -s time.google.com")
return False
return True
Error 2: 429 Too Many Requests
Error: HTTP 429: {"code":"60002","msg":"Request limit exceeded"}
Cause: OKX enforces rate limits: 20 requests/2 seconds for public endpoints, 60 requests/2 seconds for authenticated endpoints. Exceeding this triggers temporary blocks.
Fix: Implement exponential backoff and respect the X-RateLimit-Reset header:
# Robust rate limit handling
def fetch_with_rate_limit_handling(url: str, params: dict, max_retries: int = 5) -> dict:
"""Fetch with exponential backoff for rate limiting."""
base_delay = 0.5 # Start with 500ms
headers = {"User-Agent": "HolySheep-Tutorial/1.0"}
for attempt in range(max_retries):
response = requests.get(url, params=params, headers=headers, timeout=15)
if response.status_code == 200:
return response.json()
if response.status_code == 429:
# Use server-provided reset time if available
reset_time = int(response.headers.get("X-RateLimit-Reset", base_delay * 2 ** attempt))
actual_delay = max(reset_time, base_delay * 2 ** attempt)
print(f"Rate limited. Waiting {actual_delay:.1f}s (attempt {attempt + 1}/{max_retries})")
time.sleep(actual_delay)
continue
# For other errors, use exponential backoff
if response.status_code >= 500:
delay = base_delay * 2 ** attempt
print(f"Server error {response.status_code}. Retrying in {delay:.1f}s...")
time.sleep(delay)
continue
raise ConnectionError(f"Request failed with {response.status_code}: {response.text}")
raise ConnectionError(f"Failed after {max_retries} retries due to rate limiting")
Error 3: Empty Data Response / Missing Candles
Error: API returns {"code":"0","data":[],"msg":""} with empty candles array.
Cause: This happens when the requested time range has no trading activity (very rare for major pairs), or when pagination parameters are incorrectly set, or when the instrument ID format is wrong.
Fix: Verify instrument ID format and check pagination logic:
# Debug empty responses
def debug_empty_response(inst_id: str, start: str, end: str):
"""Debug why an OKX request returns empty data."""
endpoint = f"{OKX_BASE_URL}/api/v5/market/history-candles"
params = {
"instId": inst_id,
"bar": "1H",
"after": int(datetime.fromisoformat(start).timestamp() *