import logging import os import requests logger = logging.getLogger(__name__) CRYPTOPANIC_API_KEY = os.getenv("CRYPTOPANIC_API_KEY", "") CRYPTOCOMPARE_API_KEY = os.getenv("CRYPTOCOMPARE_API_KEY", "") def fetch_news(asset: str, date: str = None, limit: int = 10) -> list: """Fetch crypto news. Returns list of {title, url, published_at}.""" currency = asset.split("/")[0] # BTC, ETH # Try CryptoPanic first if CRYPTOPANIC_API_KEY: try: return _fetch_cryptopanic(currency, limit) except Exception as e: logger.warning(f"CryptoPanic failed: {e}") # Try CryptoCompare try: return _fetch_cryptocompare(currency, limit) except Exception as e: logger.warning(f"CryptoCompare failed: {e}") return [] def _fetch_cryptopanic(currency: str, limit: int) -> list: url = "https://cryptopanic.com/api/v1/posts/" params = { "auth_token": CRYPTOPANIC_API_KEY, "currencies": currency, "kind": "news", "limit": limit, } resp = requests.get(url, params=params, timeout=15) resp.raise_for_status() data = resp.json() results = [] for item in data.get("results", [])[:limit]: results.append({ "title": item.get("title", ""), "url": item.get("url", ""), "published_at": item.get("published_at", ""), }) return results def _fetch_cryptocompare(currency: str, limit: int) -> list: url = "https://min-api.cryptocompare.com/data/v2/news/" params = {"categories": currency, "lang": "EN", "limit": limit} if CRYPTOCOMPARE_API_KEY: params["api_key"] = CRYPTOCOMPARE_API_KEY resp = requests.get(url, params=params, timeout=15) resp.raise_for_status() data = resp.json() results = [] for item in data.get("Data", [])[:limit]: results.append({ "title": item.get("title", ""), "url": item.get("url", ""), "published_at": str(item.get("published_on", "")), }) return results