fix: stale-data fallback, per-source rate limits, broken feeds
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 <noreply@anthropic.com>
This commit is contained in:
Marc Shade
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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] = {
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user