mirror of
https://github.com/Saauc/fx-risk-terminal.git
synced 2026-08-21 06:38:08 +00:00
Initial commit: FX Risk Terminal
Multi-currency FX risk engine + browser dashboard: - Live USD valuation of a multi-currency equity book (ECB rates, no API key) - Value-at-Risk by 3 methods (parametric, historical, Monte Carlo) - Expected Shortfall, component VaR, diversification ratio - Monte Carlo via from-scratch Cholesky (pure Python, no numpy) - Historical stress testing + minimum-variance hedge search - Interactive in-browser portfolio builder (stateless, localStorage) - 20 offline unit tests Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
sauc
+311
@@ -0,0 +1,311 @@
|
||||
"""
|
||||
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}
|
||||
Reference in New Issue
Block a user