web_data.py

"""Public web context with bounded RSS reads and explicit cache freshness."""
import copy
import math
from itertools import zip_longest
import re
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import timezone
from email.utils import parsedate_to_datetime
from urllib.parse import urlsplit

import requests
from defusedxml import ElementTree as ET

UA = {"User-Agent": "lg-c2/0.4"}
TIMEOUT = 6
RETRY_DELAY = 60
MAX_RSS_BYTES = 2_000_000
NEWS_FEEDS = [
    ("CoinDesk", "https://www.coindesk.com/arc/outboundfeeds/rss/"),
    ("Cointelegraph", "https://cointelegraph.com/rss"),
    ("Decrypt", "https://decrypt.co/feed"),
]
SOURCE_URLS = {
    "fng": "https://api.alternative.me/fng/?limit=1",
    "gecko": "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin&vs_currencies=usd&include_24hr_change=true",
    "dom": "https://api.coingecko.com/api/v3/global",
    "spot": "https://api.coinbase.com/v2/prices/BTC-USD/spot",
}
BULL = {"surge", "rally", "soar", "gain", "record", "approve", "adopt", "inflow", "bullish", "breakout"}
BEAR = {"crash", "plunge", "drop", "hack", "ban", "lawsuit", "outflow", "bearish", "liquidation", "fraud", "fear"}


def _clean(text, limit=180):
    text = re.sub(r"[^\w\s\.,:;!?'\"%$€£()\-–/+]", "", text or "")
    return re.sub(r"\s+", " ", text).strip()[:limit]


def _safe_url(value):
    try:
        parts = urlsplit(value.strip())
        if parts.scheme not in ("http", "https") or not parts.hostname or parts.username or parts.password:
            return None
        if len(value) > 1000:
            return None
        return value.strip()
    except (ValueError, AttributeError):
        return None


def news_sentiment(items):
    bull = bear = 0
    for item in items:
        words = set(re.findall(r"[a-z\-]+", item["title"].lower()))
        b, s = len(words & BULL), len(words & BEAR)
        bull += int(b > s)
        bear += int(s > b)
    total = bull + bear
    return ((bull - bear) / total if total else 0.0), bull, bear


def _json(url):
    response = requests.get(url, headers=UA, timeout=TIMEOUT)
    try:
        response.raise_for_status()
        return response.json()
    finally:
        response.close()


def _number(value, minimum=None, maximum=None):
    if isinstance(value, bool):
        raise ValueError('Invalid numeric provider value')
    number = float(value)
    if not math.isfinite(number) or (minimum is not None and number < minimum) or (maximum is not None and number > maximum):
        raise ValueError('Invalid numeric provider value')
    return number


def _fetch_fng():
    item = _json(SOURCE_URLS["fng"])["data"][0]
    value = _number(item["value"], 0, 100)
    if not value.is_integer():
        raise ValueError("Invalid sentiment index")
    return {"value": int(value), "label": _clean(str(item["value_classification"]), 40)}


def _fetch_gecko():
    item = _json(SOURCE_URLS["gecko"])["bitcoin"]
    return {"price": _number(item["usd"], 1e-12), "change_24h": _number(item.get("usd_24h_change") or 0.0)}


def _fetch_dominance():
    return _number(_json(SOURCE_URLS["dom"])["data"]["market_cap_percentage"]["btc"], 0, 100)


def _fetch_spot():
    return _number(_json(SOURCE_URLS["spot"])["data"]["amount"], 1e-12)


def _child(element, name):
    return next((child for child in element if child.tag.rsplit("}", 1)[-1] == name), None)


def _published(value):
    try:
        date = parsedate_to_datetime(value)
        if date.tzinfo is None:
            date = date.replace(tzinfo=timezone.utc)
        return date.astimezone(timezone.utc).isoformat()
    except (TypeError, ValueError, OverflowError):
        # Retain an explicit unknown date instead of inventing one.
        return None


def _parse_feed(data, source, feed_url):
    root = ET.fromstring(data, forbid_dtd=True, forbid_entities=True, forbid_external=True)
    items = []
    for element in root.iter():
        if element.tag.rsplit("}", 1)[-1] not in ("item", "entry"):
            continue
        title_node, link_node = _child(element, "title"), _child(element, "link")
        title = _clean(title_node.text if title_node is not None else "")
        url = _safe_url((link_node.get("href") or link_node.text or "") if link_node is not None else "")
        if not title or not url:
            continue
        date_node = _child(element, "pubDate")
        items.append({"source": source, "title": title, "url": url,
                      "feed_url": feed_url,
                      "published_at": _published(date_node.text) if date_node is not None else None})
        if len(items) >= 8:
            break
    return items


