"""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()