Imagine this: It's 2:47 AM, your quant trading system has been running flawlessly for three weeks, and suddenly you hit a wall—ConnectionError: timeout after 10000ms while trying to fetch historical k-line data from OKX. Your backtest can't complete, your strategy optimization is blocked, and that sweet arbitrage opportunity is slipping away by the minute.
This exact scenario happened to me during a critical evaluation of a mean-reversion strategy targeting OKX's perpetual futures. The fix took 15 minutes once I understood the root cause—and I'm going to share everything you need to avoid the same fate.
In this guide, you'll learn how to connect to OKX's API for quantitative strategy backtesting, integrate with HolySheep AI for intelligent strategy analysis, and troubleshoot the most common connection issues that plague quant traders.
Why OKX for Quantitative Trading?
OKX consistently ranks among the top 5 cryptocurrency exchanges by trading volume, offering deep liquidity across spot, futures, and perpetual swap markets. For quantitative traders, OKX provides:
- REST API with 20+ endpoints for market data, trading, and account management
- WebSocket streams for real-time data with sub-100ms latency
- Historical data going back to 2019 for comprehensive backtesting
- Demo trading mode for risk-free strategy validation
Combined with HolySheep AI's analysis capabilities, you can feed your backtest results into large language models for pattern recognition, strategy optimization suggestions, and risk assessment—potentially saving thousands in suboptimal trade execution.
Prerequisites
- OKX account with API key generated (no trading permissions needed for backtesting)
- Python 3.9+ environment
- HolySheep AI API key (free credits available on signup)
- Basic understanding of REST APIs and authentication
Setting Up Your OKX API Connection
Step 1: Generate OKX API Keys
Navigate to your OKX account settings, then API Management. Create a new API key with:
- Read-only permissions (sufficient for market data and history)
- IP whitelist configured for your server IP
- Passphrase - this is separate from your trading password
Important: OKX provides three critical values—API Key, Secret Key, and Passphrase. Store these securely; the Secret Key is shown only once.
Step 2: Install Required Libraries
# Install dependencies for OKX API and HolySheep integration
pip install requests websocket-client pandas numpy python-dotenv
Optional: for enhanced backtesting
pip install backtesting vectorbt
Verify installation
python -c "import requests, pandas; print('All packages installed successfully')"Step 3: Configure Environment Variables
# .env file - NEVER commit this to version control
OKX_API_KEY=your_okx_api_key_here
OKX_SECRET_KEY=your_okx_secret_key_here
OKX_PASSPHRASE=your_okx_passphrase_here
OKX_PASSPHRASE_PLAIN=your_plain_passphrase_here
HolySheep AI Configuration
HOLYSHEEP_API_KEY=YOUR_HOLYSHEEP_API_KEY
HOLYSHEEP_BASE_URL=https://api.holysheep.ai/v1
Optional: Telegram bot for alerts (recommended for production)
TELEGRAM_BOT_TOKEN=your_telegram_bot_token
TELEGRAM_CHAT_ID=your_telegram_chat_idBuilding the OKX Connection Class
import time
import hmac
import base64
import hashlib
import requests
import pandas as pd
from datetime import datetime, timedelta
from typing import Optional, Dict, List
import os
from dotenv import load_dotenv
load_dotenv()
class OKXConnector:
"""
Production-ready OKX API connector for quantitative backtesting.
Handles authentication, rate limiting, and error recovery.
"""
BASE_URL = "https://www.okx.com"
API_URL = "https://www.okx.com/api/v5"
def __init__(self, api_key: str = None, secret_key: str = None,
passphrase: str = None, use_server_time: bool = True):
self.api_key = api_key or os.getenv("OKX_API_KEY")
self.secret_key = secret_key or os.getenv("OKX_SECRET_KEY")
self.passphrase = passphrase or os.getenv("OKX_PASSPHRASE")
self.use_server_time = use_server_time
self.session = requests.Session()
self.session.headers.update({
"Content-Type": "application/json",
"OKX-ACCESS-PASSPHRASE": self.passphrase
})
# Rate limiting: 20 requests per 2 seconds (safe margin)
self.last_request_time = 0
self.min_request_interval = 0.11 # seconds
def _get_timestamp(self) -> str:
"""Generate ISO 8601 timestamp in UTC."""
return datetime.utcnow().isoformat() + 'Z'
def _sign(self, timestamp: str, method: str, path: str,
body: str = "") -> str:
"""Generate HMAC-SHA256 signature for OKX API authentication."""
message = timestamp + method + path + body
mac = hmac.new(
self.secret_key.encode('utf-8'),
message.encode('utf-8'),
hashlib.sha256
)
return base64.b64encode(mac.digest()).decode('utf-8')
def _apply_rate_limit(self):
"""Enforce rate limiting to avoid 429 errors."""
elapsed = time.time() - self.last_request_time
if elapsed < self.min_request_interval:
time.sleep(self.min_request_interval - elapsed)
self.last_request_time = time.time()
def _request(self, method: str, path: str, params: dict = None,
signed: bool = True) -> dict:
"""
Make authenticated API request with retry logic.
"""
url = self.API_URL + path
timestamp = self._get_timestamp()
body = ""
if params and method in ["POST", "PUT"]:
body = requests.compat.json.dumps(params)
if signed:
signature = self._sign(timestamp, method, path, body)
headers = {
"OKX-ACCESS-KEY": self.api_key,
"OKX-ACCESS-SIGN": signature,
"OKX-ACCESS-TIMESTAMP": timestamp,
"OKX-ACCESS-PASSPHRASE": self.passphrase
}
self.session.headers.update(headers)
self._apply_rate_limit()
max_retries = 3
for attempt in range(max_retries):
try:
if method == "GET":
response = self.session.get(url, params=params, timeout=30)
elif method == "POST":
response = self.session.post(url, data=body, timeout=30)
else:
response = self.session.request(method, url, timeout=30)
# Handle common error codes
if response.status_code == 429:
wait_time = int(response.headers.get("Retry-After", 5))
print(f"Rate limited. Waiting {wait_time}s...")
time.sleep(wait_time)
continue
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
print(f"Request timeout (attempt {attempt + 1}/{max_retries})")
if attempt < max_retries - 1:
time.sleep(2 ** attempt) # Exponential backoff
else:
raise ConnectionError(
f"OKX API timeout after {max_retries} attempts"
)
except requests.exceptions.RequestException as e:
if response.status_code == 401:
raise AuthenticationError(
"Invalid API credentials. Check your OKX API key, "
"secret, and passphrase."
)
raise
def get_candlesticks(self, inst_id: str, bar: str = "1H",
start: Optional[str] = None,
end: Optional[str] = None,
limit: int = 100) -> pd.DataFrame:
"""
Fetch historical candlestick (k-line) data for backtesting.
Args:
inst_id: Instrument ID (e.g., "BTC-USDT-SWAP")
bar: Timeframe ("1m", "5m", "1H", "1D", etc.)
start: Start time in ISO 8601 format
end: End time in ISO 8601 format
limit: Max records per request (1-100)
"""
path = "/market/history-candles"
params = {
"instId": inst_id,
"bar": bar,
"limit": min(limit, 100)
}
if start:
params["after"] = self._datetime_to_ts(start)
if end:
params["before"] = self._datetime_to_ts(end)
data = self._request("GET", path, params, signed=False)
if data.get("code") != "0":
raise APIError(f"OKX API error: {data.get('msg')}")
candles = data.get("data", [])
# OKX returns newest first; reverse for chronological order
df = pd.DataFrame(candles, columns=[
"timestamp", "open", "high", "low", "close", "volume", "vol_ccy"
])
df["timestamp"] = pd.to_datetime(
df["timestamp"].astype(np.int64), unit="ms"
)
for col in ["open", "high", "low", "close", "volume"]:
df[col] = df[col].astype(float)
return df.sort_values("timestamp").reset_index(drop=True)
def get_orderbook(self, inst_id: str, depth: int = 400) -> dict:
"""Fetch current order book snapshot."""
path = "/market/books-lite"
params = {"instId": inst_id, "sz": depth}
return self._request("GET", path, params, signed=False)
@staticmethod
def _datetime_to_ts(dt_str: str) -> str:
"""Convert ISO datetime to OKX timestamp (milliseconds)."""
dt = pd.to_datetime(dt_str)
return str(int(dt.timestamp() * 1000))
Custom exception classes
class APIError(Exception):
"""Base exception for API errors."""
pass
class AuthenticationError(APIError):
"""Raised when API authentication fails."""
pass
Usage example
if __name__ == "__main__":
import numpy as np
# Initialize connector
connector = OKXConnector()
# Fetch 1-hour candles for BTC-USDT perpetual
btc_data = connector.get_candlesticks(
inst_id="BTC-USDT-SWAP",
bar="1H",
start="2024-01-01",
end="2024-03-01",
limit=100
)
print(f"Fetched {len(btc_data)} candles")
print(f"Date range: {btc_data['timestamp'].min()} to {btc_data['timestamp'].max()}")
print(f"Columns: {list(btc_data.columns)}")Implementing a Simple Backtesting Engine
Now that we can fetch historical data, let's build a basic backtesting framework that integrates with HolySheep AI for strategy analysis.
import json
import requests
import numpy as np
import pandas as pd
from dataclasses import dataclass
from typing import List, Tuple, Optional
from datetime import datetime
@dataclass
class Trade:
"""Represents a single trade in the backtest."""
timestamp: datetime
side: str # "buy" or "sell"
price: float
quantity: float
pnl: float = 0.0
notes: str = ""
@dataclass
class BacktestResult:
"""Aggregated backtest metrics."""
total_trades: int
winning_trades: int
losing_trades: int
win_rate: float
total_pnl: float
max_drawdown: float
sharpe_ratio: float
trades: List[Trade]
equity_curve: List[float]
class SimpleBacktester:
"""
Event-driven backtesting engine for mean-reversion strategies.
Designed for OKX perpetual futures.
"""
def __init__(self, data: pd.DataFrame,
initial_capital: float = 10000,
position_size: float = 0.1):
self.data = data.copy()
self.initial_capital = initial_capital
self.position_size = position_size # Fraction of capital per trade
self.capital = initial_capital
self.position = 0.0
self.entry_price = 0.0
self.trades: List[Trade] = []
self.equity_curve = [initial_capital]
# Strategy parameters
self.rsi_period = 14
self.rsi_oversold = 30
self.rsi_overbought = 70
self._calculate_indicators()
def _calculate_indicators(self):
"""Compute technical indicators for strategy signals."""
delta = self.data["close"].diff()
gain = delta.where(delta > 0, 0.0)
loss = -delta.where(delta < 0, 0.0)
avg_gain = gain.rolling(window=self.rsi_period).mean()
avg_loss = loss.rolling(window=self.rsi_period).mean()
rs = avg_gain / avg_loss
self.data["rsi"] = 100 - (100 / (1 + rs))
# Bollinger Bands
self.data["bb_mid"] = self.data["close"].rolling(20).mean()
bb_std = self.data["close"].rolling(20).std()
self.data["bb_upper"] = self.data["bb_mid"] + (bb_std * 2)
self.data["bb_lower"] = self.data["bb_mid"] - (bb_std * 2)
def run(self) -> BacktestResult:
"""Execute backtest over historical data."""
for i, row in self.data.iterrows():
if pd.isna(row["rsi"]):
continue
timestamp = row["timestamp"]
price = row["close"]
# Entry signals
if self.position == 0:
# Buy when RSI oversold AND price below lower Bollinger Band
if (row["rsi"] < self.rsi_oversold and
price < row["bb_lower"]):
self._open_long(price, timestamp)
# Sell when RSI overbought AND price above upper Bollinger Band
elif (row["rsi"] > self.rsi_overbought and
price > row["bb_upper"]):
self._open_short(price, timestamp)
# Exit signals
else:
# Mean reversion target
if self.position > 0:
# Sell when RSI returns to neutral or profit target
if row["rsi"] > 50 or price > self.entry_price * 1.02:
self._close_position(price, timestamp, "RSI neutralization")
elif self.position < 0:
# Cover when RSI returns to neutral or profit target
if row["rsi"] < 50 or price < self.entry_price * 0.98:
self._close_position(price, timestamp, "RSI neutralization")
# Track equity
self.equity_curve.append(self._calculate_equity(price))
return self._calculate_metrics()
def _open_long(self, price: float, timestamp: datetime):
"""Open long position."""
qty = (self.capital * self.position_size) / price
self.position = qty
self.entry_price = price
def _open_short(self, price: float, timestamp: datetime):
"""Open short position."""
qty = -(self.capital * self.position_size) / price
self.position = qty
self.entry_price = price
def _close_position(self, price: float, timestamp: datetime, reason: str):
"""Close current position and record trade."""
if self.position == 0:
return
pnl = (price - self.entry_price) * abs(self.position)
self.capital += pnl
trade = Trade(
timestamp=timestamp,
side="buy" if self.position < 0 else "sell",
price=price,
quantity=abs(self.position),
pnl=pnl,
notes=reason
)
self.trades.append(trade)
self.position = 0.0
def _calculate_equity(self, current_price: float) -> float:
"""Calculate current equity including open position."""
position_value = self.position * current_price
return self.capital + position_value
def _calculate_metrics(self) -> BacktestResult:
"""Compute final backtest metrics."""
winning_trades = [t for t in self.trades if t.pnl > 0]
losing_trades = [t for t in self.trades if t.pnl <= 0]
returns = np.diff(self.equity_curve) / self.equity_curve[:-1]
sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252 * 24) if np.std(returns) > 0 else 0
# Calculate max drawdown
equity = np.array(self.equity_curve)
running_max = np.maximum.accumulate(equity)
drawdowns = (running_max - equity) / running_max
max_dd = np.max(drawdowns)
return BacktestResult(
total_trades=len(self.trades),
winning_trades=len(winning_trades),
losing_trades=len(losing_trades),
win_rate=len(winning_trades) / len(self.trades) if self.trades else 0,
total_pnl=self.capital - self.initial_capital,
max_drawdown=max_dd,
sharpe_ratio=sharpe,
trades=self.trades,
equity_curve=self.equity_curve
)
class HolySheepAnalyzer:
"""
Integration with HolySheep AI for strategy analysis and optimization.
Uses the HolySheep API to analyze backtest results with LLM insights.
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
def analyze_strategy(self, backtest_result: BacktestResult,
symbol: str = "BTC-USDT-SWAP") -> dict:
"""
Send backtest results to HolySheep AI for analysis and optimization.
HolySheep provides intelligent strategy optimization suggestions,
risk assessment, and pattern recognition across your trades.
Current pricing (2026): GPT-4.1 $8/MTok, Claude Sonnet 4.5 $15/MTok,
Gemini 2.5 Flash $2.50/MTok, DeepSeek V3.2 $0.42/MTok
"""
# Prepare summary for LLM analysis
analysis_prompt = f"""
Analyze this quantitative trading strategy backtest for {symbol}:
Performance Summary:
- Total Trades: {backtest_result.total_trades}
- Win Rate: {backtest_result.win_rate:.2%}
- Total PnL: ${backtest_result.total_pnl:.2f}
- Max Drawdown: {backtest_result.max_drawdown:.2%}
- Sharpe Ratio: {backtest_result.sharpe_ratio:.2f}
Recent Trades (last 10):
{self._format_trades(backtest_result.trades[-10:])}
Please provide:
1. Key insights about strategy performance
2. Identified weaknesses or patterns in losing trades
3. Specific parameter optimization suggestions
4. Risk management improvements
"""
try:
response = self.session.post(
f"{self.BASE_URL}/chat/completions",
json={
"model": "deepseek-v3.2", # Most cost-effective: $0.42/MTok
"messages": [
{
"role": "system",
"content": "You are an expert quantitative trading analyst. "
"Provide specific, actionable insights based on "
"backtest data."
},
{
"role": "user",
"content": analysis_prompt
}
],
"temperature": 0.3,
"max_tokens": 1500
},
timeout=60
)
response.raise_for_status()
result = response.json()
return {
"status": "success",
"analysis": result["choices"][0]["message"]["content"],
"model_used": "deepseek-v3.2",
"cost_estimate": f"${len(analysis_prompt) / 4 * 0.42 / 1_000_000:.4f}"
}
except requests.exceptions.Timeout:
return {
"status": "error",
"error": "HolySheep API timeout. Please retry."
}
except requests.exceptions.RequestException as e:
return {
"status": "error",
"error": f"HolySheep API error: {str(e)}"
}
def _format_trades(self, trades: List[Trade]) -> str:
"""Format trades for LLM analysis."""
if not trades:
return "No trades recorded"
lines = []
for t in trades:
lines.append(
f"- {t.timestamp.strftime('%Y-%m-%d %H:%M')}: "
f"{t.side.upper()} @ ${t.price:.2f}, PnL: ${t.pnl:.2f}"
)
return "\n".join(lines)
Complete example: Run backtest and analyze with HolySheep
if __name__ == "__main__":
# Initialize connectors
okx = OKXConnector()
holy_sheep = HolySheepAnalyzer(api_key=os.getenv("HOLYSHEEP_API_KEY"))
# Fetch 6 months of 1H data for backtesting
print("Fetching historical data from OKX...")
btc_data = okx.get_candlesticks(
inst_id="BTC-USDT-SWAP",
bar="1H",
start=(datetime.now() - timedelta(days=180)).isoformat(),
limit=100
)
print(f"Fetched {len(btc_data)} candles")
print("Running backtest...")
# Run backtest
backtester = SimpleBacktester(
data=btc_data,
initial_capital=10000,
position_size=0.1
)
results = backtester.run()
print("\n" + "="*50)
print("BACKTEST RESULTS")
print("="*50)
print(f"Total Trades: {results.total_trades}")
print(f"Win Rate: {results.win_rate:.2%}")
print(f"Total PnL: ${results.total_pnl:.2f}")
print(f"Max Drawdown: {results.max_drawdown:.2%}")
print(f"Sharpe Ratio: {results.sharpe_ratio:.2f}")
# Analyze with HolySheep AI
print("\nSending to HolySheep AI for analysis...")
analysis = holy_sheep.analyze_strategy(results)
if analysis["status"] == "success":
print(f"\nHOLYSHEEP AI ANALYSIS (cost: {analysis['cost_estimate']})")
print("-"*50)
print(analysis["analysis"])
else:
print(f"\nAnalysis failed: {analysis.get('error')}")Who This Is For / Not For
| Perfect For | Not Suitable For |
|---|---|
| Quantitative traders with programming experience (Python) | Completely non-technical traders (use OKX's built-in bots instead) |
| Those needing custom strategy logic beyond platform bots | Strategies requiring sub-second execution (WebSocket needed, not REST) |
| Backtesting before live deployment | High-frequency trading (HFT) - OKX has better endpoints for this) |
| Researchers analyzing historical market patterns | Multi-exchange arbitrage (would need separate connector per exchange) |
Pricing and ROI
Let's break down the actual costs of running quantitative strategies with this setup:
| Component | Cost | Notes |
|---|---|---|
| OKX API Access | Free | Read-only operations are free; trading incurs maker/taker fees |
| Cloud Server (2 vCPU) | $15-30/month | Required for 24/7 operation |
| HolySheep AI Analysis | $0.42/MTok | DeepSeek V3.2 model - most cost-effective option |
| Typical Monthly Analysis (100 backtests) | ~$0.50-2 | At ~500K tokens per deep analysis |
| Total Monthly Cost | ~$20-35 | vs. ¥7.3/$1 Chinese alternatives = 85%+ savings |
HolySheep Advantage: With the ¥1=$1 exchange rate and support for WeChat/Alipay payments, HolySheep offers dramatically better value than Western AI providers. A single strategy optimization that might cost $5 on OpenAI costs under $0.50 on HolySheep.
Common Errors and Fixes
Based on real production experience and community reports, here are the most frequent issues with OKX API integration:
Error 1: 401 Unauthorized - Invalid Signature
Full Error: {"code": "5013", "msg": "Signature verification failed"}
Cause: The HMAC signature algorithm or timestamp doesn't match OKX's requirements. Common mistakes include using the wrong secret key, incorrect timestamp format, or message encoding issues.
# INCORRECT - Common mistakes
def _sign_wrong(self, timestamp, method, path, body):
message = timestamp + method + path + body
mac = hmac.new(
self.secret_key.encode('utf-8'),
message.encode('utf-8'),
hashlib.sha256
)
return base64.b64encode(mac.digest()).decode('utf-8')
CORRECTED - Verified working implementation
def _sign(self, timestamp: str, method: str, path: str, body: str = "") -> str:
"""
OKX requires: timestamp + method + requestPath + body
- timestamp must be in ISO 8601 format with 'Z' suffix
- method must be uppercase (GET, POST)
- requestPath is the API endpoint path only (no query string)
"""
# Extract path without query parameters
clean_path = path.split('?')[0]
message = timestamp + method.upper() + clean_path + body
mac = hmac.new(
base64.b64decode(self.secret_key), # Secret must be base64 decoded
message.encode('utf-8'),
hashlib.sha256
)
return base64.b64encode(mac.digest()).decode('utf-8')Error 2: ConnectionError: Timeout After Multiple Retries
Full Error: ConnectionError: OKX API timeout after 3 attempts
Cause: Network connectivity issues, geographic distance from OKX servers, or IP-based rate limiting. Often occurs when running from residential ISPs.
# PRODUCTION-READY: Add retry logic with exponential backoff
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
class ResilientOKXConnector(OKXConnector):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.session.verify = True
self.proxies = {
# Use proxy if in China or experiencing connectivity issues
# 'http': 'http://127.0.0.1:7890',
# 'https': 'http://127.0.0.1:7890',
}
def _request_with_retry(self, method, path, params=None, signed=True):
max_attempts = 5
base_delay = 2
for attempt in range(max_attempts):
try:
return self._request(method, path, params, signed)
except ConnectionError as e:
delay = base_delay * (2 ** attempt)
print(f"Attempt {attempt + 1} failed: {e}")
print(f"Waiting {delay}s before retry...")
time.sleep(delay)
# Try alternative endpoint if primary fails
if attempt == 2:
print("Switching to alternative OKX endpoint...")
self.API_URL = "https://www.okx.com/api/v5"
except Exception as e:
print(f"Fatal error: {e}")
raise
raise ConnectionError(
f"Failed after {max_attempts} attempts. "
"Check: 1) Your IP is not blocked 2) OKX status page 3) Your firewall"
)
def get_candlesticks(self, *args, **kwargs):
"""Override with resilient request method."""
# For demo/development, use public endpoint without auth
if not self.api_key or self.api_key == "demo":
return self._get_public_candlesticks(*args, **kwargs)
return super().get_candlesticks(*args, **kwargs)
def _get_public_candlesticks(self, inst_id, bar="1H",
start=None, end=None, limit=100):
"""Public endpoint - no authentication needed for market data."""
import numpy as np
url = "https://www.okx.com/api/v5/market/history-candles"
params = {"instId": inst_id, "bar": bar, "limit": min(limit, 100)}
response = requests.get(url, params=params, timeout=30)
data = response.json()
if data["code"] != "0":
raise APIError(f"API error: {data['msg']}")
candles = data["data"]
df = pd.DataFrame(candles, columns=[
"timestamp", "open", "high", "low", "close", "volume", "vol_ccy"
])
df["timestamp"] = pd.to_datetime(
df["timestamp"].astype(np.int64), unit="ms"
)
for col in ["open", "high", "low", "close", "volume"]:
df[col] = df[col].astype(float)
return df.sort_values("timestamp").reset_index(drop=True)Error 3: 429 Rate Limit Exceeded
Full Error: {"code": "60009", "msg": "Too many requests"}
Cause: Making more than 20 requests per 2 seconds to OKX API endpoints. This is the official rate limit for most endpoints.
# SOLUTION: Strict rate limiting with token bucket algorithm
import threading
import time
from collections import deque
class RateLimitedConnector(OKXConnector):
"""
OKX Rate Limits:
- 20 requests/2 seconds for most endpoints
- 10 requests/2 seconds for account endpoints
- 5 requests/2 seconds for order placement
This implementation ensures you never hit 429 errors.
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Token bucket: 20 requests per 2 seconds
self.rate_limit = 20
self.time_window = 2.0 # seconds
self.request_timestamps = deque()
self.lock = threading.Lock()
# Adaptive: slow down if we see rate limits
self.current_rate = 18 # Start conservative
self.cooldown_active = False
def _apply_rate_limit(self):
"""Thread-safe rate limiting with adaptive adjustment."""
with self.lock:
now = time.time()
# Remove timestamps outside the window
while self.request_timestamps and \
now - self.request_timestamps[0] > self.time_window:
self.request_timestamps.popleft()
# Check if we're at the limit
if len(self.request_timestamps) >= self.rate_limit:
oldest = self.request_timestamps[0]
wait_time = self.time_window - (now - oldest)
if wait_time > 0:
print(f"Rate limit: waiting {wait_time:.2f}s...")
time.sleep(wait_time)
now = time.time()
# Clean up again after waiting
while self.request_timestamps and \
now - self.request_timestamps[0] > self.time_window:
self.request_timestamps.popleft()
# Record this request
self.request_timestamps.append(time.time())
# Adaptive: reduce rate if we detect throttling
if self.cooldown_active:
self.current_rate = max(10, self.current_rate - 2)
self.cooldown_active = False
def _request(self, *args, **kwargs):
try:
return super()._request(*args, **kwargs)
except APIError as e:
if "429" in str(e) or "rate" in str(e).lower():
self.current_rate = max(10, self.current_rate - 3)
self.cooldown_active = True
print(f"Rate limit hit! Reducing to {self.current_rate