crypto-mcp/providers/coingecko.py

"""
CoinGecko data provider.
No API key required for basic endpoints.
"""

import time
import httpx

from config import COINGECKO_BASE_URL, SYMBOL_TO_COINGECKO_ID
from errors import SymbolNotFoundError, ProviderError


def _resolve_coingecko_id(symbol: str) -> str:
    sym = symbol.upper()
    cg_id = SYMBOL_TO_COINGECKO_ID.get(sym)
    if not cg_id:
        raise SymbolNotFoundError(f"Unknown symbol: {symbol}. Not in CoinGecko map.")
    return cg_id


async def fetch_price(symbol: str) -> dict:
    cg_id = _resolve_coingecko_id(symbol)
    url = f"{COINGECKO_BASE_URL}/simple/price"
    params = {"ids": cg_id, "vs_currencies": "usd", "include_last_updated_at": "true"}

    async with httpx.AsyncClient(timeout=10.0) as client:
        try:
            resp = await client.get(url, params=params)
            resp.raise_for_status()
        except httpx.HTTPStatusError as e:
            raise ProviderError(f"CoinGecko HTTP error: {e.response.status_code}") from e
        except httpx.RequestError as e:
            raise ProviderError(f"CoinGecko request failed: {e}") from e

    data = resp.json()
    if cg_id not in data:
        raise SymbolNotFoundError(f"CoinGecko returned no data for {symbol}")

    entry = data[cg_id]
    return {"symbol": symbol.upper(), "price": float(entry["usd"]), "timestamp": int(entry.get("last_updated_at", time.time()))}


async def fetch_ohlcv(symbol: str, timeframe: str, limit: int = 100) -> dict:
    cg_id = _resolve_coingecko_id(symbol)
    days_map = {"1m": 1, "5m": 1, "15m": 1, "1h": 7, "4h": 14, "1d": 90}
    days = days_map.get(timeframe, 7)

    url = f"{COINGECKO_BASE_URL}/coins/{cg_id}/ohlc"
    params = {"vs_currency": "usd", "days": days}

    async with httpx.AsyncClient(timeout=15.0) as client:
        try:
            resp = await client.get(url, params=params)
            resp.raise_for_status()
        except httpx.HTTPStatusError as e:
            raise ProviderError(f"CoinGecko OHLC HTTP error: {e.response.status_code}") from e
        except httpx.RequestError as e:
            raise ProviderError(f"CoinGecko OHLC request failed: {e}") from e

    raw = resp.json()
    if not raw:
        raise ProviderError(f"CoinGecko returned empty OHLCV for {symbol}")

    candles = [[int(row[0] / 1000), float(row[1]), float(row[2]), float(row[3]), float(row[4]), 0.0] for row in raw]
    return {"symbol": symbol.upper(), "timeframe": timeframe, "candles": candles[-limit:]}


async def fetch_top_movers(limit: int = 10) -> dict:
    url = f"{COINGECKO_BASE_URL}/coins/markets"
    params = {"vs_currency": "usd", "order": "market_cap_desc", "per_page": 100, "page": 1, "price_change_percentage": "24h", "sparkline": "false"}

    async with httpx.AsyncClient(timeout=15.0) as client:
        try:
            resp = await client.get(url, params=params)
            resp.raise_for_status()
        except httpx.HTTPStatusError as e:
            raise ProviderError(f"CoinGecko markets HTTP error: {e.response.status_code}") from e
        except httpx.RequestError as e:
            raise ProviderError(f"CoinGecko markets request failed: {e}") from e

    coins = resp.json()

    def _format(coin: dict) -> dict:
        return {"symbol": coin["symbol"].upper(), "name": coin["name"], "price": coin.get("current_price"), "change_24h_pct": coin.get("price_change_percentage_24h"), "market_cap": coin.get("market_cap")}

    sorted_by_change = sorted([c for c in coins if c.get("price_change_percentage_24h") is not None], key=lambda c: c["price_change_percentage_24h"], reverse=True)
    return {"gainers": [_format(c) for c in sorted_by_change[:limit]], "losers": [_format(c) for c in sorted_by_change[-limit:][::-1]]}