fx-risk-terminal/app.py

"""
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)