"""Catalyst scoring — news + market sentiment. Data sources (all free, no paid keys required for core functionality): 1. Fear & Greed Index — alternative.me, free, whole-market modifier 2. CryptoPanic — optional; requires CRYPTOPANIC_TOKEN env var 3. CoinGecko Trending — free, no key, top 7 trending coins by search volume 4. CoinDesk RSS — free, no key, live crypto headlines 5. Cointelegraph RSS — free, no key, live crypto headlines Design rules: - News never overrides a bad technical setup; it only modifies confidence - All results are cached to avoid hammering APIs mid-scan - If all APIs fail, we fall back to 0.5 (neutral) — no crash, no hallucination - Scores are always 0.0–1.0 before being multiplied by 10 in scorer.py """ from __future__ import annotations import os, time, math, xml.etree.ElementTree as ET from urllib.request import urlopen, Request from urllib.error import URLError import json # ── Cache store ────────────────────────────────────────────────────────────── _fng_cache: dict = {} # {"value": int, "label": str, "ts": float} _news_cache: dict = {} # {symbol: {"score": float, "items": list, "ts": float}} FNG_TTL = 3600 # 1 hour — index updates once a day NEWS_TTL = 900 # 15 min per coin # ───────────────────────────────────────────────────────────────────────────── # FEAR & GREED INDEX # ───────────────────────────────────────────────────────────────────────────── def fetch_fear_greed() -> dict: """Returns {"value": 0-100, "label": str, "score_mod": float, "ts": float} score_mod is a multiplier applied to the whole-market catalyst: Extreme Fear (0-24) → contrarian LONG boost → 0.65 (market oversold) Fear (25-44) → mild bullish → 0.55 Neutral (45-55) → no effect → 0.50 Greed (56-74) → mild caution → 0.45 Extreme Greed (75-100)→ contrarian SHORT signal → 0.35 (market overbought) """ global _fng_cache if _fng_cache and time.time() - _fng_cache.get("ts", 0) < FNG_TTL: return _fng_cache try: req = Request( "https://api.alternative.me/fng/", headers={"User-Agent": "TradeCopilot/1.0"} ) with urlopen(req, timeout=5) as r: data = json.loads(r.read()) entry = data["data"][0] value = int(entry["value"]) label = entry["value_classification"] if value <= 24: mod = 0.65 # Extreme Fear — contrarian long opportunity elif value <= 44: mod = 0.55 # Fear — mildly bullish elif value <= 55: mod = 0.50 # Neutral elif value <= 74: mod = 0.45 # Greed — mild caution else: mod = 0.35 # Extreme Greed — market likely overbought _fng_cache = {"value": value, "label": label, "score_mod": mod, "ts": time.time()} return _fng_cache except Exception as e: # API down — return neutral, don't crash return {"value": None, "label": "unavailable", "score_mod": 0.50, "ts": time.time(), "error": str(e)[:80]} # ───────────────────────────────────────────────────────────────────────────── # CRYPTOPANIC NEWS SENTIMENT # ───────────────────────────────────────────────────────────────────────────── def _cp_token() -> str | None: """Read token from environment — set CRYPTOPANIC_TOKEN in HF Spaces secrets.""" return os.environ.get("CRYPTOPANIC_TOKEN") or None def _coin_slug(symbol: str) -> str: """BTC-USDT → BTC, BTCUSDT → BTC""" s = symbol.upper() for suffix in ("-USDT", "-USD", "USDT", "USD"): if s.endswith(suffix): s = s[: len(s) - len(suffix)] break return s def fetch_coin_news(symbol: str) -> dict: """Fetch recent news for a coin and compute a sentiment score 0.0–1.0. Returns: { "score": float, # 0.0 (very bearish) → 1.0 (very bullish) "label": str, # "bullish" / "bearish" / "neutral" / "no_data" "items": list, # raw headline objects for display "source": str, # "cryptopanic" or "unavailable" "ts": float } """ global _news_cache coin = _coin_slug(symbol) cached = _news_cache.get(coin) if cached and time.time() - cached.get("ts", 0) < NEWS_TTL: return cached token = _cp_token() if not token: result = {"score": 0.50, "label": "no_key", "items": [], "source": "unavailable", "ts": time.time()} _news_cache[coin] = result return result try: url = ( f"https://cryptopanic.com/api/free/v1/posts/" f"?auth_token={token}¤cies={coin}&filter=hot&public=true" ) req = Request(url, headers={"User-Agent": "TradeCopilot/1.0"}) with urlopen(req, timeout=6) as r: data = json.loads(r.read()) posts = data.get("results", []) if not posts: result = {"score": 0.50, "label": "neutral", "items": [], "source": "cryptopanic", "ts": time.time()} _news_cache[coin] = result return result # ── Sentiment scoring ───────────────────────────────────────────── # Each post has votes: {"positive": N, "negative": N, "important": N} # Weight by recency: posts in last 1h = 1.0, 2h = 0.5, 6h = 0.15 now = time.time() total_weight = 0.0 weighted_sentiment = 0.0 items_out = [] for post in posts[:20]: # cap at 20 most recent title = post.get("title", "") votes = post.get("votes", {}) pos = votes.get("positive", 0) or 0 neg = votes.get("negative", 0) or 0 imp = votes.get("important", 0) or 0 # Parse published_at to get age in hours pub = post.get("published_at", "") try: from datetime import datetime, timezone dt = datetime.fromisoformat(pub.replace("Z", "+00:00")) age_h = (datetime.now(timezone.utc) - dt).total_seconds() / 3600 except Exception: age_h = 3.0 # Recency weight: exponential decay recency = math.exp(-0.5 * age_h) # half-life ~2h # Net sentiment per post: +1 = fully bullish, -1 = fully bearish total_votes = pos + neg + 1e-9 net = (pos - neg) / total_votes # -1 to +1 # Important flag boosts weight importance = 1.0 + 0.5 * min(imp / 5, 1.0) w = recency * importance weighted_sentiment += net * w total_weight += w items_out.append({ "title": title, "pos": pos, "neg": neg, "imp": imp, "age_h": round(age_h, 1), "url": post.get("url", "") }) # Normalise to 0.0–1.0 if total_weight > 0: raw = weighted_sentiment / total_weight # -1 to +1 score = round((raw + 1) / 2, 3) # 0.0 to 1.0 else: score = 0.50 if score >= 0.62: label = "bullish" elif score <= 0.38: label = "bearish" else: label = "neutral" result = {"score": score, "label": label, "items": items_out[:5], "source": "cryptopanic", "ts": time.time()} _news_cache[coin] = result return result except Exception as e: result = {"score": 0.50, "label": "neutral", "items": [], "source": "unavailable", "ts": time.time(), "error": str(e)[:80]} _news_cache[coin] = result return result # ───────────────────────────────────────────────────────────────────────────── # COMBINED CATALYST SCORE # ───────────────────────────────────────────────────────────────────────────── def score_catalyst(symbol: str) -> tuple[float, list[str], dict]: """Main entry point called by scorer.py. Returns: (score_0_to_1, notes_list, raw_data_dict) Combination logic: base = coin news sentiment (0.0–1.0) mod = fear & greed modifier (0.35–0.65) final = base × 0.7 + mod × 0.3 ← news matters more than market mood If CryptoPanic key is missing, we use F&G as the full signal. If both fail, returns 0.5 neutral. """ fng = fetch_fear_greed() news = fetch_coin_news(symbol) notes = [] raw = {"fear_greed": fng, "news": news} # ── Fear & Greed ────────────────────────────────────────────────────── fng_mod = fng.get("score_mod", 0.50) fng_val = fng.get("value") fng_label = fng.get("label", "unavailable") if fng_val is not None: notes.append(f"Market sentiment: {fng_label} ({fng_val}/100)") else: notes.append("Fear & Greed: unavailable") # ── CryptoPanic news ────────────────────────────────────────────────── news_score = news.get("score", 0.50) news_label = news.get("label", "neutral") news_source = news.get("source", "unavailable") if news_source == "unavailable" and news.get("label") == "no_key": notes.append("News: no CryptoPanic key — set CRYPTOPANIC_TOKEN in HF Secrets") # Fall back to F&G only final = fng_mod elif news_source == "unavailable": notes.append("News: CryptoPanic unreachable") final = fng_mod elif news_label == "neutral": notes.append(f"News: neutral (score {news_score:.2f})") final = news_score * 0.7 + fng_mod * 0.3 else: # Show top headline if available top = news.get("items", [{}])[0].get("title", "") if news.get("items") else "" snippet = f' — "{top[:60]}…"' if top else "" notes.append(f"News: {news_label} (score {news_score:.2f}){snippet}") final = news_score * 0.7 + fng_mod * 0.3 final = round(max(0.0, min(1.0, final)), 3) return final, notes, raw # ───────────────────────────────────────────────────────────────────────────── # FREE NEWS SOURCES — no API key required # Used by Market Signals panel (separate from Live Setups scoring) # ───────────────────────────────────────────────────────────────────────────── _trending_cache: dict = {} # {"coins": list, "ts": float} _headlines_cache: dict = {} # {"headlines": list, "ts": float} TRENDING_TTL = 900 # 15 min — CoinGecko trending updates hourly HEADLINES_TTL = 600 # 10 min — RSS headlines def fetch_trending_coins() -> list[dict]: """CoinGecko /search/trending — top 7 coins by search volume, no key needed. Returns list of {name, symbol, market_cap_rank, score (0=hottest)}. Cached 15 min. """ global _trending_cache if _trending_cache and time.time() - _trending_cache.get("ts", 0) < TRENDING_TTL: return _trending_cache.get("coins", []) try: req = Request( "https://api.coingecko.com/api/v3/search/trending", headers={"User-Agent": "TradeCopilot/1.0", "Accept": "application/json"} ) with urlopen(req, timeout=5) as r: data = json.loads(r.read()) coins = [] for entry in data.get("coins", []): item = entry.get("item", {}) coins.append({ "name": item.get("name", ""), "symbol": item.get("symbol", "").upper(), "market_cap_rank": item.get("market_cap_rank"), "score": item.get("score", 99), # 0 = hottest "thumb": item.get("thumb", ""), }) _trending_cache = {"coins": coins, "ts": time.time()} return coins except Exception as e: _trending_cache = {"coins": [], "ts": time.time(), "error": str(e)[:80]} return [] def _parse_rss(url: str, max_items: int = 8) -> list[dict]: """Parse an RSS 2.0 feed. Returns list of {title, link, published}.""" try: req = Request(url, headers={"User-Agent": "TradeCopilot/1.0"}) with urlopen(req, timeout=5) as r: tree = ET.parse(r) items = tree.findall(".//item")[:max_items] result = [] for item in items: title = item.find("title") link = item.find("link") pub = item.find("pubDate") result.append({ "title": (title.text or "").strip() if title is not None else "", "link": (link.text or "").strip() if link is not None else "", "published": (pub.text or "").strip() if pub is not None else "", "source": url.split("/")[2], # domain as source label }) return result except Exception: return [] def fetch_crypto_headlines() -> dict: """Fetch live crypto headlines from CoinDesk + Cointelegraph RSS. Cross-references with CoinGecko trending coins to tag which coins are mentioned. Returns: { trending_coins: [{name, symbol, score}, ...], headlines: [{title, link, published, source, coins_mentioned}, ...], ts: float, error: str | None, } Cached 10 min. Falls back gracefully if any source is down. """ global _headlines_cache if _headlines_cache and time.time() - _headlines_cache.get("ts", 0) < HEADLINES_TTL: return _headlines_cache trending = fetch_trending_coins() trending_symbols = {c["symbol"].upper() for c in trending} trending_names = {c["name"].lower(): c["symbol"] for c in trending} coindesk_headlines = _parse_rss("https://www.coindesk.com/arc/outboundfeeds/rss/") cointelegraph_headlines = _parse_rss("https://cointelegraph.com/rss") all_headlines = coindesk_headlines + cointelegraph_headlines # Tag each headline with any trending coin mentioned for h in all_headlines: title_up = h["title"].upper() title_lo = h["title"].lower() mentioned = [] for sym in trending_symbols: if sym in title_up: mentioned.append(sym) for name, sym in trending_names.items(): if name in title_lo and sym not in mentioned: mentioned.append(sym) # Also check common coins by name even if not trending for keyword, sym in [ ("bitcoin", "BTC"), ("ethereum", "ETH"), ("solana", "SOL"), ("ripple", "XRP"), ("bnb", "BNB"), ("dogecoin", "DOGE"), ]: if keyword in title_lo and sym not in mentioned: mentioned.append(sym) h["coins_mentioned"] = mentioned result = { "trending_coins": trending, "headlines": all_headlines, "total": len(all_headlines), "sources": ["coingecko_trending", "coindesk_rss", "cointelegraph_rss"], "ts": time.time(), "error": None if all_headlines else "All RSS sources unavailable", } _headlines_cache = result return result