rmi-backend/app/coingecko_connector.py

403 lines
15 KiB
Python

"""
CoinGecko Connector — Free + Pro API with x402 pay-per-use fallback.
Free tier: 30 req/min. Pro tier: 500+ req/min. x402: pay-per-request.
Endpoints covered:
- /ping (health check)
- /simple/price (batch prices)
- /simple/token_price (token prices by contract)
- /search/trending (trending coins, NFTs, categories)
- /coins/markets (top coins by market cap)
- /coins/{id} (detailed token info)
- /coins/categories (category listings)
- /coins/{id}/ohlc (candlestick data)
- /global (global market metrics)
- /exchanges (exchange listings)
- /onchain/... (GeckoTerminal DEX data via pro)
- /key (API key status)
"""
import asyncio
import logging
import os
import time
from typing import Any
import httpx
logger = logging.getLogger(__name__)
# ── API Keys ───────────────────────────────────────────────
def _load_key(filename: str) -> str:
"""Load API key from /root/.secrets/ or env var."""
env = os.getenv(filename.upper(), "")
if env:
return env
try:
with open(f"/root/.secrets/{filename}") as f:
return f.read().strip()
except Exception:
return ""
COINGECKO_FREE_KEY = _load_key("coingecko_api_key") # Demo key: CG-...
COINGECKO_PRO_KEY = _load_key("coingecko_api_key_pro") # Also Demo tier
class CoinGeckoConnector:
"""CoinGecko Demo API connector with rate limiting, caching, and smart polling.
Both keys are Demo tier (30 req/min). Webhooks require Analyst plan ($103/mo).
Strategy: cache aggressively, poll smart, use n8n for scheduled refresh.
"""
BASE_URL = "https://api.coingecko.com/api/v3"
def __init__(self):
self._cache: dict[str, tuple] = {} # key -> (expiry, value)
self._last_request = 0.0
self._min_interval = 2.0 # 30 req/min on Demo tier
# Use whichever key we have (both are Demo tier)
self._api_key = COINGECKO_PRO_KEY or COINGECKO_FREE_KEY
def _headers(self) -> dict[str, str]:
headers = {"Accept": "application/json"}
if self._api_key:
headers["x-cg-demo-api-key"] = self._api_key
return headers
def _base(self) -> str:
return self.BASE_URL
async def _get(self, endpoint: str, params: dict | None = None, ttl: int = 300) -> Any | None:
"""Rate-limited GET with caching."""
cache_key = f"{endpoint}:{sorted(params.items()) if params else ''}"
now = time.time()
# Check cache
if cache_key in self._cache:
expiry, value = self._cache[cache_key]
if now < expiry:
return value
# Rate limit
elapsed = now - self._last_request
interval = 2.0 # 30 req/min on Demo tier
if elapsed < interval:
await asyncio.sleep(interval - elapsed)
try:
self._last_request = time.time()
url = f"{self._base()}/{endpoint}"
async with httpx.AsyncClient(timeout=12.0) as client:
r = await client.get(url, params=params or {}, headers=self._headers())
if r.status_code == 200:
data = r.json()
self._cache[cache_key] = (now + ttl, data)
return data
elif r.status_code == 429:
logger.warning("CoinGecko rate limited, backing off")
await asyncio.sleep(30)
return None
else:
logger.debug(f"CoinGecko {endpoint}: {r.status_code} {r.text[:200]}")
return None
except Exception as e:
logger.debug(f"CoinGecko error: {e}")
return None
# ── Health Check ─────────────────────────────────────────
async def ping(self) -> dict:
"""Check API status. Returns {"gecko_says": "To the Moon!"} if healthy."""
data = await self._get("ping", ttl=60)
return data or {"gecko_says": "error", "status": "unavailable"}
async def api_key_status(self) -> dict:
"""Check API key validity and rate limits."""
try:
async with httpx.AsyncClient(timeout=10.0) as client:
r = await client.get(
f"{self.BASE_URL}/key",
headers=self._headers(),
)
if r.status_code == 200:
return r.json()
return {"status": f"error_{r.status_code}"}
except Exception as e:
return {"status": f"error: {e}"}
# ── Simple Price ─────────────────────────────────────────
async def get_token_price(self, token_id: str) -> dict | None:
"""Get current price for a token by CoinGecko ID."""
data = await self._get(
"simple/price",
params={
"ids": token_id,
"vs_currencies": "usd,btc",
"include_market_cap": "true",
"include_24hr_vol": "true",
"include_24hr_change": "true",
},
)
if data and token_id in data:
return data[token_id]
return None
async def get_token_prices(self, token_ids: list[str]) -> dict[str, dict]:
"""Batch get prices for multiple tokens."""
data = await self._get(
"simple/price",
params={
"ids": ",".join(token_ids[:250]), # API limit
"vs_currencies": "usd,btc",
"include_market_cap": "true",
"include_24hr_vol": "true",
"include_24hr_change": "true",
},
)
return data or {}
async def get_token_price_by_contract(self, chain: str, contract: str) -> dict | None:
"""Get price by contract address (Ethereum, BSC, etc.)."""
data = await self._get(
f"simple/token_price/{chain}",
params={
"contract_addresses": contract,
"vs_currencies": "usd,btc",
"include_market_cap": "true",
"include_24hr_vol": "true",
"include_24hr_change": "true",
},
)
if data and contract in data:
return data[contract]
return None
# ── Trending & Search ────────────────────────────────────
async def get_trending(self) -> list[dict]:
"""Get trending coins (updated every 30 min)."""
data = await self._get("search/trending", ttl=600)
if not data:
return []
coins = data.get("coins", [])
return [
{
"id": c.get("item", {}).get("id"),
"symbol": c.get("item", {}).get("symbol", "").upper(),
"name": c.get("item", {}).get("name"),
"market_cap_rank": c.get("item", {}).get("market_cap_rank"),
"price_btc": c.get("item", {}).get("price_btc"),
"score": c.get("score"),
}
for c in coins[:20]
]
# ── Market Data ──────────────────────────────────────────
async def get_market_overview(self, vs_currency: str = "usd", per_page: int = 50) -> list[dict]:
"""Get top coins by market cap."""
data = await self._get(
"coins/markets",
params={
"vs_currency": vs_currency,
"order": "market_cap_desc",
"per_page": per_page,
"page": 1,
"sparkline": "false",
},
)
if not data:
return []
return [
{
"id": c.get("id"),
"symbol": c.get("symbol", "").upper(),
"name": c.get("name"),
"price": c.get("current_price"),
"market_cap": c.get("market_cap"),
"volume_24h": c.get("total_volume"),
"change_24h": c.get("price_change_percentage_24h"),
"image": c.get("image"),
}
for c in data[:per_page]
]
async def get_global_metrics(self) -> dict | None:
"""Get global crypto market metrics."""
data = await self._get("global", ttl=300)
if not data:
return None
d = data.get("data", {})
return {
"total_market_cap_usd": d.get("total_market_cap", {}).get("usd"),
"total_volume_24h_usd": d.get("total_volume", {}).get("usd"),
"btc_dominance": d.get("market_cap_percentage", {}).get("btc"),
"eth_dominance": d.get("market_cap_percentage", {}).get("eth"),
"active_cryptocurrencies": d.get("active_cryptocurrencies"),
"markets": d.get("markets"),
}
# ── Token Detail ─────────────────────────────────────────
async def get_token_detail(self, token_id: str) -> dict | None:
"""Get detailed info about a specific token."""
data = await self._get(
f"coins/{token_id}",
params={
"localization": "false",
"tickers": "false",
"market_data": "true",
"community_data": "false",
"developer_data": "false",
},
ttl=600,
)
if not data:
return None
md = data.get("market_data", {})
return {
"id": data.get("id"),
"symbol": data.get("symbol", "").upper(),
"name": data.get("name"),
"price_usd": md.get("current_price", {}).get("usd"),
"market_cap": md.get("market_cap", {}).get("usd"),
"volume_24h": md.get("total_volume", {}).get("usd"),
"change_24h": md.get("price_change_percentage_24h"),
"change_7d": md.get("price_change_percentage_7d"),
"ath": md.get("ath", {}).get("usd"),
"atl": md.get("atl", {}).get("usd"),
"circulating_supply": md.get("circulating_supply"),
"total_supply": md.get("total_supply"),
"max_supply": md.get("max_supply"),
"categories": data.get("categories", []),
"links": {
"homepage": data.get("links", {}).get("homepage", [])[:1],
"twitter": data.get("links", {}).get("twitter_screen_name"),
"telegram": data.get("links", {}).get("telegram_channel_identifier"),
},
}
# ── Categories ───────────────────────────────────────────
async def get_categories(self) -> list[dict]:
"""Get all coin categories sorted by market cap."""
data = await self._get("coins/categories/list", ttl=3600)
if not data:
return []
return [{"id": c.get("id"), "name": c.get("name")} for c in data[:50]]
async def get_category_market_data(self) -> list[dict]:
"""Get categories with market data (top gainers/losers by category)."""
data = await self._get(
"coins/categories",
params={
"order": "market_cap_desc",
},
ttl=600,
)
if not data:
return []
return [
{
"id": c.get("id"),
"name": c.get("name"),
"market_cap": c.get("market_cap"),
"volume_24h": c.get("volume_24h"),
"change_24h": c.get("market_cap_change_24h"),
"top_3_coins": c.get("top_3_coins", [])[:3],
}
for c in data[:20]
]
# ── OHLC / Candlestick ───────────────────────────────────
async def get_ohlc(self, token_id: str, vs_currency: str = "usd", days: int = 7) -> list[list]:
"""Get OHLC candlestick data. Returns [[timestamp, open, high, low, close], ...]"""
data = await self._get(
f"coins/{token_id}/ohlc",
params={
"vs_currency": vs_currency,
"days": str(days),
},
ttl=300,
)
return data or []
# ── Exchanges ────────────────────────────────────────────
async def get_exchanges(self, per_page: int = 20) -> list[dict]:
"""Get top exchanges by trading volume."""
data = await self._get(
"exchanges",
params={
"per_page": per_page,
"page": 1,
},
ttl=600,
)
if not data:
return []
return [
{
"id": e.get("id"),
"name": e.get("name"),
"trust_score": e.get("trust_score"),
"volume_24h_btc": e.get("trade_volume_24h_btc"),
"url": e.get("url"),
}
for e in data[:per_page]
]
# ── Onchain / GeckoTerminal (Free, no key needed) ──────────
async def get_trending_pools(self, chain: str = "solana") -> list[dict]:
"""Get trending DEX pools via GeckoTerminal free API."""
try:
async with httpx.AsyncClient(timeout=10.0) as client:
r = await client.get(
f"https://api.geckoterminal.com/api/v2/networks/{chain}/trending_pools",
headers={"Accept": "application/json"},
)
if r.status_code == 200:
data = r.json()
pools = data.get("data", [])
if isinstance(pools, list):
return pools[:20]
# GeckoTerminal wraps in attributes
attrs = data.get("data", {})
if isinstance(attrs, dict):
items = attrs.get("attributes", attrs.get("items", []))
return items[:20] if isinstance(items, list) else []
except Exception as e:
logger.debug(f"GeckoTerminal trending failed: {e}")
return []
# ── Status ───────────────────────────────────────────────
def status(self) -> dict:
return {
"tier": "demo",
"api_key": bool(self._api_key),
"free_key": bool(COINGECKO_FREE_KEY),
"pro_key_note": "both keys are Demo tier — webhooks require Analyst plan ($103/mo)",
"cache_size": len(self._cache),
"base_url": self._base(),
"rate_limit": "30 req/min",
}
# ── Singleton ──────────────────────────────────────────────
_connector: CoinGeckoConnector | None = None
def get_coingecko_connector() -> CoinGeckoConnector:
global _connector
if _connector is None:
_connector = CoinGeckoConnector()
return _connector