Your Name
feat: UI improvements and error suppression - Enhanced dashboard and market pages with improved header buttons, logo, and currency symbol display - Stopped animated ticker - Removed pie chart legends - Added error suppressor for external service errors (SSE, Permissions-Policy warnings) - Improved header button prominence and icon appearance - Enhanced logo with glow effects and better design - Fixed currency symbol visibility in market tables
8b7b267
| #!/usr/bin/env python3 | |
| """ | |
| CoinGecko API Client - REAL DATA ONLY | |
| Fetches real cryptocurrency market data from CoinGecko | |
| NO MOCK DATA - All data from live CoinGecko API | |
| """ | |
| import httpx | |
| import logging | |
| from typing import Dict, Any, List, Optional | |
| from datetime import datetime | |
| from fastapi import HTTPException | |
| logger = logging.getLogger(__name__) | |
| class CoinGeckoClient: | |
| """ | |
| Real CoinGecko API Client | |
| Primary source for real-time cryptocurrency market prices | |
| """ | |
| def __init__(self): | |
| self.base_url = "https://api.coingecko.com/api/v3" | |
| self.timeout = 15.0 | |
| # Symbol to CoinGecko ID mapping | |
| self.symbol_to_id = { | |
| "BTC": "bitcoin", | |
| "ETH": "ethereum", | |
| "BNB": "binancecoin", | |
| "XRP": "ripple", | |
| "ADA": "cardano", | |
| "DOGE": "dogecoin", | |
| "SOL": "solana", | |
| "TRX": "tron", | |
| "DOT": "polkadot", | |
| "MATIC": "matic-network", | |
| "LTC": "litecoin", | |
| "SHIB": "shiba-inu", | |
| "AVAX": "avalanche-2", | |
| "UNI": "uniswap", | |
| "LINK": "chainlink", | |
| "ATOM": "cosmos", | |
| "XLM": "stellar", | |
| "ETC": "ethereum-classic", | |
| "XMR": "monero", | |
| "BCH": "bitcoin-cash" | |
| } | |
| # Reverse mapping | |
| self.id_to_symbol = {v: k for k, v in self.symbol_to_id.items()} | |
| def _symbol_to_coingecko_id(self, symbol: str) -> str: | |
| """Convert crypto symbol to CoinGecko coin ID""" | |
| symbol = symbol.upper().replace("USDT", "").replace("USD", "") | |
| return self.symbol_to_id.get(symbol, symbol.lower()) | |
| def _coingecko_id_to_symbol(self, coin_id: str) -> str: | |
| """Convert CoinGecko coin ID to symbol""" | |
| return self.id_to_symbol.get(coin_id, coin_id.upper()) | |
| async def get_market_prices( | |
| self, | |
| symbols: Optional[List[str]] = None, | |
| limit: int = 100 | |
| ) -> List[Dict[str, Any]]: | |
| """ | |
| Fetch REAL market prices from CoinGecko | |
| Args: | |
| symbols: List of crypto symbols (e.g., ["BTC", "ETH"]) | |
| limit: Maximum number of results | |
| Returns: | |
| List of real market data | |
| """ | |
| try: | |
| async with httpx.AsyncClient(timeout=self.timeout) as client: | |
| if symbols: | |
| # Get specific symbols using /simple/price endpoint | |
| coin_ids = [self._symbol_to_coingecko_id(s) for s in symbols] | |
| response = await client.get( | |
| f"{self.base_url}/simple/price", | |
| params={ | |
| "ids": ",".join(coin_ids), | |
| "vs_currencies": "usd", | |
| "include_24hr_change": "true", | |
| "include_24hr_vol": "true", | |
| "include_market_cap": "true" | |
| } | |
| ) | |
| response.raise_for_status() | |
| data = response.json() | |
| # Transform to standard format | |
| prices = [] | |
| for coin_id, coin_data in data.items(): | |
| symbol = self._coingecko_id_to_symbol(coin_id) | |
| prices.append({ | |
| "symbol": symbol, | |
| "name": symbol, # CoinGecko simple/price doesn't include name | |
| "price": coin_data.get("usd", 0), | |
| "change24h": coin_data.get("usd_24h_change", 0), | |
| "changePercent24h": coin_data.get("usd_24h_change", 0), | |
| "volume24h": coin_data.get("usd_24h_vol", 0), | |
| "marketCap": coin_data.get("usd_market_cap", 0), | |
| "source": "coingecko", | |
| "timestamp": int(datetime.utcnow().timestamp() * 1000) | |
| }) | |
| logger.info(f"✅ CoinGecko: Fetched {len(prices)} real prices for specific symbols") | |
| return prices | |
| else: | |
| # Get top coins by market cap using /coins/markets endpoint | |
| response = await client.get( | |
| f"{self.base_url}/coins/markets", | |
| params={ | |
| "vs_currency": "usd", | |
| "order": "market_cap_desc", | |
| "per_page": min(limit, 250), | |
| "page": 1, | |
| "sparkline": "false", | |
| "price_change_percentage": "24h" | |
| } | |
| ) | |
| response.raise_for_status() | |
| data = response.json() | |
| # Transform to standard format | |
| prices = [] | |
| for coin in data: | |
| prices.append({ | |
| "symbol": coin.get("symbol", "").upper(), | |
| "name": coin.get("name", ""), | |
| "price": coin.get("current_price", 0), | |
| "change24h": coin.get("price_change_24h", 0), | |
| "changePercent24h": coin.get("price_change_percentage_24h", 0), | |
| "volume24h": coin.get("total_volume", 0), | |
| "marketCap": coin.get("market_cap", 0), | |
| "source": "coingecko", | |
| "timestamp": int(datetime.utcnow().timestamp() * 1000) | |
| }) | |
| logger.info(f"✅ CoinGecko: Fetched {len(prices)} real market prices") | |
| return prices | |
| except httpx.HTTPError as e: | |
| logger.error(f"❌ CoinGecko API HTTP error: {e}") | |
| raise HTTPException( | |
| status_code=503, | |
| detail=f"CoinGecko API temporarily unavailable: {str(e)}" | |
| ) | |
| except Exception as e: | |
| logger.error(f"❌ CoinGecko API failed: {e}") | |
| raise HTTPException( | |
| status_code=503, | |
| detail=f"Failed to fetch real market data from CoinGecko: {str(e)}" | |
| ) | |
| async def get_ohlcv(self, symbol: str, days: int = 7) -> Dict[str, Any]: | |
| """ | |
| Fetch REAL OHLCV (price history) data from CoinGecko | |
| Args: | |
| symbol: Cryptocurrency symbol (e.g., "BTC", "ETH") | |
| days: Number of days of historical data (1, 7, 14, 30, 90, 180, 365, max) | |
| Returns: | |
| Dict with OHLCV data | |
| """ | |
| try: | |
| coin_id = self._symbol_to_coingecko_id(symbol) | |
| async with httpx.AsyncClient(timeout=self.timeout) as client: | |
| # Get market chart (OHLC) data | |
| response = await client.get( | |
| f"{self.base_url}/coins/{coin_id}/market_chart", | |
| params={ | |
| "vs_currency": "usd", | |
| "days": str(days), | |
| "interval": "daily" if days > 1 else "hourly" | |
| } | |
| ) | |
| response.raise_for_status() | |
| data = response.json() | |
| logger.info(f"✅ CoinGecko: Fetched {days} days of OHLCV data for {symbol}") | |
| return data | |
| except httpx.HTTPError as e: | |
| logger.error(f"❌ CoinGecko OHLCV API HTTP error: {e}") | |
| raise HTTPException( | |
| status_code=503, | |
| detail=f"CoinGecko OHLCV API unavailable: {str(e)}" | |
| ) | |
| except Exception as e: | |
| logger.error(f"❌ CoinGecko OHLCV API failed: {e}") | |
| raise HTTPException( | |
| status_code=503, | |
| detail=f"Failed to fetch OHLCV data from CoinGecko: {str(e)}" | |
| ) | |
| async def get_trending_coins(self, limit: int = 10) -> List[Dict[str, Any]]: | |
| """ | |
| Fetch REAL trending coins from CoinGecko | |
| Returns: | |
| List of real trending coins | |
| """ | |
| try: | |
| async with httpx.AsyncClient(timeout=self.timeout) as client: | |
| # Get trending coins | |
| response = await client.get(f"{self.base_url}/search/trending") | |
| response.raise_for_status() | |
| data = response.json() | |
| trending = [] | |
| coins = data.get("coins", [])[:limit] | |
| # Get price data for trending coins | |
| if coins: | |
| coin_ids = [coin["item"]["id"] for coin in coins] | |
| # Fetch current prices | |
| price_response = await client.get( | |
| f"{self.base_url}/simple/price", | |
| params={ | |
| "ids": ",".join(coin_ids), | |
| "vs_currencies": "usd", | |
| "include_24hr_change": "true" | |
| } | |
| ) | |
| price_response.raise_for_status() | |
| price_data = price_response.json() | |
| for idx, coin_obj in enumerate(coins): | |
| coin = coin_obj["item"] | |
| coin_id = coin["id"] | |
| prices = price_data.get(coin_id, {}) | |
| trending.append({ | |
| "symbol": coin.get("symbol", "").upper(), | |
| "name": coin.get("name", ""), | |
| "rank": idx + 1, | |
| "price": prices.get("usd", 0), | |
| "change24h": prices.get("usd_24h_change", 0), | |
| "marketCapRank": coin.get("market_cap_rank", 0), | |
| "source": "coingecko", | |
| "timestamp": int(datetime.utcnow().timestamp() * 1000) | |
| }) | |
| logger.info(f"✅ CoinGecko: Fetched {len(trending)} real trending coins") | |
| return trending | |
| except httpx.HTTPError as e: | |
| logger.error(f"❌ CoinGecko trending API HTTP error: {e}") | |
| raise HTTPException( | |
| status_code=503, | |
| detail=f"CoinGecko trending API unavailable: {str(e)}" | |
| ) | |
| except Exception as e: | |
| logger.error(f"❌ CoinGecko trending API failed: {e}") | |
| raise HTTPException( | |
| status_code=503, | |
| detail=f"Failed to fetch trending coins: {str(e)}" | |
| ) | |
| # Global instance | |
| coingecko_client = CoinGeckoClient() | |
| __all__ = ["CoinGeckoClient", "coingecko_client"] | |