paulpas Role: Backend System Architect — designs robust API integration patterns that respect exchange rate limits while maximizing throughput and resilience.
Philosophy: Intelligent Throttling — rate limiting should be proactive and adaptive, preventing disruptions before they occur while maintaining optimal execution speed through predictive capacity management.
Key Principles
Token Bucket Algorithm: Implement token bucket rate limiting with configurable refill rates and burst capacity for handling temporary spikes.
Sliding Window Tracking: Use sliding window counters for precise rate limiting that adapts to recent activity patterns.
Per-Endpoint Granularity: Apply different rate limits to different endpoints based on their impact and exchange-specific requirements.
Circuit Breaker Patterns: Implement circuit breakers that trip when failures exceed thresholds, preventing cascading failures and allowing recovery.
Backpressure Handling: Design for graceful degradation when rate limits are exceeded, queueing requests rather than dropping them.
Implementation Guidelines
Structure
- Core logic:
skills/exchange-integration/rate_limiter.py - Circuit breaker:
skills/exchange-integration/circuit_breaker.py - Tests:
skills/tests/test_rate_limiting.py
Patterns to Follow
- Use asyncio locks for thread-safe rate limiting in concurrent environments
- Implement rate limiters as stateful classes with clean interfaces
- Separate rate limit tracking from request execution
- Use exponential backoff with jitter for retry delays
Code Examples
Token Bucket Algorithm
import asyncio
import time
from dataclasses import dataclass
from typing import Optional
import threading
@dataclass
class TokenBucketConfig:
"""Configuration for token bucket."""
rate: float # Tokens per second
capacity: float # Maximum tokens (burst capacity)
initial_tokens: Optional[float] = None
def __post_init__(self):
if self.initial_tokens is None:
self.initial_tokens = self.capacity
class TokenBucket:
"""
Thread-safe token bucket rate limiter.
Tokens are consumed on each request and refill at a constant rate.
When no tokens are available, requests must wait.
"""
def __init__(self, config: TokenBucketConfig):
self.config = config
self.tokens = config.initial_tokens
self.last_update = time.monotonic()
self.lock = threading.Lock()
def _refill_tokens(self):
"""Refill tokens based on elapsed time."""
now = time.monotonic()
elapsed = now - self.last_update
tokens_to_add = elapsed * self.config.rate
self.tokens = min(self.config.capacity, self.tokens + tokens_to_add)
self.last_update = now
def consume(self, tokens: float = 1.0) -> bool:
"""
Attempt to consume tokens.
Returns True if tokens were consumed, False otherwise.
"""
with self.lock:
self._refill_tokens()
if self.tokens >= tokens:
self.tokens -= tokens
return True
return False
async def consume_async(self, tokens: float = 1.0) -> bool:
"""Async version that waits for tokens if none available."""
with self.lock:
self._refill_tokens()
if self.tokens >= tokens:
self.tokens -= tokens
return True
# Calculate wait time
tokens_needed = tokens - self.tokens
wait_time = tokens_needed / self.config.rate
return wait_time
async def wait_and_consume(self, tokens: float = 1.0) -> float:
"""
Wait for tokens to become available and consume them.
Returns actual wait time in seconds.
"""
start_time = time.monotonic()
while True:
with self.lock:
self._refill_tokens()
if self.tokens >= tokens:
self.tokens -= tokens
break
tokens_needed = tokens - self.tokens
wait_time = tokens_needed / self.config.rate
# Wait for calculated time plus small jitter
await asyncio.sleep(wait_time * 0.9 + wait_time * 0.2 * (hash(time.monotonic()) % 100) / 100)
return time.monotonic() - start_time
def get_available_tokens(self) -> float:
"""Get current token count (after refill)."""
with self.lock:
self._refill_tokens()
return self.tokens
def get_wait_time(self, tokens: float = 1.0) -> float:
"""Get estimated wait time for given tokens."""
with self.lock:
self._refill_tokens()
if self.tokens >= tokens:
return 0.0
tokens_needed = tokens - self.tokens
return tokens_needed / self.config.rate
class AdaptiveTokenBucket(TokenBucket):
"""
Token bucket with adaptive rate adjustment based on performance feedback.
Increases rate when consecutive successful requests occur.
Decreases rate when rate limit errors are encountered.
"""
def __init__(self,
config: TokenBucketConfig,
min_rate: float = None,
max_rate: float = None,
adjustment_factor: float = 1.2):
super().__init__(config)
self.min_rate = min_rate or (config.rate * 0.5)
self.max_rate = max_rate or (config.rate * 2.0)
self.adjustment_factor = adjustment_factor
self.success_streak = 0
self.failure_streak = 0
def on_success(self):
"""Called after successful request."""
self.success_streak += 1
self.failure_streak = 0
# Gradually increase rate after success streak
if self.success_streak >= 10:
new_rate = self.config.rate * self.adjustment_factor
if new_rate <= self.max_rate:
self.config.rate = new_rate
self.success_streak = 0
def on_failure(self):
"""Called after rate limit error or failure."""
self.failure_streak += 1
self.success_streak = 0
# Reduce rate after failures
if self.failure_streak >= 3:
new_rate = self.config.rate / self.adjustment_factor
if new_rate >= self.min_rate:
self.config.rate = new_rate
self.failure_streak = 0
def get_stats(self) -> dict:
"""Get current bucket statistics."""
return {
'tokens': self.get_available_tokens(),
'rate': self.config.rate,
'success_streak': self.success_streak,
'failure_streak': self.failure_streak
}
class SlidingWindowCounter:
"""
Sliding window rate limiter for precise limiting.
Counts requests in a sliding time window rather than fixed intervals.
More accurate than fixed window for bursty traffic patterns.
"""
def __init__(self, window_seconds: float, max_requests: int):
self.window = window_seconds
self.max_requests = max_requests
self.requests = []
self.lock = threading.Lock()
def _cleanup_old_requests(self):
"""Remove requests outside the current window."""
now = time.monotonic()
cutoff = now - self.window
self.requests = [t for t in self.requests if t > cutoff]
def is_allowed(self) -> bool:
"""Check if request is allowed without recording it."""
with self.lock:
self._cleanup_old_requests()
return len(self.requests) < self.max_requests
def record_request(self) -> bool:
"""Record a request and check if it was allowed."""
with self.lock:
self._cleanup_old_requests()
if len(self.requests) >= self.max_requests:
return False
self.requests.append(time.monotonic())
return True
def get_request_count(self) -> int:
"""Get current request count in window."""
with self.lock:
self._cleanup_old_requests()
return len(self.requests)
def get_wait_time(self) -> float:
"""Get time until next request would be allowed."""
with self.lock:
self._cleanup_old_requests()
if len(self.requests) < self.max_requests:
return 0.0
# Find oldest request and calculate when it expires
oldest = min(self.requests)
return (oldest + self.window) - time.monotonic()
class HybridRateLimiter:
"""
Hybrid rate limiter combining token bucket and sliding window.
Token bucket for burst handling, sliding window for hard limits.
"""
def __init__(self,
burst_config: TokenBucketConfig,
window_config: dict):
self.token_bucket = AdaptiveTokenBucket(burst_config)
self.sliding_window = SlidingWindowCounter(
window_config['seconds'],
window_config['max_requests']
)
def is_allowed(self) -> bool:
"""Check if request is allowed."""
return (self.token_bucket.consume() and
self.sliding_window.is_allowed())
async def wait_and_execute(self) -> float:
"""Wait for both limiters and return total wait time."""
# Wait for token bucket first
bucket_wait = await self.token_bucket.wait_and_consume()
# Check sliding window
if self.sliding_window.is_allowed():
self.sliding_window.record_request()
return bucket_wait
# Need to wait for sliding window too
window_wait = self.sliding_window.get_wait_time()
await asyncio.sleep(window_wait)
self.sliding_window.record_request()
return bucket_wait + window_wait
def get_stats(self) -> dict:
"""Get combined statistics."""
return {
'bucket': self.token_bucket.get_stats(),
'window': {
'requests': self.sliding_window.get_request_count(),
'window_seconds': self.sliding_window.window,
'max_requests': self.sliding_window.max_requests
}
}
Per-Endpoint Rate Limits
from dataclasses import dataclass, field
from typing import Dict, Optional
from enum import Enum
import threading
class EndpointType(Enum):
"""Types of exchange endpoints."""
AUTH = "auth" # Authentication/authorization
MARKET = "market" # Public market data
ORDER = "order" # Order placement/cancellation
ACCOUNT = "account" # Account information
HISTORY = "history" # Trade/candle history
@dataclass
class EndpointRateLimit:
"""Rate limit configuration for a single endpoint."""
endpoint_type: EndpointType
rate: float # Requests per second
burst_capacity: float
weight: float = 1.0 # Relative weight (higher = more expensive)
hard_limit: Optional[int] = None # Maximum requests per period
class EndpointRateLimiter:
"""
Manages rate limits per endpoint type.
Allows prioritizing critical endpoints while limiting others.
"""
def __init__(self, limits: Dict[EndpointType, EndpointRateLimit]):
self.limits = limits
self.buckets: Dict[EndpointType, TokenBucket] = {}
self.lock = threading.Lock()
# Initialize buckets for each endpoint
for endpoint_type, config in limits.items():
bucket_config = TokenBucketConfig(
rate=config.rate,
capacity=config.burst_capacity
)
self.buckets[endpoint_type] = TokenBucket(bucket_config)
def consume(self, endpoint_type: EndpointType, weight: float = 1.0) -> bool:
"""Consume tokens for an endpoint."""
with self.lock:
bucket = self.buckets.get(endpoint_type)
if bucket is None:
raise ValueError(f"No rate limit configured for {endpoint_type}")
# Scale tokens by weight
tokens_needed = weight * (1.0 / bucket_config.rate) if bucket_config else 1.0
return bucket.consume(tokens_needed)
async def wait_for_endpoint(self,
endpoint_type: EndpointType,
weight: float = 1.0) -> float:
"""Wait for tokens for specific endpoint."""
with self.lock:
bucket = self.buckets.get(endpoint_type)
if bucket is None:
raise ValueError(f"No rate limit configured for {endpoint_type}")
tokens_needed = weight * (1.0 / bucket.config.rate) if bucket.config else 1.0
return await bucket.wait_and_consume(tokens_needed)
def is_endpoint_allowed(self, endpoint_type: EndpointType) -> bool:
"""Check if endpoint is currently allowed."""
with self.lock:
bucket = self.buckets.get(endpoint_type)
if bucket is None:
return True # No limit means no restriction
return bucket.consume(0) # Check without consuming
class GlobalRateLimiter:
"""
Global rate limiter that applies across all endpoints.
Prevents exceeding overall exchange quota.
"""
def __init__(self,
global_rate: float,
global_burst: float,
endpoint_limits: Dict[EndpointType, EndpointRateLimit]):
self.global_bucket = AdaptiveTokenBucket(
TokenBucketConfig(rate=global_rate, capacity=global_burst)
)
self.endpoint_limiter = EndpointRateLimiter(endpoint_limits)
def consume(self, endpoint_type: EndpointType, weight: float = 1.0) -> bool:
"""Consume from both global and endpoint-specific limits."""
# Check global limit first
if not self.global_bucket.consume(weight * 0.1): # 10% of weight for global
return False
# Check endpoint-specific
return self.endpoint_limiter.consume(endpoint_type, weight)
async def wait_and_execute(self,
endpoint_type: EndpointType,
weight: float = 1.0) -> float:
"""Wait for both limits and execute."""
global_wait = await self.global_bucket.wait_and_consume(weight * 0.1)
endpoint_wait = await self.endpoint_limiter.wait_for_endpoint(endpoint_type, weight)
return global_wait + endpoint_wait
class PriorityRateLimiter:
"""
Rate limiter with priority support.
Higher priority endpoints can preempt lower priority ones.
"""
def __init__(self,
rates: Dict[str, float],
priorities: Dict[str, int]):
"""
Initialize with rate and priority for each endpoint type.
Args:
rates: Endpoint type -> requests per second
priorities: Endpoint type -> priority (higher = more important)
"""
self.buckets: Dict[str, AdaptiveTokenBucket] = {}
self.priorities = priorities
for endpoint_type, rate in rates.items():
config = TokenBucketConfig(rate=rate, capacity=rate * 2)
self.buckets[endpoint_type] = AdaptiveTokenBucket(config)
def consume(self, endpoint_type: str, priority_override: int = None) -> bool:
"""Consume tokens with optional priority override."""
bucket = self.buckets.get(endpoint_type)
if bucket is None:
return False
# If priority is high enough, we can use more tokens
current_priority = priority_override or self.priorities.get(endpoint_type, 0)
if current_priority >= 10: # High priority
return bucket.consume(1.5) # Use 50% more tokens
return bucket.consume(1.0)
def get_endpoint_priority(self, endpoint_type: str) -> int:
"""Get priority level for endpoint."""
return self.priorities.get(endpoint_type, 0)
def can_priority_preempt(self, high_priority: str, low_priority: str) -> bool:
"""Check if high priority can preempt low priority."""
high_prio = self.priorities.get(high_priority, 0)
low_prio = self.priorities.get(low_priority, 0)
return high_prio > low_prio + 5 # Need 5 point advantage
Circuit Breaker Patterns
from dataclasses import dataclass
from typing import Callable, Optional
from enum import Enum
import time
import threading
class CircuitState(Enum):
"""Circuit breaker states."""
CLOSED = "closed" # Normal operation
OPEN = "open" # Failing, reject requests
HALF_OPEN = "half_open" # Testing recovery
@dataclass
class CircuitBreakerConfig:
"""Configuration for circuit breaker."""
failure_threshold: int = 5 # Failures to open circuit
success_threshold: int = 3 # Successes to close circuit
timeout_seconds: float = 30.0 # Time to wait before half-open
half_open_max_calls: int = 3 # Max calls in half-open state
class CircuitBreaker:
"""
Circuit breaker pattern implementation.
Protects against cascading failures by stopping requests to failing services.
"""
def __init__(self, config: CircuitBreakerConfig):
self.config = config
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
self.last_failure_time: Optional[float] = None
self.lock = threading.Lock()
def _should_attempt_request(self) -> bool:
"""Check if request should be attempted based on circuit state."""
if self.state == CircuitState.CLOSED:
return True
if self.state == CircuitState.OPEN:
# Check if timeout has passed
if self.last_failure_time is None:
return False
elapsed = time.monotonic() - self.last_failure_time
if elapsed >= self.config.timeout_seconds:
self.state = CircuitState.HALF_OPEN
self.success_count = 0
return True
return False
# HALF_OPEN state
return True
def record_success(self):
"""Record a successful request."""
with self.lock:
if self.state == CircuitState.HALF_OPEN:
self.success_count += 1
if self.success_count >= self.config.success_threshold:
self.state = CircuitState.CLOSED
self.failure_count = 0
self.success_count = 0
elif self.state == CircuitState.CLOSED:
self.failure_count = 0 # Reset on success
def record_failure(self):
"""Record a failed request."""
with self.lock:
self.failure_count += 1
self.last_failure_time = time.monotonic()
if self.state == CircuitState.HALF_OPEN:
# Any failure in half-open goes back to open
self.state = CircuitState.OPEN
self.success_count = 0
elif self.state == CircuitState.CLOSED:
if self.failure_count >= self.config.failure_threshold:
self.state = CircuitState.OPEN
def execute(self, func: Callable, *args, **kwargs):
"""
Execute function with circuit breaker protection.
Raises CircuitBreakerOpen if circuit is open.
"""
if not self._should_attempt_request():
raise CircuitBreakerOpen(
f"Circuit is {self.state.value}, last failure: {self.last_failure_time}"
)
try:
result = func(*args, **kwargs)
self.record_success()
return result
except Exception as e:
self.record_failure()
raise
class CircuitBreakerOpen(Exception):
"""Exception raised when circuit breaker is open."""
pass
class RateLimitedCircuitBreaker(CircuitBreaker):
"""
Circuit breaker with rate limiting for recovery attempts.
Prevents flood of recovery requests when circuit opens.
"""
def __init__(self,
breaker_config: CircuitBreakerConfig,
rate_limit_config: dict = None):
super().__init__(breaker_config)
if rate_limit_config is None:
rate_limit_config = {
'rate': 0.1, # 1 request per 10 seconds
'burst': 1
}
from .rate_limiter import TokenBucketConfig, TokenBucket
self.recovery_limiter = TokenBucket(
TokenBucketConfig(
rate=rate_limit_config['rate'],
capacity=rate_limit_config['burst']
)
)
def _should_attempt_request(self) -> bool:
"""Check if request should be attempted with rate limiting."""
# First check circuit state
if not super()._should_attempt_request():
return False
# Then check recovery rate limit
return self.recovery_limiter.consume()
async def execute_with_rate_limit(self, func: Callable, *args, **kwargs):
"""Execute with rate limiting during recovery."""
if not self._should_attempt_request():
raise CircuitBreakerOpen(f"Circuit is {self.state.value}")
# Wait for rate limiter if needed
from .rate_limiter import AdaptiveTokenBucket
if isinstance(self.recovery_limiter, AdaptiveTokenBucket):
wait_time = await self.recovery_limiter.wait_and_consume()
if wait_time > 0:
import asyncio
await asyncio.sleep(wait_time)
try:
result = func(*args, **kwargs)
self.record_success()
return result
except Exception as e:
self.record_failure()
raise
class CompositeCircuitBreaker:
"""
Circuit breaker for multiple dependent services.
Opens circuit if any dependent service fails.
"""
def __init__(self, services: list):
self.services = {s: CircuitBreaker(CircuitBreakerConfig()) for s in services}
self.main_breaker = CircuitBreaker(CircuitBreakerConfig())
def get_breaker(self, service: str) -> CircuitBreaker:
"""Get circuit breaker for specific service."""
return self.services.get(service)
def on_service_failure(self, service: str):
"""Record failure for specific service."""
breaker = self.services.get(service)
if breaker:
breaker.record_failure()
# Propagate to main breaker
self.main_breaker.record_failure()
def on_service_success(self, service: str):
"""Record success for specific service."""
breaker = self.services.get(service)
if breaker:
breaker.record_success()
# Check if all services are healthy
all_healthy = all(
b.state == CircuitState.CLOSED for b in self.services.values()
)
if all_healthy:
self.main_breaker.record_success()
Adherence Checklist
Before completing your task, verify:
- Guard Clauses: All rate limiters check for zero/negative values; all circuit breakers validate configuration
- Parsed State: Rate limit configurations parsed into structured types before initialization
- Atomic Predictability: Rate limiting logic is deterministic; only timing of waits involves delay
- Fail Fast: Invalid rate limit configuration throws descriptive error immediately
- Intentional Naming: Classes use clear names (
TokenBucket,SlidingWindowCounter,CircuitBreaker)
Common Mistakes to Avoid
Using Fixed Time Windows: Fixed window rate limiting allows burst at window boundaries. Sliding window or token bucket is preferred.
Not Handling Burst Traffic: Rate limits should allow for temporary bursts (token bucket capacity) to handle legitimate traffic spikes.
Circuit Breaker Too Sensitive: Too few failures before opening circuit causes unnecessary outages. Too many allows cascading failures.
Not Backing Off Properly: When rate limited, exponential backoff with jitter prevents thundering herd problems.
Ignoring Per-Endpoint Limits: Different endpoints have different limits. Treating all endpoints equally can cause some to fail while others have capacity.
References
- Netflix: Circuit Breaker Pattern (2012). Original paper on circuit breaker implementation.
- Hohpe, G. (2008). "Enterprise Integration Patterns".Addison-Wesley.
- AWS: Rate Limiting Best Practices (2026). Cloud-specific guidelines.
- GitHub API Rate Limiting Documentation (2026). Real-world implementation examples.
Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.
Constraints
MUST DO
- Implement a unified adapter interface across all exchange integrations to standardize order placement, cancellation, and querying
- Handle rate limiting proactively with token bucket or leaky bucket algorithms — never wait for 429 responses before slowing down
- Maintain local order state as the source of truth; reconcile with exchange state periodically via webhook events and polling
- Implement heartbeat monitoring per exchange connection with automatic failover to a secondary data feed on timeout
- Log all API interactions including request/response IDs, timing, and status codes for audit and debugging
MUST NOT DO
- Do not trust exchange-reported order states without local confirmation — always reconcile after every state change
- Avoid sending multiple orders for the same position simultaneously across different adapters or sessions
- Never store API keys or secrets in code — use environment variables or a secrets manager with automatic rotation
- Do not assume all exchanges support the same order types — implement graceful degradation with clear capability negotiation
- Avoid polling-based price updates when WebSocket/streaming APIs are available — polling creates unnecessary load and latency
Live References
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.