| """ |
| CoinGecko -- free public API, no key required (300 calls/min). |
| """ |
| import httpx |
|
|
| BASE = "https://api.coingecko.com/api/v3" |
|
|
|
|
| def run(params: dict) -> dict: |
| mode = params.get("mode", "markets") |
|
|
| if mode == "trending": |
| resp = httpx.get(f"{BASE}/search/trending", timeout=15.0) |
| resp.raise_for_status() |
| coins = resp.json().get("coins", []) |
| results = [ |
| { |
| "name": c["item"]["name"], |
| "symbol": c["item"]["symbol"], |
| "market_cap_rank": c["item"]["market_cap_rank"], |
| } |
| for c in coins |
| ] |
| return {"mode": "trending", "count": len(results), "results": results} |
|
|
| if mode == "ohlc": |
| coin_id = params.get("coin_id") |
| if not coin_id: |
| raise ValueError("coin_id is required for mode='ohlc'") |
| days = int(params.get("days", 30)) |
| resp = httpx.get( |
| f"{BASE}/coins/{coin_id}/ohlc", params={"vs_currency": "usd", "days": days}, timeout=15.0 |
| ) |
| resp.raise_for_status() |
| candles = resp.json() |
| results = [{"time": c[0], "open": c[1], "high": c[2], "low": c[3], "close": c[4]} for c in candles] |
| return {"mode": "ohlc", "coin_id": coin_id, "count": len(results), "results": results} |
|
|
| |
| per_page = int(params.get("per_page", 100)) |
| resp = httpx.get( |
| f"{BASE}/coins/markets", |
| params={"vs_currency": "usd", "order": "market_cap_desc", "per_page": min(per_page, 250), "page": 1}, |
| timeout=15.0, |
| ) |
| resp.raise_for_status() |
| coins = resp.json() |
| results = [ |
| { |
| "id": c["id"], |
| "symbol": c["symbol"], |
| "price": c["current_price"], |
| "market_cap": c["market_cap"], |
| "volume_24h": c["total_volume"], |
| "change_24h_pct": c["price_change_percentage_24h"], |
| "change_7d_pct": c.get("price_change_percentage_7d_in_currency"), |
| } |
| for c in coins |
| ] |
| return {"mode": "markets", "count": len(results), "results": results} |
|
|
