diff --git a/engine/osiris_intake.py b/engine/osiris_intake.py index 5ccf46e..e9feecc 100644 --- a/engine/osiris_intake.py +++ b/engine/osiris_intake.py @@ -9,6 +9,7 @@ import asyncio import logging import re +from datetime import datetime, timezone import httpx @@ -103,28 +104,81 @@ def _text(d: dict, *keys: str) -> str: def _coord(d: dict): lat = d.get("lat") or d.get("latitude") lng = d.get("lng") or d.get("lon") or d.get("longitude") - coords = d.get("coords") or (d.get("geometry") or {}).get("coordinates") - if (lat is None or lng is None) and isinstance(coords, (list, tuple)) and len(coords) >= 2: - lng, lat = coords[0], coords[1] # GeoJSON [lng, lat] + if lat is None or lng is None: + geo = (d.get("geometry") or {}).get("coordinates") + if isinstance(geo, (list, tuple)) and len(geo) >= 2: + lng, lat = geo[0], geo[1] # GeoJSON is [lng, lat] + if lat is None or lng is None: + coords = d.get("coords") + if isinstance(coords, (list, tuple)) and len(coords) >= 2: + lat, lng = coords[0], coords[1] # Osiris `coords` is [lat, lng] (see news route) try: - return (float(lat), float(lng)) if lat is not None and lng is not None else (None, None) + lat, lng = float(lat), float(lng) except (TypeError, ValueError): return (None, None) + if abs(lat) > 90 or abs(lng) > 180: + return (None, None) + return (lat, lng) + + +# A feed's risk scale CANNOT be inferred from the value: 8 means 8/10 from the +# news route but 8/100 from country-risk, and guessing wrong INVERTS the ranking +# — which is the whole bug this map exists to kill. So each producer declares its +# own scale, keyed by source. An unlisted source is IGNORED rather than guessed: +# dropping a signal is recoverable, inverting one is not. +# news -> scoreRisk() in osiris news/route.ts: int, starts at 1, +2 per risk +# keyword, capped at 10. +# NOTE: country-risk ("risk", a 0-100 scale) is deliberately absent — it has a +# dedicated handler (_country_risk_events) that never reaches _salience. Listing +# it here would be dead config that reads as live. It was the feed that exposed +# the guessing bug: 8 means 8/10 from news but 8/100 from country-risk. +RISK_SCALE = { + "news": 10.0, +} -def _salience(title: str, summary: str, raw: dict) -> float: +def _salience(title: str, summary: str, raw: dict, source: str = "") -> float: text = f"{title} {summary}".lower() score = 0.4 for word, weight in _HOT.items(): if word in text: score = max(score, weight) - # honor an upstream risk score if Osiris provided one + # Honor an upstream risk score ONLY when we know that source's scale. rs = raw.get("risk_score") or raw.get("severity_score") - if isinstance(rs, (int, float)): - score = max(score, min(1.0, float(rs) / (100.0 if rs > 1 else 1.0))) + scale = RISK_SCALE.get(source) + if scale and isinstance(rs, (int, float)) and not isinstance(rs, bool) and rs > 0: + score = max(score, min(1.0, float(rs) / scale)) return round(min(1.0, score), 2) +def _event_ts(d: dict) -> int | None: + """The item's own event time in epoch ms, if the feed carries one. + Without this every event is stamped at fetch time, so week-old posts + surface as breaking and `since` filtering is meaningless.""" + for k in ("published", "pubDate", "pub_date", "date_added", "date", "datetime", "updated", "time"): + v = d.get(k) + if v is None or v == "": + continue + if isinstance(v, bool): + continue + if isinstance(v, (int, float)): # epoch seconds or ms + v = float(v) + if v > 1e12: + return int(v) + if v > 1e9: + return int(v * 1000) + continue + if isinstance(v, str): + try: + dt = datetime.fromisoformat(v.strip().replace("Z", "+00:00")) + except ValueError: + continue + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return int(dt.timestamp() * 1000) + return None + + def _to_event(d: dict, source: str, category: str) -> WorldEvent | None: title = _text(d, "title", "name", "headline", "question", "html", "place") mag = d.get("magnitude") or d.get("mag") @@ -145,7 +199,7 @@ def _to_event(d: dict, source: str, category: str) -> WorldEvent | None: except (TypeError, ValueError): pass lat, lng = _coord(d) - return WorldEvent( + evt = WorldEvent( title=title[:240], summary=summary[:2000], category=(category if source == "polymarket" else (_text(d, "category") or category)), @@ -153,9 +207,13 @@ def _to_event(d: dict, source: str, category: str) -> WorldEvent | None: lat=lat, lng=lng, url=_text(d, "url", "link", "feed_url"), - salience=_salience(title, summary, d), + salience=_salience(title, summary, d, source), raw={k: d[k] for k in list(d)[:25]}, ) + ts = _event_ts(d) + if ts: + evt.ts = ts + return evt def _markets_events(data: dict, source: str) -> list[WorldEvent]: @@ -303,6 +361,57 @@ def _ioda_events(data: dict) -> list[WorldEvent]: return out +# Osiris country-risk emits {code, risk_score, risk_level, tags} — no name, no +# coords — so the generic path found no title and DROPPED all 20 silently. Map +# the ISO-2 codes it actually serves to a name and an approximate centroid so +# they become real, locatable events. Codes are from RISK_FACTORS in +# osiris/src/app/api/country-risk/route.ts; extend both together. +_COUNTRY = { + "UA": ("Ukraine", 49.0, 32.0), "RU": ("Russia", 61.5, 105.3), + "IL": ("Israel", 31.0, 34.9), "PS": ("Palestinian Territories", 31.9, 35.2), + "SY": ("Syria", 34.8, 39.0), "YE": ("Yemen", 15.6, 48.5), + "MM": ("Myanmar", 21.9, 95.9), "SD": ("Sudan", 12.9, 30.2), + "AF": ("Afghanistan", 33.9, 67.7), "KP": ("North Korea", 40.3, 127.5), + "IR": ("Iran", 32.4, 53.7), "CN": ("China", 35.9, 104.2), + "TW": ("Taiwan", 23.7, 121.0), "VE": ("Venezuela", 6.4, -66.6), + "HT": ("Haiti", 18.9, -72.3), "LB": ("Lebanon", 33.9, 35.9), + "PK": ("Pakistan", 30.4, 69.3), "SO": ("Somalia", 5.2, 46.2), + "LY": ("Libya", 26.3, 17.2), "ET": ("Ethiopia", 9.1, 40.5), +} + + +def _country_risk_events(data: dict) -> list[WorldEvent]: + """Osiris country-risk → locatable events. + + This is a HAND-MAINTAINED geopolitical baseline (osiris marks it + `static: true`), not a live measurement — so salience is capped well below + breaking-news territory and derived from the score alone. It is background + context for the oracle, not a signal that something just happened. + """ + out = [] + for c in (data or {}).get("countries", []) or []: + if not isinstance(c, dict): + continue + code = str(c.get("code") or "").upper() + score = c.get("risk_score") + if code not in _COUNTRY or not isinstance(score, (int, float)): + continue + name, lat, lng = _COUNTRY[code] + tags = ", ".join(t.replace("_", " ") for t in (c.get("tags") or [])) + level = c.get("risk_level", "") + out.append(WorldEvent( + title=f"Country risk: {name} {score}/100 ({level})"[:120], + summary=(f"Standing geopolitical baseline — {tags}." if tags else + "Standing geopolitical baseline.") + " Static reference, not a live event.", + category="instability", source="risk", + lat=lat, lng=lng, + # Ceiling 0.5: a static table must never outrank live signal. A + # 90/100 baseline is context; a missile landing is news. + salience=round(min(0.5, float(score) / 200.0), 2), + )) + return out + + def _space_weather_events(data: dict) -> list[WorldEvent]: """NOAA SWPC — geomagnetic storms + solar flares (satellites, GPS, grids).""" out = [] @@ -700,6 +809,8 @@ async def _fetch_feed(self, c: httpx.AsyncClient, path: str, source: str, catego data = r.json() if source in ("markets", "crypto"): out.extend(_markets_events(data, source)) + elif source == "risk": + out.extend(_country_risk_events(data)) elif source == "futures": out.extend(_futures_events(data)) elif source == "gdacs":