Files
nexus-quant-terminal/backend/app/routers/news.py
T
Kansaram 61e145a442 feat: launch NEXUS TERMINAL — Bloomberg-style FX options analytics platform
Full-stack forex options analytics terminal with Bloomberg-inspired UI.

Backend (FastAPI + Python):
- Garman-Kohlhagen options pricing engine with full Greeks
- Goldman Sachs gs-quant AI signals (RSI, MACD, Bollinger, Hurst, OU)
- Monte Carlo GBM simulation and volatility surface generation
- CFTC COT institutional positioning + Forex Factory economic calendar
- Live data proxy: OpenSky aircraft + USGS earthquakes (CORS-safe)
- Multi-leg strategy library (straddle, iron condor, butterfly, spreads)

Frontend (React 18 + Vite):
- NEXUS animated orbital logo (3-ring SVG) + canvas favicon animation
- Bloomberg terminal design: JetBrains Mono, color-mix() tokens
- 11 dashboard tabs: Greeks, Chart, AI Signals, 3D Surfaces, Breakeven,
  Scenarios, Monte Carlo, Institutional, Calendar, Live Map, Live Feeds
- Live World Map (react-leaflet): aircraft, earthquakes, weather radar
- Live Feeds: CoinGecko crypto top-12 + Windy.com global webcams
- Economic calendar with filters + institutional flow (CFTC COT)
- Animated landing page + session-based routing
- Fully responsive dark-only terminal design system

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-06-07 18:19:23 +05:30

111 lines
3.5 KiB
Python

"""
news.py — Free economic calendar from Forex Factory.
Source: nfs.faireconomy.media (official FF data mirror, JSON format)
ff_calendar_thisweek.json — current week
(nextweek.json appears Fri/Sat only, gracefully skipped if 404)
No API key. No auth.
"""
import httpx
from datetime import datetime, timezone
from fastapi import APIRouter
router = APIRouter(prefix="/news", tags=["news"])
FF_JSON_URLS = [
"https://nfs.faireconomy.media/ff_calendar_thisweek.json",
"https://nfs.faireconomy.media/ff_calendar_nextweek.json",
]
# In-memory cache — refresh every hour
_cache: dict = {"data": None, "ts": 0.0}
_CACHE_TTL = 3600 # seconds
def _fetch_ff_json(url: str) -> list[dict]:
try:
resp = httpx.get(
url, timeout=15, follow_redirects=True,
headers={"User-Agent": "Mozilla/5.0 (compatible; QuantRiskFX/3.0)"}
)
if resp.status_code != 200:
return []
raw = resp.json()
events = []
for ev in (raw if isinstance(raw, list) else []):
# `date` is already ISO-8601 with TZ offset, e.g. "2026-06-07T08:30:00-04:00"
date_str = ev.get("date", "")
dt_utc = None
if date_str:
try:
dt_et = datetime.fromisoformat(date_str)
dt_utc = dt_et.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%MZ")
except ValueError:
pass
events.append({
"title": ev.get("title", "").strip(),
"country": ev.get("country", "").strip(),
"date_raw": date_str,
"datetime_utc": dt_utc,
"impact": ev.get("impact", "").strip(),
"forecast": ev.get("forecast", "").strip(),
"previous": ev.get("previous", "").strip(),
"actual": ev.get("actual", "").strip(),
"url": "",
})
return events
except Exception:
return []
def _load_calendar() -> dict:
import time
now = time.time()
if _cache["data"] is not None and (now - _cache["ts"]) < _CACHE_TTL:
return _cache["data"]
seen: set[tuple] = set()
all_events: list = []
for url in FF_JSON_URLS:
for ev in _fetch_ff_json(url):
key = (ev["title"], ev["country"], ev["datetime_utc"])
if key not in seen:
seen.add(key)
all_events.append(ev)
all_events.sort(key=lambda e: e["datetime_utc"] or "")
now_utc = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%MZ")
next_high = next(
(e for e in all_events
if e["impact"] == "High"
and (e["datetime_utc"] or "") >= now_utc
and not e["actual"]),
None,
)
result = {
"events": all_events,
"count": len(all_events),
"now_utc": now_utc,
"next_high": next_high,
"source": "Forex Factory Economic Calendar — nfs.faireconomy.media (free, no API key)",
"note": "Times in UTC (source is US Eastern Time).",
"cached": False,
}
_cache["data"] = result
_cache["ts"] = now
return result
@router.get("/calendar")
def get_calendar():
"""
Economic calendar from Forex Factory (current + next week).
Times returned as UTC ISO-8601 strings. Cached for 1 hour.
"""
result = _load_calendar()
return {**result, "cached": _cache["ts"] > 0}