catalyst.py

"""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}&currencies={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