metals-mcp/discover_swissquote_api.py

import requests
import time
import json
import random
from datetime import datetime, UTC
from concurrent.futures import ThreadPoolExecutor, as_completed
from threading import Lock

BASE_URL = "https://forex-data-feed.swissquote.com/public-quotes/bboquotes/instrument/{}"

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0 Safari/537.36",
    "Accept": "application/json, text/plain, */*",
    "Accept-Language": "en-US,en;q=0.9",
    "Connection": "keep-alive"
}

BASE_ASSETS = [
    "XAU", "XAG", "XPT", "XPD",
    "EUR", "USD", "GBP", "JPY", "CHF", "AUD", "CAD", "NZD"
]

QUOTE_ASSETS = [
    "USD", "EUR", "JPY", "CHF", "GBP"
]

OUTPUT_FILE = "swissquote_pairs.json"

# Rate limiting
REQUESTS_PER_SECOND = 5
MIN_INTERVAL = 1.0 / REQUESTS_PER_SECOND
last_request_time = 0
rate_lock = Lock()


def rate_limited_request():
    global last_request_time
    with rate_lock:
        now = time.time()
        elapsed = now - last_request_time
        if elapsed < MIN_INTERVAL:
            time.sleep(MIN_INTERVAL - elapsed)
        last_request_time = time.time()


def fetch(symbol, retries=2):
    url = BASE_URL.format(symbol)

    for attempt in range(retries + 1):
        try:
            rate_limited_request()
            time.sleep(random.uniform(0.02, 0.08))  # jitter

            r = requests.get(url, headers=HEADERS, timeout=3)

            if r.status_code != 200:
                raise Exception(f"HTTP {r.status_code}")

            data = r.json()

            # Normalize response (dict or list)
            if isinstance(data, list):
                if not data:
                    return None
                data = data[0]

            if not isinstance(data, dict):
                return None

            prices = data.get("spreadProfilePrices")
            if not prices:
                return None

            p = prices[0]

            bid = float(p.get("bid", 0))
            ask = float(p.get("ask", 0))
            ts = p.get("timestamp")

            if bid > 0 and ask > 0:
                return {
                    "symbol": symbol,
                    "bid": bid,
                    "ask": ask,
                    "timestamp": ts
                }

            return None

        except Exception as e:
            if attempt < retries:
                time.sleep(0.3 * (2 ** attempt))  # backoff
            else:
                print(f"[FAIL] {symbol} → {e}")

    return None


def main():
    symbols = [
        f"{b}/{q}"
        for b in BASE_ASSETS
        for q in QUOTE_ASSETS
        if b != q
    ]

    print(f"Checking {len(symbols)} symbols...\n")

    results = []

    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = {executor.submit(fetch, s): s for s in symbols}

        for future in as_completed(futures):
            symbol = futures[future]
            result = future.result()

            if result:
                print(f"✔ {symbol}")
                results.append(result)
            else:
                print(f"✘ {symbol}")

    output = {
        "generated_at": datetime.now(UTC).isoformat(),
        "total_checked": len(symbols),
        "total_valid": len(results),
        "pairs": sorted(results, key=lambda x: x["symbol"])
    }

    with open(OUTPUT_FILE, "w") as f:
        json.dump(output, f, indent=2)

    print("\n=== DONE ===")
    print(f"Valid pairs: {len(results)}")
    print(f"Saved to: {OUTPUT_FILE}")


if __name__ == "__main__":
    main()