fx-risk-terminal/fx_engine.py

"""
FX data fetching and portfolio analytics — multi-currency.

The set of currencies the dashboard tracks is derived automatically from
positions.json (every non-USD currency that appears). Rates are quoted as
CCY/USD (USD value of one unit of the currency).

Data sources (no API key required):
  - Live / historical rates: frankfurter.app  (ECB reference rates, daily)
  - Fallback for live:       open.er-api.com  (updates ~hourly)
"""

import os
import json
import math
import time
import logging
from datetime import date, timedelta
from pathlib import Path
from typing import Any

import requests

log = logging.getLogger(__name__)

FRANKFURTER = "https://api.frankfurter.app"
OPEN_ER     = "https://open.er-api.com/v6/latest"
SESSION     = requests.Session()
SESSION.headers.update({"User-Agent": "fx-dashboard/2.0"})
TIMEOUT     = 10

POSITIONS_FILE = Path(__file__).parent / "positions.json"
EXAMPLE_FILE   = Path(__file__).parent / "positions.json.example"

# Currencies available from the ECB reference set (frankfurter). Used only to
# warn the user if positions.json references something unsupported.
SUPPORTED = {
    "EUR", "SEK", "GBP", "CHF", "NOK", "DKK", "JPY", "BRL", "MXN", "KRW",
    "AUD", "CAD", "CNY", "CZK", "HKD", "HUF", "ILS", "INR", "NZD", "PLN",
    "RON", "SGD", "THB", "TRY", "ZAR", "BGN", "IDR", "ISK", "MYR", "PHP",
}


# ── In-process cache ──────────────────────────────────────────────────────────

_cache: dict[str, tuple[Any, float]] = {}


def _cached(key: str, ttl: int, fn):
    entry = _cache.get(key)
    if entry and time.monotonic() - entry[1] < ttl:
        return entry[0]
    value = fn()
    _cache[key] = (value, time.monotonic())
    return value


def invalidate(key: str | None = None) -> None:
    _cache.clear() if key is None else _cache.pop(key, None)


# ── Date helpers ──────────────────────────────────────────────────────────────

def _prev_business_day(d: date) -> date:
    d -= timedelta(days=1)
    while d.weekday() >= 5:
        d -= timedelta(days=1)
    return d


def _business_days_ago(n: int) -> date:
    d, count = date.today(), 0
    while count < n:
        d -= timedelta(days=1)
        if d.weekday() < 5:
            count += 1
    return d


# ── Positions & currency universe ─────────────────────────────────────────────

def load_positions() -> list[dict]:
    # FX_DEMO=1 forces the committed example book (used for public screenshots
    # and for anyone cloning the repo) without touching a private positions.json.
    demo = os.environ.get("FX_DEMO") in ("1", "true", "yes")
    path = EXAMPLE_FILE if demo or not POSITIONS_FILE.exists() else POSITIONS_FILE
    if path is EXAMPLE_FILE and not demo:
        log.warning("positions.json not found — loading example (demo) data")
    return json.loads(path.read_text()) if path.exists() else []


def sanitize_positions(raw: list) -> list[dict]:
    """Validate and coerce a user-supplied book (from the browser builder)."""
    clean = []
    if not isinstance(raw, list):
        return clean
    for p in raw[:50]:  # hard cap
        if not isinstance(p, dict):
            continue
        ccy = str(p.get("currency", "")).upper().strip()
        try:
            shares = float(p.get("shares", 0))
            avg = float(p.get("avg_cost", 0))
        except (TypeError, ValueError):
            continue
        if ccy not in SUPPORTED or shares <= 0 or avg <= 0:
            continue
        clean.append({
            "ticker":   str(p.get("ticker", "?"))[:16] or "?",
            "shares":   shares,
            "avg_cost": avg,
            "currency": ccy,
            "exchange": str(p.get("exchange", ""))[:24],
        })
    return clean


