""" Flask server for the FX Risk Terminal (multi-currency). Routes ------ GET / → dashboard UI GET /api/health → liveness + data-source status GET /api/rates → live rates, prev close, % change, flags (per currency) GET /api/positions → enriched positions (USD value, weights, FX P&L, exposure) GET /api/historical → 30-day rate series + per-currency vol + correlation GET /api/risk → VaR (3 methods), CVaR, component VaR, MC dist, hedge, diversification GET /api/stress → historical stress-scenario impacts POST /api/scenario → user-defined FX shocks {moves:{ccy:pct}} → portfolio impact """ import logging from flask import Flask, render_template, jsonify, request import fx_engine as fx import risk_engine as risk logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s") log = logging.getLogger("app") app = Flask(__name__) def _pct_change(cur, pre): return round((cur - pre) / pre * 100, 4) if cur and pre else None def _market(): return fx.fetch_live_rates(), fx.fetch_prev_close() def _returns(hist): """Per-currency daily log-return series from a historical payload.""" return {c: fx.log_returns(s) for c, s in hist["rates"].items()} @app.errorhandler(Exception) def handle_error(err): log.exception("Unhandled error: %s", err) if request.path.startswith("/api/"): return jsonify({"error": str(err), "type": type(err).__name__}), 500 raise err @app.route("/") def index(): return render_template("dashboard.html") @app.route("/api/health") def api_health(): live = fx.fetch_live_rates() return jsonify({"status": "ok", "source": live.get("source"), "rate_date": live.get("date"), "currencies": fx.currency_universe()}) @app.route("/api/rates") def api_rates(): live, prev = _market() change = {c: _pct_change(live["rates"].get(c), prev["rates"].get(c)) for c in live["rates"]} return jsonify({ "live": live, "prev": prev, "change": change, "flagged": {c: abs(v or 0) >= 0.5 for c, v in change.items()}, }) @app.route("/api/positions") def api_positions(): live, prev = _market() return jsonify(fx.enrich_positions(live, prev)) @app.route("/api/historical") def api_historical(): hist = fx.fetch_historical(30) rates = hist["rates"] analytics = {} for c, s in rates.items(): v = fx.annualized_volatility(s) analytics[c] = {"vol": round(v, 2) if v else None, "vol_color": fx.vol_color(v)} currencies = sorted(rates.keys()) cov = risk.covariance_matrix(_returns(hist), currencies) corr = risk.correlation_matrix(cov) if cov else [] return jsonify({"series": hist, "currencies": currencies, "analytics": analytics, "correlation": corr}) @app.route("/api/risk") def api_risk(): live, prev = _market() portfolio = fx.enrich_positions(live, prev) hist = fx.fetch_historical(30) horizon = max(1, int(request.args.get("horizon", 1))) return jsonify(risk.full_risk_report(portfolio, _returns(hist), horizon)) @app.route("/api/stress") def api_stress(): live, prev = _market() portfolio = fx.enrich_positions(live, prev) currencies = portfolio["currencies"] exposure = risk.exposure_vector(portfolio, currencies) return jsonify({"total_exposure_usd": portfolio["total_cost_usd"], "currencies": currencies, "scenarios": risk.stress_test(exposure, currencies)}) @app.route("/api/analyze", methods=["POST"]) def api_analyze(): """ Stateless full-bundle analysis of a user-supplied book (the in-browser portfolio builder). The book is never persisted server-side — it is sanitised, priced against live rates, and returned with every analytic so the frontend can render the whole dashboard from one round trip. """ body = request.get_json(silent=True) or {} positions = fx.sanitize_positions(body.get("positions", [])) horizon = max(1, int(body.get("horizon", 1))) if not positions: return jsonify({"error": "no valid positions", "type": "ValueError"}), 400 currencies = fx.currency_universe(positions) live = fx.fetch_live_rates(currencies) prev = fx.fetch_prev_close(currencies) hist = fx.fetch_historical(30, currencies) portfolio = fx.enrich_positions(live, prev, positions) change = {c: _pct_change(live["rates"].get(c), prev["rates"].get(c)) for c in live["rates"]} analytics = {c: {"vol": (lambda v: round(v, 2) if v else None)(fx.annualized_volatility(s)), "vol_color": fx.vol_color(fx.annualized_volatility(s))} for c, s in hist["rates"].items()} cov = risk.covariance_matrix(_returns(hist), currencies) corr = risk.correlation_matrix(cov) if cov else [] return jsonify({ "rates": {"live": live, "prev": prev, "change": change, "flagged": {c: abs(v or 0) >= 0.5 for c, v in change.items()}}, "positions": portfolio, "risk": risk.full_risk_report(portfolio, _returns(hist), horizon), "stress": {"total_exposure_usd": portfolio["total_cost_usd"], "currencies": currencies, "scenarios": risk.stress_test(risk.exposure_vector(portfolio, currencies), currencies)}, "historical": {"series": hist, "currencies": currencies, "analytics": analytics, "correlation": corr}, }) @app.route("/api/scenario", methods=["POST"]) def api_scenario(): body = request.get_json(silent=True) or {} moves = body.get("moves", {}) # Backward-compatible with the old {eur_move, sek_move} shape. if "eur_move" in body: moves["EUR"] = body["eur_move"] if "sek_move" in body: moves["SEK"] = body["sek_move"] moves = {c: float(v) for c, v in moves.items()} # Optional custom book from the in-browser builder; else the file book. custom = fx.sanitize_positions(body.get("positions", [])) if body.get("positions") else None currencies = fx.currency_universe(custom) if custom else None live, prev = (fx.fetch_live_rates(currencies), fx.fetch_prev_close(currencies)) if custom else _market() portfolio = fx.enrich_positions(live, prev, custom) return jsonify(fx.scenario_impact(portfolio, moves)) if __name__ == "__main__": app.run(debug=True, port=5050)