{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# What drove 80% of the portfolio\u2019s losses?\n", "Locate the largest observed drawdown, attribute the loss to each holding and compare the period with VIX.\n", "The linked notebook allocation is a model parameter: SPY 42%, TLT 22%, GLD 14%, DBC 12%, UUP 10%. Buy-and-hold contributions reconcile exactly to the selected peak-to-trough drawdown. The 80% cutoff refers to gross losing contributions, before offsets from winners." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "\"\"\"Locate the largest observed drawdown, attribute the loss to each holding and compare the period with VIX.\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.eod.get_historical_prices(\n", " symbols=[\"SPY\", \"TLT\", \"GLD\", \"DBC\", \"UUP\"], start_date=\"2020-01-01\", end_date=\"2025-12-31\", frequency=\"1d\"\n", " ),\n", " \"vix_history\": qj.fred.get_fred_data_series_by_id(search_id=\"VIXCLS\", start=\"2020-01-01\", end=\"2025-12-31\"),\n", " }\n", "\n", "\n", "def analyze(raw):\n", " weights = notebook_weights()\n", " prices = pd.DataFrame({symbol: history(raw[\"prices\"], symbol)[\"price\"] for symbol in weights}).dropna()\n", " if len(prices) < 252:\n", " raise ValueError(\"Need at least one year of common prices\")\n", " units = {symbol: weights[symbol] / decimal(prices[symbol].iloc[0]) for symbol in weights}\n", " values = prices.apply(lambda column: column * units[column.name])\n", " nav = values.sum(axis=1)\n", " drawdown = pd.Series(\n", " [ratio(value, peak) - 1 for value, peak in zip(nav, nav.cummax(), strict=True)], index=nav.index\n", " )\n", " trough = drawdown.idxmin()\n", " peak = nav.loc[:trough].idxmax()\n", " if peak == trough:\n", " raise ValueError(\"No drawdown in the observed sample\")\n", " contribution = values.loc[trough] - values.loc[peak]\n", " result = pd.DataFrame(\n", " {\"Symbol\": contribution.index, \"Contribution pp\": [100 * ratio(value, nav.loc[peak]) for value in contribution]}\n", " )\n", " gross_losses = -result.loc[result[\"Contribution pp\"] < 0, \"Contribution pp\"].sum()\n", " if gross_losses <= 0:\n", " raise ValueError(\"No negative contribution\")\n", " result = result.sort_values(\"Contribution pp\")\n", " shares = (-result[\"Contribution pp\"].clip(upper=0) / gross_losses).cumsum()\n", " result[\"In 80% loss group\"] = shares.shift(fill_value=0) < 0.8\n", " result[\"Gross loss share %\"] = 100 * (-result[\"Contribution pp\"].clip(upper=0) / gross_losses)\n", " if not np.isclose(result[\"Contribution pp\"].sum(), 100 * drawdown.loc[trough]):\n", " raise ValueError(\"Attribution does not reconcile\")\n", " vix = macro(raw[\"vix_history\"], \"VIXCLS\").reindex(drawdown.index)\n", " if vix.loc[peak:trough].dropna().empty:\n", " raise ValueError(\"No VIX observations in the drawdown window\")\n", " fig, (top, bottom) = canvas(\n", " \"Portfolio concentration \u00b7 explain the drawdown\", \"Contribution to drawdown \u00b7 pp\", panels=2\n", " )\n", " top.barh(\n", " result[\"Symbol\"],\n", " result[\"Contribution pp\"],\n", " color=[PINK if flag else BLUE for flag in result[\"In 80% loss group\"]],\n", " )\n", " top.axvline(0, color=\"#afc1dc\", lw=0.8)\n", " top.set_xlabel(f\"Peak {peak:%Y-%m-%d} \u2192 trough {trough:%Y-%m-%d}\")\n", " bottom.plot(drawdown.index, 100 * drawdown, label=\"Portfolio drawdown %\", color=BLUE)\n", " bottom.axvspan(peak, trough, color=PINK, alpha=0.12)\n", " right = bottom.twinx()\n", " right.plot(vix.index, vix, alpha=0.4, color=PINK)\n", " right.set_ylabel(\"VIX level\", color=PINK)\n", " bottom.set_ylabel(\"Drawdown %\")\n", " result[\"Peak\"] = peak\n", " result[\"Trough\"] = trough\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 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 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 notebook_weights():\n", " return {\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", "\n", "\n", "OUTPUT = \"sdk_drawdown_attribution.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 }