| |
| """ |
| CryptoCompare API Client - Comprehensive crypto data with API key authentication |
| Official API: https://min-api.cryptocompare.com/ |
| """ |
|
|
| import httpx |
| import logging |
| import os |
| from typing import Dict, Any, List, Optional |
| from datetime import datetime |
|
|
| logger = logging.getLogger(__name__) |
|
|
| |
| CRYPTOCOMPARE_API_KEY = os.getenv("CRYPTOCOMPARE_KEY", "e79c8e6d4c5b4a3f2e1d0c9b8a7f6e5d4c3b2a1f") |
|
|
|
|
| class CryptoCompareClient: |
| """ |
| CryptoCompare API Client - Market data, news, social stats, and more |
| """ |
| |
| def __init__(self, api_key: str = CRYPTOCOMPARE_API_KEY): |
| self.base_url = "https://min-api.cryptocompare.com/data" |
| self.api_key = api_key |
| self.timeout = 15.0 |
| |
| def _get_headers(self) -> Dict[str, str]: |
| """Get request headers with API key""" |
| headers = { |
| "Content-Type": "application/json" |
| } |
| if self.api_key: |
| headers["authorization"] = f"Apikey {self.api_key}" |
| return headers |
| |
| async def get_price(self, symbols: List[str], currency: str = "USD") -> Dict[str, Any]: |
| """ |
| Get current prices for multiple symbols |
| |
| Args: |
| symbols: List of cryptocurrency symbols (e.g., ["BTC", "ETH"]) |
| currency: Target currency (default: USD) |
| |
| Returns: |
| Price data from CryptoCompare |
| """ |
| try: |
| async with httpx.AsyncClient(timeout=self.timeout) as client: |
| fsyms = ",".join([s.upper() for s in symbols]) |
| |
| response = await client.get( |
| f"{self.base_url}/pricemultifull", |
| params={ |
| "fsyms": fsyms, |
| "tsyms": currency |
| }, |
| headers=self._get_headers() |
| ) |
| response.raise_for_status() |
| data = response.json() |
| |
| result = { |
| "data": data.get("RAW", {}), |
| "display": data.get("DISPLAY", {}), |
| "source": "CryptoCompare", |
| "timestamp": datetime.utcnow().isoformat() |
| } |
| |
| logger.info(f"✅ CryptoCompare: Fetched prices for {len(symbols)} symbols") |
| return result |
| |
| except httpx.HTTPStatusError as e: |
| logger.error(f"❌ CryptoCompare API HTTP error: {e.response.status_code}") |
| raise |
| except Exception as e: |
| logger.error(f"❌ CryptoCompare API failed: {e}") |
| raise |
| |
| async def get_ohlcv( |
| self, |
| symbol: str, |
| currency: str = "USD", |
| limit: int = 100, |
| aggregate: int = 1 |
| ) -> Dict[str, Any]: |
| """ |
| Get OHLCV (candlestick) data |
| |
| Args: |
| symbol: Cryptocurrency symbol (e.g., "BTC") |
| currency: Target currency (default: USD) |
| limit: Number of data points (max 2000) |
| aggregate: Data aggregation (e.g., 1 = 1 hour, 24 = 1 day) |
| |
| Returns: |
| OHLCV data |
| """ |
| try: |
| async with httpx.AsyncClient(timeout=self.timeout) as client: |
| response = await client.get( |
| f"{self.base_url}/v2/histohour", |
| params={ |
| "fsym": symbol.upper(), |
| "tsym": currency.upper(), |
| "limit": limit, |
| "aggregate": aggregate |
| }, |
| headers=self._get_headers() |
| ) |
| response.raise_for_status() |
| data = response.json() |
| |
| if data.get("Response") == "Success": |
| logger.info(f"✅ CryptoCompare: Fetched OHLCV for {symbol}") |
| return { |
| "symbol": symbol.upper(), |
| "data": data.get("Data", {}).get("Data", []), |
| "source": "CryptoCompare", |
| "timestamp": datetime.utcnow().isoformat() |
| } |
| else: |
| raise Exception(f"CryptoCompare API error: {data.get('Message', 'Unknown error')}") |
| |
| except httpx.HTTPStatusError as e: |
| logger.error(f"❌ CryptoCompare OHLCV HTTP error: {e.response.status_code}") |
| raise |
| except Exception as e: |
| logger.error(f"❌ CryptoCompare OHLCV failed: {e}") |
| raise |
| |
| async def get_news(self, limit: int = 50) -> Dict[str, Any]: |
| """ |
| Get latest crypto news from CryptoCompare |
| |
| Args: |
| limit: Number of news articles (max 200) |
| |
| Returns: |
| News articles |
| """ |
| try: |
| async with httpx.AsyncClient(timeout=self.timeout) as client: |
| response = await client.get( |
| f"{self.base_url}/v2/news/", |
| params={ |
| "lang": "EN", |
| "limit": limit |
| }, |
| headers=self._get_headers() |
| ) |
| response.raise_for_status() |
| data = response.json() |
| |
| if data.get("Type") == 100: |
| articles = data.get("Data", []) |
| logger.info(f"✅ CryptoCompare: Fetched {len(articles)} news articles") |
| return { |
| "articles": articles, |
| "count": len(articles), |
| "source": "CryptoCompare", |
| "timestamp": datetime.utcnow().isoformat() |
| } |
| else: |
| raise Exception(f"CryptoCompare news error: {data.get('Message', 'Unknown error')}") |
| |
| except httpx.HTTPStatusError as e: |
| logger.error(f"❌ CryptoCompare news HTTP error: {e.response.status_code}") |
| raise |
| except Exception as e: |
| logger.error(f"❌ CryptoCompare news failed: {e}") |
| raise |
| |
| async def get_social_stats(self, coin_id: int) -> Dict[str, Any]: |
| """ |
| Get social statistics for a cryptocurrency |
| |
| Args: |
| coin_id: CryptoCompare coin ID |
| |
| Returns: |
| Social stats (Twitter, Reddit, etc.) |
| """ |
| try: |
| async with httpx.AsyncClient(timeout=self.timeout) as client: |
| response = await client.get( |
| f"{self.base_url}/social/coin/latest", |
| params={"coinId": coin_id}, |
| headers=self._get_headers() |
| ) |
| response.raise_for_status() |
| data = response.json() |
| |
| if data.get("Response") == "Success": |
| logger.info(f"✅ CryptoCompare: Fetched social stats for coin {coin_id}") |
| return { |
| "data": data.get("Data", {}), |
| "source": "CryptoCompare", |
| "timestamp": datetime.utcnow().isoformat() |
| } |
| else: |
| raise Exception(f"CryptoCompare social stats error: {data.get('Message', 'Unknown error')}") |
| |
| except httpx.HTTPStatusError as e: |
| logger.error(f"❌ CryptoCompare social stats HTTP error: {e.response.status_code}") |
| raise |
| except Exception as e: |
| logger.error(f"❌ CryptoCompare social stats failed: {e}") |
| raise |
| |
| async def get_top_exchanges_by_volume(self, symbol: str, currency: str = "USD", limit: int = 10) -> Dict[str, Any]: |
| """ |
| Get top exchanges by trading volume for a symbol |
| |
| Args: |
| symbol: Cryptocurrency symbol |
| currency: Target currency |
| limit: Number of exchanges to return |
| |
| Returns: |
| Top exchanges data |
| """ |
| try: |
| async with httpx.AsyncClient(timeout=self.timeout) as client: |
| response = await client.get( |
| f"{self.base_url}/top/exchanges/full", |
| params={ |
| "fsym": symbol.upper(), |
| "tsym": currency.upper(), |
| "limit": limit |
| }, |
| headers=self._get_headers() |
| ) |
| response.raise_for_status() |
| data = response.json() |
| |
| if data.get("Response") == "Success": |
| exchanges = data.get("Data", {}).get("Exchanges", []) |
| logger.info(f"✅ CryptoCompare: Fetched top {len(exchanges)} exchanges for {symbol}") |
| return { |
| "exchanges": exchanges, |
| "symbol": symbol.upper(), |
| "source": "CryptoCompare", |
| "timestamp": datetime.utcnow().isoformat() |
| } |
| else: |
| raise Exception(f"CryptoCompare exchanges error: {data.get('Message', 'Unknown error')}") |
| |
| except httpx.HTTPStatusError as e: |
| logger.error(f"❌ CryptoCompare exchanges HTTP error: {e.response.status_code}") |
| raise |
| except Exception as e: |
| logger.error(f"❌ CryptoCompare exchanges failed: {e}") |
| raise |
|
|
|
|
| |
| cryptocompare_client = CryptoCompareClient() |
|
|
|
|
| |
| async def fetch_cryptocompare_price(symbols: List[str]) -> Dict[str, Any]: |
| """Get prices from CryptoCompare""" |
| return await cryptocompare_client.get_price(symbols) |
|
|
|
|
| async def fetch_cryptocompare_news(limit: int = 50) -> Dict[str, Any]: |
| """Get news from CryptoCompare""" |
| return await cryptocompare_client.get_news(limit) |
|
|
|
|
| async def fetch_cryptocompare_ohlcv(symbol: str, limit: int = 100) -> Dict[str, Any]: |
| """Get OHLCV data from CryptoCompare""" |
| return await cryptocompare_client.get_ohlcv(symbol, limit=limit) |
|
|
|
|
| __all__ = [ |
| "CryptoCompareClient", |
| "cryptocompare_client", |
| "fetch_cryptocompare_price", |
| "fetch_cryptocompare_news", |
| "fetch_cryptocompare_ohlcv" |
| ] |
|
|
