| """CryptoCompare public/API-key optional fallback adapter.""" |
|
|
| from __future__ import annotations |
|
|
| import os |
| from typing import Dict, List |
|
|
| from .base import ProviderResult, base_asset, fail, fetch_json, normalize_spot_symbol, ok, to_float, to_int |
|
|
| BASE = "https://min-api.cryptocompare.com/data" |
|
|
|
|
| def _params(params: Dict[str, object]) -> Dict[str, object]: |
| key = os.getenv("CRYPTOCOMPARE_KEY") or os.getenv("CRYPTOCOMPARE_API_KEY") |
| if key: |
| params = dict(params) |
| params["api_key"] = key |
| return params |
|
|
|
|
| async def ticker(symbol: str) -> ProviderResult: |
| base = base_asset(symbol) |
| payload, error, status, latency = await fetch_json("cryptocompare", "ticker", f"{BASE}/pricemultifull", params=_params({"fsyms": base, "tsyms": "USD"}), timeout=8.0) |
| if error or not isinstance(payload, dict): |
| return fail("cryptocompare", "ticker", error or "price payload invalid", latency, status) |
| raw = ((payload.get("RAW") or {}).get(base) or {}).get("USD") or {} |
| data = { |
| "symbol": normalize_spot_symbol(symbol), |
| "price": to_float(raw.get("PRICE")), |
| "lastPrice": to_float(raw.get("PRICE")), |
| "volume24h": to_float(raw.get("VOLUME24HOUR")), |
| "quoteVolume24h": to_float(raw.get("VOLUME24HOURTO")), |
| "priceChangePercent": to_float(raw.get("CHANGEPCT24HOUR")), |
| "high24h": to_float(raw.get("HIGH24HOUR")), |
| "low24h": to_float(raw.get("LOW24HOUR")), |
| "provider": "cryptocompare", |
| } |
| if data["price"] is None: |
| return fail("cryptocompare", "ticker", f"price missing for {base}", latency, status) |
| return ok("cryptocompare", "ticker", data, latency) |
|
|
|
|
| async def ohlcv(symbol: str, interval: str = "1h", limit: int = 100) -> ProviderResult: |
| base = base_asset(symbol) |
| tf = (interval or "1h").lower() |
| if tf.endswith("m"): |
| endpoint = "histominute"; aggregate = max(1, to_int(tf[:-1], 1) or 1) |
| elif tf.endswith("d"): |
| endpoint = "histoday"; aggregate = max(1, to_int(tf[:-1], 1) or 1) |
| else: |
| endpoint = "histohour"; aggregate = max(1, to_int(tf[:-1].replace('h',''), 1) or 1) |
| payload, error, status, latency = await fetch_json( |
| "cryptocompare", |
| "ohlcv", |
| f"{BASE}/v2/{endpoint}", |
| params=_params({"fsym": base, "tsym": "USD", "limit": min(max(int(limit), 1), 2000), "aggregate": aggregate}), |
| timeout=10.0, |
| ) |
| if error or not isinstance(payload, dict): |
| return fail("cryptocompare", "ohlcv", error or "ohlcv payload invalid", latency, status) |
| rows = ((payload.get("Data") or {}).get("Data") or [])[-limit:] |
| candles = [] |
| for row in rows: |
| if not isinstance(row, dict): |
| continue |
| candles.append({ |
| "timestamp": to_int(row.get("time")) * 1000 if to_int(row.get("time")) else None, |
| "open": to_float(row.get("open")), |
| "high": to_float(row.get("high")), |
| "low": to_float(row.get("low")), |
| "close": to_float(row.get("close")), |
| "volume": to_float(row.get("volumefrom")), |
| "quoteVolume": to_float(row.get("volumeto")), |
| "provider": "cryptocompare", |
| }) |
| candles = [c for c in candles if c.get("timestamp") is not None] |
| if not candles: |
| return fail("cryptocompare", "ohlcv", f"no candles for {base}", latency, status) |
| return ok("cryptocompare", "ohlcv", candles, latency) |
|
|
