From e47f0a70b232bbd59133fb48ee0352fe5ae0ead2 Mon Sep 17 00:00:00 2001 From: Marc Shade Date: Mon, 23 Feb 2026 20:34:36 -0500 Subject: [PATCH] fix: stale-data fallback, per-source rate limits, broken feeds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cache: add get_stale() for last-known-good fallback when APIs fail. Fetcher: on API failure or tripped circuit breaker, serve stale cached data instead of returning None — dashboards never go blank after first successful fetch. Add per-source rate limits (11 sources configured). Fix Yahoo Finance: compute change_pct from previousClose when regularMarketChangePercent is absent (Yahoo v8 API change). Fix health RSS: replace dead WHO DON/ProMED/CIDRAP URLs with working WHO news, CDC outbreaks, and Outbreak News Today feeds. Fix wildfires: extend FIRMS query from 1-day to 2-day (NRT data lag), include nominal-confidence fires (was filtering to high-only). Co-Authored-By: Claude Opus 4.6 --- src/world_intel_mcp/cache.py | 12 ++++- src/world_intel_mcp/fetcher.py | 71 ++++++++++++++++++++----- src/world_intel_mcp/sources/health.py | 6 +-- src/world_intel_mcp/sources/markets.py | 11 +++- src/world_intel_mcp/sources/wildfire.py | 6 +-- 5 files changed, 83 insertions(+), 23 deletions(-) diff --git a/src/world_intel_mcp/cache.py b/src/world_intel_mcp/cache.py index 54d2bbb..442114a 100644 --- a/src/world_intel_mcp/cache.py +++ b/src/world_intel_mcp/cache.py @@ -56,11 +56,19 @@ class Cache: return None value, expires_at = row if time.time() > expires_at: - conn.execute("DELETE FROM cache WHERE key = ?", (key,)) - conn.commit() return None return json.loads(value) + def get_stale(self, key: str) -> Any | None: + """Get cached value even if expired (last-known-good fallback).""" + conn = self._get_conn() + row = conn.execute( + "SELECT value FROM cache WHERE key = ?", (key,) + ).fetchone() + if row is None: + return None + return json.loads(row[0]) + def set(self, key: str, value: Any, ttl_seconds: int) -> None: """Store a value with TTL in seconds.""" conn = self._get_conn() diff --git a/src/world_intel_mcp/fetcher.py b/src/world_intel_mcp/fetcher.py index f651e0a..3f889cf 100644 --- a/src/world_intel_mcp/fetcher.py +++ b/src/world_intel_mcp/fetcher.py @@ -1,6 +1,8 @@ """Async HTTP fetcher with timeout, retry, rate limiting, and circuit breaker integration. All external HTTP calls in world-intel-mcp go through this module. +Stale-data fallback: when an API call fails, the last-known-good cached +response is returned (marked with _stale=True) so dashboards never go blank. """ import asyncio @@ -20,6 +22,24 @@ _yahoo_lock = asyncio.Lock() _yahoo_last_call: float = 0.0 _YAHOO_MIN_INTERVAL = 0.6 # seconds +# Per-source rate limits (min seconds between calls). +# Sources not listed here have no enforced limit. +_SOURCE_RATE_LIMITS: dict[str, float] = { + "yahoo-finance": 0.6, # unofficial — ~100 req/min safe + "opensky": 6.0, # free tier: 10 req/min + "coingecko": 2.0, # free tier: 30 calls/min + "cloudflare-radar": 3.0, # 20 req/min + "reddit": 1.5, # ~60 req/min (be conservative) + "nasa-firms": 2.0, # API key: ~1000 req/day + "polymarket": 1.0, # be polite + "faa": 1.0, # govt API + "usgs": 1.0, # generous but be polite + "acled": 2.0, # API key based + "nga": 2.0, # govt API +} +_source_locks: dict[str, asyncio.Lock] = {} +_source_last_call: dict[str, float] = {} + class Fetcher: """Centralized HTTP fetcher with caching, retries, and circuit breaking.""" @@ -80,20 +100,21 @@ class Fetcher: Returns: Parsed JSON or None on failure. """ - # Check circuit breaker - if not self.breaker.is_available(source): - logger.debug("Circuit open for %s, skipping", source) - return None - - # Check cache + # Check cache (live) effective_key = cache_key or f"{source}:{url}:{params}" cached = self.cache.get(effective_key) if cached is not None: return cached - # Yahoo rate limiting + # Check circuit breaker — fall back to stale data if open + if not self.breaker.is_available(source): + logger.debug("Circuit open for %s, trying stale cache", source) + return self._stale_fallback(effective_key, source) + + # Per-source rate limiting if yahoo_rate_limit: await self._yahoo_throttle() + await self._source_throttle(source) # Fetch with retries client = await self._get_client() @@ -120,10 +141,10 @@ class Fetcher: attempt + 1, self.max_retries, source, exc, wait) await asyncio.sleep(wait) - # All retries failed + # All retries failed — try stale cache before giving up self.breaker.record_failure(source) logger.warning("Fetch failed for %s: %s (url=%s)", source, last_error, url) - return None + return self._stale_fallback(effective_key, source) async def get_text( self, @@ -136,14 +157,16 @@ class Fetcher: timeout: float | None = None, ) -> str | None: """Fetch raw text with caching and circuit breaking.""" - if not self.breaker.is_available(source): - return None - effective_key = cache_key or f"{source}:text:{url}:{params}" cached = self.cache.get(effective_key) if cached is not None: return cached + if not self.breaker.is_available(source): + return self._stale_fallback(effective_key, source) + + await self._source_throttle(source) + client = await self._get_client() last_error: Exception | None = None @@ -167,7 +190,7 @@ class Fetcher: self.breaker.record_failure(source) logger.warning("Text fetch failed for %s: %s", source, last_error) - return None + return self._stale_fallback(effective_key, source) async def get_xml( self, @@ -180,6 +203,28 @@ class Fetcher: """Fetch XML content (returns raw text for feedparser/ET parsing).""" return await self.get_text(url, source, cache_key, cache_ttl, timeout=timeout) + def _stale_fallback(self, cache_key: str, source: str) -> Any | None: + """Return stale (expired) cached data as last-known-good fallback.""" + stale = self.cache.get_stale(cache_key) + if stale is not None: + logger.info("Serving stale cache for %s (key=%s)", source, cache_key) + return stale + + async def _source_throttle(self, source: str) -> None: + """Enforce per-source rate limit from _SOURCE_RATE_LIMITS.""" + min_interval = _SOURCE_RATE_LIMITS.get(source) + if min_interval is None: + return + if source not in _source_locks: + _source_locks[source] = asyncio.Lock() + async with _source_locks[source]: + now = time.time() + last = _source_last_call.get(source, 0.0) + elapsed = now - last + if elapsed < min_interval: + await asyncio.sleep(min_interval - elapsed) + _source_last_call[source] = time.time() + async def _yahoo_throttle(self) -> None: """Enforce Yahoo Finance rate limit (600ms between calls).""" global _yahoo_last_call diff --git a/src/world_intel_mcp/sources/health.py b/src/world_intel_mcp/sources/health.py index 54333b5..9c94be6 100644 --- a/src/world_intel_mcp/sources/health.py +++ b/src/world_intel_mcp/sources/health.py @@ -22,9 +22,9 @@ logger = logging.getLogger("world-intel-mcp.sources.health") # --------------------------------------------------------------------------- _HEALTH_FEEDS: list[tuple[str, str]] = [ - ("WHO DON", "https://www.who.int/feeds/entity/don/en/rss.xml"), - ("ProMED", "https://promedmail.org/feed/"), - ("CIDRAP", "https://www.cidrap.umn.edu/infectious-disease-topics/rss.xml"), + ("WHO", "https://www.who.int/rss-feeds/news-english.xml"), + ("CDC", "https://tools.cdc.gov/api/v2/resources/media/132608.rss"), + ("Outbreak News", "https://outbreaknewstoday.com/feed/"), ] HIGH_CONCERN_PATHOGENS: set[str] = { diff --git a/src/world_intel_mcp/sources/markets.py b/src/world_intel_mcp/sources/markets.py index 92dc2f5..c6fd78e 100644 --- a/src/world_intel_mcp/sources/markets.py +++ b/src/world_intel_mcp/sources/markets.py @@ -83,10 +83,17 @@ async def _fetch_yahoo_quote(fetcher: Fetcher, symbol: str, cache_key: str, cach try: meta = data["chart"]["result"][0]["meta"] + price = meta.get("regularMarketPrice") + change_pct = meta.get("regularMarketChangePercent") + # Yahoo v8 chart often omits regularMarketChangePercent — compute from previousClose + if change_pct is None and price is not None: + prev = meta.get("previousClose") or meta.get("chartPreviousClose") + if prev and prev > 0: + change_pct = round(((price - prev) / prev) * 100, 4) return { "symbol": symbol, - "price": meta.get("regularMarketPrice"), - "change_pct": meta.get("regularMarketChangePercent"), + "price": price, + "change_pct": change_pct, "currency": meta.get("currency"), } except (KeyError, IndexError, TypeError): diff --git a/src/world_intel_mcp/sources/wildfire.py b/src/world_intel_mcp/sources/wildfire.py index a0fef55..432849c 100644 --- a/src/world_intel_mcp/sources/wildfire.py +++ b/src/world_intel_mcp/sources/wildfire.py @@ -79,8 +79,8 @@ def _parse_fires_csv(csv_text: str) -> list[dict]: if len(fields) <= max(idx_lat, idx_lon, idx_conf): continue - confidence = fields[idx_conf].strip() - if confidence.lower() not in ("high", "h"): + confidence = fields[idx_conf].strip().lower() + if confidence in ("low", "l"): continue try: @@ -171,7 +171,7 @@ async def fetch_wildfires( async def _fetch_region(region_name: str, bbox: str) -> tuple[str, list[dict] | None]: """Fetch fires for a single region, return (name, fires_list_or_None).""" - url = f"{_FIRMS_BASE_URL}/{key}/VIIRS_SNPP_NRT/{bbox}/1" + url = f"{_FIRMS_BASE_URL}/{key}/VIIRS_SNPP_NRT/{bbox}/2" csv_text = await fetcher.get_text( url=url,