def currency_universe(positions: list[dict] | None = None) -> list[str]:
    """Sorted unique non-USD currencies present in the book."""
    positions = load_positions() if positions is None else positions
    ccys = {p["currency"] for p in positions if p.get("currency") != "USD"}
    unsupported = ccys - SUPPORTED
    if unsupported:
        log.warning("unsupported currencies (no free ECB feed): %s", unsupported)
    return sorted(ccys & SUPPORTED)


def _to_param(ccys: list[str]) -> str:
    return ",".join(ccys)


# ── Live rates ────────────────────────────────────────────────────────────────

def _fetch_live_frankfurter(ccys: list[str]) -> dict:
    r = SESSION.get(f"{FRANKFURTER}/latest?from=USD&to={_to_param(ccys)}", timeout=TIMEOUT)
    r.raise_for_status()
    data = r.json()
    return {
        "rates": {c: 1 / data["rates"][c] for c in ccys if c in data["rates"]},
        "date":  data["date"],
        "source": "frankfurter.app",
    }


def _fetch_live_open_er(ccys: list[str]) -> dict:
    r = SESSION.get(f"{OPEN_ER}/USD", timeout=TIMEOUT)
    r.raise_for_status()
    data = r.json()
    return {
        "rates": {c: 1 / data["rates"][c] for c in ccys if c in data["rates"]},
        "date":  date.today().isoformat(),
        "source": "open.er-api.com",
    }


def fetch_live_rates(currencies: list[str] | None = None) -> dict:
    ccys = currency_universe() if currencies is None else sorted(set(currencies) & SUPPORTED)
    key = "live:" + _to_param(ccys)
    def _fetch():
        try:
            return _fetch_live_frankfurter(ccys)
        except Exception as e:
            log.warning("frankfurter live failed (%s) — falling back", e)
            return _fetch_live_open_er(ccys)
    return _cached(key, 60, _fetch)


# ── Previous close ────────────────────────────────────────────────────────────

def fetch_prev_close(currencies: list[str] | None = None) -> dict:
    ccys = currency_universe() if currencies is None else sorted(set(currencies) & SUPPORTED)
    key = "prev:" + _to_param(ccys)
    def _fetch():
        prev = _prev_business_day(date.today()).isoformat()
        try:
            r = SESSION.get(f"{FRANKFURTER}/{prev}?from=USD&to={_to_param(ccys)}", timeout=TIMEOUT)
            r.raise_for_status()
            data = r.json()
            return {"rates": {c: 1 / data["rates"][c] for c in ccys if c in data["rates"]},
                    "date": data["date"]}
        except Exception as e:
            log.warning("prev close failed (%s) — using live", e)
            live = fetch_live_rates()
            return {"rates": dict(live["rates"]), "date": "—"}
    return _cached(key, 3600, _fetch)


# ── Historical series ─────────────────────────────────────────────────────────

def fetch_historical(days: int = 30, currencies: list[str] | None = None) -> dict:
    """{dates, rates:{ccy:[...]}} for the past `days` business days."""
    ccys = currency_universe() if currencies is None else sorted(set(currencies) & SUPPORTED)
    key = f"hist:{days}:" + _to_param(ccys)
    def _fetch():
        start = _business_days_ago(days).isoformat()
        end   = date.today().isoformat()
        try:
            r = SESSION.get(f"{FRANKFURTER}/{start}..{end}?from=USD&to={_to_param(ccys)}", timeout=TIMEOUT)
            r.raise_for_status()
            data = r.json()
            dates = sorted(data["rates"])
            series = {c: [] for c in ccys}
            kept = []
            for d in dates:
                row = data["rates"][d]
                if all(c in row for c in ccys):
                    kept.append(d)
                    for c in ccys:
                        series[c].append(round(1 / row[c], 6))
            return {"dates": kept, "rates": series}
        except Exception as e:
            log.error("historical fetch failed: %s", e)
            return {"dates": [], "rates": {c: [] for c in ccys}}
    return _cached(key, 3600, _fetch)


