File size: 2,082 Bytes
0ee2df7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 | """
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}
# default: markets
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}
|