{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# How does FX change your portfolio\u2019s tail risk?\n",
    "Translate a multi-asset allocation into EUR, estimate historical VaR and Expected Shortfall, and inspect high-VIX observations.\n",
    "Notebook allocation: SPY 42%, TLT 22%, GLD 14%, DBC 12%, UUP 10%, daily rebalanced before costs. EUR translation uses ECB reference rates. 95% historical one-day loss estimates; high VIX uses the previous session and a sample 85th-percentile cutoff. Not a forecast."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"Translate a multi-asset allocation into EUR, estimate historical VaR and Expected Shortfall, and inspect high-VIX observations.\n",
    "\n",
    "Research example. Live observations can change.\n",
    "\"\"\"\n",
    "\n",
    "from datetime import UTC, datetime\n",
    "from decimal import Decimal\n",
    "from textwrap import fill\n",
    "\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "AS_OF = datetime.now(UTC).date()\n",
    "NAVY, BLUE, PINK = \"#132244\", \"#8fb9ff\", \"#e7a2cc\"\n",
    "\n",
    "\n",
    "def fetch(qj):\n",
    "    return {\n",
    "        \"prices\": qj.domains.call(\n",
    "            \"equity.pricing.get_historical_prices\",\n",
    "            provider=\"eod\",\n",
    "            symbols=[\"SPY\", \"TLT\", \"GLD\", \"DBC\", \"UUP\"],\n",
    "            start_date=\"2020-01-01\",\n",
    "            end_date=\"2025-12-31\",\n",
    "            frequency=\"1d\",\n",
    "        ),\n",
    "        \"ecb_fx\": qj.domains.call(\n",
    "            \"macro.eu.get_ecb_series\",\n",
    "            provider=\"ecb\",\n",
    "            series_id=\"EXR.D.USD.EUR.SP00.A\",\n",
    "            start_date=\"2020-01-01\",\n",
    "            end_date=\"2025-12-31\",\n",
    "        ),\n",
    "        \"vix_history\": qj.domains.call(\n",
    "            \"macro.indicators.get_fred_data_series_by_id\",\n",
    "            provider=\"fred\",\n",
    "            search_id=\"VIXCLS\",\n",
    "            start=\"2020-01-01\",\n",
    "            end=\"2025-12-31\",\n",
    "        ),\n",
    "    }\n",
    "\n",
    "\n",
    "def analyze(raw):\n",
    "    usd = portfolio_returns(raw).rename(\"USD return\")\n",
    "    # Use exact daily FX boundaries; do not fill missing reference rates.\n",
    "    fx = frame(raw[\"ecb_fx\"]).set_index(\"date\")[\"value\"].map(decimal).reindex(usd.index)\n",
    "    fx_change = change(fx)\n",
    "    euro = ((1 + usd) / (1 + fx_change) - 1).rename(\"EUR return\")\n",
    "    vix = macro(raw[\"vix_history\"], \"VIXCLS\").reindex(usd.index).shift(1).rename(\"Prior VIX\")\n",
    "    data = pd.concat([usd, euro, vix], axis=1).dropna()\n",
    "    if len(data) < 252:\n",
    "        raise ValueError(\"Insufficient common portfolio, FX and VIX observations\")\n",
    "    stressed = data.loc[data[\"Prior VIX\"] >= data[\"Prior VIX\"].quantile(0.85), \"EUR return\"]\n",
    "    rows = []\n",
    "    for name, returns in [(\"USD\", data[\"USD return\"]), (\"EUR\", data[\"EUR return\"]), (\"EUR \u00b7 high VIX\", stressed)]:\n",
    "        losses = -returns.to_numpy()\n",
    "        var = np.quantile(losses, 0.95)\n",
    "        es = losses[losses >= var].mean()\n",
    "        rows.append(\n",
    "            {\"Sample\": name, \"95% VaR %\": 100 * var, \"Expected shortfall %\": 100 * es, \"Observations\": len(losses)}\n",
    "        )\n",
    "    result = pd.DataFrame(rows)\n",
    "    fig, (top, bottom) = canvas(\"Portfolio \u00b7 currency and tail risk\", \"Observed daily loss frequency\", panels=2)\n",
    "    for column, label in [(\"USD return\", \"USD investor\"), (\"EUR return\", \"EUR investor\")]:\n",
    "        top.hist(-100 * data[column], bins=65, histtype=\"step\", lw=1.7, label=label)\n",
    "    top.set_xlabel(\"One-day loss %\")\n",
    "    legend(top)\n",
    "    x = np.arange(len(result))\n",
    "    bottom.bar(x - 0.18, result[\"95% VaR %\"], 0.36, label=\"95% VaR\")\n",
    "    bottom.bar(x + 0.18, result[\"Expected shortfall %\"], 0.36, label=\"Expected shortfall\")\n",
    "    bottom.set_xticks(x, result[\"Sample\"])\n",
    "    bottom.set_ylabel(\"Historical one-day loss %\")\n",
    "    legend(bottom)\n",
    "    return result, fig\n",
    "\n",
    "\n",
    "def payload(response):\n",
    "    if hasattr(response, \"to_dict\"):\n",
    "        response = response.to_dict()\n",
    "    while isinstance(response, dict):\n",
    "        if response.get(\"error\") or response.get(\"error_code\") or response.get(\"success\") is False:\n",
    "            raise ValueError(\"Provider request failed\")\n",
    "        if \"data\" in response:\n",
    "            response = response[\"data\"]\n",
    "        elif \"value\" in response:\n",
    "            response = response[\"value\"]\n",
    "        else:\n",
    "            break\n",
    "    return response\n",
    "\n",
    "\n",
    "def frame(response, date=\"date\"):\n",
    "    rows = payload(response)\n",
    "    if not isinstance(rows, list) or not rows:\n",
    "        raise ValueError(\"No provider observations returned\")\n",
    "    result = pd.DataFrame(rows)\n",
    "    if date:\n",
    "        result[date] = pd.to_datetime(result[date], utc=True)\n",
    "        result = result.sort_values(date)\n",
    "    return result\n",
    "\n",
    "\n",
    "def decimal(value):\n",
    "    if pd.isna(value):\n",
    "        raise ValueError(\"Required monetary value or quantity is missing\")\n",
    "    return Decimal(str(value))\n",
    "\n",
    "\n",
    "def ratio(numerator, denominator):\n",
    "    denominator = decimal(denominator)\n",
    "    if denominator == 0:\n",
    "        raise ValueError(\"Required denominator is zero\")\n",
    "    return float(decimal(numerator) / denominator)\n",
    "\n",
    "\n",
    "def change(values, periods=1):\n",
    "    previous = values.shift(periods)\n",
    "    return pd.Series(\n",
    "        [ratio(x, y) - 1 if pd.notna(x) and pd.notna(y) else np.nan for x, y in zip(values, previous, strict=True)],\n",
    "        index=values.index,\n",
    "    )\n",
    "\n",
    "\n",
    "def history(response, symbol, adjusted=True):\n",
    "    values = payload(response)\n",
    "    if not isinstance(values, dict) or symbol not in values:\n",
    "        raise ValueError(f\"No price coverage for {symbol}\")\n",
    "    result = frame(values[symbol]).set_index(\"date\")\n",
    "    field = \"adjusted_close\" if adjusted else \"close\"\n",
    "    result[\"price\"] = result[field].map(decimal)\n",
    "    result[\"volume\"] = result[\"volume\"].map(decimal)\n",
    "    if result.index.has_duplicates:\n",
    "        raise ValueError(\"Duplicate trading dates\")\n",
    "    return result\n",
    "\n",
    "\n",
    "def macro(response, column):\n",
    "    values = frame(response).set_index(\"date\")[column]\n",
    "    values = pd.to_numeric(values, errors=\"coerce\").dropna()\n",
    "    if values.empty:\n",
    "        raise ValueError(f\"No numeric observations for {column}\")\n",
    "    return values\n",
    "\n",
    "\n",
    "def canvas(title, ylabel, panels=1):\n",
    "    plt.rcParams.update(\n",
    "        {\n",
    "            \"figure.facecolor\": NAVY,\n",
    "            \"axes.facecolor\": NAVY,\n",
    "            \"savefig.facecolor\": NAVY,\n",
    "            \"text.color\": \"#e9eef8\",\n",
    "            \"axes.labelcolor\": \"#afc1dc\",\n",
    "            \"xtick.color\": \"#afc1dc\",\n",
    "            \"ytick.color\": \"#afc1dc\",\n",
    "            \"axes.edgecolor\": \"#34486b\",\n",
    "            \"grid.color\": \"#34486b\",\n",
    "            \"font.size\": 10,\n",
    "            \"axes.prop_cycle\": plt.cycler(color=[BLUE, PINK, \"#ba9dea\"]),\n",
    "        }\n",
    "    )\n",
    "    fig, axes = plt.subplots(panels, 1, figsize=(9, 6.1), squeeze=False, layout=\"constrained\")\n",
    "    fig.suptitle(fill(title, width=58), fontsize=17, x=0.04, ha=\"left\", weight=\"medium\")\n",
    "    for ax in axes.flat:\n",
    "        ax.grid(axis=\"y\", alpha=0.45)\n",
    "        ax.set_axisbelow(True)\n",
    "        ax.spines[[\"top\", \"right\"]].set_visible(False)\n",
    "    axes.flat[0].set_ylabel(ylabel)\n",
    "    return fig, list(axes.flat)\n",
    "\n",
    "\n",
    "def legend(ax):\n",
    "    ax.legend(frameon=False, labelcolor=\"#e9eef8\", fontsize=9)\n",
    "\n",
    "\n",
    "def run():\n",
    "    from quantjourney.sdk import QuantJourneyAPI\n",
    "\n",
    "    qj = QuantJourneyAPI.from_env()\n",
    "    raw = fetch(qj)\n",
    "    table, fig = analyze(raw)\n",
    "    print(table.to_string(index=False))\n",
    "    fig.savefig(OUTPUT, dpi=180)\n",
    "    plt.show()\n",
    "\n",
    "\n",
    "def portfolio_returns(raw):\n",
    "    weights = {\n",
    "        \"SPY\": Decimal(\"0.42\"),\n",
    "        \"TLT\": Decimal(\"0.22\"),\n",
    "        \"GLD\": Decimal(\"0.14\"),\n",
    "        \"DBC\": Decimal(\"0.12\"),\n",
    "        \"UUP\": Decimal(\"0.10\"),\n",
    "    }\n",
    "    if sum(weights.values()) != Decimal(1):\n",
    "        raise ValueError(\"Portfolio weights must sum to one\")\n",
    "    returns = pd.DataFrame({symbol: change(history(raw[\"prices\"], symbol)[\"price\"]) for symbol in weights}).dropna()\n",
    "    if len(returns) < 252:\n",
    "        raise ValueError(\"Need at least one year of common daily portfolio observations\")\n",
    "    return returns.dot(pd.Series({symbol: float(weight) for symbol, weight in weights.items()}))\n",
    "\n",
    "\n",
    "OUTPUT = \"sdk_portfolio_var.png\"\n",
    "\n",
    "if __name__ == \"__main__\":\n",
    "    run()\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "name": "python"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}