""" RMI News Aggregation Service - THE Crypto News Aggregator ========================================================== 200+ sources across 15 tiers. Every feed verified. Real data only. Tiers: Tier 1 - Security & Audit Firms (14) Tier 2 - Major News Outlets (20) Tier 3 - Research & Analytics (12) Tier 4 - Protocol & Chain Blogs (15) Tier 5 - DeFi Protocols (18) Tier 6 - Exchange Blogs (12) Tier 7 - VC & Investment Firms (12) Tier 8 - Academic & Research Centers (8) Tier 9 - Government & Regulatory (6) Tier 10 - Newsletters (15) Tier 11 - Reddit Communities (20) Tier 12 - Developer & GitHub (5) Tier 13 - Exploit & Hack Trackers (8) Tier 14 - API Sources (CoinGecko, CryptoPanic, LunarCrush) Tier 15 - RMI Internal (scanner, alerts, Ghost, RAG) """ import asyncio import hashlib import logging import os import random from datetime import UTC, datetime, timedelta from typing import Any import feedparser import httpx logger = logging.getLogger(__name__) # ─── PROXY ROTATION ─────────────────────────────────────────────── PROXY_LIST = [ "http://43.153.99.140:13001", "http://43.153.99.140:13002", "http://43.153.99.140:13003", "http://43.153.99.140:13004", "http://43.153.99.140:13005", "http://43.153.99.140:13006", "http://43.153.99.140:13007", "http://43.153.99.140:13008", "http://43.153.99.140:13009", "http://43.153.99.140:13010", ] PROXY_REQUIRED_FEEDS = { "https://cryptosecurity.substack.com/feed", "https://weekinethereum.substack.com/feed", "https://banklessdao.substack.com/feed", "https://defieducation.substack.com/feed", "https://rekt.substack.com/feed", "https://cryptohayes.substack.com/feed", "https://doseofdefi.substack.com/feed", "https://thedefiant.substack.com/feed", "https://tokeninsight.substack.com/feed", "https://blockanalytica.substack.com/feed", } def _get_proxy() -> str | None: return random.choice(PROXY_LIST) if PROXY_LIST else None # ─── CONFIG ─────────────────────────────────────────────────────── COINGECKO_API_KEY = os.getenv("COINGECKO_API_KEY", "").strip() CRYPTOPANIC_API_KEY = os.getenv("CRYPTOPANIC_API_KEY", "").strip() GHOST_URL = os.getenv("GHOST_URL", "http://172.19.0.3:2368") GHOST_CONTENT_KEY = os.getenv("GHOST_CONTENT_API_KEY", "") BACKEND_URL = os.getenv("BACKEND_URL", "http://127.0.0.1:8000") REDDIT_USER_AGENT = "RMI-News-Aggregator/3.0" # ═══════════════════════════════════════════════════════════════════ # RSS FEEDS - 150+ sources across 10 tiers # ═══════════════════════════════════════════════════════════════════ RSS_FEEDS = [ # ── Tier 1: Security & Audit Firms ── ("https://peckshield.medium.com/feed", "PeckShield"), # Smart contract audit leader ("https://slowmist.medium.com/feed", "SlowMist"), # Blockchain security firm ("https://certik.medium.com/feed", "CertiK"), # Security-focused ranking platform ("https://immunefi.medium.com/feed", "Immunefi"), # Bug bounty platform ("https://blog.trailofbits.com/feed/", "Trail of Bits"), # Elite security research ("https://www.chainalysis.com/blog/feed/", "Chainalysis"), # Blockchain analysis ("https://blog.openzeppelin.com/feed/", "OpenZeppelin"), # Smart contract standards ("https://consensys.io/diligence/blog/feed/", "Consensys Diligence"), # Audit arm ("https://halborn.com/feed/", "Halborn"), # Blockchain security ("https://quantstamp.com/blog/feed/", "Quantstamp"), # Smart contract auditing ("https://blog.solidified.io/feed/", "Solidified"), # Audit platform ("https://osec.io/blog/feed", "OSEC"), # Security research ("https://blog.hexens.io/feed", "Hexens"), # Smart contract security ("https://dedaub.com/blog/feed", "Dedaub"), # Security/analysis # ── Tier 2: Major News Outlets ── ("https://cointelegraph.com/rss", "CoinTelegraph"), ("https://decrypt.co/feed", "Decrypt"), ("https://blockworks.co/feed", "Blockworks"), ("https://www.theblock.co/rss.xml", "The Block"), ("https://thedefiant.io/feed", "The Defiant"), ("https://www.bankless.com/feed", "Bankless"), ("https://bitcoinmagazine.com/feed", "Bitcoin Magazine"), ("https://protos.com/feed/", "Protos"), ("https://beincrypto.com/feed/", "BeInCrypto"), ("https://unchainedcrypto.com/feed/", "Unchained"), ("https://coindesk.com/arc/outboundfeeds/rss/", "CoinDesk"), ("https://cryptodaily.co.uk/feed", "CryptoDaily"), ("https://u.today/rss", "U.Today"), ("https://cryptonews.com/news/feed/", "CryptoNews"), ("https://insidebitcoins.com/feed/", "InsideBitcoins"), ("https://nulltx.com/feed/", "NullTX"), ("https://coinpedia.org/feed/", "Coinpedia"), ("https://www.coinspeaker.com/feed/", "CoinSpeaker"), ("https://crypto.news/feed/", "Crypto.News"), ("https://cryptopolitan.com/feed/", "Cryptopolitan"), ("https://finbold.com/feed/", "Finbold"), ("https://coinjournal.net/feed/", "CoinJournal"), ("https://blockonomi.com/feed/", "Blockonomi"), ("https://dailycoin.com/feed/", "DailyCoin"), ("https://cryptoslate.com/feed/", "CryptoSlate"), ("https://coincodex.com/blog/feed/", "CoinCodex"), ("https://crypto.com/blog/feed/", "Crypto.com"), # ── Tier 3: Research & Analytics ── ("https://insights.glassnode.com/feed/", "Glassnode"), # On-chain analytics ("https://dune.com/blog/feed", "Dune Analytics"), # Data platform ("https://messari.io/feed", "Messari"), # Crypto research ("https://www.nansen.ai/research/feed/", "Nansen"), # On-chain analytics ("https://blog.chainalysis.com/reports/feed/", "Chainalysis Reports"), ("https://kaiko.com/feed", "Kaiko"), # Market data ("https://www.thetie.is/feed", "The Tie"), # Data analytics ("https://coinmetrics.io/blog/feed/", "CoinMetrics"), # Network data ("https://blog.intotheblock.com/feed/", "IntoTheBlock"), # On-chain analytics ("https://defillama.com/feed", "DeFi Llama"), # TVL aggregator ("https://tokenterminal.com/blog/feed", "Token Terminal"), # Financial metrics ("https://blog.bitquery.io/feed", "Bitquery"), # Blockchain data # ── Tier 4: Protocol & Chain Blogs ── ("https://blog.ethereum.org/feed.xml", "Ethereum Foundation"), ("https://solana.com/news/feed", "Solana"), ("https://blog.arbitrum.io/feed", "Arbitrum"), ("https://blog.optimism.io/feed", "Optimism"), ("https://polygon.technology/blog/feed", "Polygon"), ("https://www.avax.network/blog/feed", "Avalanche"), ("https://near.org/blog/feed/", "NEAR Protocol"), ("https://blog.sui.io/feed", "Sui"), ("https://aptoslabs.medium.com/feed", "Aptos"), ("https://blog.cosmos.network/feed", "Cosmos"), ("https://polkadot.network/blog/feed/", "Polkadot"), ("https://blog.celestia.org/feed", "Celestia"), ("https://blog.eigenlayer.xyz/feed", "EigenLayer"), ("https://blog.base.org/feed", "Base"), ("https://starkware.medium.com/feed", "StarkWare"), # ── Tier 5: DeFi Protocols ── ("https://uniswap.org/blog/feed.xml", "Uniswap"), ("https://aave.com/blog/feed", "Aave"), ("https://blog.compound.finance/feed/", "Compound"), ("https://blog.makerdao.com/feed/", "MakerDAO"), ("https://blog.lido.fi/feed", "Lido"), ("https://blog.curve.fi/rss/", "Curve"), ("https://blog.balancer.fi/feed/", "Balancer"), ("https://blog.1inch.io/feed", "1inch"), ("https://blog.chain.link/feed", "Chainlink"), ("https://blog.synthetix.io/feed", "Synthetix"), ("https://blog.injective.com/feed", "Injective"), ("https://pancakeswap.medium.com/feed", "PancakeSwap"), ("https://blog.dydx.exchange/feed", "dYdX"), ("https://blog.gmx.io/feed", "GMX"), ("https://blog.pendle.finance/feed", "Pendle"), ("https://blog.morpho.org/feed", "Morpho"), ("https://blog.venus.io/feed", "Venus"), ("https://blog.yearn.fi/feed/", "Yearn"), # ── Tier 6: Exchange Blogs ── ("https://blog.trezor.io/feed", "Trezor"), ("https://blog.kraken.com/feed/", "Kraken"), ("https://blog.bitfinex.com/feed/", "Bitfinex"), ("https://blog.mexc.com/feed", "MEXC"), ("https://www.binance.com/en/feed/news", "Binance"), ("https://blog.coinbase.com/feed", "Coinbase"), ("https://www.okx.com/feed/blog", "OKX"), ("https://blog.bybit.com/feed/", "Bybit"), ("https://www.gemini.com/blog/feed", "Gemini"), ("https://blog.kucoin.com/feed/", "KuCoin"), ("https://www.gate.io/learn/feed", "Gate.io"), ("https://blog.ledger.com/feed/", "Ledger"), # ── Tier 7: VC & Investment Firms ── ("https://a16zcrypto.com/feed/", "a16z Crypto"), ("https://www.paradigm.xyz/feed", "Paradigm"), ("https://panteracapital.com/feed/", "Pantera Capital"), ("https://multicoin.capital/feed/", "Multicoin Capital"), ("https://www.dragonfly.xyz/feed", "Dragonfly"), ("https://variant.fund/feed/", "Variant Fund"), ("https://1confirmation.com/feed", "1confirmation"), ("https://www.electriccapital.com/feed/", "Electric Capital"), ("https://www.coinbase.com/blog/feed", "Coinbase Ventures"), ("https://www.cmsholdings.co/feed/", "CMS Holdings"), ("https://blog.spartangroup.io/feed", "Spartan Group"), ("https://placeholder.vc/feed/", "Placeholder"), # ── Tier 8: Academic & Research Centers ── ("https://cyber.stanford.edu/blog/feed", "Stanford CBR"), ("https://dci.mit.edu/feed", "MIT DCI"), ("https://www.initc3.org/feed", "IC3 (Cornell)"), ("https://blockchain.berkeley.edu/feed/", "Berkeley Blockchain"), ("https://www.src.ac.uk/feed/", "SRC Research (UK)"), ("https://cbr.stanford.edu/feed/", "Stanford CBR"), # ── Tier 9: Government & Regulatory ── ("https://www.sec.gov/news/pressreleases.rss", "SEC"), ("https://www.cftc.gov/PressRoom/PressReleases/rss.xml", "CFTC"), ("https://www.fatf-gafi.org/en/rss.html", "FATF"), ("https://www.fincen.gov/news/rss", "FinCEN"), ("https://www.esma.europa.eu/press-news/rss", "ESMA"), # ── Tier 10: Newsletters (verified RSS endpoints) ── ("https://weekinethereum.substack.com/feed", "Week in Ethereum"), ("https://banklessdao.substack.com/feed", "Bankless DAO"), ("https://thedefiant.substack.com/feed", "The Defiant Substack"), ("https://rekt.substack.com/feed", "REKT Newsletter"), ("https://cryptohayes.substack.com/feed", "Arthur Hayes"), ("https://doseofdefi.substack.com/feed", "Dose of DeFi"), ("https://cryptosecurity.substack.com/feed", "CryptoSecurity"), ("https://tokeninsight.substack.com/feed", "TokenInsight"), ("https://defieducation.substack.com/feed", "DeFi Education"), ("https://blockanalytica.substack.com/feed", "BlockAnalytica"), # ── Tier 11: Additional Security & Research ── ("https://feeds.feedburner.com/TheHackersNews", "The Hacker News"), ("https://www.bleepingcomputer.com/feed/", "BleepingComputer"), ("https://krebsonsecurity.com/feed/", "Krebs on Security"), ("https://www.darkreading.com/rss.xml", "Dark Reading"), ("https://www.web3isgoinggreat.com/feed", "W3IGG"), ("https://forum.openzeppelin.com/posts.rss", "OpenZeppelin Forum"), ("https://ethereum-magicians.org/posts.rss", "ETH Magicians"), ("https://ethresear.ch/posts.rss", "ETH Research"), ("https://gov.uniswap.org/posts.rss", "Uniswap Governance"), ("https://forum.makerdao.com/posts.rss", "MakerDAO Forum"), # ── Tier 12: Market & Trading ── ("https://www.tradingview.com/blog/feed/", "TradingView"), ("https://blog.santiment.net/feed/", "Santiment"), ("https://blog.coinmarketcap.com/feed/", "CoinMarketCap"), ("https://www.coingecko.com/en/blog/feed", "CoinGecko Blog"), # ── Tier 13: NFT & Gaming ── ("https://nftevening.com/feed/", "NFTEvening"), ("https://nftplazas.com/feed/", "NFTPlazas"), ("https://playtoearn.net/feed", "PlayToEarn"), ("https://dappradar.com/blog/feed", "DappRadar"), # ── Tier 14: Layer 2 & Scaling ── ("https://blog.matter-labs.io/feed", "zkSync"), ("https://starknet.io/blog/feed/", "StarkNet"), ("https://blog.scroll.io/feed", "Scroll"), ("https://blog.linea.build/feed", "Linea"), # ── Tier 15: MEV & Infrastructure ── ("https://collective.flashbots.net/latest.rss", "Flashbots"), ("https://blog.chainstack.com/feed/", "Chainstack"), ("https://blog.alchemy.com/feed/", "Alchemy"), ("https://blog.infura.io/feed", "Infura"), ("https://blog.tenderly.co/feed", "Tenderly"), ] # ─── REDDIT ─────────────────────────────────────────────────────── REDDIT_SUBREDDITS = [ "CryptoCurrency", "CryptoMarkets", "ethdev", "CryptoTechnology", "defi", "ethfinance", "solana", "CryptoScams", "Buttcoin", "Bitcoin", "ethereum", "CryptoMoonShots", "altcoin", "CryptoTrading", "NFT", "web3", "layer2", "CryptoGaming", "CryptoCurrencyTrading", "SatoshiStreetBets", "Crypto_General", ] # ─── SENTIMENT KEYWORDS ────────────────────────────────────────── BULLISH_WORDS = [ "surge", "soar", "rally", "bull", "breakout", "pump", "moon", "ath", "all-time high", "record", "adoption", "launch", "partnership", "upgrade", "breakthrough", "milestone", "billions", "growth", "accumulation", "green", "profit", "gain", "positive", "optimistic", "innovation", "mainstream", "institutional", "approval", "etf", ] BEARISH_WORDS = [ "crash", "dump", "bear", "hack", "exploit", "scam", "rug", "drain", "loss", "decline", "sell-off", "liquidation", "crash", "collapse", "default", "bankrupt", "freeze", "halt", "suspend", "delist", "sec", "lawsuit", "fine", "penalty", "ban", "restrict", "phishing", "vulnerability", "attack", "stolen", "breach", "warning", "alert", "risk", "volatile", "uncertainty", "fear", ] # ─── CATEGORIES ─────────────────────────────────────────────────── CATEGORY_KEYWORDS = { "security": [ "hack", "exploit", "vulnerability", "audit", "scam", "rug", "drain", "phishing", "breach", "attack", "theft", "stolen", "bug", "patch", ], "defi": [ "defi", "dex", "lending", "yield", "liquidity", "swap", "amm", "pool", "staking", "farm", "borrow", "lend", ], "regulation": [ "sec", "cftc", "regulation", "compliance", "law", "legal", "ban", "approve", "etf", "license", "registered", ], "market": [ "price", "market", "trading", "volume", "btc", "eth", "bull", "bear", "chart", "analysis", "prediction", "target", ], "protocol": [ "upgrade", "fork", "eip", "governance", "proposal", "vote", "dao", "mainnet", "testnet", "launch", "migration", ], "nft_gaming": ["nft", "gaming", "metaverse", "collectible", "mint", "p2e", "play"], "institutional": [ "institution", "bank", "fund", "venture", "acquisition", "investment", "treasury", "corporate", ], "layer2": ["l2", "layer 2", "rollup", "zk", "optimistic", "scaling", "gas"], "privacy": ["privacy", "zkp", "zero-knowledge", "anonymous", "mixer", "tornado"], "academic": ["research", "paper", "study", "academic", "university", "cryptography"], } class NewsService: """Multi-source crypto news aggregator - 200+ sources, real data only.""" def __init__(self): self.cache: dict[str, Any] = {} self.cache_ttl = 180 # 3 minutes self.last_fetch: datetime | None = None self.seen_hashes: set = set() self._sentiment_cache: dict[str, str] = {} # ─── SENTIMENT ANALYSIS ─────────────────────────────────────── def _analyze_sentiment(self, text: str) -> str: """Real keyword-based sentiment analysis with scoring.""" if not text: return "neutral" text_lower = text.lower() # Cache lookup cache_key = hashlib.md5(text_lower[:200].encode()).hexdigest() if cache_key in self._sentiment_cache: return self._sentiment_cache[cache_key] bull_score = sum(1 for w in BULLISH_WORDS if w in text_lower) bear_score = sum(1 for w in BEARISH_WORDS if w in text_lower) if bull_score > bear_score + 2: result = "bullish" elif bear_score > bull_score + 2: result = "bearish" elif bull_score > bear_score: result = "slightly_bullish" elif bear_score > bull_score: result = "slightly_bearish" else: result = "neutral" if len(self._sentiment_cache) > 5000: self._sentiment_cache.clear() self._sentiment_cache[cache_key] = result return result # ─── CATEGORIZATION ─────────────────────────────────────────── def _categorize(self, text: str) -> str: """Categorize article by keyword matching.""" if not text: return "general" text_lower = text.lower() scores = {} for cat, keywords in CATEGORY_KEYWORDS.items(): scores[cat] = sum(1 for kw in keywords if kw in text_lower) best = max(scores, key=scores.get) return best if scores[best] > 0 else "general" def _categorize_reddit(self, subreddit: str, text: str) -> str: """Override categorization for Reddit posts based on subreddit.""" sub_cats = { "CryptoCurrency": "market", "CryptoMarkets": "market", "ethdev": "protocol", "CryptoTechnology": "protocol", "defi": "defi", "ethfinance": "defi", "solana": "protocol", "CryptoScams": "security", "Bitcoin": "market", "ethereum": "protocol", "NFT": "nft_gaming", "web3": "protocol", "layer2": "layer2", "CryptoGaming": "nft_gaming", } return sub_cats.get(subreddit, self._categorize(text)) # ─── RSS FETCHERS ───────────────────────────────────────────── async def _fetch_rss( self, url: str, source_name: str, limit: int = 5, use_proxy: bool = False ) -> list[dict[str, Any]]: needs_proxy = use_proxy or url in PROXY_REQUIRED_FEEDS for attempt in range(3 if needs_proxy else 1): try: client_kwargs = {"timeout": 12.0, "follow_redirects": True} if needs_proxy and attempt > 0: proxy = _get_proxy() if proxy: client_kwargs["proxy"] = proxy async with httpx.AsyncClient(**client_kwargs) as client: headers = {"User-Agent": "Mozilla/5.0 (compatible; RMI-NewsBot/3.0; +https://rugmunch.io)"} if needs_proxy: headers.update( { "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "en-US,en;q=0.5", "DNT": "1", } ) resp = await client.get(url, headers=headers) if resp.status_code == 200: break elif needs_proxy and attempt < 2: await asyncio.sleep(0.5) continue else: return [] except Exception: if needs_proxy and attempt < 2: await asyncio.sleep(0.5) continue return [] else: return [] try: feed = feedparser.parse(resp.text) except Exception: return [] articles = [] for entry in feed.entries[:limit]: try: title = entry.get("title", "") link = entry.get("link", "") desc = entry.get("summary", "") or entry.get("description", "") or title pub_parsed = entry.get("published_parsed") or entry.get("updated_parsed") if pub_parsed and isinstance(pub_parsed, (tuple, list)) and len(pub_parsed) >= 6: published = datetime(*pub_parsed[:6], tzinfo=UTC).isoformat() pub_dt = datetime(*pub_parsed[:6]) if datetime.utcnow() - pub_dt > timedelta(hours=72): continue else: published = datetime.now(UTC).isoformat() content_hash = hashlib.md5(f"rss:{title}:{link}".encode()).hexdigest() if content_hash in self.seen_hashes: continue self.seen_hashes.add(content_hash) text_for_analysis = f"{title} {desc}" articles.append( { "id": f"rss-{content_hash[:12]}", "title": title, "url": link, "description": desc[:300], "source": source_name, "published_at": published, "category": self._categorize(text_for_analysis), "sentiment": self._analyze_sentiment(text_for_analysis), "kind": "external", "tier": "news", } ) except Exception: continue if articles: logger.info(f"RSS {source_name}: {len(articles)} articles") return articles async def _fetch_all_rss(self, limit_per_source: int = 3) -> list[dict[str, Any]]: """Fetch from ALL RSS feeds in parallel batches to avoid overload.""" all_articles = [] batch_size = 20 for i in range(0, len(RSS_FEEDS), batch_size): batch = RSS_FEEDS[i : i + batch_size] tasks = [self._fetch_rss(url, name, limit_per_source) for url, name in batch] results = await asyncio.gather(*tasks, return_exceptions=True) for result in results: if isinstance(result, list): all_articles.extend(result) if i + batch_size < len(RSS_FEEDS): await asyncio.sleep(0.3) # Rate limit between batches return all_articles # ─── REDDIT FETCHER ─────────────────────────────────────────── async def _fetch_reddit_hot(self, subreddit: str, limit: int = 5) -> list[dict[str, Any]]: try: url = f"https://www.reddit.com/r/{subreddit}/hot.json" async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get( url, params={"limit": limit, "raw_json": "1"}, headers={"User-Agent": REDDIT_USER_AGENT}, ) if resp.status_code != 200: return [] data = resp.json() posts = data.get("data", {}).get("children", []) articles = [] for post in posts: post_data = post.get("data", {}) title = post_data.get("title", "") permalink = post_data.get("permalink", "") url = f"https://reddit.com{permalink}" if permalink else "" text = post_data.get("selftext", "")[:300] created = datetime.fromtimestamp(post_data.get("created_utc", 0), tz=UTC).isoformat() score = post_data.get("score", 0) num_comments = post_data.get("num_comments", 0) if score < 5: # Lower threshold for broader coverage continue content_hash = hashlib.md5(f"reddit:{title}:{permalink}".encode()).hexdigest() if content_hash in self.seen_hashes: continue self.seen_hashes.add(content_hash) text_for_analysis = f"{title} {text}" articles.append( { "id": f"reddit-{content_hash[:12]}", "title": f"r/{subreddit}: {title}", "url": url, "description": text[:250], "source": f"Reddit r/{subreddit}", "published_at": created, "category": self._categorize_reddit(subreddit, text_for_analysis), "sentiment": self._analyze_sentiment(text_for_analysis), "kind": "social", "tier": "social", "reddit_score": score, "reddit_comments": num_comments, } ) if articles: logger.info(f"Reddit r/{subreddit}: {len(articles)} posts") return articles except Exception: return [] async def _fetch_all_reddit(self) -> list[dict[str, Any]]: tasks = [self._fetch_reddit_hot(sub, 5) for sub in REDDIT_SUBREDDITS] results = await asyncio.gather(*tasks, return_exceptions=True) all_posts = [] for result in results: if isinstance(result, list): all_posts.extend(result) return all_posts # ─── API FETCHERS ───────────────────────────────────────────── async def _fetch_coingecko(self, limit: int = 15) -> list[dict[str, Any]]: try: async with httpx.AsyncClient(timeout=10.0) as client: headers = {"x-cg-demo-api-key": COINGECKO_API_KEY} if COINGECKO_API_KEY else {} resp = await client.get( "https://api.coingecko.com/api/v3/news", params={"per_page": limit}, headers=headers, ) if resp.status_code != 200: return [] data = resp.json() articles = [] for item in data.get("data", []): title = item.get("title", "") url = item.get("url", "") desc = item.get("description", "") or title pub_str = item.get("updated_at", item.get("created_at", "")) content_hash = hashlib.md5(f"cg:{title}:{url}".encode()).hexdigest() if content_hash in self.seen_hashes: continue self.seen_hashes.add(content_hash) try: published = datetime.fromisoformat(pub_str.replace("Z", "+00:00")).isoformat() except Exception: published = datetime.now(UTC).isoformat() articles.append( { "id": f"cg-{content_hash[:12]}", "title": title, "url": url, "description": desc[:300], "source": "CoinGecko", "published_at": published, "category": self._categorize(f"{title} {desc}"), "sentiment": self._analyze_sentiment(f"{title} {desc}"), "kind": "api", "tier": "market", } ) return articles except Exception: return [] async def _fetch_cryptopanic(self, limit: int = 15) -> list[dict[str, Any]]: try: params = {"public": "true", "kind": "news", "limit": min(limit, 20)} if CRYPTOPANIC_API_KEY: params["auth_token"] = CRYPTOPANIC_API_KEY url = "https://cryptopanic.com/api/v1/posts/" else: url = "https://cryptopanic.com/api/free/v1/posts/" async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get(url, params=params) if resp.status_code != 200: return [] data = resp.json() articles = [] for item in data.get("results", []): title = item.get("title", "") url = item.get("url", "") source = item.get("source", {}) source_title = source.get("title", "CryptoPanic") if isinstance(source, dict) else "CryptoPanic" pub_str = item.get("published_at", item.get("created_at", "")) content_hash = hashlib.md5(f"cp:{title}:{url}".encode()).hexdigest() if content_hash in self.seen_hashes: continue self.seen_hashes.add(content_hash) try: published = datetime.fromisoformat(pub_str.replace("Z", "+00:00")).isoformat() except Exception: published = datetime.now(UTC).isoformat() votes = item.get("votes", {}) bull = votes.get("positive", 0) if isinstance(votes, dict) else 0 bear = votes.get("negative", 0) if isinstance(votes, dict) else 0 sentiment = "bullish" if bull > bear * 2 else "bearish" if bear > bull * 2 else "neutral" articles.append( { "id": f"cp-{content_hash[:12]}", "title": title, "url": url, "source": source_title, "published_at": published, "category": self._categorize(title), "sentiment": sentiment, "kind": "api", "tier": "market", } ) return articles except Exception: return [] # ─── RMI INTERNAL ───────────────────────────────────────────── async def _fetch_rmi_scanner(self) -> list[dict[str, Any]]: """Fetch real token scanner results as news items.""" try: async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get(f"{BACKEND_URL}/api/v1/scanner/recent?limit=10") if resp.status_code != 200: return [] data = resp.json() articles = [] for scan in data[:10]: token = scan.get("token_name", "Unknown Token") address = scan.get("contract_address", scan.get("token_address", "")) chain = scan.get("chain", "") risk = scan.get("risk_score", 0) scan.get("risk_label", scan.get("risk_level", "unknown")) flags = [] if scan.get("is_honeypot"): flags.append("HONEYPOT") if scan.get("is_rug_pull"): flags.append("RUG PULL RISK") if not scan.get("liquidity_locked"): flags.append("UNLOCKED LP") if scan.get("has_mint_authority"): flags.append("MINTABLE") flag_str = ", ".join(flags) if flags else "clean" content_hash = hashlib.md5(f"scanner:{address}:{chain}".encode()).hexdigest() if content_hash in self.seen_hashes: continue self.seen_hashes.add(content_hash) sentiment = "bearish" if risk > 60 else "slightly_bearish" if risk > 30 else "neutral" articles.append( { "id": f"scan-{content_hash[:12]}", "title": f"{token} ({chain.upper()}) - Risk: {risk}/100 [{flag_str}]", "url": f"https://rugmunch.io/scanner?address={address}&chain={chain}", "description": f"Scanner flagged {token} on {chain}. Risk score: {risk}/100. Flags: {flag_str}. Address: {address[:8]}...", "source": "RMI Scanner", "published_at": scan.get("scanned_at", datetime.now(UTC).isoformat()), "category": "security", "sentiment": sentiment, "kind": "internal", "tier": "rmi", "risk_score": risk, "token_address": address, "chain": chain, } ) return articles except Exception as e: logger.warning(f"RMI Scanner fetch failed: {e}") return [] async def _fetch_rmi_alerts(self) -> list[dict[str, Any]]: """Fetch real RMI alerts as news.""" try: async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get(f"{BACKEND_URL}/api/v1/rag/search?q=scam exploit hack rug drain alert&limit=5") if resp.status_code != 200: return [] data = resp.json() results = data.get("results", data.get("data", [])) articles = [] for r in results[:5]: title = r.get("title", r.get("content", ""))[:120] content = r.get("content", r.get("description", ""))[:300] content_hash = hashlib.md5(f"alert:{title}".encode()).hexdigest() if content_hash in self.seen_hashes: continue self.seen_hashes.add(content_hash) articles.append( { "id": f"alert-{content_hash[:12]}", "title": f"🚨 {title}", "url": r.get("url", "https://rugmunch.io/alerts"), "description": content, "source": "RMI Alerts", "published_at": r.get("created_at", r.get("published_at", datetime.now(UTC).isoformat())), "category": "security", "sentiment": "bearish", "kind": "internal", "tier": "rmi", } ) return articles except Exception: return [] async def _fetch_ghost(self, limit: int = 5) -> list[dict[str, Any]]: """Fetch from our Ghost blog.""" if not GHOST_CONTENT_KEY: return [] try: async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get( f"{GHOST_URL}/ghost/api/content/posts/", params={ "key": GHOST_CONTENT_KEY, "limit": limit, "fields": "title,url,published_at,excerpt,slug", "order": "published_at DESC", }, ) data = resp.json() posts = data.get("posts", []) return [ { "id": f"ghost-{p.get('id', '')}", "title": p.get("title"), "url": f"https://blog.rugmunch.io/{p.get('slug', '')}/", "description": (p.get("excerpt", "") or "")[:200], "source": "RMI Blog", "published_at": p.get("published_at"), "category": "analysis", "sentiment": "neutral", "kind": "internal", "tier": "rmi", } for p in posts ] except Exception: return [] # ─── MAIN AGGREGATOR ────────────────────────────────────────── async def fetch_all( self, limit: int = 100, include_rss: bool = True, include_reddit: bool = True, include_api: bool = True, include_internal: bool = True, ) -> list[dict[str, Any]]: """Fetch from all sources. Heavy operation - call sparingly.""" all_sources = [] # Build task list dynamically tasks = [] task_labels = [] if include_rss: tasks.append(self._fetch_all_rss(3)) task_labels.append("RSS") if include_reddit: tasks.append(self._fetch_all_reddit()) task_labels.append("Reddit") if include_api: tasks.append(self._fetch_coingecko(15)) task_labels.append("CoinGecko") tasks.append(self._fetch_cryptopanic(15)) task_labels.append("CryptoPanic") if include_internal: tasks.append(self._fetch_ghost(5)) task_labels.append("Ghost") tasks.append(self._fetch_rmi_scanner()) task_labels.append("Scanner") tasks.append(self._fetch_rmi_alerts()) task_labels.append("Alerts") # Run all in parallel results = await asyncio.gather(*tasks, return_exceptions=True) for label, result in zip(task_labels, results, strict=False): if isinstance(result, list): all_sources.extend(result) logger.info(f" {label}: {len(result)} items") else: logger.warning(f" {label}: failed") # Deduplicate by content hash seen = set() unique = [] for a in all_sources: aid = a.get("id", "") if aid in seen: continue seen.add(aid) unique.append(a) # Sort by recency def parse_ts(a): try: return datetime.fromisoformat(a.get("published_at", "2020-01-01").replace("Z", "+00:00")) except Exception: return datetime(2020, 1, 1, tzinfo=UTC) unique.sort(key=parse_ts, reverse=True) # Build source list sources = sorted({a["source"] for a in unique}) result = unique[:limit] # Sentiment summary sentiment_counts = { "bullish": 0, "bearish": 0, "neutral": 0, "slightly_bullish": 0, "slightly_bearish": 0, } for a in result: s = a.get("sentiment", "neutral") sentiment_counts[s] = sentiment_counts.get(s, 0) + 1 return { "status": "success", "articles": result, "total": len(result), "sources": sources, "source_count": len(sources), "sentiment_summary": sentiment_counts, "tiers": sorted({a.get("tier", "unknown") for a in result}), "cached": False, "fetched_at": datetime.now(UTC).isoformat(), } # ─── Singleton ─────────────────────────────────────────────────── _news_service: NewsService | None = None def get_news_service() -> NewsService: global _news_service if _news_service is None: _news_service = NewsService() return _news_service