{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# What risks sit behind the stock returns?\n", "Align monthly adjusted returns and the five Fama\u2013French factors, subtract the risk-free return, then estimate exposures.\n", "AAPL, MSFT and SPY, monthly common observations in 2020\u20132025. Five-factor OLS with an intercept. Exposures describe the sample; they are not forecasts or portfolio recommendations." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "\"\"\"Align monthly adjusted returns and the five Fama\u2013French factors, subtract the risk-free return, then estimate exposures.\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=[\"AAPL\", \"MSFT\", \"SPY\"], 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", " }\n", "\n", "\n", "def analyze(raw):\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", " factors = factors[names + [\"RF\"]].apply(pd.to_numeric) / 100\n", " rows = []\n", " for symbol in [\"AAPL\", \"MSFT\", \"SPY\"]:\n", " prices = history(raw[\"prices\"], symbol)[\"price\"].resample(pd.offsets.MonthEnd()).last()\n", " returns = change(prices).rename(\"Stock\")\n", " joined = factors.join(returns, how=\"inner\").dropna()\n", " if len(joined) < 24:\n", " raise ValueError(\"Need at least 24 matched monthly observations\")\n", " design = np.column_stack([np.ones(len(joined)), joined[names].to_numpy()])\n", " excess = joined[\"Stock\"].to_numpy() - joined[\"RF\"].to_numpy()\n", " coefficients, _, rank, _ = np.linalg.lstsq(design, excess, rcond=None)\n", " if rank != design.shape[1]:\n", " raise ValueError(\"Factor design is rank deficient\")\n", " rows.append({\"Symbol\": symbol, **dict(zip(names, coefficients[1:], strict=True)), \"Months\": len(joined)})\n", " result = pd.DataFrame(rows).set_index(\"Symbol\")\n", " fig, (ax,) = canvas(\"Five-factor exposures \u00b7 what the returns load on\", \"\")\n", " from matplotlib.colors import LinearSegmentedColormap\n", "\n", " cmap = LinearSegmentedColormap.from_list(\"exposures\", [PINK, NAVY, BLUE])\n", " matrix = result[names].to_numpy()\n", " bound = max(abs(matrix.min()), abs(matrix.max()))\n", " ax.imshow(matrix, cmap=cmap, vmin=-bound, vmax=bound, aspect=\"auto\")\n", " ax.set_xticks(range(len(names)), names)\n", " ax.set_yticks(range(len(result)), result.index)\n", " ax.grid(False)\n", " for i, j in np.ndindex(matrix.shape):\n", " ax.text(j, i, f\"{matrix[i, j]:.2f}\", ha=\"center\", va=\"center\", color=\"#ffffff\")\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 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", "OUTPUT = \"sdk_factor_exposure.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 }