wip misc; cleaup launch
This commit is contained in:
74
core/coingecko_client.py
Normal file
74
core/coingecko_client.py
Normal file
@@ -0,0 +1,74 @@
|
||||
"""
|
||||
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 {}
|
||||
|
||||
|
Reference in New Issue
Block a user