{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Is the portfolio becoming more exposed to one factor?\n",
    "Estimate rolling monthly factor exposures for the notebook allocation and compare the market loading with VIX.\n",
    "The notebook allocation uses fixed buy-and-hold starting weights. Each 24-month regression uses only its trailing observations. Five equity factors explain a projection of this multi-asset allocation, not every risk. Current-vintage factors and prices support descriptive analysis, not a point-in-time trading claim."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"Estimate rolling monthly factor exposures for the notebook allocation and compare the market loading 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",
    "        \"factors\": qj.ff.get_factors(region=\"america\", factor=\"5_factors\"),\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",
    "    nav = sum((prices[symbol] * (weights[symbol] / decimal(prices[symbol].iloc[0])) for symbol in weights))\n",
    "    returns = change(nav.resample(pd.offsets.MonthEnd()).last()).rename(\"Portfolio\")\n",
    "    factors = frame(raw[\"factors\"], date=None)\n",
    "    factors.index = pd.PeriodIndex(factors[\"Date\"], freq=\"M\").to_timestamp(\"M\").tz_localize(\"UTC\")\n",
    "    names = [\"Mkt-RF\", \"SMB\", \"HML\", \"RMW\", \"CMA\"]\n",
    "    data = (factors[names + [\"RF\"]].apply(pd.to_numeric) / 100).join(returns, how=\"inner\").dropna()\n",
    "    rows = []\n",
    "    for end in range(24, len(data) + 1):\n",
    "        sample = data.iloc[end - 24 : end]\n",
    "        design = np.column_stack([np.ones(len(sample)), sample[names]])\n",
    "        coefficients, _, rank, _ = np.linalg.lstsq(design, sample[\"Portfolio\"] - sample[\"RF\"], rcond=None)\n",
    "        if rank != design.shape[1]:\n",
    "            raise ValueError(\"Factor design is rank deficient\")\n",
    "        rows.append({\"Date\": sample.index[-1], **dict(zip(names, coefficients[1:], strict=True))})\n",
    "    result = pd.DataFrame(rows).set_index(\"Date\") if rows else pd.DataFrame()\n",
    "    if result.empty:\n",
    "        raise ValueError(\"Need 24 matched monthly observations\")\n",
    "    vix = macro(raw[\"vix_history\"], \"VIXCLS\").resample(pd.offsets.MonthEnd()).mean().reindex(result.index)\n",
    "    if vix.dropna().empty:\n",
    "        raise ValueError(\"No matching VIX observations\")\n",
    "    result[\"Monthly mean VIX\"] = vix\n",
    "    fig, (top, bottom) = canvas(\"Factor regime \u00b7 rolling portfolio exposures\", \"24-month factor coefficient\", panels=2)\n",
    "    for index, name in enumerate(names):\n",
    "        top.plot(result.index, result[name], label=name, linestyle=\"-\" if index < 3 else \"--\")\n",
    "    legend(top)\n",
    "    bottom.plot(result.index, result[\"Mkt-RF\"], color=BLUE, label=\"Market loading\")\n",
    "    right = bottom.twinx()\n",
    "    right.plot(result.index, vix, color=PINK, alpha=0.6)\n",
    "    right.set_ylabel(\"Monthly mean VIX\", color=PINK)\n",
    "    bottom.set_ylabel(\"Market coefficient\")\n",
    "    return result.reset_index(), 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 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_factor_regime.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
}