def _fetch_feed(source, url):
    response = requests.get(url, headers=UA, timeout=(3, TIMEOUT), stream=True)
    try:
        response.raise_for_status()
        size = 0
        chunks = []
        started = time.monotonic()
        for chunk in response.iter_content(chunk_size=4096):
            size += len(chunk)
            if size > MAX_RSS_BYTES or time.monotonic() - started > 12:
                raise ValueError("RSS size or read-time limit")
            chunks.append(chunk)
        return _parse_feed(b"".join(chunks), source, url)
    finally:
        response.close()


def _fetch_news():
    items, seen, groups = [], set(), []
    # Concurrent reads, then round-robin selection to retain source diversity.
    with ThreadPoolExecutor(max_workers=len(NEWS_FEEDS)) as executor:
        futures = [executor.submit(_fetch_feed, source, url) for source, url in NEWS_FEEDS]
        for future in futures:
            try:
                groups.append(future.result())
            except Exception:
                groups.append([])
    for row in zip_longest(*groups):
        for item in row:
            if item is None:
                continue
            key = item["title"].casefold()
            if key not in seen:
                seen.add(key)
                items.append(item)
    if not items:
        raise RuntimeError("RSS unavailable")
    return items


JOBS = {"fng": (900, _fetch_fng), "gecko": (180, _fetch_gecko),
        "dom": (900, _fetch_dominance), "spot": (30, _fetch_spot),
        "news": (300, _fetch_news)}


class WebIntel:
    def __init__(self, jobs=None, clock=None):
        self._lock = threading.Lock()
        self._cache, self._next_try, self.ok = {}, {}, {}
        self.jobs = JOBS if jobs is None else jobs
        self.clock = clock or time.time

    def _get(self, name, ttl, fn):
        now = self.clock()
        with self._lock:
            ts, value = self._cache.get(name, (0.0, None))
            next_try = self._next_try.get(name, 0.0)
        if now < next_try or (value is not None and now - ts < ttl):
            return value
        try:
            value = fn()
        except Exception:
            with self._lock:
                self._next_try[name] = self.clock() + RETRY_DELAY
                self.ok[name] = False
            return None
        with self._lock:
            self._cache[name] = (self.clock(), value)
            self._next_try[name] = 0.0
            self.ok[name] = True
        return value

    def snapshot(self):
        with ThreadPoolExecutor(max_workers=max(1, len(self.jobs))) as executor:
            futures = [executor.submit(self._get, key, ttl, fn) for key, (ttl, fn) in self.jobs.items()]
            for future in futures:
                future.result()
        now = self.clock()
        with self._lock:
            cache, status = copy.deepcopy(self._cache), dict(self.ok)
        metadata, values = {}, {}
        for key, (ttl, _) in self.jobs.items():
            ts, value = cache.get(key, (None, None))
            age = max(0.0, now - ts) if ts is not None else None
            fresh = value is not None and status.get(key, False) and age is not None and age < ttl
            metadata[key] = {"ok": status.get(key, False), "fresh": bool(fresh),
                             "fetched_at": ts, "age_seconds": age,
                             "source_url": SOURCE_URLS.get(key)}
            values[key] = value if fresh else None
        fng, gecko, dom, spot = (values.get(key) for key in ("fng", "gecko", "dom", "spot"))
        news = (values.get("news") or [])[:5]
        score, bull, bear = news_sentiment(news)
        parts, bits = [], []
        if fng:
            parts.append((.4, (fng["value"] - 50) / 50))
            bits.append(f"F&G {fng['value']} ({fng['label']})")
        if news:
            parts.append((.35, score)); bits.append(f"news {score:+.2f} ({bull}↑/{bear}↓)")
        if gecko:
            parts.append((.25, max(-1., min(1., gecko["change_24h"] / 5))))
            bits.append(f"24h {gecko['change_24h']:+.1f}%")
        if dom is not None:
            bits.append(f"dom. BTC {dom:.1f}%")
        weight = sum(w for w, _ in parts)
        fresh_count = sum(int(item["fresh"]) for item in metadata.values())
        return {"available": fresh_count > 0, "retrieved_at": now,
                "status": "available" if fresh_count == len(self.jobs) and fresh_count else "partial" if fresh_count else "unavailable",
                "fng": fng, "gecko": gecko, "dominance": dom, "spot": spot,
                "headlines": news, "news_score": score, "bull": bull, "bear": bear,
                "composite": sum(w * v for w, v in parts) / weight if weight else 0.,
                "line": " | ".join(bits) or "web indisponible",
                "sources_ok": status, "source_status": metadata}

    @staticmethod
    def price_gap(price, web):
        spot = (web or {}).get("spot")
        return abs(spot - price) / price if spot and price and price > 0 else None


WEB = WebIntel()