exec-mcp/src/exec_mcp/bitstamp_metadata.py

from __future__ import annotations

import requests

from .storage import get_connection

BITSTAMP_BASE_URL = "https://www.bitstamp.net"
METADATA_REFRESH_SECONDS = 24 * 60 * 60


def fetch_currencies() -> list[dict]:
    response = requests.get(f"{BITSTAMP_BASE_URL}/api/v2/currencies/", timeout=30)
    response.raise_for_status()
    return response.json()


def fetch_markets() -> list[dict]:
    response = requests.get(f"{BITSTAMP_BASE_URL}/api/v2/markets/", timeout=30)
    response.raise_for_status()
    return response.json()


def save_metadata(kind: str, payload: list[dict]) -> None:
    with get_connection() as conn:
        conn.execute("DELETE FROM bitstamp_metadata WHERE kind = ?", (kind,))
        for item in payload:
            conn.execute(
                "INSERT INTO bitstamp_metadata (kind, item_key, payload_json) VALUES (?, ?, ?)",
                (kind, _item_key(kind, item), __import__("json").dumps(item)),
            )
        conn.commit()


def load_metadata(kind: str) -> list[dict]:
    with get_connection() as conn:
        rows = conn.execute(
            "SELECT payload_json FROM bitstamp_metadata WHERE kind = ? ORDER BY item_key ASC",
            (kind,),
        ).fetchall()
    return [__import__("json").loads(row["payload_json"]) for row in rows]


def load_market_by_symbol(market_symbol: str) -> dict | None:
    symbol = market_symbol.lower()
    for item in load_metadata("markets"):
        if str(item.get("market_symbol", "")).lower() == symbol:
            return item
    return None


def list_markets() -> list[dict]:
    return load_metadata("markets")


def list_currencies() -> list[dict]:
    return load_metadata("currencies")


def refresh_metadata() -> dict:
    currencies = fetch_currencies()
    markets = fetch_markets()
    save_metadata("currencies", currencies)
    save_metadata("markets", markets)
    return {"currencies": len(currencies), "markets": len(markets)}


def _item_key(kind: str, item: dict) -> str:
    if kind == "currencies":
        return str(item.get("code") or item.get("currency") or item.get("id") or "")
    if kind == "markets":
        return str(item.get("name") or item.get("pair") or item.get("url_symbol") or item.get("id") or "")
    return str(item.get("id") or item.get("name") or "")