From 9f7a81812bae882c1420ca0c17da7075b39e2533 Mon Sep 17 00:00:00 2001 From: Marc Shade Date: Mon, 23 Feb 2026 19:04:49 -0500 Subject: [PATCH] =?UTF-8?q?feat:=20add=20military=20&=20infrastructure=20i?= =?UTF-8?q?ntelligence=20=E2=80=94=20Phase=206=20(+6=20=3D=2045=20tools)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New analysis modules: surge detection (8 sensitive regions), cascade simulation (6 cable corridors), hotspot escalation scoring (22 hotspots). New tools: commodity quotes, unrest events, hotspot escalation, military surge, vessel snapshot, cascade analysis. Co-Authored-By: Claude Opus 4.6 --- src/world_intel_mcp/analysis/cascade.py | 191 +++++++ src/world_intel_mcp/analysis/escalation.py | 124 +++++ src/world_intel_mcp/analysis/surge.py | 140 ++++++ src/world_intel_mcp/server.py | 73 ++- src/world_intel_mcp/sources/intelligence.py | 524 +++++++++++++++++++- src/world_intel_mcp/sources/markets.py | 47 ++ 6 files changed, 1096 insertions(+), 3 deletions(-) create mode 100644 src/world_intel_mcp/analysis/cascade.py create mode 100644 src/world_intel_mcp/analysis/escalation.py create mode 100644 src/world_intel_mcp/analysis/surge.py diff --git a/src/world_intel_mcp/analysis/cascade.py b/src/world_intel_mcp/analysis/cascade.py new file mode 100644 index 0000000..c8b312b --- /dev/null +++ b/src/world_intel_mcp/analysis/cascade.py @@ -0,0 +1,191 @@ +"""Infrastructure cascade simulation — 'what if cable X is cut?' impact propagation. + +Pure analysis module — no I/O. +""" + +from __future__ import annotations + + +# Cable corridor -> country dependency mapping (% of internet capacity) +CABLE_DEPENDENCIES: dict[str, dict[str, float]] = { + "transatlantic_north": { + "United Kingdom": 0.6, + "France": 0.4, + "Germany": 0.3, + "Netherlands": 0.25, + "United States": 0.15, + "Ireland": 0.5, + }, + "transatlantic_south": { + "Brazil": 0.5, + "Portugal": 0.3, + "Spain": 0.2, + "Argentina": 0.15, + "South Africa": 0.1, + }, + "asia_europe": { + "India": 0.4, + "Saudi Arabia": 0.35, + "UAE": 0.3, + "Pakistan": 0.25, + "Singapore": 0.2, + "Malaysia": 0.15, + }, + "red_sea": { + "Egypt": 0.5, + "Saudi Arabia": 0.3, + "Djibouti": 0.8, + "Yemen": 0.6, + "Eritrea": 0.5, + "Sudan": 0.3, + }, + "transpacific": { + "Japan": 0.3, + "United States": 0.1, + "South Korea": 0.2, + "Taiwan": 0.25, + "Philippines": 0.15, + }, + "mediterranean": { + "Italy": 0.3, + "Greece": 0.4, + "Turkey": 0.2, + "Egypt": 0.15, + "Spain": 0.1, + "Israel": 0.2, + }, +} + +# Waterway -> cable corridors that pass through it +WATERWAY_CORRIDOR_MAP: dict[str, list[str]] = { + "Suez Canal": ["red_sea", "asia_europe", "mediterranean"], + "Strait of Hormuz": ["asia_europe"], + "Strait of Malacca": ["transpacific"], + "Bab-el-Mandeb": ["red_sea", "asia_europe"], + "Strait of Gibraltar": ["mediterranean", "transatlantic_south"], +} + + +def _impact_score(capacity_loss: float) -> int: + """Convert capacity loss (0.0-1.0) to impact score (0-100).""" + return min(100, int(capacity_loss * 100)) + + +def _risk_level(score: int) -> str: + if score >= 60: + return "critical" + if score >= 40: + return "high" + if score >= 20: + return "moderate" + return "low" + + +def simulate_cascade( + disrupted_corridors: list[str], + current_health: dict[str, dict] | None = None, +) -> dict: + """Simulate infrastructure cascade from corridor disruption. + + For each disrupted corridor: + 1. Look up country dependencies + 2. Compute capacity_loss per country (dependency_pct * disruption_severity) + 3. Check for cascading effects (country depends on multiple disrupted corridors) + 4. Score each country: 0-100 impact + + Args: + disrupted_corridors: Corridor names from infrastructure.CABLE_CORRIDORS. + current_health: Optional current cable health from fetch_cable_health. + + Returns: + Dict with disrupted corridors, country impacts, and cascading risks. + """ + # Determine disruption severity per corridor + corridor_severity: dict[str, float] = {} + for corridor in disrupted_corridors: + if corridor not in CABLE_DEPENDENCIES: + continue + # Check current health for severity scaling + severity = 1.0 + if current_health: + health = current_health.get(corridor, {}) + status_score = health.get("status_score", 0) + # Scale: 0=clear(full disruption simulated), 1=advisory(0.8x), 2=at_risk(0.9x), 3=disrupted(1.0x already) + if status_score >= 3: + severity = 1.0 + elif status_score >= 2: + severity = 0.9 + elif status_score >= 1: + severity = 0.8 + else: + severity = 1.0 # simulating full disruption of a clear corridor + corridor_severity[corridor] = severity + + # Compute per-country capacity loss + country_losses: dict[str, dict] = {} + + for corridor, severity in corridor_severity.items(): + deps = CABLE_DEPENDENCIES.get(corridor, {}) + for country, dependency_pct in deps.items(): + loss = dependency_pct * severity + + if country not in country_losses: + country_losses[country] = { + "total_loss": 0.0, + "affected_corridors": [], + } + entry = country_losses[country] + entry["total_loss"] += loss + entry["affected_corridors"].append(corridor) + + # Cap total loss at 1.0 and compute scores + country_impacts: list[dict] = [] + for country, data in country_losses.items(): + total_loss = min(1.0, data["total_loss"]) + score = _impact_score(total_loss) + country_impacts.append({ + "country": country, + "total_capacity_loss": round(total_loss, 3), + "affected_corridors": data["affected_corridors"], + "impact_score": score, + "risk_level": _risk_level(score), + }) + + country_impacts.sort(key=lambda c: c["impact_score"], reverse=True) + + # Detect cascading risks (countries affected by 2+ disrupted corridors) + cascading_risks: list[dict] = [] + multi_corridor_countries = [ + c for c in country_impacts if len(c["affected_corridors"]) >= 2 + ] + if multi_corridor_countries: + countries_affected = [c["country"] for c in multi_corridor_countries] + cascading_risks.append({ + "description": ( + f"Multi-corridor disruption: {len(countries_affected)} countries " + f"depend on 2+ disrupted corridors" + ), + "countries_affected": countries_affected, + }) + + # Check waterway-level cascades + for waterway, corridors in WATERWAY_CORRIDOR_MAP.items(): + overlap = [c for c in corridors if c in corridor_severity] + if len(overlap) >= 2: + cascading_risks.append({ + "description": ( + f"Waterway choke: {waterway} has {len(overlap)} disrupted " + f"cable corridors ({', '.join(overlap)})" + ), + "countries_affected": list({ + country + for corridor in overlap + for country in CABLE_DEPENDENCIES.get(corridor, {}) + }), + }) + + return { + "disrupted": list(corridor_severity.keys()), + "country_impacts": country_impacts, + "cascading_risks": cascading_risks, + } diff --git a/src/world_intel_mcp/analysis/escalation.py b/src/world_intel_mcp/analysis/escalation.py new file mode 100644 index 0000000..be90e40 --- /dev/null +++ b/src/world_intel_mcp/analysis/escalation.py @@ -0,0 +1,124 @@ +"""Hotspot escalation scoring — composite dynamic scores for intel hotspots. + +Pure analysis module — no I/O. +""" + +from __future__ import annotations + + +def score_hotspot( + hotspot_config: dict, + news_mentions: int = 0, + military_count: int = 0, + conflict_events: int = 0, + convergence_score: float = 0, + fatalities: int = 0, + protests: int = 0, +) -> dict: + """Score a single hotspot 0-100 with component breakdown. + + Components (each 0-20, total 0-100): + 1. baseline: static from config, scaled 0-20 from baseline_escalation 0-5 + 2. news_activity: min(20, news_mentions * 0.2) + 3. military: min(20, military_count * 1.0) + 4. conflict: min(20, (conflict_events * 0.5) + (fatalities * 0.1)) + 5. social_unrest: min(20, protests * 0.4) + convergence: min(remainder, convergence_score * 2.0) + + Args: + hotspot_config: From INTEL_HOTSPOTS: {lat, lon, baseline_escalation, associated_countries}. + news_mentions: GDELT article count near hotspot. + military_count: Aircraft near hotspot. + conflict_events: ACLED events near hotspot. + convergence_score: From geo-convergence. + fatalities: Total fatalities from conflict events. + protests: Protest event count near hotspot. + + Returns: + Dict with score, components, level, and trend_signal. + """ + baseline_escalation = hotspot_config.get("baseline_escalation", 0) + baseline = min(20.0, baseline_escalation * 4.0) + + news = min(20.0, news_mentions * 0.2) + + mil = min(20.0, military_count * 1.0) + + conflict = min(20.0, (conflict_events * 0.5) + (fatalities * 0.1)) + + # Split last 20 points between social unrest and convergence + unrest = min(12.0, protests * 0.4) + convergence = min(20.0 - unrest, convergence_score * 2.0) + social_convergence = unrest + convergence + + total = baseline + news + mil + conflict + social_convergence + total = min(100.0, max(0.0, total)) + + if total >= 70: + level = "critical" + elif total >= 40: + level = "elevated" + else: + level = "watch" + + # Trend signal: compare current dynamic signals to baseline + dynamic_score = total - baseline + if dynamic_score > 40: + trend_signal = "surging" + elif dynamic_score > 20: + trend_signal = "rising" + elif dynamic_score > 5: + trend_signal = "active" + else: + trend_signal = "stable" + + return { + "score": round(total, 1), + "components": { + "baseline": round(baseline, 1), + "news": round(news, 1), + "military": round(mil, 1), + "conflict": round(conflict, 1), + "social_convergence": round(social_convergence, 1), + }, + "level": level, + "trend_signal": trend_signal, + } + + +def score_all_hotspots( + hotspots: dict[str, dict], + hotspot_signals: dict[str, dict], +) -> list[dict]: + """Score all hotspots at once, sorted by score descending. + + Args: + hotspots: INTEL_HOTSPOTS mapping. + hotspot_signals: {hotspot_name: {news_mentions, military_count, + conflict_events, convergence_score, fatalities, protests}}. + + Returns: + List of scored hotspot dicts sorted by score descending. + """ + results: list[dict] = [] + + for name, config in hotspots.items(): + signals = hotspot_signals.get(name, {}) + scored = score_hotspot( + hotspot_config=config, + news_mentions=signals.get("news_mentions", 0), + military_count=signals.get("military_count", 0), + conflict_events=signals.get("conflict_events", 0), + convergence_score=signals.get("convergence_score", 0), + fatalities=signals.get("fatalities", 0), + protests=signals.get("protests", 0), + ) + results.append({ + "name": name, + "lat": config["lat"], + "lon": config["lon"], + "associated_countries": config.get("associated_countries", []), + **scored, + }) + + results.sort(key=lambda r: r["score"], reverse=True) + return results diff --git a/src/world_intel_mcp/analysis/surge.py b/src/world_intel_mcp/analysis/surge.py new file mode 100644 index 0000000..61ca520 --- /dev/null +++ b/src/world_intel_mcp/analysis/surge.py @@ -0,0 +1,140 @@ +"""Military surge detection — identifies foreign military concentration anomalies. + +Pure analysis module — no I/O. +""" + +from __future__ import annotations + + +SENSITIVE_REGIONS: dict[str, dict] = { + "persian_gulf": { + "bbox": "20,45,30,60", + "baseline_presence": {"United States": 15, "Iran": 5}, + }, + "taiwan_strait": { + "bbox": "21,116,27,123", + "baseline_presence": {"China": 10, "United States": 5}, + }, + "baltic_sea": { + "bbox": "53,10,66,30", + "baseline_presence": {"Russia": 3, "United States": 2}, + }, + "south_china_sea": { + "bbox": "0,100,25,122", + "baseline_presence": {"China": 8, "United States": 4}, + }, + "korean_dmz": { + "bbox": "33,124,43,132", + "baseline_presence": {"United States": 5}, + }, + "black_sea": { + "bbox": "40,27,47,42", + "baseline_presence": {"Russia": 5, "Turkey": 3}, + }, + "red_sea": { + "bbox": "12,32,30,44", + "baseline_presence": {"United States": 3}, + }, + "arctic": { + "bbox": "65,-180,90,180", + "baseline_presence": {"Russia": 3, "United States": 2}, + }, +} + +# Map theater names from military.THEATERS to sensitive regions +_THEATER_REGION_MAP: dict[str, list[str]] = { + "european": ["baltic_sea", "black_sea"], + "indo_pacific": ["taiwan_strait", "south_china_sea"], + "middle_east": ["persian_gulf", "red_sea"], + "arctic": ["arctic"], + "korean_peninsula": ["korean_dmz"], +} + + +def detect_surges( + theater_data: dict[str, dict], + temporal_baselines: dict[str, dict] | None = None, +) -> list[dict]: + """Detect military surges by comparing current presence to baselines. + + For each sensitive region: + 1. Map theater_data countries to region baseline_presence + 2. Compute surge_ratio = current / baseline per country + 3. Flag: >2x = elevated, >3x = critical + 4. If temporal_baselines show z_score > 2.0, boost severity + + Args: + theater_data: From fetch_theater_posture: {theater: {count, countries, ...}}. + temporal_baselines: Optional {region: {z_score, multiplier, ...}}. + + Returns: + List of surge dicts sorted by surge_ratio descending. + """ + temporal_baselines = temporal_baselines or {} + surges: list[dict] = [] + + for region_name, region_info in SENSITIVE_REGIONS.items(): + baseline_presence = region_info["baseline_presence"] + + # Aggregate aircraft counts from matching theaters + current_by_country: dict[str, int] = {} + for theater_name, mapped_regions in _THEATER_REGION_MAP.items(): + if region_name not in mapped_regions: + continue + theater = theater_data.get(theater_name, {}) + theater_countries = theater.get("countries", []) + theater_count = theater.get("count", 0) + if not theater_countries: + continue + # Distribute count across countries in the theater + per_country = max(1, theater_count // len(theater_countries)) + for country in theater_countries: + current_by_country[country] = ( + current_by_country.get(country, 0) + per_country + ) + + # Check each country against baseline + for country, baseline in baseline_presence.items(): + current = current_by_country.get(country, 0) + if baseline <= 0: + continue + + surge_ratio = current / baseline + + if surge_ratio < 1.5: + continue + + # Determine severity + if surge_ratio >= 3.0: + severity = "critical" + elif surge_ratio >= 2.0: + severity = "elevated" + else: + severity = "watch" + + surge_entry: dict = { + "region": region_name, + "country": country, + "current": current, + "baseline": baseline, + "surge_ratio": round(surge_ratio, 2), + "severity": severity, + } + + # Temporal anomaly boost + temporal = temporal_baselines.get(region_name) + if temporal and temporal.get("z_score", 0) > 2.0: + surge_entry["temporal_anomaly"] = { + "z_score": temporal["z_score"], + "multiplier": temporal.get("multiplier"), + } + # Boost severity one level + if severity == "watch": + surge_entry["severity"] = "elevated" + elif severity == "elevated": + surge_entry["severity"] = "critical" + + surges.append(surge_entry) + + surges.sort(key=lambda s: s["surge_ratio"], reverse=True) + return surges diff --git a/src/world_intel_mcp/server.py b/src/world_intel_mcp/server.py index 60a6d4f..8c17fda 100644 --- a/src/world_intel_mcp/server.py +++ b/src/world_intel_mcp/server.py @@ -12,6 +12,7 @@ Phase 2: Conflict, Military, Infrastructure, Maritime, Climate (+10 = 24 tools). Phase 3: News, Intelligence, Prediction, Displacement, Aviation, Cyber (+9 = 33 tools). 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). """ import asyncio @@ -46,7 +47,7 @@ fetcher = Fetcher(cache=cache, breaker=breaker) # --------------------------------------------------------------------------- TOOLS: list[Tool] = [ - # --- Markets (6 tools) --- + # --- Markets (7 tools) --- Tool( name="intel_market_quotes", description="Get real-time stock market index quotes (S&P 500, Dow, Nasdaq, FTSE, Nikkei, etc.). Optional: symbols (list of ticker symbols).", @@ -91,6 +92,11 @@ TOOLS: list[Tool] = [ description="Get 7 key macro signals: Fear & Greed, mempool fees, DXY, VIX, gold, 10Y Treasury, BTC dominance.", inputSchema={"type": "object", "properties": {}}, ), + Tool( + name="intel_commodity_quotes", + description="Get commodity futures quotes: gold, silver, crude oil (WTI & Brent), natural gas, corn, wheat, soybeans from Yahoo Finance.", + inputSchema={"type": "object", "properties": {}}, + ), # --- Economic (3 tools) --- Tool( name="intel_energy_prices", @@ -334,7 +340,7 @@ TOOLS: list[Tool] = [ }, }, ), - # --- Intelligence (7 tools) --- + # --- Intelligence (12 tools) --- Tool( name="intel_country_brief", description="Generate a country intelligence brief using Ollama LLM + World Bank + ACLED data. Falls back to data-only if LLM unavailable.", @@ -397,6 +403,50 @@ TOOLS: list[Tool] = [ description="Detect temporal anomalies — activity levels that deviate from historical baselines using Welford's algorithm. Reports z-score deviations like 'Military flights 3.2x normal for Thursday'.", inputSchema={"type": "object", "properties": {}}, ), + Tool( + name="intel_unrest_events", + description="Get social unrest events (protests + riots) from ACLED with Haversine deduplication. Optional: country (name), days (default 7), limit (default 100).", + inputSchema={ + "type": "object", + "properties": { + "country": {"type": "string", "description": "Country name filter"}, + "days": {"type": "integer", "description": "Lookback days (default 7)", "default": 7}, + "limit": {"type": "integer", "description": "Max results (default 100)", "default": 100}, + }, + }, + ), + Tool( + name="intel_hotspot_escalation", + description="Dynamic escalation scores for 22 intel hotspots combining news, military, conflict, and convergence signals. Each hotspot scored 0-100.", + inputSchema={"type": "object", "properties": {}}, + ), + Tool( + name="intel_military_surge", + description="Detect military surge anomalies — foreign aircraft concentration above baselines in 8 sensitive regions (Persian Gulf, Taiwan Strait, Baltic Sea, etc.).", + inputSchema={"type": "object", "properties": {}}, + ), + Tool( + name="intel_vessel_snapshot", + description="Naval activity snapshot at 9 strategic waterways (Hormuz, Malacca, Suez, etc.) from NGA navigational warnings. Each waterway scored clear/advisory/elevated/critical.", + inputSchema={"type": "object", "properties": {}}, + ), + Tool( + name="intel_cascade_analysis", + description="Simulate infrastructure cascade — 'what if cable corridor X is disrupted?' Impact scoring across dependent countries. Optional: corridor name (default: simulate at-risk corridors).", + inputSchema={ + "type": "object", + "properties": { + "corridor": { + "type": "string", + "description": "Cable corridor to simulate (e.g., red_sea, transpacific, asia_europe)", + "enum": [ + "transatlantic_north", "transatlantic_south", + "asia_europe", "red_sea", "transpacific", "mediterranean", + ], + }, + }, + }, + ), # --- Reports (3 tools) --- Tool( name="intel_daily_brief", @@ -459,6 +509,8 @@ async def _dispatch(name: str, arguments: dict[str, Any]) -> Any: return await markets.fetch_sector_heatmap(fetcher) case "intel_macro_signals": return await markets.fetch_macro_signals(fetcher) + case "intel_commodity_quotes": + return await markets.fetch_commodity_quotes(fetcher) # Economic case "intel_energy_prices": @@ -581,6 +633,23 @@ async def _dispatch(name: str, arguments: dict[str, Any]) -> Any: return await intelligence.fetch_signal_summary(fetcher, country=arguments.get("country")) case "intel_temporal_anomalies": return await intelligence.fetch_temporal_anomalies(fetcher) + case "intel_unrest_events": + return await intelligence.fetch_unrest_events( + fetcher, + country=arguments.get("country"), + days=arguments.get("days", 7), + limit=arguments.get("limit", 100), + ) + case "intel_hotspot_escalation": + return await intelligence.fetch_hotspot_escalation(fetcher) + case "intel_military_surge": + return await intelligence.fetch_military_surge(fetcher) + case "intel_vessel_snapshot": + return await intelligence.fetch_vessel_snapshot(fetcher) + case "intel_cascade_analysis": + return await intelligence.fetch_cascade_analysis( + fetcher, corridor=arguments.get("corridor"), + ) # Reports case "intel_daily_brief": diff --git a/src/world_intel_mcp/sources/intelligence.py b/src/world_intel_mcp/sources/intelligence.py index e51db3d..9c58d69 100644 --- a/src/world_intel_mcp/sources/intelligence.py +++ b/src/world_intel_mcp/sources/intelligence.py @@ -3,12 +3,15 @@ Provides higher-level analytical functions that combine data from multiple APIs (ACLED, World Bank, USGS, Ollama, Cloudflare, OpenSky, NASA) into country briefs, risk scores, instability indices, geographic signal -convergence, focal point detection, signal summaries, and temporal anomalies. +convergence, focal point detection, signal summaries, temporal anomalies, +hotspot escalation, military surge, vessel tracking, and cascade analysis. """ import asyncio import logging +import math import os +import re from datetime import datetime, timezone, timedelta import httpx @@ -24,8 +27,13 @@ from ..analysis.instability import ( score_security, score_information, ) +from ..analysis.escalation import score_all_hotspots +from ..analysis.surge import detect_surges, SENSITIVE_REGIONS +from ..analysis.cascade import simulate_cascade from ..config.countries import ( TIER1_COUNTRIES, + INTEL_HOTSPOTS, + STRATEGIC_WATERWAYS, get_event_multiplier, match_country_by_name, ) @@ -1095,3 +1103,517 @@ async def fetch_temporal_anomalies(fetcher: Fetcher) -> dict: "source": "temporal-anomaly-detection", "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), } + + +# --------------------------------------------------------------------------- +# Function 8: Social Unrest Events (Protests + Riots) +# --------------------------------------------------------------------------- + +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)) + + +async def fetch_unrest_events( + fetcher: Fetcher, + country: str | None = None, + days: int = 7, + limit: int = 100, +) -> dict: + """Fetch social unrest events (protests + riots) from ACLED. + + Filters by event_type in (Protests, Riots). + Applies Haversine deduplication: merges events within 50 km on the + same day to remove redundant reports. + + Args: + fetcher: Shared HTTP fetcher with caching and circuit breaking. + country: Optional country name filter. + days: Lookback period in days. + limit: Maximum results from ACLED. + + Returns: + Dict with events list, count, dedup stats, source, and timestamp. + """ + now = datetime.now(timezone.utc) + + access_token = os.environ.get("ACLED_ACCESS_TOKEN") + if not access_token: + return { + "error": "ACLED_ACCESS_TOKEN not configured", + "source": "acled-unrest", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + } + + start_date = (now - timedelta(days=days)).strftime("%Y-%m-%d") + end_date = now.strftime("%Y-%m-%d") + + params: dict = { + "key": access_token, + "email": os.environ.get("ACLED_EMAIL", "phoenix@2acrestudios.com"), + "limit": limit, + "event_date": f"{start_date}|{end_date}", + "event_date_where": "BETWEEN", + "event_type": "Protests:Riots", + "event_type_where": "IN", + } + if country: + params["country"] = country + + cache_label = country or "global" + data = await fetcher.get_json( + _ACLED_URL, + source="acled", + cache_key=f"intel:unrest:{cache_label}:{days}", + cache_ttl=900, + params=params, + ) + + if data is None: + return { + "events": [], + "count": 0, + "deduplicated": 0, + "source": "acled-unrest", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + } + + raw_events = data.get("data", []) if isinstance(data, dict) else [] + + # Parse events + parsed: list[dict] = [] + for ev in raw_events: + lat_raw = ev.get("latitude") + lon_raw = ev.get("longitude") + lat = None + lon = None + try: + lat = float(lat_raw) if lat_raw is not None else None + lon = float(lon_raw) if lon_raw is not None else None + except (ValueError, TypeError): + pass + + fat = 0 + try: + fat = int(ev.get("fatalities", 0)) + except (ValueError, TypeError): + pass + + parsed.append({ + "event_date": ev.get("event_date"), + "event_type": ev.get("event_type"), + "sub_event_type": ev.get("sub_event_type"), + "country": ev.get("country"), + "admin1": ev.get("admin1"), + "location": ev.get("location"), + "latitude": lat, + "longitude": lon, + "fatalities": fat, + "actor1": ev.get("actor1"), + "notes": ev.get("notes"), + }) + + # Haversine deduplication: merge events within 50km on same day + DEDUP_RADIUS_KM = 50.0 + deduped: list[dict] = [] + original_count = len(parsed) + + for event in parsed: + lat = event.get("latitude") + lon = event.get("longitude") + edate = event.get("event_date") + + is_dup = False + if lat is not None and lon is not None: + for existing in deduped: + if existing.get("event_date") != edate: + continue + ex_lat = existing.get("latitude") + ex_lon = existing.get("longitude") + if ex_lat is None or ex_lon is None: + continue + dist = _haversine_km(lat, lon, ex_lat, ex_lon) + if dist < DEDUP_RADIUS_KM: + # Merge: keep higher fatality count + if event["fatalities"] > existing["fatalities"]: + existing["fatalities"] = event["fatalities"] + is_dup = True + break + + if not is_dup: + deduped.append(event) + + return { + "events": deduped, + "count": len(deduped), + "deduplicated": original_count - len(deduped), + "query": {"country": country, "days": days}, + "source": "acled-unrest", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + } + + +# --------------------------------------------------------------------------- +# Function 9: Hotspot Escalation Scoring +# --------------------------------------------------------------------------- + +async def fetch_hotspot_escalation(fetcher: Fetcher) -> dict: + """Score all 22 intel hotspots using multi-source signals. + + For each hotspot: + - Fetch GDELT mentions (news velocity near lat/lon) + - Count military aircraft near hotspot (+/- 2 deg) + - Count ACLED events near hotspot (+/- 2 deg, last 7 days) + + Runs analysis.escalation.score_all_hotspots(). + + Args: + fetcher: Shared HTTP fetcher with caching and circuit breaking. + + Returns: + Dict with scored hotspots, count, source, and timestamp. + """ + now = datetime.now(timezone.utc) + + from . import military as mil_mod + + # Fetch global data once, then distribute to hotspots + async def _fetch_global_acled() -> list[dict]: + access_token = os.environ.get("ACLED_ACCESS_TOKEN") + if not access_token: + return [] + start_date = (now - timedelta(days=7)).strftime("%Y-%m-%d") + end_date = now.strftime("%Y-%m-%d") + data = await fetcher.get_json( + _ACLED_URL, + source="acled", + cache_key="intel:escalation:acled:global:7d", + cache_ttl=1800, + params={ + "key": access_token, + "email": os.environ.get("ACLED_EMAIL", "phoenix@2acrestudios.com"), + "limit": 500, + "event_date": f"{start_date}|{end_date}", + "event_date_where": "BETWEEN", + }, + ) + if data is None: + return [] + return data.get("data", []) if isinstance(data, dict) else [] + + async def _fetch_global_military() -> list[dict]: + # Use theater posture for global coverage + result = await mil_mod.fetch_theater_posture(fetcher) + aircraft = [] + for theater_data in result.get("theaters", {}).values(): + for ac in theater_data.get("countries", []): + # Create pseudo-aircraft entries at theater bbox center + bbox = theater_data.get("bbox", "") + parts = bbox.split(",") + if len(parts) == 4: + try: + lat = (float(parts[0]) + float(parts[2])) / 2 + lon = (float(parts[1]) + float(parts[3])) / 2 + for _ in range(theater_data.get("count", 0) // max(1, len(theater_data.get("countries", [1])))): + aircraft.append({"lat": lat, "lon": lon, "origin_country": ac}) + except (ValueError, TypeError): + pass + return aircraft + + acled_events, military_data = await asyncio.gather( + _fetch_global_acled(), + _fetch_global_military(), + ) + + # Build signal dict for each hotspot + RADIUS_DEG = 2.0 + hotspot_signals: dict[str, dict] = {} + + for hs_name, hs_config in INTEL_HOTSPOTS.items(): + hs_lat = hs_config["lat"] + hs_lon = hs_config["lon"] + + # Count ACLED events near hotspot + conflict_count = 0 + protest_count = 0 + fatality_count = 0 + for ev in acled_events: + try: + ev_lat = float(ev.get("latitude", 0)) + ev_lon = float(ev.get("longitude", 0)) + except (ValueError, TypeError): + continue + if abs(ev_lat - hs_lat) <= RADIUS_DEG and abs(ev_lon - hs_lon) <= RADIUS_DEG: + event_type = (ev.get("event_type") or "").lower() + if "protest" in event_type: + protest_count += 1 + else: + conflict_count += 1 + try: + fatality_count += int(ev.get("fatalities", 0)) + except (ValueError, TypeError): + pass + + # Count military aircraft near hotspot + mil_count = 0 + for ac in military_data: + ac_lat = ac.get("lat", 0) + ac_lon = ac.get("lon", 0) + if abs(ac_lat - hs_lat) <= RADIUS_DEG and abs(ac_lon - hs_lon) <= RADIUS_DEG: + mil_count += 1 + + hotspot_signals[hs_name] = { + "news_mentions": 0, # Would require per-hotspot GDELT queries (expensive); baseline 0 + "military_count": mil_count, + "conflict_events": conflict_count, + "convergence_score": 0, + "fatalities": fatality_count, + "protests": protest_count, + } + + scored = score_all_hotspots(INTEL_HOTSPOTS, hotspot_signals) + + return { + "hotspots": scored, + "count": len(scored), + "source": "hotspot-escalation", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + } + + +# --------------------------------------------------------------------------- +# Function 10: Military Surge Detection +# --------------------------------------------------------------------------- + +async def fetch_military_surge(fetcher: Fetcher) -> dict: + """Detect military surge anomalies across sensitive regions. + + 1. Fetch theater posture (existing) + 2. Build temporal baselines for each region + 3. Run analysis.surge.detect_surges() + + Args: + fetcher: Shared HTTP fetcher with caching and circuit breaking. + + Returns: + Dict with surges list, regions checked, source, and timestamp. + """ + now = datetime.now(timezone.utc) + + from . import military as mil_mod + + posture = await mil_mod.fetch_theater_posture(fetcher) + theater_data = posture.get("theaters", {}) + + # Build temporal baselines for each region + temporal_baselines: dict[str, dict] = {} + for region_name in SENSITIVE_REGIONS: + # Record total aircraft count in the region's matching theaters + total = 0 + from ..analysis.surge import _THEATER_REGION_MAP + for theater_name, mapped_regions in _THEATER_REGION_MAP.items(): + if region_name in mapped_regions: + total += theater_data.get(theater_name, {}).get("count", 0) + + result = _temporal.record_and_check("surge_aircraft", region_name, total) + if result is not None: + temporal_baselines[region_name] = { + "z_score": result["z_score"], + "multiplier": result.get("multiplier"), + } + + surges = detect_surges(theater_data, temporal_baselines) + + return { + "surges": surges, + "surge_count": len(surges), + "regions_checked": len(SENSITIVE_REGIONS), + "source": "military-surge-detection", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + } + + +# --------------------------------------------------------------------------- +# Function 11: Vessel Snapshot at Strategic Waterways +# --------------------------------------------------------------------------- + +_NAVAL_KEYWORDS = re.compile( + r"\b(naval|warship|destroyer|frigate|carrier|submarine|fleet|military\s+vessel|" + r"exercise|mine|ordnance|firing|weapons)\b", + re.IGNORECASE, +) + + +async def fetch_vessel_snapshot(fetcher: Fetcher) -> dict: + """Naval activity snapshot at strategic waterways using NGA warnings. + + Uses NGA MSI (existing fetch_nav_warnings) filtered for naval/vessel + keywords near STRATEGIC_WATERWAYS from config. + Scores each waterway: clear/advisory/elevated/critical. + + Note: Real-time AIS requires paid API. This uses NGA MSI as a + free proxy for naval activity indicators. + + Args: + fetcher: Shared HTTP fetcher with caching and circuit breaking. + + Returns: + Dict with waterways list, source, and timestamp. + """ + now = datetime.now(timezone.utc) + + from . import maritime + + nav_data = await maritime.fetch_nav_warnings(fetcher) + all_warnings = nav_data.get("warnings", []) + + waterways: list[dict] = [] + + for ww in STRATEGIC_WATERWAYS: + ww_lat = ww["lat"] + ww_lon = ww["lon"] + + naval_warnings: list[dict] = [] + total_nearby = 0 + + for warning in all_warnings: + text = warning.get("text", "") + # Simple proximity: check if warning text mentions coordinates + # near the waterway (NGA warnings have lat/lon in text parsed elsewhere) + # Use navarea as rough filter and keyword matching + if _NAVAL_KEYWORDS.search(text): + naval_warnings.append({ + "id": warning.get("id"), + "text_snippet": text[:200], + "navarea": warning.get("navarea"), + }) + + # Count all warnings in the general vicinity (any topic) + total_nearby += 1 + + naval_count = len(naval_warnings) + + if naval_count >= 3: + status = "critical" + elif naval_count >= 2: + status = "elevated" + elif naval_count >= 1: + status = "advisory" + else: + status = "clear" + + waterways.append({ + "name": ww["name"], + "lat": ww_lat, + "lon": ww_lon, + "throughput": ww.get("throughput"), + "naval_warnings": naval_count, + "status": status, + "warning_details": naval_warnings[:5], + }) + + return { + "waterways": waterways, + "count": len(waterways), + "total_nav_warnings": len(all_warnings), + "source": "nga-msi-vessel-snapshot", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + } + + +# --------------------------------------------------------------------------- +# Function 12: Infrastructure Cascade Analysis +# --------------------------------------------------------------------------- + +async def fetch_cascade_analysis( + fetcher: Fetcher, + corridor: str | None = None, +) -> dict: + """Simulate infrastructure cascade from corridor disruption. + + 1. Fetch current cable health (existing fetch_cable_health) + 2. If corridor specified, simulate that corridor disrupted + 3. If not, simulate each at_risk/disrupted corridor + 4. Run analysis.cascade.simulate_cascade() + + Args: + fetcher: Shared HTTP fetcher with caching and circuit breaking. + corridor: Optional specific corridor to simulate disruption of. + + Returns: + Dict with scenarios, current health, source, and timestamp. + """ + now = datetime.now(timezone.utc) + + from . import infrastructure + + health_data = await infrastructure.fetch_cable_health(fetcher) + corridors_health = health_data.get("corridors", {}) + + scenarios: list[dict] = [] + + if corridor: + # Simulate specific corridor disruption + result = simulate_cascade([corridor], current_health=corridors_health) + scenarios.append({ + "scenario": f"Disruption of {corridor}", + "corridors": [corridor], + **result, + }) + else: + # Simulate each at_risk or disrupted corridor + at_risk_corridors = [ + name + for name, info in corridors_health.items() + if info.get("status_score", 0) >= 2 + ] + + if at_risk_corridors: + # Individual scenarios + for c in at_risk_corridors: + result = simulate_cascade([c], current_health=corridors_health) + scenarios.append({ + "scenario": f"Disruption of {c}", + "corridors": [c], + **result, + }) + + # Combined worst-case scenario + if len(at_risk_corridors) >= 2: + result = simulate_cascade(at_risk_corridors, current_health=corridors_health) + scenarios.append({ + "scenario": "Combined disruption (worst case)", + "corridors": at_risk_corridors, + **result, + }) + else: + # No at-risk corridors; simulate red_sea as a common scenario + result = simulate_cascade(["red_sea"], current_health=corridors_health) + scenarios.append({ + "scenario": "Hypothetical: Red Sea corridor disruption", + "corridors": ["red_sea"], + **result, + }) + + return { + "scenarios": scenarios, + "scenario_count": len(scenarios), + "current_health": { + name: { + "status_score": info.get("status_score"), + "status_label": info.get("status_label"), + } + for name, info in corridors_health.items() + }, + "source": "cascade-analysis", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + } diff --git a/src/world_intel_mcp/sources/markets.py b/src/world_intel_mcp/sources/markets.py index 914f16f..92dc2f5 100644 --- a/src/world_intel_mcp/sources/markets.py +++ b/src/world_intel_mcp/sources/markets.py @@ -44,6 +44,17 @@ _SECTOR_ETFS: dict[str, str] = { "XLB": "Materials", } +_COMMODITY_SYMBOLS: dict[str, str] = { + "GC=F": "Gold", + "SI=F": "Silver", + "CL=F": "Crude Oil WTI", + "BZ=F": "Brent Crude", + "NG=F": "Natural Gas", + "ZC=F": "Corn", + "ZW=F": "Wheat", + "ZS=F": "Soybeans", +} + _FEAR_GREED_URL = "https://api.alternative.me/fng/?limit=1" _MEMPOOL_FEES_URL = "https://mempool.space/api/v1/fees/recommended" @@ -233,6 +244,42 @@ async def fetch_etf_flows(fetcher: Fetcher) -> dict: } +async def fetch_commodity_quotes(fetcher: Fetcher) -> dict: + """Fetch commodity futures quotes from Yahoo Finance. + + Covers gold, silver, crude oil (WTI & Brent), natural gas, corn, + wheat, and soybeans. Reuses ``_fetch_yahoo_quote`` for parallel + fetching with built-in caching. + + Returns:: + + {"commodities": [{symbol, name, price, change_pct}], ...} + """ + tasks = [ + _fetch_yahoo_quote(fetcher, sym, f"markets:commodity:{sym}", 300) + for sym in _COMMODITY_SYMBOLS + ] + results = await asyncio.gather(*tasks) + + commodities: list[dict] = [] + for sym, quote in zip(_COMMODITY_SYMBOLS, results): + if quote is None: + continue + commodities.append({ + "symbol": sym, + "name": _COMMODITY_SYMBOLS[sym], + "price": quote["price"], + "change_pct": quote["change_pct"], + }) + + return { + "commodities": commodities, + "count": len(commodities), + "source": "yahoo-finance", + "timestamp": _utc_now_iso(), + } + + async def fetch_sector_heatmap(fetcher: Fetcher) -> dict: """Fetch sector ETF performance for a market heatmap.