Rate Limiting Strategies: How to Maximize Your API Quota

import requests
import time
from functools import lru_cache

API_KEY = "your_api_key_here"
BASE_URL = "https://tickatlas.com/v1"
HEADERS = {"X-API-Key": API_KEY}

class CachedClient:
    def __init__(self):
        self._cache = {}
        self._cache_ttl = {}

    def get_indicators(self, symbol: str, timeframe: str,
                       ttl: int = 60) -> dict:
        """Fetch the full indicator set with local caching."""
        cache_key = f"{symbol}:{timeframe}"
        now = time.time()

        # Return cached if still fresh
        if cache_key in self._cache:
            if now - self._cache_ttl[cache_key] < ttl:
                return self._cache[cache_key]

        # One call returns all 42 indicators for this symbol/timeframe.
        resp = requests.get(f"{BASE_URL}/indicators", params={
            "symbol": symbol,
            "timeframe": timeframe,
        }, headers=HEADERS)
        resp.raise_for_status()
        data = resp.json()

        self._cache[cache_key] = data
        self._cache_ttl[cache_key] = now
        return data

# Usage: H1 indicators cached for 55 minutes
client = CachedClient()
data = client.get_indicators("EURUSD", "H1", ttl=3300)

# BAD: one request per indicator (5 requests, 5x the quota)
rsi = client.get("/indicator?symbol=EURUSD&indicator=RSI_14&timeframe=H1")
macd = client.get("/indicator?symbol=EURUSD&indicator=MACD_hist&timeframe=H1")
bb = client.get("/indicator?symbol=EURUSD&indicator=BB_upper&timeframe=H1")
adx = client.get("/indicator?symbol=EURUSD&indicator=ADX&timeframe=H1")
atr = client.get("/indicator?symbol=EURUSD&indicator=ATR_14&timeframe=H1")

# GOOD: /v1/indicators returns ALL 42 for the symbol in one request
all_data = client.get("/indicators?symbol=EURUSD&timeframe=H1")
ind = all_data["data"]["indicators"]
rsi, macd, adx = ind["RSI_14"], ind["MACD_hist"], ind["ADX"]
# 80% reduction in API usage

import time

def request_with_backoff(url: str, params: dict,
                         max_retries: int = 5) -> dict:
    """Make API request with exponential backoff on rate limits."""
    for attempt in range(max_retries):
        resp = requests.get(url, params=params, headers=HEADERS)

        if resp.status_code == 200:
            return resp.json()

        if resp.status_code == 429:
            wait = 2 ** attempt  # 1, 2, 4, 8, 16 seconds
            retry_after = resp.headers.get("Retry-After", wait)
            print(f"Rate limited. Waiting {retry_after}s (attempt {attempt + 1})")
            time.sleep(float(retry_after))
        else:
            resp.raise_for_status()

    raise Exception(f"Max retries ({max_retries}) exceeded")

class UsageTracker:
    """Track API usage to stay within limits."""
    def __init__(self, daily_limit: int):
        self.daily_limit = daily_limit
        self.calls_today = 0
        self.reset_date = None

    def can_make_request(self) -> bool:
        today = datetime.utcnow().date()
        if self.reset_date != today:
            self.calls_today = 0
            self.reset_date = today

        return self.calls_today < self.daily_limit

    def record_request(self):
        self.calls_today += 1

    def remaining(self) -> int:
        return max(0, self.daily_limit - self.calls_today)

    def usage_percent(self) -> float:
        return (self.calls_today / self.daily_limit) * 100

# Usage
tracker = UsageTracker(daily_limit=10000)  # Starter plan

if tracker.can_make_request():
    data = client.get_indicators("EURUSD", "H1", "RSI_14,MACD")
    tracker.record_request()
    print(f"Usage: {tracker.usage_percent():.1f}% ({tracker.remaining()} remaining)")
else:
    print("Daily limit reached. Serving from cache only.")

def priority_scan(tracker: UsageTracker,
                  watchlist: list[str]) -> list[dict]:
    """Scan with priority when quota is low."""
    usage_pct = tracker.usage_percent()

    if usage_pct < 50:
        # Plenty of quota — scan everything
        symbols_to_check = watchlist
    elif usage_pct < 80:
        # Getting tight — only check top 5 pairs
        symbols_to_check = watchlist[:5]
    else:
        # Almost out — only check pairs with recent signals
        symbols_to_check = get_pairs_with_recent_signals()

    results = []
    for symbol in symbols_to_check:
        if tracker.can_make_request():
            data = client.get_indicators(symbol, "H1", "RSI_14")
            tracker.record_request()
            results.append(data)

    return results