# ── Volatility / correlation helpers ──────────────────────────────────────────

def log_returns(series: list[float]) -> list[float]:
    return [math.log(series[i] / series[i - 1])
            for i in range(1, len(series))
            if series[i - 1] > 0 and series[i] > 0]


def _mean(xs): return sum(xs) / len(xs) if xs else 0.0


def _std(xs):
    if len(xs) < 2:
        return 0.0
    m = _mean(xs)
    return math.sqrt(sum((x - m) ** 2 for x in xs) / (len(xs) - 1))


def annualized_volatility(series: list[float]) -> float | None:
    if len(series) < 3:
        return None
    return _std(log_returns(series)) * math.sqrt(252) * 100


def vol_color(v: float | None) -> str:
    if v is None: return "neutral"
    if v < 6:  return "green"
    if v < 10: return "yellow"
    return "red"


# ── Positions enrichment ──────────────────────────────────────────────────────

def enrich_positions(live: dict, prev: dict, positions: list[dict] | None = None) -> dict:
    rate_map = live["rates"]
    prev_map = prev["rates"]
    enriched = []

    for p in (load_positions() if positions is None else positions):
        ccy    = p["currency"]
        rate   = rate_map.get(ccy)
        p_rate = prev_map.get(ccy)

        cost_local = round(p["shares"] * p["avg_cost"], 2)
        cost_usd   = round(cost_local * rate, 2) if rate else None
        fx_pnl_usd = round(cost_local * (rate - p_rate), 2) if rate and p_rate else None
        fx_pct     = round((rate - p_rate) / p_rate * 100, 3) if rate and p_rate else None

        enriched.append({**p,
            "cost_local": cost_local, "cost_usd": cost_usd,
            "fx_pnl_usd": fx_pnl_usd, "fx_pct": fx_pct,
            "rate": round(rate, 6) if rate else None,
            "flagged": abs(fx_pct) >= 0.5 if fx_pct is not None else False})

    total_cost = sum(p["cost_usd"] for p in enriched if p["cost_usd"])
    total_pnl  = round(sum(p["fx_pnl_usd"] for p in enriched if p["fx_pnl_usd"] is not None), 2)

    for p in enriched:
        p["weight_pct"] = round(p["cost_usd"] / total_cost * 100, 1) if total_cost and p["cost_usd"] else None

    exposure: dict[str, float] = {}
    for p in enriched:
        if p["cost_usd"]:
            exposure[p["currency"]] = round(exposure.get(p["currency"], 0) + p["cost_usd"], 2)

    exposure_pct = {c: round(v / total_cost * 100, 1) if total_cost else None
                    for c, v in exposure.items()}

    return {
        "positions": enriched,
        "total_cost_usd": round(total_cost, 2),
        "total_fx_pnl_usd": total_pnl,
        "exposure_usd": exposure,
        "exposure_pct": exposure_pct,
        "currencies": sorted(exposure.keys()),
    }


# ── Scenario calculator ───────────────────────────────────────────────────────

def scenario_impact(portfolio: dict, moves_pct: dict[str, float]) -> dict:
    """moves_pct: {ccy: pct_move}. Returns per-position and total USD impact."""
    total_cost = portfolio["total_cost_usd"]
    items, total = [], 0.0
    for p in portfolio["positions"]:
        if p.get("cost_usd") is None:
            continue
        move = moves_pct.get(p["currency"], 0) / 100
        impact = round(p["cost_usd"] * move, 2)
        total += impact
        items.append({"ticker": p["ticker"], "currency": p["currency"],
                      "impact_usd": impact, "weight_pct": p.get("weight_pct")})
    total = round(total, 2)
    return {"positions": items, "total_usd": total,
            "pct_of_portfolio": round(total / total_cost * 100, 3) if total_cost else None}