75 lines
2.5 KiB
Python
75 lines
2.5 KiB
Python
"""
|
|
CoinGecko client (public free endpoints) to fetch BTC/ETH prices for plotting.
|
|
|
|
Rules:
|
|
- No synthetic data. Return empty structures or None when unavailable.
|
|
- Use shared RateLimiter for polite access and retries.
|
|
- Default to public endpoints that do not require API key; if a key is provided via env, include it.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from .api_rate_limiter import get_rate_limiter
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class CoinGeckoClient:
|
|
def __init__(self,
|
|
api_base_url: str = "https://api.coingecko.com/api/v3",
|
|
api_key_env: str = "COINGECKO_API_KEY",
|
|
user_agent: str = "gogo2-dashboard/1.0") -> None:
|
|
self.api_base_url = api_base_url.rstrip("/")
|
|
self.api_key = os.environ.get(api_key_env) or ""
|
|
self.user_agent = user_agent
|
|
self._rl = get_rate_limiter()
|
|
|
|
def get_simple_price(self, ids: List[str], vs_currency: str = "usd") -> Dict[str, Any]:
|
|
if not ids:
|
|
return {}
|
|
url = f"{self.api_base_url}/simple/price"
|
|
params = {
|
|
"ids": ",".join(ids),
|
|
"vs_currencies": vs_currency,
|
|
}
|
|
headers = {"User-Agent": self.user_agent}
|
|
# Optional key
|
|
if self.api_key:
|
|
params["x_cg_pro_api_key"] = self.api_key
|
|
resp = self._rl.make_request("coingecko_api", url, method="GET", params=params, headers=headers)
|
|
if resp is None:
|
|
return {}
|
|
try:
|
|
return resp.json() # type: ignore
|
|
except Exception as ex:
|
|
logger.error("CoinGecko simple price JSON error: %s", ex)
|
|
return {}
|
|
|
|
def get_market_chart(self, coin_id: str, vs_currency: str = "usd", days: int = 5, interval: str = "hourly") -> Dict[str, Any]:
|
|
if not coin_id:
|
|
return {}
|
|
url = f"{self.api_base_url}/coins/{coin_id}/market_chart"
|
|
params = {
|
|
"vs_currency": vs_currency,
|
|
"days": str(max(1, int(days))),
|
|
"interval": interval,
|
|
}
|
|
headers = {"User-Agent": self.user_agent}
|
|
if self.api_key:
|
|
params["x_cg_pro_api_key"] = self.api_key
|
|
resp = self._rl.make_request("coingecko_api", url, method="GET", params=params, headers=headers)
|
|
if resp is None:
|
|
return {}
|
|
try:
|
|
return resp.json() # type: ignore
|
|
except Exception as ex:
|
|
logger.error("CoinGecko market_chart JSON error: %s", ex)
|
|
return {}
|
|
|
|
|