How to Monitor Spreads in Real-Time

curl -H "X-API-Key: YOUR_API_KEY" \
  "https://tickatlas.com/v1/spread?symbol=EURUSD"

// GET /v1/spread?symbol=EURUSD
{
  "success": true,
  "data": {
    "symbol": "EURUSD",
    "current": {
      "spread_pips": 0.9,
      "spread_points": 9
    },
    "statistics": {
      "period": "24h",
      "avg_spread": 1.1,
      "min_spread": 0.6,
      "max_spread": 4.8,
      "std_deviation": 0.42
    },
    "by_session": {
      "asian": 1.4,
      "london": 0.8,
      "new_york": 1.0
    }
  }
}

import requests
import time
import logging
from collections import deque
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("spread_monitor")

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://tickatlas.com/v1"

class SpreadMonitor:
    def __init__(self, symbols: list[str], alert_multiplier: float = 2.0):
        self.symbols = symbols
        self.alert_multiplier = alert_multiplier
        # Track recent spreads for each symbol (rolling window)
        self.history: dict[str, deque] = {
            sym: deque(maxlen=100) for sym in symbols
        }

    def fetch_spread(self, symbol: str) -> dict:
        resp = requests.get(
            f"{BASE_URL}/spread",
            headers={"X-API-Key": API_KEY},
            params={"symbol": symbol},
        )
        resp.raise_for_status()
        return resp.json()["data"]

    def check_all(self) -> list[dict]:
        """Check spreads for all symbols. Return any alerts."""
        alerts = []
        for symbol in self.symbols:
            data = self.fetch_spread(symbol)
            spread = data["current"]["spread_pips"]
            self.history[symbol].append(spread)

            # Calculate rolling average
            history = list(self.history[symbol])
            if len(history) < 10:
                continue  # Need enough data

            avg_spread = sum(history) / len(history)
            threshold = avg_spread * self.alert_multiplier

            if spread > threshold:
                alert = {
                    "symbol": symbol,
                    "current_spread": spread,
                    "average_spread": round(avg_spread, 2),
                    "multiplier": round(spread / avg_spread, 1),
                    "timestamp": datetime.utcnow().isoformat(),
                }
                alerts.append(alert)
                logger.warning(
                    f"SPREAD ALERT: {symbol} at {spread} pips "
                    f"({alert['multiplier']}x average)"
                )

        return alerts

# Usage
monitor = SpreadMonitor(
    ["EURUSD", "GBPUSD", "USDJPY", "XAUUSD"],
    alert_multiplier=2.0,
)

while True:
    alerts = monitor.check_all()
    for a in alerts:
        print(f"WARNING: {a['symbol']} spread {a['current_spread']} pips")
    time.sleep(30)  # Check every 30 seconds

def should_trade(symbol: str, max_spread_pips: float) -> bool:
    """Check if spread conditions allow trading."""
    resp = requests.get(
        f"{BASE_URL}/spread",
        headers={"X-API-Key": API_KEY},
        params={"symbol": symbol},
    )
    spread = resp.json()["data"]["current"]["spread_pips"]

    if spread > max_spread_pips:
        logger.info(f"Skipping {symbol}: spread {spread} > {max_spread_pips}")
        return False
    return True

# In your trading bot:
if should_trade("EURUSD", max_spread_pips=2.0):
    execute_strategy("EURUSD")
else:
    logger.info("Waiting for better spread conditions")