REST API Best Practices for Financial Data Applications

import requests
import time
import logging

logger = logging.getLogger(__name__)

def api_call_with_retry(
    url: str,
    headers: dict,
    params: dict,
    max_retries: int = 3,
    base_delay: float = 1.0,
) -> dict:
    """Make an API call with exponential backoff retry."""
    for attempt in range(max_retries + 1):
        try:
            resp = requests.get(url, headers=headers, params=params, timeout=10)

            if resp.status_code == 429:  # Rate limited
                retry_after = int(resp.headers.get("Retry-After", base_delay * 2))
                logger.warning(f"Rate limited. Waiting {retry_after}s")
                time.sleep(retry_after)
                continue

            if resp.status_code >= 500:  # Server error -- retry
                raise requests.exceptions.HTTPError(f"Server error: {resp.status_code}")

            resp.raise_for_status()
            return resp.json()

        except (requests.exceptions.ConnectionError,
                requests.exceptions.Timeout,
                requests.exceptions.HTTPError) as e:
            if attempt == max_retries:
                logger.error(f"Failed after {max_retries} retries: {e}")
                raise
            delay = base_delay * (2 ** attempt)
            logger.warning(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay}s")
            time.sleep(delay)

# Usage
data = api_call_with_retry(
    "https://tickatlas.com/v1/indicator",
    headers={"X-API-Key": "YOUR_KEY"},
    params={"symbol": "EURUSD", "indicator": "RSI_14", "timeframe": "H4"},
)

import threading
import time

class RateLimiter:
    """Token bucket rate limiter for API calls."""

    def __init__(self, calls_per_minute: int):
        self.calls_per_minute = calls_per_minute
        self.interval = 60.0 / calls_per_minute
        self.last_call = 0.0
        self.lock = threading.Lock()

    def wait(self):
        with self.lock:
            now = time.monotonic()
            elapsed = now - self.last_call
            if elapsed < self.interval:
                time.sleep(self.interval - elapsed)
            self.last_call = time.monotonic()

# Create a limiter matching your plan
limiter = RateLimiter(calls_per_minute=120)  # Starter plan

def safe_api_call(url: str, headers: dict, params: dict) -> dict:
    limiter.wait()  # Block until we can make a call
    return api_call_with_retry(url, headers, params)

from datetime import datetime, timezone, timedelta

def validate_indicator_response(data: dict, symbol: str) -> bool:
    """Validate that indicator data is fresh and reasonable."""
    if not data.get("success"):
        return False

    inner = data.get("data", {})

    # Check symbol matches
    if inner.get("symbol") != symbol:
        logger.error(f"Symbol mismatch: expected {symbol}, got {inner.get('symbol')}")
        return False

    # Check freshness (server_time is the time of the data, ISO-8601 UTC)
    ts = inner.get("server_time")
    if ts:
        data_time = datetime.fromisoformat(ts.replace("Z", "+00:00"))
        age = datetime.now(timezone.utc) - data_time
        if age > timedelta(hours=2):
            logger.warning(f"Stale data: {age} old for {symbol}")
            return False

    # Check the value exists and is numeric
    value = inner.get("value")
    if not isinstance(value, (int, float)):
        logger.error(f"Non-numeric value for {inner.get('indicator')}: {value}")
        return False

    return True

class TickAtlasClient:
    """Production-grade API client with retry, rate limiting, and validation."""

    def __init__(self, api_key: str, rate_limit: int = 60):
        self.base_url = "https://tickatlas.com/v1"
        self.headers = {"X-API-Key": api_key}
        self.limiter = RateLimiter(rate_limit)

    def get_indicator(self, symbol: str, indicator: str, timeframe: str) -> dict:
        self.limiter.wait()
        data = api_call_with_retry(
            f"{self.base_url}/indicator",
            self.headers,
            {"symbol": symbol, "indicator": indicator, "timeframe": timeframe},
        )
        if not validate_indicator_response(data, symbol):
            raise ValueError(f"Invalid response for {symbol}/{indicator}")
        return data["data"]

    def get_ohlcv(self, symbol: str, timeframe: str, limit: int = 50) -> list:
        self.limiter.wait()
        data = api_call_with_retry(
            f"{self.base_url}/ohlcv",
            self.headers,
            {"symbol": symbol, "timeframe": timeframe, "limit": limit},
        )
        return data["data"]["candles"]

    def get_spread(self, symbol: str) -> dict:
        self.limiter.wait()
        data = api_call_with_retry(
            f"{self.base_url}/spread",
            self.headers,
            {"symbol": symbol},
        )
        return data["data"]

# Usage
client = TickAtlasClient("YOUR_API_KEY", rate_limit=60)
rsi = client.get_indicator("EURUSD", "RSI_14", "H4")
print(f"RSI: {rsi['value']}")