from __future__ import annotations import os import re import xml.etree.ElementTree as ET from typing import Any import httpx CRYPTOPANIC_URL = "https://cryptopanic.com/api/v1/posts/" ALTERNATIVE_ME_URL = "https://api.alternative.me/fng/" COINDESK_RSS = "https://www.coindesk.com/arc/outboundfeeds/rss/" LUNARCRUSH_COIN_URL = "https://lunarcrush.com/api4/public/coins/{symbol}/v1" CRYPTOCOMPARE_NEWS_URL = "https://min-api.cryptocompare.com/data/v2/news/" MESSARI_NEWS_URL = "https://data.messari.io/api/v1/news" async def _fetch_coindesk_headlines(limit: int = 20) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] try: async with httpx.AsyncClient(timeout=10, follow_redirects=True) as client: resp = await client.get(COINDESK_RSS) if resp.status_code != 200: return items root = ET.fromstring(resp.text) for node in root.findall(".//item")[:limit]: items.append( { "title": node.findtext("title", ""), "source": "CoinDesk", "published_at": node.findtext("pubDate", ""), "url": node.findtext("link", ""), } ) except Exception: pass return items WORLD_NEWS_FEEDS = [ ("Reuters Business", "https://feeds.reuters.com/reuters/businessNews"), ("CNBC Top News", "https://search.cnbc.com/rs/search/combinedcms/view.xml?partnerId=wrss01&id=100003114"), ("Bloomberg Markets", "https://feeds.bloomberg.com/markets/news.rss"), ("CoinDesk", "https://www.coindesk.com/arc/outboundfeeds/rss/"), ] # Public funding rate endpoints (no auth required) BINANCE_FUNDING_URL = "https://fapi.binance.com/fapi/v1/premiumIndex" BYBIT_FUNDING_URL = "https://api.bybit.com/v5/market/tickers" OKX_FUNDING_URL = "https://www.okx.com/api/v5/public/funding-rate" BINANCE_OI_HIST_URL = "https://fapi.binance.com/futures/data/openInterestHist" async def _fetch_cryptocompare_news(limit: int = 20) -> list[dict[str, Any]]: """CER-017: CryptoCompare news free (no key needed).""" items: list[dict[str, Any]] = [] try: async with httpx.AsyncClient(timeout=10) as client: resp = await client.get(CRYPTOCOMPARE_NEWS_URL, params={"lang": "EN"}) if resp.status_code != 200: return items data = resp.json() for r in (data.get("Data") or [])[:limit]: ts = r.get("published_on") try: import datetime as _dt pub = _dt.datetime.fromtimestamp(int(ts), _dt.UTC).isoformat() if ts else "" except (TypeError, ValueError): pub = "" items.append({ "title": r.get("title", ""), "source": r.get("source", "CryptoCompare"), "published_at": pub, "url": r.get("url", ""), "provider": "cryptocompare", }) except Exception: pass return items async def _fetch_messari_news(limit: int = 20) -> list[dict[str, Any]]: """CER-017: Messari news free (no key needed for basic feed).""" items: list[dict[str, Any]] = [] try: async with httpx.AsyncClient(timeout=10) as client: resp = await client.get(MESSARI_NEWS_URL) if resp.status_code != 200: return items data = resp.json() for r in (data.get("data") or [])[:limit]: items.append({ "title": r.get("title", ""), "source": (r.get("author") or {}).get("name") or "Messari", "published_at": r.get("published_at", ""), "url": r.get("url", ""), "provider": "messari", }) except Exception: pass return items def _normalize_title(t: str) -> str: """Lowercase + strip non-alnum per dedup tra provider.""" return "".join(ch for ch in t.lower() if ch.isalnum() or ch.isspace()).strip() async def fetch_crypto_news(api_key: str = "", limit: int = 20) -> dict[str, Any]: """CER-017: multi-source aggregator (CoinDesk + CryptoCompare + Messari) + dedup. Se `api_key` Cryptopanic presente, include anche quella come 4° source. """ import asyncio # CoinDesk + CryptoCompare + Messari sempre (free, no key) tasks = [ _fetch_coindesk_headlines(limit), _fetch_cryptocompare_news(limit), _fetch_messari_news(limit), ] include_cp = bool(api_key) and api_key.lower() not in ("placeholder", "none", "changeme") if include_cp: tasks.append(_fetch_cryptopanic_news(api_key, limit)) results = await asyncio.gather(*tasks, return_exceptions=True) all_items: list[dict[str, Any]] = [] providers_ok: list[str] = [] providers_failed: list[str] = [] provider_names = ["coindesk", "cryptocompare", "messari"] if include_cp: provider_names.append("cryptopanic") for name, res in zip(provider_names, results, strict=True): if isinstance(res, Exception) or not res: providers_failed.append(name) continue providers_ok.append(name) for item in res: if "provider" not in item: item["provider"] = name all_items.append(item) # Dedup per normalized title — preserva primo match seen: set[str] = set() deduped: list[dict[str, Any]] = [] for h in all_items: key = _normalize_title(h.get("title", "")) if not key or key in seen: continue seen.add(key) deduped.append(h) # Sort per published_at DESC (stringhe ISO confrontabili; quelle vuote in fondo) deduped.sort(key=lambda x: x.get("published_at") or "", reverse=True) return { "headlines": deduped[:limit], "sources": providers_ok, "sources_failed": providers_failed, "total_before_dedup": len(all_items), "total_after_dedup": len(deduped), } async def _fetch_cryptopanic_news(api_key: str, limit: int) -> list[dict[str, Any]]: """Cryptopanic as one of the sources. Failure → [].""" try: async with httpx.AsyncClient(timeout=10) as client: resp = await client.get( CRYPTOPANIC_URL, params={"auth_token": api_key, "public": "true"}, ) if resp.status_code >= 400: return [] data = resp.json() except Exception: return [] return [ { "title": r.get("title", ""), "source": (r.get("source") or {}).get("title", "Cryptopanic"), "published_at": r.get("published_at", ""), "url": r.get("url", ""), "provider": "cryptopanic", } for r in (data.get("results") or [])[:limit] ] async def _fetch_lunarcrush(symbol: str, api_key: str) -> dict | None: """CER-P2-005: LunarCrush v4 social metrics. Ritorna None se fail.""" try: async with httpx.AsyncClient(timeout=10) as client: resp = await client.get( LUNARCRUSH_COIN_URL.format(symbol=symbol.upper()), headers={"Authorization": f"Bearer {api_key}"}, ) if resp.status_code != 200: return None data = (resp.json() or {}).get("data") or {} return { "galaxy_score": data.get("galaxy_score"), "alt_rank": data.get("alt_rank"), "sentiment": data.get("sentiment"), # 0-100 scale "social_volume_24h": data.get("social_volume_24h"), "social_dominance": data.get("social_dominance"), } except Exception: return None def _fng_to_sentiment(value: int) -> float: """Normalize fear&greed 0-100 to [-1, 1] proxy sentiment.""" return round((value - 50) / 50.0, 3) async def fetch_social_sentiment(symbol: str = "BTC") -> dict[str, Any]: """CER-P2-005: provider chain LunarCrush + fear_greed proxy. Se LUNARCRUSH_API_KEY env è presente e risponde, usa valori reali. Altrimenti deriva proxy da fear_greed per popolare twitter/reddit sentiment (marcato come derived=True così l'agent sa che è proxy). """ async with httpx.AsyncClient(timeout=10) as client: fng_resp = await client.get(ALTERNATIVE_ME_URL, params={"limit": 1}) fng_data = fng_resp.json() fng_list = fng_data.get("data", []) fng = fng_list[0] if fng_list else {} fng_value = int(fng.get("value", 0)) proxy = _fng_to_sentiment(fng_value) if fng_value else 0.0 result: dict[str, Any] = { "fear_greed_index": fng_value, "fear_greed_label": fng.get("value_classification", ""), "symbol": symbol.upper(), "social_volume": 0, "twitter_sentiment": 0.0, "reddit_sentiment": 0.0, "source": "fear_greed_only", "derived": True, } lc_key = os.environ.get("LUNARCRUSH_API_KEY", "").strip() if lc_key: lc = await _fetch_lunarcrush(symbol, lc_key) if lc is not None: # LunarCrush sentiment 0-100 → normalize to [-1, 1] lc_sent = lc.get("sentiment") norm = round((float(lc_sent) - 50) / 50.0, 3) if lc_sent is not None else None result.update({ "twitter_sentiment": norm if norm is not None else proxy, "reddit_sentiment": norm if norm is not None else proxy, "social_volume": int(lc.get("social_volume_24h") or 0), "galaxy_score": lc.get("galaxy_score"), "alt_rank": lc.get("alt_rank"), "social_dominance": lc.get("social_dominance"), "source": "lunarcrush+fear_greed", "derived": False, }) return result # Proxy-only path result["twitter_sentiment"] = proxy result["reddit_sentiment"] = proxy result["note"] = ( "twitter/reddit derived from fear_greed_index; configure LUNARCRUSH_API_KEY " "for real social metrics" ) return result async def fetch_funding_rates(asset: str = "BTC") -> dict[str, Any]: """Fetch perpetual funding rates from Binance, Bybit and OKX public APIs.""" asset = asset.upper() usdt_symbol = f"{asset}USDT" okx_inst = f"{asset}-USDT-SWAP" rates: list[dict[str, Any]] = [] async with httpx.AsyncClient(timeout=10) as client: # Binance try: resp = await client.get(BINANCE_FUNDING_URL, params={"symbol": usdt_symbol}) if resp.status_code == 200: d = resp.json() rates.append( { "exchange": "binance", "asset": asset, "rate": float(d.get("lastFundingRate", 0)), "next_funding_time": d.get("nextFundingTime", ""), } ) except Exception: pass # Bybit try: resp = await client.get( BYBIT_FUNDING_URL, params={"category": "linear", "symbol": usdt_symbol}, ) if resp.status_code == 200: items = resp.json().get("result", {}).get("list", []) if items: d = items[0] rates.append( { "exchange": "bybit", "asset": asset, "rate": float(d.get("fundingRate", 0)), "next_funding_time": d.get("nextFundingTime", ""), } ) except Exception: pass # OKX try: resp = await client.get(OKX_FUNDING_URL, params={"instId": okx_inst}) if resp.status_code == 200: items = resp.json().get("data", []) if items: d = items[0] rates.append( { "exchange": "okx", "asset": asset, "rate": float(d.get("fundingRate", 0)), "next_funding_time": d.get("nextFundingTime", ""), } ) except Exception: pass return {"asset": asset, "rates": rates} async def fetch_cross_exchange_funding(assets: list[str] | None = None) -> dict[str, Any]: """Snapshot multi-asset funding rates con spread e arbitrage detection.""" from datetime import UTC, datetime as _dt assets = [a.upper() for a in (assets or ["BTC", "ETH", "SOL"])] snapshot: dict[str, dict[str, Any]] = {} async with httpx.AsyncClient(timeout=10) as client: for asset in assets: rates: dict[str, float | None] = { "binance": None, "bybit": None, "okx": None, "hyperliquid": None, } try: resp = await client.get( BINANCE_FUNDING_URL, params={"symbol": f"{asset}USDT"} ) if resp.status_code == 200: rates["binance"] = float(resp.json().get("lastFundingRate", 0)) except Exception: pass try: resp = await client.get( BYBIT_FUNDING_URL, params={"category": "linear", "symbol": f"{asset}USDT"}, ) if resp.status_code == 200: items = resp.json().get("result", {}).get("list", []) if items: rates["bybit"] = float(items[0].get("fundingRate", 0)) except Exception: pass try: resp = await client.get( OKX_FUNDING_URL, params={"instId": f"{asset}-USDT-SWAP"} ) if resp.status_code == 200: items = resp.json().get("data", []) if items: rates["okx"] = float(items[0].get("fundingRate", 0)) except Exception: pass try: resp = await client.post( "https://api.hyperliquid.xyz/info", json={"type": "metaAndAssetCtxs"}, ) if resp.status_code == 200: data = resp.json() universe = data[0].get("universe") or [] ctx_list = data[1] if len(data) > 1 else [] for meta, ctx in zip(universe, ctx_list, strict=False): if meta.get("name", "").upper() == asset: rates["hyperliquid"] = float(ctx.get("funding", 0)) break except Exception: pass present = [v for v in rates.values() if v is not None] spread_max_min = max(present) - min(present) if present else None anomaly = None if present and spread_max_min is not None: mean_r = sum(present) / len(present) for name, v in rates.items(): if v is None: continue if abs(v - mean_r) > 2 * (spread_max_min / 2 or 1e-9): anomaly = f"{name}_outlier" break snapshot[asset] = { **rates, "spread_max_min": spread_max_min, "anomaly": anomaly, } # Arbitrage opportunities arbs = [] for asset, data in snapshot.items(): values = [(k, v) for k, v in data.items() if k in ("binance", "bybit", "okx", "hyperliquid") and v is not None] if len(values) < 2: continue values.sort(key=lambda x: x[1]) low_ex, low_v = values[0] high_ex, high_v = values[-1] diff = high_v - low_v ann_pct = diff * 24 * 365 * 100 # hourly funding → annual pct if ann_pct > 50: arbs.append({ "asset": asset, "pair": f"long_{low_ex}_short_{high_ex}", "funding_differential_ann": round(ann_pct, 2), "risk_adjusted": "acceptable" if ann_pct > 100 else "marginal", }) return { "assets": assets, "snapshot": snapshot, "arbitrage_opportunities": arbs, "data_timestamp": _dt.now(UTC).isoformat(), } async def fetch_world_news() -> dict[str, Any]: """Fetch world financial news from free RSS feeds.""" articles: list[dict[str, Any]] = [] async with httpx.AsyncClient(timeout=10, follow_redirects=True) as client: for source_name, url in WORLD_NEWS_FEEDS: try: resp = await client.get(url) if resp.status_code != 200: continue root = ET.fromstring(resp.text) for item in root.findall(".//item")[:5]: title = item.findtext("title", "") link = item.findtext("link", "") pub_date = item.findtext("pubDate", "") description = item.findtext("description", "") if "<" in description: description = re.sub(r"<[^>]+>", "", description).strip() articles.append( { "source": source_name, "title": title, "url": link, "published": pub_date, "summary": description[:200] if description else "", } ) except Exception: continue return {"articles": articles, "count": len(articles)} async def fetch_oi_history(asset: str = "BTC", period: str = "5m", limit: int = 288) -> dict[str, Any]: """Open interest history perpetual da Binance futures (public). period: 5m|15m|30m|1h|2h|4h|6h|12h|1d (Binance API). limit: 1..500, default 288 = 24h a 5min. """ asset = asset.upper() symbol = f"{asset}USDT" limit = max(1, min(int(limit), 500)) points: list[dict[str, Any]] = [] try: async with httpx.AsyncClient(timeout=10) as client: resp = await client.get( BINANCE_OI_HIST_URL, params={"symbol": symbol, "period": period, "limit": limit}, ) if resp.status_code == 200: for row in resp.json() or []: points.append( { "timestamp": int(row.get("timestamp", 0)), "oi": float(row.get("sumOpenInterest", 0)), "oi_value_usd": float(row.get("sumOpenInterestValue", 0)), } ) except Exception: pass def _delta_pct(points: list[dict[str, Any]], minutes_back: int) -> float | None: if len(points) < 2: return None current = points[-1] cutoff_ts = current["timestamp"] - minutes_back * 60 * 1000 past = next((p for p in reversed(points) if p["timestamp"] <= cutoff_ts), None) if past is None or past["oi"] == 0: return None return round(100.0 * (current["oi"] - past["oi"]) / past["oi"], 3) return { "asset": asset, "exchange": "binance", "symbol": symbol, "period": period, "points": points, "current_oi": points[-1]["oi"] if points else None, "current_oi_value_usd": points[-1]["oi_value_usd"] if points else None, "delta_pct_1h": _delta_pct(points, 60), "delta_pct_4h": _delta_pct(points, 240), "delta_pct_24h": _delta_pct(points, 1440), "data_points": len(points), }