diff --git a/src/world_intel_mcp/analysis/alerts.py b/src/world_intel_mcp/analysis/alerts.py new file mode 100644 index 0000000..173acfc --- /dev/null +++ b/src/world_intel_mcp/analysis/alerts.py @@ -0,0 +1,314 @@ +"""Alert digest and weekly trends analysis for world-intel-mcp. + +Provides cross-domain alert aggregation (intel_alert_digest) and +temporal trend analysis (intel_weekly_trends). Both use lazy imports +to avoid circular dependencies with source modules. +""" + +import asyncio +import logging +import math +import sqlite3 +from datetime import datetime, timezone + +logger = logging.getLogger("world-intel-mcp.analysis.alerts") + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +async def _safe_fetch(coro, label: str) -> dict: + """Run a coroutine safely, returning empty dict on failure.""" + try: + return await coro + except Exception as exc: + logger.warning("Alert digest: %s failed: %s", label, exc) + return {} + + +# --------------------------------------------------------------------------- +# Alert Digest +# --------------------------------------------------------------------------- + +_ALERT_THRESHOLDS: dict[str, dict] = { + "space_weather": { + "field": "current_kp", + "threshold": 5.0, + "priority": "high", + "domain": "space", + "message_template": "Geomagnetic storm: Kp={value} ({level})", + }, + "instability": { + "field": "countries", + "threshold": 70, + "priority": "critical", + "domain": "political", + "message_template": "{count} countries above instability threshold", + }, + "military_surge": { + "field": "surge_count", + "threshold": 1, + "priority": "high", + "domain": "military", + "message_template": "{count} military surge anomalies detected", + }, + "cable_health": { + "field": "corridors", + "threshold": 2, + "priority": "high", + "domain": "infrastructure", + "message_template": "{count} cable corridors at risk", + }, + "hotspot_escalation": { + "field": "hotspots", + "threshold": 60, + "priority": "critical", + "domain": "security", + "message_template": "{count} hotspots above escalation threshold", + }, + "internet_outages": { + "field": "outage_count", + "threshold": 5, + "priority": "medium", + "domain": "infrastructure", + "message_template": "{count} internet outages active", + }, + "shipping_stress": { + "field": "stress_score", + "threshold": 30.0, + "priority": "medium", + "domain": "economic", + "message_template": "Shipping stress index at {value}", + }, +} + + +async def fetch_alert_digest(fetcher) -> dict: + """Aggregate alerts from 7 intelligence domains. + + Calls existing source functions in parallel, applies threshold-based + alerting, and returns a prioritized alert list. + + Args: + fetcher: Shared HTTP fetcher with caching and circuit breaking. + + Returns: + Dict with alerts, alert_count, by_priority, domains_checked, source. + """ + # Lazy imports to avoid circular deps + from ..sources import space_weather, infrastructure + from ..sources import intelligence + from ..sources import shipping + + # Fetch all sources in parallel + ( + sw_data, + instability_data, + surge_data, + cable_data, + hotspot_data, + outage_data, + shipping_data, + ) = await asyncio.gather( + _safe_fetch(space_weather.fetch_space_weather(fetcher), "space_weather"), + _safe_fetch(intelligence.fetch_instability_index(fetcher), "instability"), + _safe_fetch(intelligence.fetch_military_surge(fetcher), "military_surge"), + _safe_fetch(infrastructure.fetch_cable_health(fetcher), "cable_health"), + _safe_fetch(intelligence.fetch_hotspot_escalation(fetcher), "hotspot_escalation"), + _safe_fetch(infrastructure.fetch_internet_outages(fetcher), "internet_outages"), + _safe_fetch(shipping.fetch_shipping_index(fetcher), "shipping"), + ) + + alerts: list[dict] = [] + + # Space weather: Kp >= 5 + kp = sw_data.get("current_kp") + if kp is not None and kp >= _ALERT_THRESHOLDS["space_weather"]["threshold"]: + alerts.append({ + "domain": "space", + "priority": "high", + "message": f"Geomagnetic storm: Kp={kp} ({sw_data.get('kp_level', 'Unknown')})", + "value": kp, + }) + + # Instability: countries above threshold + countries = instability_data.get("countries", []) + high_instability = [c for c in countries if c.get("instability_index", 0) >= 70] + if high_instability: + alerts.append({ + "domain": "political", + "priority": "critical", + "message": f"{len(high_instability)} countries above instability threshold (>=70)", + "countries": [c.get("country_name", c.get("country_code")) for c in high_instability[:5]], + "value": len(high_instability), + }) + + # Military surge + surge_count = surge_data.get("surge_count", 0) + if surge_count >= 1: + alerts.append({ + "domain": "military", + "priority": "high", + "message": f"{surge_count} military surge anomalies detected", + "surges": surge_data.get("surges", [])[:3], + "value": surge_count, + }) + + # Cable health: corridors with status_score >= 2 + corridors = cable_data.get("corridors", {}) + at_risk = [ + name for name, info in corridors.items() + if isinstance(info, dict) and info.get("status_score", 0) >= 2 + ] + if at_risk: + alerts.append({ + "domain": "infrastructure", + "priority": "high", + "message": f"{len(at_risk)} cable corridors at elevated risk: {', '.join(at_risk[:3])}", + "corridors": at_risk, + "value": len(at_risk), + }) + + # Hotspot escalation: hotspots with score >= 60 + hotspots = hotspot_data.get("hotspots", []) + hot = [h for h in hotspots if h.get("score", 0) >= 60] + if hot: + alerts.append({ + "domain": "security", + "priority": "critical", + "message": f"{len(hot)} hotspots above escalation threshold", + "hotspots": [h.get("name") for h in hot[:5]], + "value": len(hot), + }) + + # Internet outages + outage_count = outage_data.get("outage_count", 0) + if outage_count >= 5: + alerts.append({ + "domain": "infrastructure", + "priority": "medium", + "message": f"{outage_count} internet outages active", + "value": outage_count, + }) + + # Shipping stress + stress = shipping_data.get("stress_score", 0) + if stress >= 30: + alerts.append({ + "domain": "economic", + "priority": "medium" if stress < 60 else "high", + "message": f"Shipping stress index at {stress} ({shipping_data.get('assessment', 'unknown')})", + "value": stress, + }) + + # Sort: critical > high > medium + priority_order = {"critical": 0, "high": 1, "medium": 2, "low": 3} + alerts.sort(key=lambda a: priority_order.get(a.get("priority", "low"), 4)) + + # Group by priority + by_priority: dict[str, int] = {} + for a in alerts: + p = a.get("priority", "unknown") + by_priority[p] = by_priority.get(p, 0) + 1 + + return { + "alerts": alerts, + "alert_count": len(alerts), + "by_priority": by_priority, + "domains_checked": [ + "space_weather", "instability", "military_surge", + "cable_health", "hotspot_escalation", "internet_outages", + "shipping_stress", + ], + "source": "alert-digest", + "timestamp": _utc_now_iso(), + } + + +# --------------------------------------------------------------------------- +# Weekly Trends +# --------------------------------------------------------------------------- + +async def fetch_weekly_trends(fetcher) -> dict: + """Analyze weekly trends from temporal baselines. + + Reads the TemporalBaseline SQLite database to compute volatility + (coefficient of variation) for each tracked metric, and calls + temporal_anomalies to get current deviations. + + Args: + fetcher: Shared HTTP fetcher with caching and circuit breaking. + + Returns: + Dict with trends, trend_count, analysis_period, source. + """ + from ..analysis.temporal import TemporalBaseline, _DB_PATH + from ..sources import intelligence + + now = datetime.now(timezone.utc) + + # Get current anomalies + anomaly_data = await _safe_fetch( + intelligence.fetch_temporal_anomalies(fetcher), "temporal_anomalies" + ) + + # Read baselines from SQLite + trends: list[dict] = [] + + try: + conn = sqlite3.connect(_DB_PATH) + rows = conn.execute( + "SELECT key, count, mean, m2, updated_at FROM baselines WHERE count >= 5" + ).fetchall() + conn.close() + + for key, n, mean, m2, updated_at in rows: + parts = key.split(":") + if len(parts) < 2: + continue + + event_type = parts[0] + region = parts[1] + weekday = parts[2] if len(parts) > 2 else "" + month = parts[3] if len(parts) > 3 else "" + + # Compute coefficient of variation (volatility) + variance = m2 / (n - 1) if n > 1 else 0.0 + std = math.sqrt(variance) if variance > 0 else 0.0 + cv = (std / mean * 100) if mean > 0 else 0.0 + + trends.append({ + "metric": event_type, + "region": region, + "weekday": weekday, + "month": month, + "observations": n, + "mean": round(mean, 2), + "std_dev": round(std, 2), + "volatility_cv": round(cv, 1), + "last_updated": updated_at, + }) + + except (sqlite3.Error, OSError) as exc: + logger.warning("Failed to read temporal baselines: %s", exc) + + # Sort by volatility descending + trends.sort(key=lambda t: t.get("volatility_cv", 0), reverse=True) + + # Attach current anomalies + current_anomalies = anomaly_data.get("anomalies", []) + + return { + "trends": trends[:50], + "trend_count": len(trends), + "current_anomalies": current_anomalies, + "current_anomaly_count": len(current_anomalies), + "analysis_period": "weekly (by weekday+month seasonality)", + "source": "temporal-weekly-trends", + "timestamp": _utc_now_iso(), + } diff --git a/src/world_intel_mcp/config/countries.py b/src/world_intel_mcp/config/countries.py index 0b0740b..62bfd7e 100644 --- a/src/world_intel_mcp/config/countries.py +++ b/src/world_intel_mcp/config/countries.py @@ -108,3 +108,66 @@ def match_country_by_name(name: str) -> str | None: if keyword in lower or lower in keyword: return iso3 return None + + +# --------------------------------------------------------------------------- +# Election calendar +# --------------------------------------------------------------------------- + +UPCOMING_ELECTIONS: list[dict] = [ + {"country": "Germany", "iso3": "DEU", "election_type": "federal", "date": "2025-02-23", "description": "Federal election (Bundestag)", "instability_impact": "low"}, + {"country": "Ecuador", "iso3": "ECU", "election_type": "presidential", "date": "2025-02-09", "description": "Presidential runoff", "instability_impact": "medium"}, + {"country": "Belarus", "iso3": "BLR", "election_type": "presidential", "date": "2025-01-26", "description": "Presidential election", "instability_impact": "high"}, + {"country": "Canada", "iso3": "CAN", "election_type": "federal", "date": "2025-10-20", "description": "Federal election", "instability_impact": "low"}, + {"country": "Iraq", "iso3": "IRQ", "election_type": "parliamentary", "date": "2025-10-01", "description": "Parliamentary election", "instability_impact": "high"}, + {"country": "Chile", "iso3": "CHL", "election_type": "presidential", "date": "2025-11-16", "description": "Presidential election", "instability_impact": "medium"}, + {"country": "Poland", "iso3": "POL", "election_type": "presidential", "date": "2025-05-18", "description": "Presidential election", "instability_impact": "low"}, + {"country": "Philippines", "iso3": "PHL", "election_type": "midterm", "date": "2025-05-12", "description": "Midterm elections", "instability_impact": "medium"}, + {"country": "Singapore", "iso3": "SGP", "election_type": "general", "date": "2025-05-03", "description": "General election", "instability_impact": "low"}, + {"country": "Australia", "iso3": "AUS", "election_type": "federal", "date": "2025-05-17", "description": "Federal election", "instability_impact": "low"}, + {"country": "South Korea", "iso3": "KOR", "election_type": "presidential", "date": "2025-06-03", "description": "Snap presidential election", "instability_impact": "medium"}, + {"country": "Ivory Coast", "iso3": "CIV", "election_type": "presidential", "date": "2025-10-01", "description": "Presidential election", "instability_impact": "high"}, + {"country": "Norway", "iso3": "NOR", "election_type": "parliamentary", "date": "2025-09-08", "description": "Parliamentary election", "instability_impact": "low"}, + {"country": "United States", "iso3": "USA", "election_type": "midterm", "date": "2026-11-03", "description": "Midterm elections", "instability_impact": "medium"}, + {"country": "Brazil", "iso3": "BRA", "election_type": "municipal", "date": "2026-10-04", "description": "Municipal elections", "instability_impact": "medium"}, + {"country": "Mexico", "iso3": "MEX", "election_type": "midterm", "date": "2027-06-06", "description": "Midterm elections", "instability_impact": "medium"}, + {"country": "France", "iso3": "FRA", "election_type": "presidential", "date": "2027-04-10", "description": "Presidential election", "instability_impact": "medium"}, + {"country": "India", "iso3": "IND", "election_type": "general", "date": "2029-04-01", "description": "General election (projected)", "instability_impact": "medium"}, +] + + +def get_election_risk(iso3: str) -> dict | None: + """Return the next upcoming election for a country with days_until.""" + from datetime import date + + today = date.today() + upper = iso3.upper() + best: dict | None = None + best_days: int = 999999 + + for entry in UPCOMING_ELECTIONS: + if entry["iso3"] != upper: + continue + try: + election_date = date.fromisoformat(entry["date"]) + except ValueError: + continue + days_until = (election_date - today).days + if days_until < best_days: + best_days = days_until + best = {**entry, "days_until": days_until} + + return best + + +# --------------------------------------------------------------------------- +# Nuclear test sites +# --------------------------------------------------------------------------- + +NUCLEAR_TEST_SITES: list[dict] = [ + {"name": "Punggye-ri", "country": "North Korea", "iso3": "PRK", "lat": 41.28, "lon": 129.08, "status": "active", "last_test": "2017-09-03", "notes": "DPRK primary test site, 6 tests conducted"}, + {"name": "Lop Nur", "country": "China", "iso3": "CHN", "lat": 41.75, "lon": 88.35, "status": "dormant", "last_test": "1996-07-29", "notes": "Chinese test site, 45 tests"}, + {"name": "Novaya Zemlya", "country": "Russia", "iso3": "RUS", "lat": 73.37, "lon": 54.78, "status": "dormant", "last_test": "1990-10-24", "notes": "Soviet/Russian arctic test site, Tsar Bomba"}, + {"name": "Nevada NTS", "country": "United States", "iso3": "USA", "lat": 37.07, "lon": -116.05, "status": "dormant_reference", "last_test": "1992-09-23", "notes": "US primary test site, 928 tests"}, + {"name": "Semipalatinsk", "country": "Kazakhstan", "iso3": "KAZ", "lat": 50.07, "lon": 78.43, "status": "closed", "last_test": "1989-10-19", "notes": "Soviet test site, 456 tests, closed 1991"}, +] diff --git a/src/world_intel_mcp/server.py b/src/world_intel_mcp/server.py index 8c17fda..40668c7 100644 --- a/src/world_intel_mcp/server.py +++ b/src/world_intel_mcp/server.py @@ -3,7 +3,7 @@ World Intelligence MCP Server ============================== -Real-time global intelligence across 17 domains: +Real-time global intelligence across 23 domains: financial markets, economic indicators, earthquakes, wildfires, conflict, military flights, infrastructure, and more. @@ -13,6 +13,7 @@ Phase 3: News, Intelligence, Prediction, Displacement, Aviation, Cyber (+9 = 33 Phase 4: Reports — daily brief, country dossier, threat landscape (+3 = 36 tools). Phase 5: Analysis — focal points, signal summary, temporal anomalies, CII v2 (+3 = 39 tools). Phase 6: Military & infrastructure intelligence (+6 = 45 tools). +Phase 7: Health, sanctions, elections, shipping, social, nuclear, alerts, trends (+10 = 55 tools). """ import asyncio @@ -28,7 +29,7 @@ from mcp.types import Tool, TextContent from .cache import Cache from .circuit_breaker import CircuitBreaker from .fetcher import Fetcher -from .sources import markets, economic, seismology, wildfire, conflict, military, infrastructure, maritime, climate, news, intelligence, prediction, displacement, aviation, cyber +from .sources import markets, economic, seismology, wildfire, conflict, military, infrastructure, maritime, climate, news, intelligence, prediction, displacement, aviation, cyber, space_weather, ai_watch, health, sanctions, elections, shipping, social, nuclear from .reports import generator as report_gen logging.basicConfig( @@ -480,6 +481,99 @@ TOOLS: list[Tool] = [ }, }, ), + # --- Space Weather (1 tool) --- + Tool( + name="intel_space_weather", + description="Get solar activity: Kp geomagnetic index, X-ray flare class, solar wind, and SWPC alerts from NOAA.", + inputSchema={"type": "object", "properties": {}}, + ), + # --- AI Watch (1 tool) --- + Tool( + name="intel_ai_releases", + description="Track AI/AGI developments from arXiv, HuggingFace, and AI news feeds. Lab mention trending.", + inputSchema={ + "type": "object", + "properties": { + "limit": {"type": "integer", "description": "Max items (default 50)", "default": 50}, + }, + }, + ), + # --- Health (1 tool) --- + Tool( + name="intel_disease_outbreaks", + description="Aggregate disease outbreak alerts from WHO DON, ProMED, and CIDRAP. Flags high-concern pathogens (Ebola, H5N1, mpox, etc.).", + inputSchema={ + "type": "object", + "properties": { + "limit": {"type": "integer", "description": "Max items (default 50)", "default": 50}, + }, + }, + ), + # --- Sanctions (1 tool) --- + Tool( + name="intel_sanctions_search", + description="Search the US Treasury OFAC Specially Designated Nationals (SDN) sanctions list. Substring match on name, country, program.", + inputSchema={ + "type": "object", + "properties": { + "query": {"type": "string", "description": "Name substring to search"}, + "country": {"type": "string", "description": "Country filter"}, + "program": {"type": "string", "description": "Sanctions program filter"}, + "limit": {"type": "integer", "description": "Max results (default 50)", "default": 50}, + }, + }, + ), + # --- Elections (1 tool) --- + Tool( + name="intel_election_calendar", + description="Get upcoming global election calendar with proximity-based instability risk scoring. Covers 2025-2029.", + inputSchema={ + "type": "object", + "properties": { + "country": {"type": "string", "description": "ISO-3 code or country name filter"}, + }, + }, + ), + # --- Shipping (1 tool) --- + Tool( + name="intel_shipping_index", + description="Compute shipping stress index from dry bulk ETFs (BDRY, SBLK, EGLE, ZIM). Stress score 0-100.", + inputSchema={"type": "object", "properties": {}}, + ), + # --- Social (1 tool) --- + Tool( + name="intel_social_signals", + description="Monitor geopolitical discussion velocity on Reddit (r/worldnews, r/geopolitics). Engagement metrics and trending posts.", + inputSchema={ + "type": "object", + "properties": { + "limit": {"type": "integer", "description": "Max posts per subreddit (default 25)", "default": 25}, + }, + }, + ), + # --- Nuclear (1 tool) --- + Tool( + name="intel_nuclear_monitor", + description="Monitor seismic activity near 5 known nuclear test sites (Punggye-ri, Lop Nur, Novaya Zemlya, Nevada NTS, Semipalatinsk). Concern scoring based on depth, magnitude, distance.", + inputSchema={ + "type": "object", + "properties": { + "hours": {"type": "integer", "description": "Lookback hours (default 72)", "default": 72}, + }, + }, + ), + # --- Alert Digest (1 tool) --- + Tool( + name="intel_alert_digest", + description="Cross-domain alert aggregation from 7 intelligence sources: space weather, instability, military surge, cable health, hotspot escalation, internet outages, shipping stress. Threshold-based prioritized alerts.", + inputSchema={"type": "object", "properties": {}}, + ), + # --- Weekly Trends (1 tool) --- + Tool( + name="intel_weekly_trends", + description="Analyze weekly trends from temporal baselines. Reports volatility (coefficient of variation) and current anomalies across all tracked metrics.", + inputSchema={"type": "object", "properties": {}}, + ), # --- System (1 tool) --- Tool( name="intel_status", @@ -651,6 +745,60 @@ async def _dispatch(name: str, arguments: dict[str, Any]) -> Any: fetcher, corridor=arguments.get("corridor"), ) + # Space Weather + case "intel_space_weather": + return await space_weather.fetch_space_weather(fetcher) + + # AI Watch + case "intel_ai_releases": + return await ai_watch.fetch_ai_watch(fetcher, limit=arguments.get("limit", 50)) + + # Health + case "intel_disease_outbreaks": + return await health.fetch_disease_outbreaks(fetcher, limit=arguments.get("limit", 50)) + + # Sanctions + case "intel_sanctions_search": + return await sanctions.fetch_sanctions_search( + fetcher, + query=arguments.get("query", ""), + country=arguments.get("country"), + program=arguments.get("program"), + limit=arguments.get("limit", 50), + ) + + # Elections + case "intel_election_calendar": + return await elections.fetch_election_calendar( + fetcher, country=arguments.get("country"), + ) + + # Shipping + case "intel_shipping_index": + return await shipping.fetch_shipping_index(fetcher) + + # Social + case "intel_social_signals": + return await social.fetch_social_signals( + fetcher, limit=arguments.get("limit", 25), + ) + + # Nuclear + case "intel_nuclear_monitor": + return await nuclear.fetch_nuclear_monitor( + fetcher, hours=arguments.get("hours", 72), + ) + + # Alert Digest + case "intel_alert_digest": + from .analysis.alerts import fetch_alert_digest + return await fetch_alert_digest(fetcher) + + # Weekly Trends + case "intel_weekly_trends": + from .analysis.alerts import fetch_weekly_trends + return await fetch_weekly_trends(fetcher) + # Reports case "intel_daily_brief": return await report_gen.generate_daily_brief(output_dir=arguments.get("output_dir")) @@ -682,6 +830,14 @@ async def _dispatch(name: str, arguments: dict[str, Any]) -> Any: "displacement": ["unhcr"], "aviation": ["faa"], "cyber": ["feodo-tracker", "cisa-kev", "sans-dshield", "urlhaus"], + "space_weather": ["noaa-swpc"], + "ai_watch": ["arxiv", "huggingface", "ai-news-rss"], + "health": ["who-don", "promed", "cidrap"], + "sanctions": ["ofac-sdn"], + "elections": ["election-calendar"], + "shipping": ["yahoo-finance"], + "social": ["reddit-public"], + "nuclear": ["usgs-nuclear-monitor"], }, } diff --git a/src/world_intel_mcp/sources/elections.py b/src/world_intel_mcp/sources/elections.py new file mode 100644 index 0000000..34d9176 --- /dev/null +++ b/src/world_intel_mcp/sources/elections.py @@ -0,0 +1,112 @@ +"""Election calendar and risk scoring source for world-intel-mcp. + +Pure data module — reads from config.countries.UPCOMING_ELECTIONS and +computes proximity-based risk scores. No I/O, no API keys. +""" + +import logging +from datetime import date, datetime, timezone + +from ..config.countries import UPCOMING_ELECTIONS + +logger = logging.getLogger("world-intel-mcp.sources.elections") + +# --------------------------------------------------------------------------- +# Risk scoring +# --------------------------------------------------------------------------- + +_PROXIMITY_SCORES: list[tuple[int, int]] = [ + (30, 100), + (90, 70), + (180, 40), + (365, 20), +] +_FALLBACK_PROXIMITY = 5 + +_IMPACT_MULTIPLIERS: dict[str, float] = { + "high": 1.5, + "medium": 1.0, + "low": 0.6, +} + + +def _proximity_score(days_until: int) -> int: + """Score based on how close the election is.""" + abs_days = abs(days_until) + for threshold, score in _PROXIMITY_SCORES: + if abs_days <= threshold: + return score + return _FALLBACK_PROXIMITY + + +def _compute_risk(days_until: int, impact: str) -> float: + """Compute election risk: proximity_score * impact_multiplier.""" + base = _proximity_score(days_until) + multiplier = _IMPACT_MULTIPLIERS.get(impact, 1.0) + return round(base * multiplier, 1) + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +async def fetch_election_calendar( + fetcher: object, # noqa: ARG001 — kept for API consistency + country: str | None = None, +) -> dict: + """Return the election calendar with proximity risk scoring. + + Args: + fetcher: Unused (pure data), kept for dispatch consistency. + country: Optional ISO-3 code or country name filter. + + Returns: + Dict with elections list, count, highest_risk entry, source. + """ + today = date.today() + now = datetime.now(timezone.utc) + + elections: list[dict] = [] + country_lower = country.lower().strip() if country else "" + + for entry in UPCOMING_ELECTIONS: + # Filter + if country_lower: + if ( + country_lower not in entry["country"].lower() + and country_lower != entry["iso3"].lower() + ): + continue + + try: + election_date = date.fromisoformat(entry["date"]) + except ValueError: + continue + + days_until = (election_date - today).days + risk_score = _compute_risk(days_until, entry["instability_impact"]) + + elections.append({ + "country": entry["country"], + "iso3": entry["iso3"], + "election_type": entry["election_type"], + "date": entry["date"], + "description": entry["description"], + "days_until": days_until, + "status": "past" if days_until < 0 else "upcoming", + "instability_impact": entry["instability_impact"], + "risk_score": risk_score, + }) + + # Sort by risk_score descending + elections.sort(key=lambda e: e["risk_score"], reverse=True) + + highest_risk = elections[0] if elections else None + + return { + "elections": elections, + "count": len(elections), + "highest_risk": highest_risk, + "source": "election-calendar", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + } diff --git a/src/world_intel_mcp/sources/health.py b/src/world_intel_mcp/sources/health.py new file mode 100644 index 0000000..54333b5 --- /dev/null +++ b/src/world_intel_mcp/sources/health.py @@ -0,0 +1,154 @@ +"""Disease outbreak monitoring source for world-intel-mcp. + +Aggregates disease outbreak alerts from WHO Disease Outbreak News (DON), +ProMED-mail, and CIDRAP. Uses RSS/Atom feeds via feedparser. No API keys required. +""" + +import asyncio +import logging +from datetime import datetime, timezone + +from ..fetcher import Fetcher + +try: + import feedparser +except ImportError: + feedparser = None # type: ignore[assignment] + +logger = logging.getLogger("world-intel-mcp.sources.health") + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +_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"), +] + +HIGH_CONCERN_PATHOGENS: set[str] = { + "ebola", "marburg", "mpox", "h5n1", "avian influenza", "bird flu", + "nipah", "mers", "sars", "cholera", "plague", "anthrax", + "polio", "yellow fever", "hantavirus", "lassa", "rift valley", + "dengue", "zika", "chikungunya", +} + +_CACHE_TTL = 600 # 10 minutes + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _utc_now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _parse_published(entry: dict) -> str | None: + """Parse RSS entry published date to ISO 8601.""" + import time as _time + + for field in ("published_parsed", "updated_parsed"): + parsed_tuple = entry.get(field) + if parsed_tuple is not None: + try: + epoch = _time.mktime(parsed_tuple[:9]) + dt = datetime.fromtimestamp(epoch, tz=timezone.utc) + return dt.strftime("%Y-%m-%dT%H:%M:%SZ") + except (ValueError, TypeError, OverflowError): + pass + + return entry.get("published") or entry.get("updated") + + +def _flag_severity(title: str, summary: str) -> tuple[bool, list[str]]: + """Check if title/summary mentions high-concern pathogens.""" + combined = f"{title} {summary}".lower() + matched = [p for p in HIGH_CONCERN_PATHOGENS if p in combined] + return bool(matched), matched + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +async def fetch_disease_outbreaks( + fetcher: Fetcher, + limit: int = 50, +) -> dict: + """Aggregate disease outbreak alerts from WHO, ProMED, and CIDRAP. + + Args: + fetcher: Shared HTTP fetcher with caching and circuit breaking. + limit: Maximum items to return. + + Returns: + Dict with items, count, high_concern_count, by_organization, source. + """ + if feedparser is None: + return { + "error": "feedparser not installed — run: pip install feedparser", + "items": [], + "count": 0, + } + + all_items: list[dict] = [] + + async def _fetch_feed(name: str, url: str) -> list[dict]: + safe_name = name.lower().replace(" ", "_") + xml_text = await fetcher.get_xml( + url, + source=f"health:{safe_name}", + cache_key=f"health:rss:{safe_name}", + cache_ttl=_CACHE_TTL, + ) + + if xml_text is None: + logger.debug("No data from health feed %s", name) + return [] + + parsed = feedparser.parse(xml_text) + items: list[dict] = [] + + for entry in parsed.get("entries", [])[:30]: + title = entry.get("title", "") + summary = entry.get("summary") or entry.get("description") or "" + is_high_concern, pathogens = _flag_severity(title, summary) + + items.append({ + "title": title, + "link": entry.get("link", ""), + "published": _parse_published(entry), + "summary": summary[:200] if len(summary) > 200 else summary, + "organization": name, + "is_high_concern": is_high_concern, + "pathogens_mentioned": pathogens, + }) + + return items + + tasks = [_fetch_feed(name, url) for name, url in _HEALTH_FEEDS] + results = await asyncio.gather(*tasks) + for items in results: + all_items.extend(items) + + # Sort by published date descending + all_items.sort(key=lambda item: item.get("published") or "", reverse=True) + all_items = all_items[:limit] + + # Counts + high_concern_count = sum(1 for i in all_items if i.get("is_high_concern")) + by_organization: dict[str, int] = {} + for item in all_items: + org = item.get("organization", "unknown") + by_organization[org] = by_organization.get(org, 0) + 1 + + return { + "items": all_items, + "count": len(all_items), + "high_concern_count": high_concern_count, + "by_organization": by_organization, + "source": "health-outbreak-monitor", + "timestamp": _utc_now_iso(), + } diff --git a/src/world_intel_mcp/sources/nuclear.py b/src/world_intel_mcp/sources/nuclear.py new file mode 100644 index 0000000..3e10ccb --- /dev/null +++ b/src/world_intel_mcp/sources/nuclear.py @@ -0,0 +1,239 @@ +"""Nuclear test site seismic monitoring source for world-intel-mcp. + +Monitors seismic activity near known nuclear test sites using USGS +GeoJSON API. Applies Haversine distance filtering and concern scoring +based on depth, magnitude, distance, and site status. +No API keys required. +""" + +import logging +import math +from datetime import datetime, timezone, timedelta + +from ..fetcher import Fetcher +from ..config.countries import NUCLEAR_TEST_SITES + +logger = logging.getLogger("world-intel-mcp.sources.nuclear") + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +_USGS_ENDPOINT = "https://earthquake.usgs.gov/fdsnws/event/1/query" + +_BBOX_PADDING_DEG = 1.5 # degrees around each site for initial USGS query +_MAX_DISTANCE_KM = 100.0 # Haversine filter radius +_CACHE_TTL = 600 # 10 minutes + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float: + """Great-circle distance between two points in km.""" + R = 6371.0 + dlat = math.radians(lat2 - lat1) + dlon = math.radians(lon2 - lon1) + a = ( + math.sin(dlat / 2) ** 2 + + math.cos(math.radians(lat1)) + * math.cos(math.radians(lat2)) + * math.sin(dlon / 2) ** 2 + ) + return R * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a)) + + +def _concern_score( + magnitude: float, + depth_km: float, + distance_km: float, + site_status: str, +) -> tuple[float, str]: + """Score concern level for a seismic event near a test site. + + Returns (score 0-100, level string). + """ + score = 0.0 + + # Magnitude contribution (0-40) + score += min(40.0, magnitude * 8.0) + + # Shallow depth bonus (nuclear tests are <5km depth) — 0-25 + if depth_km <= 2.0: + score += 25.0 + elif depth_km <= 5.0: + score += 20.0 + elif depth_km <= 10.0: + score += 10.0 + elif depth_km <= 30.0: + score += 5.0 + + # Proximity bonus (0-20) + if distance_km <= 10.0: + score += 20.0 + elif distance_km <= 30.0: + score += 15.0 + elif distance_km <= 50.0: + score += 10.0 + elif distance_km <= 100.0: + score += 5.0 + + # Active site multiplier (0-15) + if site_status == "active": + score += 15.0 + elif site_status == "dormant": + score += 5.0 + + score = min(100.0, score) + + if score >= 70: + level = "critical" + elif score >= 50: + level = "high" + elif score >= 30: + level = "elevated" + else: + level = "low" + + return round(score, 1), level + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +async def fetch_nuclear_monitor( + fetcher: Fetcher, + hours: int = 72, +) -> dict: + """Monitor seismic activity near known nuclear test sites. + + For each NUCLEAR_TEST_SITE, queries USGS within a bounding box, + applies Haversine distance filter (<=100km), and scores concern. + + Args: + fetcher: Shared HTTP fetcher with caching and circuit breaking. + hours: Lookback period in hours (default 72). + + Returns: + Dict with sites, total_flagged_events, critical_flags, + flagged_events, source. + """ + import asyncio + + now = datetime.now(timezone.utc) + starttime = (now - timedelta(hours=hours)).strftime("%Y-%m-%dT%H:%M:%S") + + flagged_events: list[dict] = [] + sites: list[dict] = [] + + async def _check_site(site: dict) -> dict: + lat = site["lat"] + lon = site["lon"] + + data = await fetcher.get_json( + _USGS_ENDPOINT, + source="usgs", + cache_key=f"nuclear:usgs:{site['name']}:{hours}", + cache_ttl=_CACHE_TTL, + params={ + "format": "geojson", + "minmagnitude": 1.0, + "starttime": starttime, + "minlatitude": lat - _BBOX_PADDING_DEG, + "maxlatitude": lat + _BBOX_PADDING_DEG, + "minlongitude": lon - _BBOX_PADDING_DEG, + "maxlongitude": lon + _BBOX_PADDING_DEG, + "limit": 50, + }, + ) + + nearby_events: list[dict] = [] + + if data is not None: + for feature in data.get("features", []): + props = feature.get("properties", {}) + geom = feature.get("geometry", {}) + coords = geom.get("coordinates", [0, 0, 0]) + + eq_lon = coords[0] if len(coords) > 0 else 0 + eq_lat = coords[1] if len(coords) > 1 else 0 + eq_depth = coords[2] if len(coords) > 2 else 0 + + distance = _haversine_km(lat, lon, eq_lat, eq_lon) + + if distance > _MAX_DISTANCE_KM: + continue + + magnitude = props.get("mag", 0) or 0 + + score, level = _concern_score( + magnitude=magnitude, + depth_km=eq_depth, + distance_km=distance, + site_status=site["status"], + ) + + # Convert time + epoch_ms = props.get("time") + eq_time = None + if epoch_ms is not None: + try: + eq_time = datetime.fromtimestamp( + epoch_ms / 1000, tz=timezone.utc + ).strftime("%Y-%m-%dT%H:%M:%SZ") + except (ValueError, TypeError, OSError): + pass + + event = { + "site": site["name"], + "site_country": site["country"], + "magnitude": magnitude, + "depth_km": round(eq_depth, 1), + "distance_km": round(distance, 1), + "latitude": eq_lat, + "longitude": eq_lon, + "time": eq_time, + "place": props.get("place"), + "concern_score": score, + "concern_level": level, + } + nearby_events.append(event) + + # Sort by concern_score descending + nearby_events.sort(key=lambda e: e["concern_score"], reverse=True) + + return { + "name": site["name"], + "country": site["country"], + "iso3": site["iso3"], + "lat": lat, + "lon": lon, + "status": site["status"], + "last_test": site["last_test"], + "events_detected": len(nearby_events), + "highest_concern": nearby_events[0] if nearby_events else None, + "events": nearby_events[:5], + } + + tasks = [_check_site(site) for site in NUCLEAR_TEST_SITES] + sites = await asyncio.gather(*tasks) + + # Collect all flagged events + for site_result in sites: + for ev in site_result.get("events", []): + flagged_events.append(ev) + + flagged_events.sort(key=lambda e: e["concern_score"], reverse=True) + critical_flags = sum(1 for e in flagged_events if e["concern_level"] == "critical") + + return { + "sites": list(sites), + "total_flagged_events": len(flagged_events), + "critical_flags": critical_flags, + "flagged_events": flagged_events[:20], + "query": {"hours": hours, "max_distance_km": _MAX_DISTANCE_KM}, + "source": "usgs-nuclear-monitor", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + } diff --git a/src/world_intel_mcp/sources/sanctions.py b/src/world_intel_mcp/sources/sanctions.py new file mode 100644 index 0000000..e5489ce --- /dev/null +++ b/src/world_intel_mcp/sources/sanctions.py @@ -0,0 +1,126 @@ +"""OFAC sanctions search source for world-intel-mcp. + +Searches the US Treasury OFAC Specially Designated Nationals (SDN) list +via the consolidated CSV download. No API key required. +""" + +import csv +import io +import logging +from datetime import datetime, timezone + +from ..fetcher import Fetcher + +logger = logging.getLogger("world-intel-mcp.sources.sanctions") + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +_OFAC_CSV_URL = "https://www.treasury.gov/ofac/downloads/sdn.csv" + +_CACHE_TTL = 86400 # 24 hours + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +async def fetch_sanctions_search( + fetcher: Fetcher, + query: str = "", + country: str | None = None, + program: str | None = None, + limit: int = 50, +) -> dict: + """Search the OFAC SDN consolidated list. + + Downloads the SDN CSV, caches for 24h, and performs substring matching + on name, country, and program fields. + + Args: + fetcher: Shared HTTP fetcher with caching and circuit breaking. + query: Substring to match against entity names. + country: Country filter (substring match). + program: Sanctions program filter (substring match). + limit: Maximum results to return. + + Returns: + Dict with matches, count, total_entities, query info, source. + """ + now = datetime.now(timezone.utc) + + csv_text = await fetcher.get_text( + _OFAC_CSV_URL, + source="ofac-sdn", + cache_key="sanctions:ofac:sdn_csv", + cache_ttl=_CACHE_TTL, + timeout=30.0, + ) + + if csv_text is None: + logger.warning("Failed to download OFAC SDN list") + return { + "matches": [], + "count": 0, + "total_entities": 0, + "query": {"query": query, "country": country, "program": program}, + "source": "ofac-sdn", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + } + + # Parse CSV — SDN format: ent_num, SDN_Name, SDN_Type, Program, Title, + # Call_Sign, Vess_type, Tonnage, GRT, Vess_flag, Vess_owner, Remarks + reader = csv.reader(io.StringIO(csv_text)) + + query_lower = query.lower().strip() + country_lower = country.lower().strip() if country else "" + program_lower = program.lower().strip() if program else "" + + matches: list[dict] = [] + total_entities = 0 + + for row in reader: + if len(row) < 4: + continue + + total_entities += 1 + name = row[1].strip() + sdn_type = row[2].strip() + programs = row[3].strip() + remarks = row[11].strip() if len(row) > 11 else "" + + # Extract country from remarks if present + entry_country = "" + if remarks: + for part in remarks.split(";"): + part = part.strip() + if part.startswith("nationality") or part.startswith("country"): + entry_country = part + + # Apply filters + if query_lower and query_lower not in name.lower(): + continue + if country_lower and country_lower not in entry_country.lower() and country_lower not in remarks.lower(): + continue + if program_lower and program_lower not in programs.lower(): + continue + + matches.append({ + "name": name, + "type": sdn_type, + "programs": programs, + "remarks": remarks[:300] if len(remarks) > 300 else remarks, + }) + + if len(matches) >= limit: + break + + return { + "matches": matches, + "count": len(matches), + "total_entities": total_entities, + "query": {"query": query, "country": country, "program": program}, + "source": "ofac-sdn", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + } diff --git a/src/world_intel_mcp/sources/shipping.py b/src/world_intel_mcp/sources/shipping.py new file mode 100644 index 0000000..70410ae --- /dev/null +++ b/src/world_intel_mcp/sources/shipping.py @@ -0,0 +1,108 @@ +"""Shipping stress index source for world-intel-mcp. + +Tracks dry bulk shipping ETFs via Yahoo Finance to compute a freight +stress index. Reuses _fetch_yahoo_quote from markets.py. +No additional API keys required. +""" + +import asyncio +import logging +from datetime import datetime, timezone + +from ..fetcher import Fetcher +from .markets import _fetch_yahoo_quote + +logger = logging.getLogger("world-intel-mcp.sources.shipping") + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +_SHIPPING_SYMBOLS: dict[str, str] = { + "BDRY": "Breakwave Dry Bulk Shipping ETF", + "SBLK": "Star Bulk Carriers", + "EGLE": "Eagle Bulk Shipping", + "ZIM": "ZIM Integrated Shipping", +} + +# Stress thresholds: if average daily change exceeds these, flag stress +_STRESS_THRESHOLDS: list[tuple[float, str]] = [ + (5.0, "extreme"), + (3.0, "high"), + (1.5, "elevated"), + (0.5, "moderate"), +] + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +async def fetch_shipping_index(fetcher: Fetcher) -> dict: + """Compute a shipping stress index from dry bulk ETF quotes. + + Fetches BDRY, SBLK, EGLE, ZIM from Yahoo Finance, computes an + aggregate stress score (0-100) based on price volatility. + + Args: + fetcher: Shared HTTP fetcher with caching and circuit breaking. + + Returns: + Dict with quotes, stress_score, assessment, signals, source. + """ + now = datetime.now(timezone.utc) + + tasks = [ + _fetch_yahoo_quote(fetcher, sym, f"shipping:quote:{sym}", 300) + for sym in _SHIPPING_SYMBOLS + ] + results = await asyncio.gather(*tasks) + + quotes: list[dict] = [] + change_pcts: list[float] = [] + + for sym, quote in zip(_SHIPPING_SYMBOLS, results): + if quote is None: + continue + change_pct = quote.get("change_pct") + quotes.append({ + "symbol": sym, + "name": _SHIPPING_SYMBOLS[sym], + "price": quote.get("price"), + "change_pct": change_pct, + }) + if change_pct is not None: + change_pcts.append(abs(change_pct)) + + # Compute stress score (0-100) + if change_pcts: + avg_change = sum(change_pcts) / len(change_pcts) + # Scale: 0% change = 0 stress, 10%+ change = 100 stress + stress_score = min(100, round(avg_change * 10, 1)) + else: + avg_change = 0.0 + stress_score = 0.0 + + # Assessment + assessment = "low" + for threshold, label in _STRESS_THRESHOLDS: + if avg_change >= threshold: + assessment = label + break + + # Signals + signals: list[str] = [] + for q in quotes: + pct = q.get("change_pct") + if pct is not None and abs(pct) > 3.0: + direction = "up" if pct > 0 else "down" + signals.append(f"{q['symbol']} {direction} {abs(pct):.1f}%") + + return { + "quotes": quotes, + "stress_score": stress_score, + "assessment": assessment, + "signals": signals, + "source": "yahoo-finance", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + } diff --git a/src/world_intel_mcp/sources/social.py b/src/world_intel_mcp/sources/social.py new file mode 100644 index 0000000..f1c9392 --- /dev/null +++ b/src/world_intel_mcp/sources/social.py @@ -0,0 +1,132 @@ +"""Social signals source for world-intel-mcp. + +Monitors Reddit public JSON endpoints for geopolitical discussion +velocity across r/worldnews and r/geopolitics. No API keys required. +""" + +import asyncio +import logging +from datetime import datetime, timezone + +from ..fetcher import Fetcher + +logger = logging.getLogger("world-intel-mcp.sources.social") + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +_SUBREDDITS: list[str] = ["worldnews", "geopolitics"] + +_REDDIT_HOT_URL = "https://www.reddit.com/r/{subreddit}/hot.json" + +_CACHE_TTL = 300 # 5 minutes + +_HEADERS = { + "User-Agent": "PhoenixAGI-WorldIntel/0.1 (intelligence monitoring)", +} + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +async def fetch_social_signals( + fetcher: Fetcher, + limit: int = 25, +) -> dict: + """Fetch hot posts from geopolitical subreddits for velocity analysis. + + Args: + fetcher: Shared HTTP fetcher with caching and circuit breaking. + limit: Max posts per subreddit. + + Returns: + Dict with posts, velocity_metrics, subreddits_queried, source. + """ + now = datetime.now(timezone.utc) + + all_posts: list[dict] = [] + + async def _fetch_subreddit(subreddit: str) -> list[dict]: + url = _REDDIT_HOT_URL.format(subreddit=subreddit) + data = await fetcher.get_json( + url, + source="reddit", + cache_key=f"social:reddit:{subreddit}:hot", + cache_ttl=_CACHE_TTL, + headers=_HEADERS, + params={"limit": str(limit), "raw_json": "1"}, + ) + + if data is None: + logger.debug("No data from r/%s", subreddit) + return [] + + posts: list[dict] = [] + children = data.get("data", {}).get("children", []) + + for child in children: + post_data = child.get("data", {}) + if not post_data: + continue + + created_utc = post_data.get("created_utc") + created_iso = None + if created_utc is not None: + try: + created_iso = datetime.fromtimestamp( + float(created_utc), tz=timezone.utc + ).strftime("%Y-%m-%dT%H:%M:%SZ") + except (ValueError, TypeError, OSError): + pass + + posts.append({ + "title": post_data.get("title", ""), + "subreddit": subreddit, + "score": post_data.get("score", 0), + "num_comments": post_data.get("num_comments", 0), + "upvote_ratio": post_data.get("upvote_ratio"), + "created": created_iso, + "url": f"https://reddit.com{post_data.get('permalink', '')}", + "is_self": post_data.get("is_self", False), + }) + + return posts + + tasks = [_fetch_subreddit(sub) for sub in _SUBREDDITS] + results = await asyncio.gather(*tasks) + for posts in results: + all_posts.extend(posts) + + # Sort by score descending + all_posts.sort(key=lambda p: p.get("score", 0), reverse=True) + + # Velocity metrics + total_score = sum(p.get("score", 0) for p in all_posts) + total_comments = sum(p.get("num_comments", 0) for p in all_posts) + avg_score = round(total_score / len(all_posts), 1) if all_posts else 0 + avg_comments = round(total_comments / len(all_posts), 1) if all_posts else 0 + + # High engagement threshold + high_engagement = [ + p for p in all_posts + if p.get("score", 0) > 1000 or p.get("num_comments", 0) > 200 + ] + + velocity_metrics = { + "total_posts": len(all_posts), + "total_score": total_score, + "total_comments": total_comments, + "avg_score": avg_score, + "avg_comments": avg_comments, + "high_engagement_count": len(high_engagement), + } + + return { + "posts": all_posts, + "velocity_metrics": velocity_metrics, + "subreddits_queried": _SUBREDDITS, + "source": "reddit-public", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + }