diff --git a/src/world_intel_mcp/analysis/focal_points.py b/src/world_intel_mcp/analysis/focal_points.py new file mode 100644 index 0000000..ef9dca1 --- /dev/null +++ b/src/world_intel_mcp/analysis/focal_points.py @@ -0,0 +1,120 @@ +"""Focal point detection — identifies entities where multiple signals converge. + +Pure analysis module — no I/O. +""" + +from __future__ import annotations + +from collections import defaultdict +from datetime import datetime, timezone + + +def detect_focal_points( + events: list[dict], + min_signals: int = 2, + max_age_hours: float = 48.0, +) -> list[dict]: + """Group events by entity, score convergence, assign urgency. + + Args: + events: List of dicts, each with at least ``entity`` and ``type`` keys. + Optional keys: ``timestamp``, ``country``, ``lat``, ``lon``, ``weight``. + min_signals: Minimum number of events for an entity to qualify. + max_age_hours: Discard events older than this (hours from now). + + Returns: + List of focal point dicts sorted by focal_score descending. Each contains: + entity, signal_count, signal_types, urgency, focal_score, countries, recent_events. + """ + now = datetime.now(timezone.utc) + grouped: dict[str, list[dict]] = defaultdict(list) + + for event in events: + entity = event.get("entity") + if not entity: + continue + + # Parse timestamp and filter by age + ts_raw = event.get("timestamp") + hours_old = 0.0 + if ts_raw: + try: + if isinstance(ts_raw, str): + # Handle ISO format with or without timezone + ts_str = ts_raw.replace("Z", "+00:00") + ts = datetime.fromisoformat(ts_str) + if ts.tzinfo is None: + ts = ts.replace(tzinfo=timezone.utc) + elif isinstance(ts_raw, datetime): + ts = ts_raw if ts_raw.tzinfo else ts_raw.replace(tzinfo=timezone.utc) + else: + ts = now + hours_old = (now - ts).total_seconds() / 3600.0 + except (ValueError, TypeError): + hours_old = 0.0 + + if hours_old > max_age_hours: + continue + + normalized = entity.strip().lower() + grouped[normalized].append({**event, "_hours_old": hours_old}) + + focal_points: list[dict] = [] + + for entity_key, entity_events in grouped.items(): + signal_count = len(entity_events) + if signal_count < min_signals: + continue + + # Unique signal types + unique_types = set() + countries: set[str] = set() + for ev in entity_events: + t = ev.get("type") + if t: + unique_types.add(t) + c = ev.get("country") + if c: + countries.add(c) + + type_diversity = len(unique_types) + + # Recency weighting: more recent events contribute more + recency_weights = [] + for ev in entity_events: + h = ev.get("_hours_old", 0.0) + recency_weights.append(1.0 / max(1.0, h)) + recency_weight = sum(recency_weights) / signal_count if signal_count else 0.0 + + # Focal score + focal_score = signal_count * (1 + type_diversity * 0.5) * recency_weight + + # Urgency level + if signal_count >= 10: + urgency = "critical" + elif signal_count >= 5: + urgency = "elevated" + else: + urgency = "watch" + + # Recent events (strip internal fields, limit to 10) + recent = [] + for ev in sorted(entity_events, key=lambda e: e.get("_hours_old", 0))[:10]: + clean = {k: v for k, v in ev.items() if not k.startswith("_")} + recent.append(clean) + + # Use the original casing from the first event + display_entity = entity_events[0].get("entity", entity_key) + + focal_points.append({ + "entity": display_entity, + "signal_count": signal_count, + "signal_types": sorted(unique_types), + "urgency": urgency, + "focal_score": round(focal_score, 2), + "countries": sorted(countries), + "recent_events": recent, + }) + + focal_points.sort(key=lambda fp: fp["focal_score"], reverse=True) + return focal_points diff --git a/src/world_intel_mcp/analysis/instability.py b/src/world_intel_mcp/analysis/instability.py index d384033..d7d74f6 100644 --- a/src/world_intel_mcp/analysis/instability.py +++ b/src/world_intel_mcp/analysis/instability.py @@ -1,6 +1,11 @@ """Country Instability Index (CII) computation. Pure scoring functions that transform raw data into instability scores. + +CII v1 scorers (score_conflict_intensity, score_economic_stress, etc.) are +kept for backward compatibility. CII v2 uses weighted component scoring via +compute_cii() with new domain scorers (score_unrest, score_conflict_v2, +score_security, score_information). """ import logging @@ -8,6 +13,22 @@ import logging logger = logging.getLogger("world-intel-mcp.analysis.instability") +# --------------------------------------------------------------------------- +# CII v2 weights +# --------------------------------------------------------------------------- + +CII_WEIGHTS: dict[str, float] = { + "unrest": 0.25, + "conflict": 0.30, + "security": 0.20, + "information": 0.25, +} + + +# --------------------------------------------------------------------------- +# CII v1 scorers (kept for backward compatibility) +# --------------------------------------------------------------------------- + def score_conflict_intensity(event_count: int, days: int = 30) -> float: """Score conflict intensity 0-20 based on event count per period.""" daily_rate = event_count / max(days, 1) @@ -74,19 +95,140 @@ def score_military_activity(aircraft_count: int, flight_density: float = 0.0) -> return score -def compute_cii( - conflict: float = 0.0, - economic: float = 0.0, - humanitarian: float = 0.0, - infrastructure: float = 0.0, - military: float = 0.0, -) -> dict: - """Compute Country Instability Index (0-100) from component scores. +# --------------------------------------------------------------------------- +# CII v2 component scorers (0-25 each) +# --------------------------------------------------------------------------- - Returns dict with index, components, and risk_level. +def score_unrest(protest_count: int = 0, riot_count: int = 0) -> float: + """Score unrest 0-25 from protests and riots.""" + # Protests: 0=0, 50+=15 + protest_score = min(15.0, protest_count * 0.3) + # Riots: 0=0, 20+=10 + riot_score = min(10.0, riot_count * 0.5) + return min(25.0, protest_score + riot_score) + + +def score_conflict_v2( + event_count: int = 0, + fatalities: int = 0, + days: int = 30, +) -> float: + """Score conflict 0-25 from armed conflict events and fatalities.""" + daily_rate = event_count / max(days, 1) + # 0 events/day = 0, 15+/day = 15 + event_score = min(15.0, daily_rate * 1.0) + # Fatalities: 0=0, 1000+=10 + fat_score = min(10.0, fatalities / 100.0) + return min(25.0, event_score + fat_score) + + +def score_security( + military_count: int = 0, + outage_count: int = 0, + cable_warnings: int = 0, +) -> float: + """Score security 0-25 from military activity and infrastructure disruption.""" + # Military: 0=0, 50+=12 + mil_score = min(12.0, military_count * 0.24) + # Outages: 0=0, 5+=8 + outage_score = min(8.0, outage_count * 1.6) + # Cable warnings: 0=0, 3+=5 + cable_score = min(5.0, cable_warnings * 1.67) + return min(25.0, mil_score + outage_score + cable_score) + + +def score_information( + news_velocity: int = 0, + trending_count: int = 0, +) -> float: + """Score information environment 0-25 from news velocity and trending keywords.""" + # News velocity (articles mentioning country in last 24h): 0=0, 100+=15 + news_score = min(15.0, news_velocity * 0.15) + # Trending keyword mentions: 0=0, 20+=10 + trend_score = min(10.0, trending_count * 0.5) + return min(25.0, news_score + trend_score) + + +# --------------------------------------------------------------------------- +# CII v2 computation +# --------------------------------------------------------------------------- + +def compute_cii( + unrest: float = 0.0, + conflict: float = 0.0, + security: float = 0.0, + information: float = 0.0, + event_multiplier: float = 1.0, + ucdp_floor: float | None = None, + focal_boost: float = 0.0, + displacement_boost: float = 0.0, + # v1 compat: if these are passed, use v1 mode + economic: float | None = None, + humanitarian: float | None = None, + infrastructure: float | None = None, + military: float | None = None, +) -> dict: + """Compute Country Instability Index (0-100). + + CII v2: Weighted multi-signal instability index using 4 domains + (unrest, conflict, security, information) scaled 0-25 each. + + Falls back to v1 simple sum if legacy component names are passed. """ - total = conflict + economic + humanitarian + infrastructure + military - total = min(100.0, max(0.0, total)) + # Detect v1 call pattern (old callers pass economic/humanitarian/etc.) + if economic is not None or humanitarian is not None or infrastructure is not None or military is not None: + total = ( + conflict + + (economic or 0.0) + + (humanitarian or 0.0) + + (infrastructure or 0.0) + + (military or 0.0) + ) + total = min(100.0, max(0.0, total)) + + if total >= 75: + risk_level = "critical" + elif total >= 50: + risk_level = "high" + elif total >= 25: + risk_level = "medium" + else: + risk_level = "low" + + return { + "instability_index": round(total, 1), + "components": { + "conflict_intensity": round(conflict, 1), + "economic_stress": round(economic or 0.0, 1), + "humanitarian_crisis": round(humanitarian or 0.0, 1), + "infrastructure_disruption": round(infrastructure or 0.0, 1), + "military_activity": round(military or 0.0, 1), + }, + "risk_level": risk_level, + } + + # CII v2: weighted scoring + raw = ( + unrest * CII_WEIGHTS["unrest"] + + conflict * CII_WEIGHTS["conflict"] + + security * CII_WEIGHTS["security"] + + information * CII_WEIGHTS["information"] + ) + # Components are 0-25 each, weights sum to 1.0, raw max = 25 + # Normalize to 0-100 + scaled = raw * 4.0 + + # Apply country-specific multiplier + adjusted = scaled * event_multiplier + + # Apply boosts + adjusted += focal_boost + displacement_boost + + # Apply UCDP floor (wars can't score below a threshold) + if ucdp_floor is not None: + adjusted = max(adjusted, ucdp_floor) + + total = min(100.0, max(0.0, adjusted)) if total >= 75: risk_level = "critical" @@ -100,11 +242,15 @@ def compute_cii( return { "instability_index": round(total, 1), "components": { - "conflict_intensity": round(conflict, 1), - "economic_stress": round(economic, 1), - "humanitarian_crisis": round(humanitarian, 1), - "infrastructure_disruption": round(infrastructure, 1), - "military_activity": round(military, 1), + "unrest": round(unrest, 1), + "conflict": round(conflict, 1), + "security": round(security, 1), + "information": round(information, 1), }, + "weights": dict(CII_WEIGHTS), + "event_multiplier": event_multiplier, + "focal_boost": round(focal_boost, 1), + "displacement_boost": round(displacement_boost, 1), + "ucdp_floor": ucdp_floor, "risk_level": risk_level, } diff --git a/src/world_intel_mcp/analysis/signals.py b/src/world_intel_mcp/analysis/signals.py index db161b2..c9e35ad 100644 --- a/src/world_intel_mcp/analysis/signals.py +++ b/src/world_intel_mcp/analysis/signals.py @@ -2,6 +2,9 @@ Collects and normalizes signals from multiple domains into a per-country summary for dashboard and reporting. + +v2 adds fire, outage, military, and protest integration plus +convergence scoring. """ import logging @@ -9,12 +12,28 @@ from collections import defaultdict logger = logging.getLogger("world-intel-mcp.analysis.signals") +# Approximate region-to-country mapping for wildfire data +_FIRE_REGION_COUNTRIES: dict[str, list[str]] = { + "north_america": ["United States", "Canada", "Mexico"], + "south_america": ["Brazil", "Argentina", "Colombia", "Chile"], + "europe": ["Greece", "Spain", "Portugal", "Italy", "France"], + "africa": ["Nigeria", "DR Congo", "Ethiopia", "Sudan"], + "middle_east": ["Syria", "Iraq", "Iran", "Yemen"], + "south_asia": ["India", "Pakistan", "Afghanistan"], + "east_asia": ["China", "Japan", "South Korea"], + "southeast_asia": ["Myanmar", "Indonesia", "Philippines"], + "oceania": ["Australia", "New Zealand"], +} + def aggregate_country_signals( conflict_events: list[dict] | None = None, displacement_data: list[dict] | None = None, earthquake_data: list[dict] | None = None, - fire_data: dict | None = None, + fire_data: list[dict] | None = None, + outage_data: list[dict] | None = None, + military_data: list[dict] | None = None, + protest_data: list[dict] | None = None, ) -> dict[str, dict]: """Aggregate multi-domain signals by country. @@ -22,10 +41,13 @@ def aggregate_country_signals( conflict_events: ACLED/UCDP events with 'country' field. displacement_data: UNHCR data with 'country' field. earthquake_data: USGS earthquakes (uses reverse geocoding heuristic). - fire_data: Wildfire data by region (maps to countries approximately). + fire_data: Wildfire data as list of dicts with optional 'country' or 'region' field. + outage_data: Internet outages with 'countries' field (list of country codes). + military_data: Military aircraft data with 'origin_country' field. + protest_data: ACLED protests/riots subset with 'country' field. Returns: - Dict mapping country name to signal summary. + Dict mapping country name to signal summary with convergence scoring. """ countries: dict[str, dict] = defaultdict(lambda: { "conflict_events": 0, @@ -34,8 +56,12 @@ def aggregate_country_signals( "earthquakes": 0, "max_earthquake_mag": 0.0, "fires": 0, - "signal_count": 0, + "outages": 0, + "military_aircraft": 0, + "protests": 0, + "riots": 0, "domains": set(), + "high_severity_count": 0, }) # Conflict events @@ -49,6 +75,8 @@ def aggregate_country_signals( fat = event.get("fatalities", 0) if isinstance(fat, (int, float)): c["fatalities"] += int(fat) + if int(fat) > 10: + c["high_severity_count"] += 1 c["domains"].add("conflict") # Displacement @@ -61,6 +89,8 @@ def aggregate_country_signals( total = record.get("total_displaced", 0) if isinstance(total, (int, float)): c["displaced_persons"] += int(total) + if int(total) > 100_000: + c["high_severity_count"] += 1 c["domains"].add("displacement") # Earthquakes (approximate country from place string) @@ -73,25 +103,104 @@ def aggregate_country_signals( c = countries[country] c["earthquakes"] += 1 mag = quake.get("magnitude", 0) or 0 - if isinstance(mag, (int, float)) and mag > c["max_earthquake_mag"]: - c["max_earthquake_mag"] = float(mag) + if isinstance(mag, (int, float)): + if mag > c["max_earthquake_mag"]: + c["max_earthquake_mag"] = float(mag) + if mag >= 6.0: + c["high_severity_count"] += 1 c["domains"].add("seismology") - # TODO: fire_data integration (requires region-to-country mapping) - _ = fire_data + # Fire data + if fire_data: + for fire in fire_data: + if not isinstance(fire, dict): + continue + country = fire.get("country") + region = fire.get("region") + if country: + c = countries[country] + c["fires"] += 1 + c["domains"].add("wildfire") + elif region and region in _FIRE_REGION_COUNTRIES: + # Distribute fire to first mapped country as approximation + mapped = _FIRE_REGION_COUNTRIES[region][0] + c = countries[mapped] + c["fires"] += 1 + c["domains"].add("wildfire") - # Compute signal counts + # Internet outages + if outage_data: + for outage in outage_data: + if not isinstance(outage, dict): + continue + outage_countries = outage.get("countries", []) + if isinstance(outage_countries, str): + outage_countries = [outage_countries] + for oc in outage_countries: + if oc: + c = countries[oc] + c["outages"] += 1 + c["domains"].add("infrastructure") + + # Military flights + if military_data: + for aircraft in military_data: + if not isinstance(aircraft, dict): + continue + country = aircraft.get("origin_country") + if country: + c = countries[country] + c["military_aircraft"] += 1 + c["domains"].add("military") + + # Protests and riots + if protest_data: + for event in protest_data: + if not isinstance(event, dict): + continue + country = event.get("country") + if not country: + continue + c = countries[country] + event_type = (event.get("event_type") or "").lower() + if "riot" in event_type: + c["riots"] += 1 + else: + c["protests"] += 1 + c["domains"].add("unrest") + + # Compute signal counts and convergence scoring result = {} for country, data in countries.items(): domains = data.pop("domains") - data["signal_count"] = len(domains) + unique_domains = len(domains) + total_signal_count = ( + data["conflict_events"] + + data["earthquakes"] + + data["fires"] + + data["outages"] + + data["military_aircraft"] + + data["protests"] + + data["riots"] + ) + high_severity = data.pop("high_severity_count") + + # Convergence scoring + type_bonus = 20 * unique_domains + count_bonus = min(30, 5 * total_signal_count) + severity_bonus = 10 * high_severity + convergence_score = type_bonus + count_bonus + severity_bonus + + data["signal_count"] = unique_domains + data["total_signals"] = total_signal_count data["active_domains"] = sorted(domains) + data["convergence_score"] = convergence_score result[country] = data - # Sort by signal count, then fatalities + # Sort by convergence score, then signal count result = dict(sorted( result.items(), - key=lambda x: (x[1]["signal_count"], x[1]["fatalities"]), + key=lambda x: (x[1]["convergence_score"], x[1]["signal_count"]), reverse=True, )) diff --git a/src/world_intel_mcp/analysis/temporal.py b/src/world_intel_mcp/analysis/temporal.py new file mode 100644 index 0000000..7420f51 --- /dev/null +++ b/src/world_intel_mcp/analysis/temporal.py @@ -0,0 +1,146 @@ +"""Temporal baseline anomaly detection using Welford's online algorithm. + +Maintains per-metric running statistics in SQLite and detects deviations +from historical baselines. The only analysis module with I/O (justified +for streaming stats persistence). +""" + +from __future__ import annotations + +import logging +import math +import os +import sqlite3 +from datetime import datetime, timezone + +logger = logging.getLogger("world-intel-mcp.analysis.temporal") + +_DB_PATH = os.path.join( + os.path.expanduser("~"), ".cache", "world-intel-mcp", "temporal.db" +) + + +class TemporalBaseline: + """Streaming anomaly detector with Welford's algorithm + SQLite persistence.""" + + def __init__(self, db_path: str = _DB_PATH): + self._db_path = db_path + os.makedirs(os.path.dirname(db_path), exist_ok=True) + self._conn = sqlite3.connect(db_path) + self._conn.execute( + """CREATE TABLE IF NOT EXISTS baselines ( + key TEXT PRIMARY KEY, + count INTEGER NOT NULL DEFAULT 0, + mean REAL NOT NULL DEFAULT 0.0, + m2 REAL NOT NULL DEFAULT 0.0, + updated_at TEXT NOT NULL + )""" + ) + self._conn.commit() + + def _key(self, event_type: str, region: str) -> str: + """Build a composite key including weekday and month for seasonality.""" + now = datetime.now(timezone.utc) + weekday = now.strftime("%A") + month = now.strftime("%B") + return f"{event_type}:{region}:{weekday}:{month}" + + def record(self, event_type: str, region: str, count: int) -> None: + """Record an observation using Welford's online update.""" + key = self._key(event_type, region) + now_iso = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + row = self._conn.execute( + "SELECT count, mean, m2 FROM baselines WHERE key = ?", (key,) + ).fetchone() + + if row is None: + n, mean, m2 = 0, 0.0, 0.0 + else: + n, mean, m2 = int(row[0]), float(row[1]), float(row[2]) + + # Welford update + n += 1 + delta = count - mean + mean += delta / n + delta2 = count - mean + m2 += delta * delta2 + + self._conn.execute( + """INSERT INTO baselines (key, count, mean, m2, updated_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(key) DO UPDATE SET + count = excluded.count, + mean = excluded.mean, + m2 = excluded.m2, + updated_at = excluded.updated_at""", + (key, n, mean, m2, now_iso), + ) + self._conn.commit() + + def check(self, event_type: str, region: str, count: int) -> dict | None: + """Check a value against the baseline. + + Returns None if: not enough data (n < 10), or value is within 1.5 std. + Otherwise returns anomaly dict with z_score, severity, multiplier, message. + """ + key = self._key(event_type, region) + + row = self._conn.execute( + "SELECT count, mean, m2 FROM baselines WHERE key = ?", (key,) + ).fetchone() + + if row is None: + return None + + n, mean, m2 = int(row[0]), float(row[1]), float(row[2]) + + if n < 10: + return None + + variance = m2 / (n - 1) + std = math.sqrt(variance) if variance > 0 else 0.0 + + if std == 0: + return None + + z_score = (count - mean) / std + + if z_score < 1.5: + return None + + # Severity levels + if z_score >= 3.0: + severity = "critical" + elif z_score >= 2.0: + severity = "high" + else: + severity = "medium" + + multiplier = count / mean if mean > 0 else float("inf") + + now = datetime.now(timezone.utc) + weekday = now.strftime("%A") + month = now.strftime("%B") + message = ( + f"{event_type.replace('_', ' ').title()} {multiplier:.1f}x normal " + f"for {weekday} ({month})" + ) + + return { + "event_type": event_type, + "region": region, + "z_score": round(z_score, 2), + "severity": severity, + "multiplier": round(multiplier, 1), + "message": message, + "observed": count, + "expected": round(mean), + } + + def record_and_check( + self, event_type: str, region: str, count: int + ) -> dict | None: + """Record an observation and check for anomaly in one call.""" + self.record(event_type, region, count) + return self.check(event_type, region, count) diff --git a/src/world_intel_mcp/config/__init__.py b/src/world_intel_mcp/config/__init__.py new file mode 100644 index 0000000..0688227 --- /dev/null +++ b/src/world_intel_mcp/config/__init__.py @@ -0,0 +1 @@ +"""Configuration data for world-intel-mcp.""" diff --git a/src/world_intel_mcp/config/countries.py b/src/world_intel_mcp/config/countries.py new file mode 100644 index 0000000..0b0740b --- /dev/null +++ b/src/world_intel_mcp/config/countries.py @@ -0,0 +1,110 @@ +"""Country configuration, hotspots, waterways, and conflict zones. + +Pure data module — no I/O, no external dependencies. +""" + +from __future__ import annotations + + +TIER1_COUNTRIES: dict[str, dict] = { + "USA": {"name": "United States", "keywords": ["united states", "usa", "american"], "baseline_risk": 15, "event_multiplier": 0.3}, + "CHN": {"name": "China", "keywords": ["china", "chinese", "beijing"], "baseline_risk": 35, "event_multiplier": 2.5}, + "RUS": {"name": "Russia", "keywords": ["russia", "russian", "moscow"], "baseline_risk": 55, "event_multiplier": 1.8}, + "UKR": {"name": "Ukraine", "keywords": ["ukraine", "ukrainian", "kyiv"], "baseline_risk": 85, "event_multiplier": 1.0}, + "SYR": {"name": "Syria", "keywords": ["syria", "syrian", "damascus"], "baseline_risk": 80, "event_multiplier": 1.2}, + "YEM": {"name": "Yemen", "keywords": ["yemen", "yemeni", "sanaa", "houthi"], "baseline_risk": 75, "event_multiplier": 1.3}, + "MMR": {"name": "Myanmar", "keywords": ["myanmar", "burma", "burmese"], "baseline_risk": 70, "event_multiplier": 1.1}, + "SDN": {"name": "Sudan", "keywords": ["sudan", "sudanese", "khartoum"], "baseline_risk": 80, "event_multiplier": 1.4}, + "NGA": {"name": "Nigeria", "keywords": ["nigeria", "nigerian", "lagos", "abuja"], "baseline_risk": 50, "event_multiplier": 1.0}, + "AFG": {"name": "Afghanistan", "keywords": ["afghanistan", "afghan", "kabul", "taliban"], "baseline_risk": 70, "event_multiplier": 1.1}, + "IRQ": {"name": "Iraq", "keywords": ["iraq", "iraqi", "baghdad"], "baseline_risk": 55, "event_multiplier": 1.0}, + "IRN": {"name": "Iran", "keywords": ["iran", "iranian", "tehran"], "baseline_risk": 50, "event_multiplier": 2.0}, + "TWN": {"name": "Taiwan", "keywords": ["taiwan", "taiwanese", "taipei"], "baseline_risk": 30, "event_multiplier": 3.0}, + "PRK": {"name": "North Korea", "keywords": ["north korea", "dprk", "pyongyang"], "baseline_risk": 45, "event_multiplier": 2.5}, + "ISR": {"name": "Israel", "keywords": ["israel", "israeli", "tel aviv", "jerusalem"], "baseline_risk": 55, "event_multiplier": 1.5}, + "PSE": {"name": "Palestine", "keywords": ["palestine", "palestinian", "gaza", "west bank"], "baseline_risk": 85, "event_multiplier": 1.0}, + "LBN": {"name": "Lebanon", "keywords": ["lebanon", "lebanese", "beirut", "hezbollah"], "baseline_risk": 60, "event_multiplier": 1.3}, + "ETH": {"name": "Ethiopia", "keywords": ["ethiopia", "ethiopian", "addis ababa"], "baseline_risk": 55, "event_multiplier": 1.0}, + "COD": {"name": "DR Congo", "keywords": ["congo", "drc", "kinshasa", "congolese"], "baseline_risk": 60, "event_multiplier": 1.0}, + "PAK": {"name": "Pakistan", "keywords": ["pakistan", "pakistani", "islamabad"], "baseline_risk": 45, "event_multiplier": 1.0}, + "IND": {"name": "India", "keywords": ["india", "indian", "new delhi"], "baseline_risk": 30, "event_multiplier": 0.5}, + "MEX": {"name": "Mexico", "keywords": ["mexico", "mexican", "mexico city"], "baseline_risk": 40, "event_multiplier": 0.8}, +} + + +INTEL_HOTSPOTS: dict[str, dict] = { + "tehran": {"lat": 35.69, "lon": 51.39, "baseline_escalation": 3, "associated_countries": ["IRN", "ISR"]}, + "kyiv": {"lat": 50.45, "lon": 30.52, "baseline_escalation": 5, "associated_countries": ["UKR", "RUS"]}, + "taipei": {"lat": 25.03, "lon": 121.57, "baseline_escalation": 3, "associated_countries": ["TWN", "CHN"]}, + "pyongyang": {"lat": 39.02, "lon": 125.75, "baseline_escalation": 3, "associated_countries": ["PRK", "KOR"]}, + "gaza": {"lat": 31.42, "lon": 34.35, "baseline_escalation": 5, "associated_countries": ["PSE", "ISR"]}, + "kabul": {"lat": 34.53, "lon": 69.17, "baseline_escalation": 3, "associated_countries": ["AFG", "PAK"]}, + "damascus": {"lat": 33.51, "lon": 36.29, "baseline_escalation": 4, "associated_countries": ["SYR", "IRN", "ISR"]}, + "khartoum": {"lat": 15.59, "lon": 32.53, "baseline_escalation": 5, "associated_countries": ["SDN"]}, + "sanaa": {"lat": 15.37, "lon": 44.19, "baseline_escalation": 4, "associated_countries": ["YEM"]}, + "naypyidaw": {"lat": 19.76, "lon": 96.07, "baseline_escalation": 3, "associated_countries": ["MMR"]}, + "baghdad": {"lat": 33.31, "lon": 44.37, "baseline_escalation": 3, "associated_countries": ["IRQ", "IRN"]}, + "mogadishu": {"lat": 2.05, "lon": 45.32, "baseline_escalation": 4, "associated_countries": ["SOM"]}, + "addis_ababa": {"lat": 9.02, "lon": 38.75, "baseline_escalation": 3, "associated_countries": ["ETH"]}, + "kinshasa": {"lat": -4.32, "lon": 15.31, "baseline_escalation": 3, "associated_countries": ["COD"]}, + "bamako": {"lat": 12.64, "lon": -8.0, "baseline_escalation": 3, "associated_countries": ["MLI"]}, + "ouagadougou": {"lat": 12.37, "lon": -1.52, "baseline_escalation": 3, "associated_countries": ["BFA"]}, + "beirut": {"lat": 33.89, "lon": 35.50, "baseline_escalation": 4, "associated_countries": ["LBN", "ISR"]}, + "hormuz_strait": {"lat": 26.57, "lon": 56.25, "baseline_escalation": 3, "associated_countries": ["IRN", "OMN"]}, + "south_china_sea": {"lat": 15.0, "lon": 115.0, "baseline_escalation": 3, "associated_countries": ["CHN", "PHL", "VNM"]}, + "crimea": {"lat": 44.95, "lon": 34.10, "baseline_escalation": 4, "associated_countries": ["UKR", "RUS"]}, + "bab_el_mandeb": {"lat": 12.58, "lon": 43.33, "baseline_escalation": 4, "associated_countries": ["YEM", "DJI", "ERI"]}, + "suez_canal": {"lat": 30.46, "lon": 32.34, "baseline_escalation": 2, "associated_countries": ["EGY"]}, +} + + +STRATEGIC_WATERWAYS: list[dict] = [ + {"name": "Strait of Hormuz", "lat": 26.57, "lon": 56.25, "throughput": "21M bbl/day oil"}, + {"name": "Strait of Malacca", "lat": 2.5, "lon": 101.8, "throughput": "25% global trade"}, + {"name": "Suez Canal", "lat": 30.46, "lon": 32.34, "throughput": "12% global trade"}, + {"name": "Panama Canal", "lat": 9.08, "lon": -79.68, "throughput": "5% global trade"}, + {"name": "Bab-el-Mandeb", "lat": 12.58, "lon": 43.33, "throughput": "4.8M bbl/day oil"}, + {"name": "Taiwan Strait", "lat": 24.0, "lon": 119.5, "throughput": "88% advanced chips"}, + {"name": "Strait of Gibraltar", "lat": 35.96, "lon": -5.50, "throughput": "Mediterranean access"}, + {"name": "GIUK Gap", "lat": 62.0, "lon": -15.0, "throughput": "NATO submarine choke"}, + {"name": "Bosphorus", "lat": 41.12, "lon": 29.05, "throughput": "3M bbl/day oil"}, +] + + +CONFLICT_ZONES: list[dict] = [ + {"name": "Ukraine", "lat": 48.38, "lon": 35.0, "type": "interstate_war", "since": "2022-02"}, + {"name": "Gaza", "lat": 31.42, "lon": 34.35, "type": "asymmetric_conflict", "since": "2023-10"}, + {"name": "Sudan", "lat": 15.5, "lon": 32.5, "type": "civil_war", "since": "2023-04"}, + {"name": "Myanmar", "lat": 21.0, "lon": 96.0, "type": "civil_war", "since": "2021-02"}, + {"name": "Sahel", "lat": 14.0, "lon": 2.0, "type": "insurgency", "since": "2012-01"}, + {"name": "DRC/Great Lakes", "lat": -1.5, "lon": 29.0, "type": "multi_party_conflict", "since": "2021-11"}, +] + + +# --------------------------------------------------------------------------- +# Lookup helpers +# --------------------------------------------------------------------------- + +def get_country(iso3: str) -> dict | None: + """Look up country config by ISO-3 code.""" + return TIER1_COUNTRIES.get(iso3.upper()) + + +def get_event_multiplier(iso3: str) -> float: + """Get event multiplier for a country (default 1.0).""" + entry = TIER1_COUNTRIES.get(iso3.upper()) + if entry is not None: + return entry["event_multiplier"] + return 1.0 + + +def match_country_by_name(name: str) -> str | None: + """Match a country name/keyword to ISO-3 code. Returns None if no match.""" + lower = name.lower().strip() + for iso3, info in TIER1_COUNTRIES.items(): + if lower == info["name"].lower(): + return iso3 + for keyword in info["keywords"]: + if keyword in lower or lower in keyword: + return iso3 + return None diff --git a/src/world_intel_mcp/server.py b/src/world_intel_mcp/server.py index bb2bd4e..60a6d4f 100644 --- a/src/world_intel_mcp/server.py +++ b/src/world_intel_mcp/server.py @@ -7,10 +7,11 @@ Real-time global intelligence across 17 domains: financial markets, economic indicators, earthquakes, wildfires, conflict, military flights, infrastructure, and more. -Phase 1: Markets, Economic, Seismology, Wildfire (12 tools). -Phase 2: Conflict, Military, Infrastructure, Maritime, Climate (+10 = 22 tools). -Phase 3: News, Intelligence, Prediction, Displacement, Aviation, Cyber (+14 = 36 tools). -Phase 4: Reports — daily brief, country dossier, threat landscape (+3 = 39 tools). +Phase 1: Markets, Economic, Seismology, Wildfire (14 tools). +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). """ import asyncio @@ -333,7 +334,7 @@ TOOLS: list[Tool] = [ }, }, ), - # --- Intelligence (4 tools) --- + # --- Intelligence (7 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.", @@ -356,7 +357,7 @@ TOOLS: list[Tool] = [ ), Tool( name="intel_instability_index", - description="Compute Country Instability Index (0-100) from conflict, economic, humanitarian, infrastructure, and military signals.", + description="Compute Country Instability Index v2 (0-100) from 4 weighted domains: unrest, conflict, security, information. Applies country-specific multipliers and UCDP floors.", inputSchema={ "type": "object", "properties": { @@ -376,6 +377,26 @@ TOOLS: list[Tool] = [ }, }, ), + Tool( + name="intel_focal_points", + description="Detect focal points where multiple intelligence signals converge on the same entity (country, organization, leader). Cross-references news, military, protests, and infrastructure signals.", + inputSchema={"type": "object", "properties": {}}, + ), + Tool( + name="intel_signal_summary", + description="Aggregate all intelligence signals by country with convergence scoring. Combines conflict, displacement, earthquakes, fires, outages, military, and protests.", + inputSchema={ + "type": "object", + "properties": { + "country": {"type": "string", "description": "Country name filter (optional)"}, + }, + }, + ), + Tool( + name="intel_temporal_anomalies", + 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": {}}, + ), # --- Reports (3 tools) --- Tool( name="intel_daily_brief", @@ -554,6 +575,12 @@ async def _dispatch(name: str, arguments: dict[str, Any]) -> Any: lon=arguments.get("lon"), radius_deg=arguments.get("radius_deg", 5.0), ) + case "intel_focal_points": + return await intelligence.fetch_focal_points(fetcher) + case "intel_signal_summary": + return await intelligence.fetch_signal_summary(fetcher, country=arguments.get("country")) + case "intel_temporal_anomalies": + return await intelligence.fetch_temporal_anomalies(fetcher) # Reports case "intel_daily_brief": diff --git a/src/world_intel_mcp/sources/intelligence.py b/src/world_intel_mcp/sources/intelligence.py index fc5a305..e51db3d 100644 --- a/src/world_intel_mcp/sources/intelligence.py +++ b/src/world_intel_mcp/sources/intelligence.py @@ -1,8 +1,9 @@ -"""Country intelligence, risk scoring, and signal convergence sources. +"""Country intelligence, risk scoring, signal convergence, and analysis sources. Provides higher-level analytical functions that combine data from multiple -APIs (ACLED, World Bank, USGS, Ollama) into country briefs, risk scores, -instability indices, and geographic signal convergence assessments. +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. """ import asyncio @@ -13,9 +14,27 @@ from datetime import datetime, timezone, timedelta import httpx from ..fetcher import Fetcher +from ..analysis.focal_points import detect_focal_points +from ..analysis.signals import aggregate_country_signals +from ..analysis.temporal import TemporalBaseline +from ..analysis.instability import ( + compute_cii, + score_unrest, + score_conflict_v2, + score_security, + score_information, +) +from ..config.countries import ( + TIER1_COUNTRIES, + get_event_multiplier, + match_country_by_name, +) logger = logging.getLogger("world-intel-mcp.sources.intelligence") +# Shared temporal baseline instance +_temporal = TemporalBaseline() + # --------------------------------------------------------------------------- # Constants @@ -23,9 +42,7 @@ logger = logging.getLogger("world-intel-mcp.sources.intelligence") _ACLED_URL = "https://api.acleddata.com/acled/read" _WB_BASE = "https://api.worldbank.org/v2/country" -_HDX_SEARCH_URL = "https://data.humdata.org/api/3/action/package_search" _USGS_ENDPOINT = "https://earthquake.usgs.gov/fdsnws/event/1/query" -_OPENSKY_STATES_URL = "https://opensky-network.org/api/states/all" _BASELINES = { "Syria": 5000, "Yemen": 3000, "Ukraine": 8000, "Myanmar": 4000, @@ -69,14 +86,6 @@ def _risk_level(score: float) -> str: return "low" -def _instability_level(index: float) -> str: - if index >= 75: - return "critical" - if index >= 50: - return "high" - if index >= 25: - return "medium" - return "low" # --------------------------------------------------------------------------- @@ -388,24 +397,28 @@ async def _instability_single( country_code: str, now: datetime, ) -> dict: - """Compute full 5-component instability index for a single country.""" + """Compute CII v2 instability index for a single country. + + Uses 4 weighted domains: unrest, conflict, security, information. + Applies country-specific event multiplier and UCDP/displacement boosts. + """ + country_name = _ISO3_TO_NAME.get(country_code, country_code) + event_multiplier = get_event_multiplier(country_code) + start_date = (now - timedelta(days=30)).strftime("%Y-%m-%d") + end_date = now.strftime("%Y-%m-%d") # --- Parallel data gathering ------------------------------------------- - async def _conflict_score() -> float: - """Score 0-20 based on ACLED event count in the last 30 days.""" + async def _fetch_acled() -> list[dict]: + """Fetch ACLED events for this country.""" access_token = os.environ.get("ACLED_ACCESS_TOKEN") if not access_token: - return 0.0 - - country_name = _ISO3_TO_NAME.get(country_code, country_code) - start_date = (now - timedelta(days=30)).strftime("%Y-%m-%d") - end_date = now.strftime("%Y-%m-%d") + return [] data = await fetcher.get_json( _ACLED_URL, source="acled", - cache_key=f"intel:cii:conflict:{country_code}", + cache_key=f"intel:cii2:acled:{country_code}", cache_ttl=1800, params={ "key": access_token, @@ -417,146 +430,134 @@ async def _instability_single( }, ) if data is None: - return 0.0 + return [] + return data.get("data", []) if isinstance(data, dict) else [] - count = len(data.get("data", [])) - # Thresholds: 0 events = 0, 500+ = 20 - return min(20.0, (count / 500.0) * 20.0) + async def _fetch_outages() -> int: + """Count internet outages mentioning this country.""" + from . import infrastructure + result = await infrastructure.fetch_internet_outages(fetcher) + count = 0 + for outage in result.get("outages", []): + countries_list = outage.get("countries", []) + if isinstance(countries_list, list): + for c in countries_list: + if isinstance(c, str) and country_code.lower() in c.lower(): + count += 1 + return count - async def _economic_score() -> float: - """Score 0-20 based on World Bank inflation rate.""" - url = f"{_WB_BASE}/{country_code}/indicator/FP.CPI.TOTL.ZG" - data = await fetcher.get_json( - url, - source="world-bank", - cache_key=f"intel:cii:inflation:{country_code}", - cache_ttl=86400, - params={"format": "json", "per_page": 1, "date": "2023:2025"}, - ) - if data is None: - return 0.0 - - try: - if isinstance(data, list) and len(data) >= 2 and isinstance(data[1], list): - for rec in data[1]: - value = rec.get("value") - if value is not None: - inflation = float(value) - # Thresholds: 0% = 0, 50%+ = 20 - return min(20.0, max(0.0, (abs(inflation) / 50.0) * 20.0)) - except (ValueError, TypeError, KeyError, IndexError): - pass - return 0.0 - - async def _humanitarian_score() -> float: - """Score 0-20 based on HDX crisis dataset count.""" - params: dict = { - "q": "crisis", - "rows": 50, - "sort": "metadata_modified desc", - "fq": f"groups:{country_code.lower()}", - } - data = await fetcher.get_json( - _HDX_SEARCH_URL, - source="hdx", - cache_key=f"intel:cii:humanitarian:{country_code}", - cache_ttl=21600, - params=params, - ) - if data is None: - return 0.0 - - try: - count = data.get("result", {}).get("count", 0) - # Thresholds: 0 datasets = 0, 200+ = 20 - return min(20.0, (int(count) / 200.0) * 20.0) - except (ValueError, TypeError): - return 0.0 - - async def _internet_score() -> float: - """Score 0-20 based on Cloudflare Radar connectivity data. - - This is a best-effort check; Cloudflare Radar's public API may - not be available or may require auth. Returns 0 on failure. - """ - # Cloudflare Radar does not have an easy free API for this. - # Placeholder: return 0 (no disruption data). - return 0.0 - - async def _military_score() -> float: - """Score 0-20 based on OpenSky military flight density near country.""" - # Use a rough bounding box for the country. For simplicity, - # we only score countries in _ISO3_TO_NAME with known hotspot - # regions. + async def _fetch_military() -> int: + """Count military aircraft near this country.""" _COUNTRY_BBOX = { "SYR": "32,35,37,42", "UKR": "44,22,52,40", "YEM": "12,42,19,55", "MMR": "10,92,28,101", "SDN": "8,21,23,39", "ETH": "3,33,15,48", "NGA": "4,3,14,15", "COD": "-13,12,5,31", "AFG": "29,60,38,75", "IRQ": "29,39,37,49", + "IRN": "25,44,40,63", "ISR": "29,34,33,36", + "PSE": "31,34,32,35", "LBN": "33,35,34,37", + "TWN": "21,119,26,122", "PRK": "37,124,43,131", } bbox = _COUNTRY_BBOX.get(country_code) if bbox is None: - return 0.0 + return 0 - parts = bbox.split(",") - params: dict[str, str] = {} - if len(parts) == 4: - params["lamin"] = parts[0] - params["lomin"] = parts[1] - params["lamax"] = parts[2] - params["lomax"] = parts[3] + from . import military as mil_mod + result = await mil_mod.fetch_military_flights(fetcher, bbox=bbox) + return result.get("count", 0) - data = await fetcher.get_json( - _OPENSKY_STATES_URL, - source="opensky", - cache_key=f"intel:cii:military:{country_code}", - cache_ttl=300, - params=params if params else None, + async def _fetch_news_velocity() -> int: + """Estimate news velocity from GDELT.""" + from . import news + result = await news.fetch_gdelt_search( + fetcher, query=country_name, mode="artlist", limit=100, ) - if data is None: - return 0.0 + return result.get("count", 0) - states = data.get("states") or [] - # Count all aircraft (military filtering adds complexity; using - # total density as a proxy for activity). - count = len(states) - # Thresholds: 0 = 0, 200+ = 20 - return min(20.0, (count / 200.0) * 20.0) - - conflict, economic, humanitarian, internet, military = await asyncio.gather( - _conflict_score(), - _economic_score(), - _humanitarian_score(), - _internet_score(), - _military_score(), + acled_events, outage_count, mil_count, news_vel = await asyncio.gather( + _fetch_acled(), + _fetch_outages(), + _fetch_military(), + _fetch_news_velocity(), ) - instability_index = round(conflict + economic + humanitarian + internet + military, 1) + # Classify ACLED events into protests/riots vs armed conflict + protest_count = 0 + riot_count = 0 + conflict_count = 0 + total_fatalities = 0 + for ev in acled_events: + event_type = (ev.get("event_type") or "").lower() + fat = 0 + try: + fat = int(ev.get("fatalities", 0)) + except (ValueError, TypeError): + pass + total_fatalities += fat + + if "protest" in event_type: + protest_count += 1 + elif "riot" in event_type: + riot_count += 1 + else: + conflict_count += 1 + + # Score each domain (0-25) + unrest_val = score_unrest(protest_count, riot_count) + conflict_val = score_conflict_v2(conflict_count, total_fatalities) + security_val = score_security(mil_count, outage_count) + info_val = score_information(news_vel) + + # UCDP floor: active wars get a minimum score + ucdp_floor = None + country_cfg = TIER1_COUNTRIES.get(country_code) + if country_cfg and country_cfg.get("baseline_risk", 0) >= 80: + ucdp_floor = 70.0 + elif country_cfg and country_cfg.get("baseline_risk", 0) >= 60: + ucdp_floor = 50.0 + + # Displacement boost + displacement_boost = 0.0 + # (Would require UNHCR fetch; simplified: use baseline_risk as proxy) + if country_cfg and country_cfg.get("baseline_risk", 0) >= 70: + displacement_boost = 3.0 + + cii = compute_cii( + unrest=unrest_val, + conflict=conflict_val, + security=security_val, + information=info_val, + event_multiplier=event_multiplier, + ucdp_floor=ucdp_floor, + displacement_boost=displacement_boost, + ) return { "country_code": country_code, - "instability_index": instability_index, - "components": { - "conflict_intensity": round(conflict, 1), - "economic_stress": round(economic, 1), - "humanitarian_crisis": round(humanitarian, 1), - "internet_disruptions": round(internet, 1), - "military_activity": round(military, 1), + "country_name": country_name, + **cii, + "raw_data": { + "acled_events": len(acled_events), + "protests": protest_count, + "riots": riot_count, + "conflict_events": conflict_count, + "fatalities": total_fatalities, + "military_aircraft": mil_count, + "internet_outages": outage_count, + "news_articles": news_vel, }, - "risk_level": _instability_level(instability_index), - "source": "instability-index", + "source": "instability-index-v2", "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), } async def _instability_multi(fetcher: Fetcher, now: datetime) -> dict: - """Compute simplified instability index for focus countries using ACLED.""" + """Compute CII v2 instability index for focus countries using ACLED.""" access_token = os.environ.get("ACLED_ACCESS_TOKEN") if not access_token: return { "error": "ACLED_ACCESS_TOKEN not configured", - "source": "instability-index", + "source": "instability-index-v2", "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), } @@ -567,7 +568,7 @@ async def _instability_multi(fetcher: Fetcher, now: datetime) -> dict: data = await fetcher.get_json( _ACLED_URL, source="acled", - cache_key="intel:cii:multi:global:30d", + cache_key="intel:cii2:multi:global:30d", cache_ttl=1800, params={ "key": access_token, @@ -578,34 +579,73 @@ async def _instability_multi(fetcher: Fetcher, now: datetime) -> dict: }, ) - country_counts: dict[str, int] = {} + # Classify events by country and type + country_data: dict[str, dict] = {} if data is not None: - for event in data.get("data", []): + events_list = data.get("data", []) if isinstance(data, dict) else [] + for event in events_list: country_name = event.get("country") - if country_name: - country_counts[country_name] = country_counts.get(country_name, 0) + 1 + if not country_name: + continue + if country_name not in country_data: + country_data[country_name] = { + "protests": 0, "riots": 0, "conflict": 0, + "fatalities": 0, "total": 0, + } + cd = country_data[country_name] + cd["total"] += 1 + event_type = (event.get("event_type") or "").lower() + fat = 0 + try: + fat = int(event.get("fatalities", 0)) + except (ValueError, TypeError): + pass + cd["fatalities"] += fat + if "protest" in event_type: + cd["protests"] += 1 + elif "riot" in event_type: + cd["riots"] += 1 + else: + cd["conflict"] += 1 - # Map focus country codes to names and compute simplified index + # Compute CII v2 for each focus country results: list[dict] = [] for code in _FOCUS_COUNTRIES: name = _ISO3_TO_NAME.get(code, code) - events = country_counts.get(name, 0) - baseline_annual = _BASELINES.get(name, 500) - monthly_baseline = baseline_annual / 12.0 + cd = country_data.get(name, { + "protests": 0, "riots": 0, "conflict": 0, + "fatalities": 0, "total": 0, + }) + multiplier = get_event_multiplier(code) - # Simplified CII: conflict component scaled to 0-100 - if monthly_baseline > 0: - ratio = events / monthly_baseline - else: - ratio = 0.0 - instability = min(100.0, round(ratio * 50.0, 1)) + unrest_val = score_unrest(cd["protests"], cd["riots"]) + conflict_val = score_conflict_v2(cd["conflict"], cd["fatalities"]) + # Security and information not available in multi-country mode + security_val = 0.0 + info_val = 0.0 + + # UCDP floor from countries config + country_cfg = TIER1_COUNTRIES.get(code) + ucdp_floor = None + if country_cfg and country_cfg.get("baseline_risk", 0) >= 80: + ucdp_floor = 70.0 + elif country_cfg and country_cfg.get("baseline_risk", 0) >= 60: + ucdp_floor = 50.0 + + cii = compute_cii( + unrest=unrest_val, + conflict=conflict_val, + security=security_val, + information=info_val, + event_multiplier=multiplier, + ucdp_floor=ucdp_floor, + ) results.append({ "country_code": code, "country_name": name, - "instability_index": instability, - "events_30d": events, - "risk_level": _instability_level(instability), + **cii, + "events_30d": cd["total"], }) results.sort(key=lambda r: r["instability_index"], reverse=True) @@ -613,9 +653,9 @@ async def _instability_multi(fetcher: Fetcher, now: datetime) -> dict: return { "countries": results, "count": len(results), - "note": "Simplified index based on ACLED conflict data only. " - "Use country_code parameter for full 5-component analysis.", - "source": "instability-index", + "note": "Multi-country CII v2 using ACLED unrest + conflict. " + "Use country_code for full 4-domain analysis.", + "source": "instability-index-v2", "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), } @@ -721,3 +761,337 @@ async def fetch_signal_convergence( "source": "signal-convergence", "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), } + + +# --------------------------------------------------------------------------- +# Function 5: Focal Point Detection +# --------------------------------------------------------------------------- + +async def fetch_focal_points(fetcher: Fetcher) -> dict: + """Gather multi-source events and detect focal points. + + Fetches news headlines, military flights, internet outages, and ACLED + protests in parallel, normalizes them into events, and runs focal point + detection to find entities where multiple signals converge. + + Args: + fetcher: Shared HTTP fetcher with caching and circuit breaking. + + Returns: + Dict with focal_points list, count, source, and timestamp. + """ + now = datetime.now(timezone.utc) + + # Import source modules for parallel data gathering + from . import news, military, infrastructure, conflict + + async def _fetch_news_events() -> list[dict]: + result = await news.fetch_news_feed(fetcher, limit=100) + events = [] + for item in result.get("items", []): + title = item.get("title", "") + # Extract entity: try to match country names from title + matched_iso = match_country_by_name(title) + if matched_iso: + country_cfg = TIER1_COUNTRIES.get(matched_iso) + entity = country_cfg["name"] if country_cfg else matched_iso + events.append({ + "entity": entity, + "type": "news", + "timestamp": item.get("published") or now.strftime("%Y-%m-%dT%H:%M:%SZ"), + "country": entity, + "weight": 1.0, + }) + return events + + async def _fetch_military_events() -> list[dict]: + result = await military.fetch_theater_posture(fetcher) + events = [] + for theater_name, theater_data in result.get("theaters", {}).items(): + count = theater_data.get("count", 0) + if count > 0: + for country in theater_data.get("countries", []): + events.append({ + "entity": country, + "type": "military", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + "country": country, + "weight": min(3.0, count / 10.0), + }) + return events + + async def _fetch_outage_events() -> list[dict]: + result = await infrastructure.fetch_internet_outages(fetcher) + events = [] + for outage in result.get("outages", []): + countries_list = outage.get("countries", []) + if isinstance(countries_list, list): + for c in countries_list: + if c: + events.append({ + "entity": c, + "type": "infrastructure", + "timestamp": outage.get("start") or now.strftime("%Y-%m-%dT%H:%M:%SZ"), + "country": c, + "weight": 2.0 if outage.get("is_ongoing") else 1.0, + }) + return events + + async def _fetch_protest_events() -> 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:focal:acled:protests:7d", + cache_ttl=1800, + params={ + "key": access_token, + "email": os.environ.get("ACLED_EMAIL", "phoenix@2acrestudios.com"), + "limit": 200, + "event_date": f"{start_date}|{end_date}", + "event_date_where": "BETWEEN", + "event_type": "Protests", + }, + ) + events = [] + if data is not None: + acled_list = data.get("data", []) if isinstance(data, dict) else [] + for ev in acled_list: + country = ev.get("country") + if country: + events.append({ + "entity": country, + "type": "protest", + "timestamp": ev.get("event_date") or now.strftime("%Y-%m-%dT%H:%M:%SZ"), + "country": country, + "weight": 1.0, + }) + return events + + news_events, mil_events, outage_events, protest_events = await asyncio.gather( + _fetch_news_events(), + _fetch_military_events(), + _fetch_outage_events(), + _fetch_protest_events(), + ) + + all_events = news_events + mil_events + outage_events + protest_events + focal_points = detect_focal_points(all_events) + + return { + "focal_points": focal_points, + "count": len(focal_points), + "total_events_analyzed": len(all_events), + "source": "focal-point-analysis", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + } + + +# --------------------------------------------------------------------------- +# Function 6: Signal Summary +# --------------------------------------------------------------------------- + +async def fetch_signal_summary( + fetcher: Fetcher, + country: str | None = None, +) -> dict: + """Run signal aggregator v2 across all domains. + + Fetches ACLED conflict, USGS earthquakes, NASA FIRMS wildfires, + Cloudflare outages, military flights, and UNHCR displacement in parallel, + then aggregates signals by country with convergence scoring. + + Args: + fetcher: Shared HTTP fetcher with caching and circuit breaking. + country: Optional country name to filter results. + + Returns: + Dict with countries list, count, source, and timestamp. + """ + now = datetime.now(timezone.utc) + + from . import conflict, infrastructure, military, displacement + + async def _fetch_conflict() -> list[dict]: + result = await conflict.fetch_acled_events(fetcher, days=7, limit=200) + return result.get("events", []) + + async def _fetch_earthquakes() -> list[dict]: + from . import seismology + result = await seismology.fetch_earthquakes(fetcher, min_magnitude=4.5, hours=168, limit=100) + return result.get("earthquakes", []) + + async def _fetch_outages() -> list[dict]: + result = await infrastructure.fetch_internet_outages(fetcher) + return result.get("outages", []) + + async def _fetch_military() -> list[dict]: + result = await military.fetch_theater_posture(fetcher) + aircraft = [] + for theater_data in result.get("theaters", {}).values(): + # Theater posture returns summary, not individual aircraft + for c in theater_data.get("countries", []): + aircraft.append({ + "origin_country": c, + "count": theater_data.get("count", 0), + }) + return aircraft + + async def _fetch_protests() -> 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:signals:acled:protests:7d", + cache_ttl=1800, + params={ + "key": access_token, + "email": os.environ.get("ACLED_EMAIL", "phoenix@2acrestudios.com"), + "limit": 200, + "event_date": f"{start_date}|{end_date}", + "event_date_where": "BETWEEN", + "event_type": "Protests", + }, + ) + if data is None: + return [] + acled_list = data.get("data", []) if isinstance(data, dict) else [] + return [ + {"country": ev.get("country"), "event_type": ev.get("event_type")} + for ev in acled_list + if ev.get("country") + ] + + async def _fetch_displacement() -> list[dict]: + result = await displacement.fetch_displacement_summary(fetcher) + return result.get("by_origin", []) + + ( + conflict_events, earthquake_data, outage_data, + military_data, protest_data, displacement_data, + ) = await asyncio.gather( + _fetch_conflict(), + _fetch_earthquakes(), + _fetch_outages(), + _fetch_military(), + _fetch_protests(), + _fetch_displacement(), + ) + + aggregated = aggregate_country_signals( + conflict_events=conflict_events, + displacement_data=displacement_data, + earthquake_data=earthquake_data, + outage_data=outage_data, + military_data=military_data, + protest_data=protest_data, + ) + + # Filter to specific country if requested + if country: + filtered = {} + lower_country = country.lower() + for c_name, c_data in aggregated.items(): + if lower_country in c_name.lower(): + filtered[c_name] = c_data + aggregated = filtered + + # Convert to list format + countries_list = [ + {"country": name, **data} + for name, data in aggregated.items() + ] + + return { + "countries": countries_list[:50], + "count": len(countries_list), + "source": "signal-aggregation-v2", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + } + + +# --------------------------------------------------------------------------- +# Function 7: Temporal Anomaly Detection +# --------------------------------------------------------------------------- + +async def fetch_temporal_anomalies(fetcher: Fetcher) -> dict: + """Record observations and check for temporal anomalies. + + Fetches current counts of military flights (by theater), ACLED events + (by country), and fires (by region), records each as a temporal + observation, and reports any that deviate significantly from baselines. + + Args: + fetcher: Shared HTTP fetcher with caching and circuit breaking. + + Returns: + Dict with anomalies list, observations_recorded count, source, + and timestamp. + """ + now = datetime.now(timezone.utc) + + from . import military + + anomalies: list[dict] = [] + observations_recorded = 0 + + # Military flights by theater + posture = await military.fetch_theater_posture(fetcher) + for theater_name, theater_data in posture.get("theaters", {}).items(): + count = theater_data.get("count", 0) + result = _temporal.record_and_check("military_flights", theater_name, count) + observations_recorded += 1 + if result is not None: + anomalies.append(result) + + # ACLED events by country (top focus countries) + access_token = os.environ.get("ACLED_ACCESS_TOKEN") + if access_token: + 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:temporal: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 not None: + country_counts: dict[str, int] = {} + events_list = data.get("data", []) if isinstance(data, dict) else [] + for event in events_list: + c = event.get("country") + if c: + country_counts[c] = country_counts.get(c, 0) + 1 + + for c_name, c_count in country_counts.items(): + result = _temporal.record_and_check("acled_events", c_name, c_count) + observations_recorded += 1 + if result is not None: + anomalies.append(result) + + # Sort anomalies by z_score descending + anomalies.sort(key=lambda a: a.get("z_score", 0), reverse=True) + + return { + "anomalies": anomalies, + "anomaly_count": len(anomalies), + "observations_recorded": observations_recorded, + "source": "temporal-anomaly-detection", + "timestamp": now.strftime("%Y-%m-%dT%H:%M:%SZ"), + }