thinkingEverytime/QuantOracle
1
1#!/usr/bin/env python32"""Build + publish daily geopolitical/news intelligence snapshot.3 4Output (local):5 data/news_intel_latest.json6 7Output (Supabase public bucket, when --upload):8 <prefix>/latest.json9 <prefix>/history/<YYYY-MM-DD>.json10"""11 12# ruff: noqa: E402 (sys.path bootstrap must run before local imports)13 14from __future__ import annotations15 16import argparse17import json18import os19import re20import sys21from dataclasses import dataclass22from datetime import UTC, datetime23from pathlib import Path24from typing import Any25from urllib.parse import quote, urlparse26 27import feedparser28import requests29 30ROOT = Path(__file__).resolve().parents[1]31if str(ROOT) not in sys.path:32 sys.path.insert(0, str(ROOT))33 34from quant.registry import data_root35from scripts.supabase_storage import from_env36 37OFFICIAL_DOMAINS = {38 "rbi.org.in",39 "sebi.gov.in",40 "mca.gov.in",41 "mea.gov.in",42 "petroleum.nic.in",43 "ppac.gov.in",44 "opec.org",45 "iea.org",46 "eia.gov",47 "imf.org",48 "worldbank.org",49 "bis.org",50 "federalreserve.gov",51 "ecb.europa.eu",52}53 54WIRE_DOMAINS = {55 "reuters.com",56 "bloomberg.com",57 "apnews.com",58 "ft.com",59 "wsj.com",60}61 62RSS_QUERIES = [63 "geopolitics oil market sanctions OPEC",64 "India policy RBI SEBI market regulation",65 "oil refinery India diesel gasoline margins",66 "shipping crude tanker Hormuz Red Sea",67]68 69HIGH_RISK_KW = {70 "war",71 "missile",72 "attack",73 "sanction",74 "embargo",75 "blockade",76 "hormuz",77 "red sea",78 "drone strike",79}80 81MED_RISK_KW = {82 "election",83 "tariff",84 "export ban",85 "rate hike",86 "rate cut",87 "inflation",88 "policy",89 "quota",90 "production cut",91}92 93ASSET_RULES = [94 (re.compile(r"\b(brent|wti|crude|oil)\b", re.I), ["BRENT", "WTI"]),95 (re.compile(r"\b(nifty|sensex|equity|stocks?)\b", re.I), ["NIFTY50", "SENSEX"]),96 (re.compile(r"\b(rupee|usd\/inr|usd inr|dollar)\b", re.I), ["USDINR"]),97 (re.compile(r"\b(yield|treasury|bond)\b", re.I), ["US10Y"]),98 (re.compile(r"\b(lng|natural gas|gas prices?)\b", re.I), ["Natural Gas"]),99]100 101SECTOR_RULES = [102 (103 re.compile(r"\b(refinery|refining|gasoline|diesel|petrol|jet fuel)\b", re.I),104 ["Refining"],105 ),106 (107 re.compile(r"\b(upstream|exploration|production)\b", re.I),108 ["Upstream Oil & Gas"],109 ),110 (re.compile(r"\b(petrochemical|petchem|chemical)\b", re.I), ["Petrochemicals"]),111 (re.compile(r"\b(shipping|tanker|freight|container)\b", re.I), ["Shipping"]),112 (re.compile(r"\b(power|utilities|coal|gas-fired)\b", re.I), ["Power Utilities"]),113]114 115REGION_RULES = [116 (re.compile(r"\b(india|nse|rbi|sebi|delhi|mumbai)\b", re.I), ["India"]),117 (118 re.compile(r"\b(saudi|iran|iraq|uae|qatar|gulf|middle east|opec)\b", re.I),119 ["Middle East"],120 ),121 (122 re.compile(r"\b(usa|united states|federal reserve|treasury)\b", re.I),123 ["United States"],124 ),125 (re.compile(r"\b(russia|ukraine|europe|ecb)\b", re.I), ["Europe"]),126 (re.compile(r"\b(china|beijing)\b", re.I), ["China"]),127]128 129 130@dataclass(frozen=True)131class RawItem:132 headline: str133 summary: str134 url: str135 source: str136 datetime: str137 provider: str138 139 140def _clean(s: str) -> str:141 return re.sub(r"\s+", " ", re.sub(r"<[^>]*>", "", s or "")).strip()142 143 144def _domain(url: str) -> str:145 try:146 host = (urlparse(url).hostname or "").lower().strip()147 except Exception:148 return ""149 if host.startswith("www."):150 host = host[4:]151 return host152 153 154def _domain_matches(host: str, allowed: set[str]) -> bool:155 return any(host == d or host.endswith(f".{d}") for d in allowed)156 157 158def _source_tier(url: str) -> str:159 host = _domain(url)160 if _domain_matches(host, OFFICIAL_DOMAINS):161 return "official"162 if _domain_matches(host, WIRE_DOMAINS):163 return "wire"164 if host:165 return "media"166 return "unknown"167 168 169def _parse_ts(raw: str) -> datetime:170 s = (raw or "").strip()171 if not s:172 return datetime.now(UTC)173 if s.endswith("Z"):174 s = s[:-1] + "+00:00"175 if len(s) >= 5 and s[-5] in "+-" and s[-3] != ":":176 s = s[:-2] + ":" + s[-2:]177 try:178 dt = datetime.fromisoformat(s)179 if dt.tzinfo is None:180 return dt.replace(tzinfo=UTC)181 return dt.astimezone(UTC)182 except Exception:183 return datetime.now(UTC)184 185 186def _impact(187 headline: str, summary: str188) -> tuple[dict[str, Any], list[str], dict[str, Any]]:189 text = f"{headline} {summary}".lower()190 191 assets: set[str] = set()192 sectors: set[str] = set()193 regions: set[str] = set()194 195 for pat, vals in ASSET_RULES:196 if pat.search(text):197 assets.update(vals)198 for pat, vals in SECTOR_RULES:199 if pat.search(text):200 sectors.update(vals)201 for pat, vals in REGION_RULES:202 if pat.search(text):203 regions.update(vals)204 205 risk = "low"206 if any(k in text for k in HIGH_RISK_KW):207 risk = "high"208 elif any(k in text for k in MED_RISK_KW):209 risk = "medium"210 211 oil_kw = {212 "oil",213 "crude",214 "brent",215 "wti",216 "refinery",217 "refining",218 "diesel",219 "gasoline",220 "petrochemical",221 "petchem",222 "opec",223 "tanker",224 }225 oil_score = sum(1 for k in oil_kw if k in text)226 oil_relevance = oil_score > 0227 228 tags: set[str] = set()229 if oil_relevance:230 tags.add("oil")231 if "refining" in sectors:232 tags.add("refinery")233 if "shipping" in text or "tanker" in text:234 tags.add("shipping")235 if "sanction" in text:236 tags.add("sanctions")237 if "opec" in text:238 tags.add("opec")239 240 impact = {241 "affected_assets": sorted(assets),242 "affected_sectors": sorted(sectors),243 "affected_regions": sorted(regions),244 "risk_level": risk,245 "summary": ", ".join(246 [247 f"assets: {', '.join(sorted(assets))}" if assets else "",248 f"sectors: {', '.join(sorted(sectors))}" if sectors else "",249 f"regions: {', '.join(sorted(regions))}" if regions else "",250 ]251 ).strip(", ")252 or "broad market sensitivity",253 }254 oil_meta = {"relevant": oil_relevance, "score": oil_score}255 return impact, sorted(tags), oil_meta256 257 258def _from_newsdata(limit: int) -> list[RawItem]:259 key = (os.getenv("NEWSDATA_API_KEY") or "").strip()260 if not key:261 return []262 q = "(geopolitics OR sanctions OR oil OR refinery OR OPEC OR RBI OR SEBI)"263 url = "https://newsdata.io/api/1/latest"264 try:265 r = requests.get(266 url,267 params={"apikey": key, "language": "en", "q": q},268 timeout=12,269 )270 if r.status_code != 200:271 return []272 payload = r.json() or {}273 out: list[RawItem] = []274 for n in (payload.get("results") or [])[:limit]:275 out.append(276 RawItem(277 headline=_clean(str(n.get("title") or "Untitled")),278 summary=_clean(str(n.get("description") or "")),279 url=str(n.get("link") or "").strip(),280 source=str(n.get("source_id") or "NewsData").strip(),281 datetime=str(n.get("pubDate") or "").strip(),282 provider="newsdata",283 )284 )285 return out286 except Exception:287 return []288 289 290def _from_gnews(limit: int) -> list[RawItem]:291 key = (os.getenv("GNEWS_API_KEY") or "").strip()292 if not key:293 return []294 q = "geopolitics OR sanctions OR oil refinery OR OPEC OR RBI"295 try:296 r = requests.get(297 "https://gnews.io/api/v4/search",298 params={"apikey": key, "q": q, "lang": "en", "max": limit},299 timeout=12,300 )301 if r.status_code != 200:302 return []303 data = r.json() or {}304 out: list[RawItem] = []305 for n in (data.get("articles") or [])[:limit]:306 source = (n.get("source") or {}).get("name") or "GNews"307 out.append(308 RawItem(309 headline=_clean(str(n.get("title") or "Untitled")),310 summary=_clean(str(n.get("description") or "")),311 url=str(n.get("url") or "").strip(),312 source=str(source).strip(),313 datetime=str(n.get("publishedAt") or "").strip(),314 provider="gnews",315 )316 )317 return out318 except Exception:319 return []320 321 322def _rss_google_search(query: str, limit: int) -> list[RawItem]:323 url = (324 f"https://news.google.com/rss/search?q={quote(query)}&hl=en-IN&gl=IN&ceid=IN:en"325 )326 try:327 r = requests.get(url, timeout=10)328 if r.status_code != 200:329 return []330 feed = feedparser.parse(r.content)331 out: list[RawItem] = []332 for e in (feed.entries or [])[:limit]:333 source = (334 (getattr(e, "source", None) or {}).get("title")335 if isinstance(getattr(e, "source", None), dict)336 else "Google News RSS"337 )338 out.append(339 RawItem(340 headline=_clean(str(e.get("title", "Untitled"))),341 summary=_clean(str(e.get("summary", ""))),342 url=str(e.get("link", "")).strip(),343 source=_clean(str(source or "Google News RSS")),344 datetime=str(e.get("published", "")).strip(),345 provider="rss",346 )347 )348 return out349 except Exception:350 return []351 352 353def _collect(limit: int) -> tuple[list[RawItem], dict[str, int], list[str]]:354 provider_counts: dict[str, int] = {}355 order: list[str] = []356 357 out: list[RawItem] = []358 359 api_sources = [360 ("newsdata", _from_newsdata),361 ("gnews", _from_gnews),362 ]363 364 for name, fn in api_sources:365 items = fn(limit)366 if name not in order:367 order.append(name)368 provider_counts[name] = provider_counts.get(name, 0) + len(items)369 out.extend(items)370 371 if "rss" not in order:372 order.append("rss")373 for q in RSS_QUERIES:374 items = _rss_google_search(q, max(8, limit // 2))375 provider_counts["rss"] = provider_counts.get("rss", 0) + len(items)376 out.extend(items)377 378 return out, provider_counts, order379 380 381def _normalize(items: list[RawItem], max_items: int) -> list[dict[str, Any]]:382 deduped: dict[str, RawItem] = {}383 for it in items:384 if not it.url or not it.headline:385 continue386 key = f"{_domain(it.url)}|{it.headline.lower().strip()}"387 if key not in deduped:388 deduped[key] = it389 390 ranked = sorted(deduped.values(), key=lambda x: _parse_ts(x.datetime), reverse=True)391 392 out: list[dict[str, Any]] = []393 for it in ranked[:max_items]:394 source_tier = _source_tier(it.url)395 impact, tags, oil_meta = _impact(it.headline, it.summary)396 out.append(397 {398 "headline": it.headline,399 "summary": it.summary,400 "url": it.url,401 "source": it.source,402 "datetime": it.datetime or "Recent",403 "provider": it.provider,404 "source_tier": source_tier,405 "impact": impact,406 "tags": tags,407 "oil_refinery": oil_meta,408 }409 )410 return out411 412 413def main() -> int:414 ap = argparse.ArgumentParser()415 ap.add_argument("--max-items", type=int, default=40)416 ap.add_argument("--upload", action="store_true")417 ap.add_argument("--prefix", default="")418 args = ap.parse_args()419 420 raw, provider_breakdown, provider_order = _collect(limit=max(20, args.max_items))421 items = _normalize(raw, max_items=args.max_items)422 now = datetime.now(UTC)423 now_s = now.isoformat().replace("+00:00", "Z")424 425 official_count = sum(1 for i in items if i.get("source_tier") == "official")426 wire_count = sum(1 for i in items if i.get("source_tier") == "wire")427 oil_refinery_count = sum(428 1 for i in items if bool((i.get("oil_refinery") or {}).get("relevant"))429 )430 431 payload = {432 "as_of_utc": now_s,433 "generated_at_utc": now_s,434 "count": len(items),435 "provider_order": provider_order,436 "provider_breakdown": provider_breakdown,437 "quality": {438 "official_count": official_count,439 "wire_count": wire_count,440 "oil_refinery_count": oil_refinery_count,441 },442 "items": items,443 }444 445 root = data_root()446 root.mkdir(parents=True, exist_ok=True)447 out_path = root / "news_intel_latest.json"448 out_path.write_text(json.dumps(payload, indent=2), encoding="utf-8")449 print(f"Wrote {out_path} ({len(items)} items)")450 451 if not args.upload:452 return 0453 454 sb = from_env(require_write=True)455 if not sb:456 raise SystemExit(457 "Missing SUPABASE_URL/SUPABASE_BUCKET/SUPABASE_SERVICE_ROLE_KEY for upload"458 )459 460 prefix = args.prefix.strip().strip("/") or (461 os.getenv("QUANTORACLE_NEWS_PREFIX") or "news/intel"462 ).strip("/")463 day = now.strftime("%Y-%m-%d")464 blob = out_path.read_bytes()465 466 sb.upload_bytes(467 f"{prefix}/history/{day}.json", blob, content_type="application/json"468 )469 sb.upload_bytes(f"{prefix}/latest.json", blob, content_type="application/json")470 print(f"Uploaded -> {sb.public_url(f'{prefix}/latest.json')}")471 return 0472 473 474if __name__ == "__main__":475 raise SystemExit(main())476