| """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): |
| |
| 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(), [] |
| |
| 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() |
|
|
