Handling API Rate Limits:
Resilient Agent Design Patterns

import asyncio
import time
import random
import logging
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Callable, Any
from functools import wraps

logger = logging.getLogger("rate_limiter")


# ============================================================
# TOKEN BUCKET RATE LIMITER
# ============================================================

class TokenBucket:
    """
    Async token bucket rate limiter.
    Refills at `rate` tokens/second, up to `capacity` tokens.
    Supports burst: initial tokens = capacity (allows initial burst).
    """

    def __init__(
        self,
        rate: float,          # tokens per second
        capacity: float,      # max tokens (burst size)
        initial_tokens: Optional[float] = None
    ):
        self.rate = rate
        self.capacity = capacity
        self.tokens = initial_tokens if initial_tokens is not None else capacity
        self.last_refill = time.monotonic()
        self._lock = asyncio.Lock()

    def _refill(self):
        now = time.monotonic()
        elapsed = now - self.last_refill
        new_tokens = elapsed * self.rate
        self.tokens = min(self.capacity, self.tokens + new_tokens)
        self.last_refill = now

    async def acquire(self, tokens: float = 1.0) -> float:
        """
        Acquire tokens. Returns wait time (0 if immediate).
        If wait > 0, caller should sleep before proceeding.
        """
        async with self._lock:
            self._refill()
            if self.tokens >= tokens:
                self.tokens -= tokens
                return 0.0
            # Calculate wait time until enough tokens available
            deficit = tokens - self.tokens
            wait = deficit / self.rate
            return wait

    async def wait_and_acquire(self, tokens: float = 1.0):
        """Acquire tokens, sleeping if necessary."""
        wait = await self.acquire(tokens)
        if wait > 0:
            logger.debug(f"Rate limiting: sleeping {wait:.3f}s")
            await asyncio.sleep(wait)
            # Re-acquire after sleeping
            async with self._lock:
                self._refill()
                self.tokens -= tokens

    def available_tokens(self) -> float:
        async with self._lock:
            self._refill()
            return self.tokens

    def utilization_pct(self) -> float:
        return (1.0 - self.tokens / self.capacity) * 100

import asyncio
import heapq
import httpx
from dataclasses import dataclass, field

# ============================================================
# PRIORITY QUEUE + COMPLETE RESILIENT AGENT
# ============================================================

@dataclass(order=True)
class QueuedRequest:
    priority: int                          # lower = higher priority
    enqueued_at: float
    url: str = field(compare=False)
    kwargs: dict = field(compare=False)
    future: asyncio.Future = field(compare=False)
    max_wait: float = field(compare=False, default=30.0)

class ResilientPurpleFlеaClient:
    """
    Complete resilient HTTP client for Purple Flea APIs.
    Combines: token bucket + priority queue + circuit breaker + backoff.

    Usage:
        client = ResilientPurpleFlеaClient(api_key="...", rate=5.0, burst=20)
        price = await client.request("GET", "/trading-api/ticker/BTC-USDC", priority=2)
        await client.request("POST", "/trading-api/order", json={...}, priority=1)
    """

    PRIORITY_NAMES = {0: "CRITICAL", 1: "HIGH", 2: "NORMAL", 3: "LOW", 4: "BACKGROUND"}
    MAX_WAIT_BY_PRIORITY = {0: 0.0, 1: 0.5, 2: 5.0, 3: 60.0, 4: 300.0}

    def __init__(
        self,
        api_key: str,
        base_url: str = "https://purpleflea.com",
        rate: float = 4.5,      # 4.5/s = 270/min (90% of 300/min standard)
        burst: float = 20,       # burst size
        max_concurrent: int = 20  # concurrent in-flight
    ):
        self.base_url = base_url
        self.http = httpx.AsyncClient(
            base_url=base_url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=15.0
        )
        self.bucket = TokenBucket(rate=rate, capacity=burst)
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.circuit = CircuitBreaker(name="purpleflea")
        self._queue: list = []
        self._worker_task: Optional[asyncio.Task] = None

    async def start(self):
        self._worker_task = asyncio.create_task(self._process_queue())

    async def stop(self):
        if self._worker_task:
            self._worker_task.cancel()
        await self.http.aclose()

    async def request(
        self,
        method: str,
        path: str,
        priority: int = 2,
        **kwargs
    ) -> dict:
        loop = asyncio.get_event_loop()
        future = loop.create_future()
        max_wait = self.MAX_WAIT_BY_PRIORITY.get(priority, 30.0)

        req = QueuedRequest(
            priority=priority,
            enqueued_at=asyncio.get_event_loop().time(),
            url=f"{method} {path}",
            kwargs={"method": method, "url": path, **kwargs},
            future=future,
            max_wait=max_wait
        )

        # P0 (critical): bypass queue entirely
        if priority == 0:
            return await self._execute(req)

        heapq.heappush(self._queue, req)
        return await future

    @with_retry(max_attempts=4, base_delay=0.5, jitter='decorrelated')
    async def _execute(self, req: QueuedRequest) -> dict:
        # Apply rate limiting
        await self.bucket.wait_and_acquire()
        # Concurrency limit
        async with self.semaphore:
            async def _do_request():
                resp = await self.http.request(**req.kwargs)
                resp.raise_for_status()
                return resp.json()
            return await self.circuit.call(_do_request)

    async def _process_queue(self):
        while True:
            if not self._queue:
                await asyncio.sleep(0.01)
                continue

            now = asyncio.get_event_loop().time()
            req = self._queue[0]

            # Check for timeout (priority aging)
            age = now - req.enqueued_at
            if age > req.max_wait and req.priority > 0:
                # Promote priority
                req.priority -= 1
                heapq.heapify(self._queue)
                logger.info(f"Promoted {req.url} to priority {req.priority}")
                continue

            req = heapq.heappop(self._queue)
            try:
                result = await self._execute(req)
                if not req.future.done():
                    req.future.set_result(result)
            except Exception as e:
                if not req.future.done():
                    req.future.set_exception(e)

# ====================================================
# Example usage with Purple Flea APIs
# ====================================================
async def main():
    client = ResilientPurpleFlеaClient(api_key="your-api-key")
    await client.start()

    try:
        # Critical: emergency stop (bypasses queue)
        await client.request("POST", "/trading-api/order",
            json={"symbol": "BTC-USDC", "side": "sell",
                  "type": "market", "reduce_only": True},
            priority=0
        )

        # Normal: price check (queued normally)
        price_data = await client.request(
            "GET", "/trading-api/ticker/BTC-USDC", priority=2
        )
        print(f"BTC mark price: ${price_data['mark_price']}")

        # Background: analytics (low priority)
        history = await client.request(
            "GET", "/trading-api/history?limit=1000", priority=4
        )

    finally:
        await client.stop()

if __name__ == "__main__":
    asyncio.run(main())