{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# How much of the post-earnings move was residual?\n",
    "Fit market, sector and style exposures before an earnings event, then reconcile their contributions to the observed stock return.\n",
    "AAPL. A 120-session regression ends before the event date. The next five sessions use frozen coefficients. Contributions reconcile to the sum of daily arithmetic returns; they do not claim a compounded holding-period return or causal attribution. Sector is XLK minus SPY; style factors are SMB, HML, RMW and CMA."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"Fit market, sector and style exposures before an earnings event, then reconcile their contributions to the observed stock return.\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\", \"SPY\", \"XLK\"], start_date=\"2023-01-01\", end_date=\"2025-12-31\", frequency=\"1d\"\n",
    "        ),\n",
    "        \"daily_factors\": qj.ff.get_factors(\n",
    "            region=\"america\", factor=\"5_factors\", frequency=\"daily\", start_date=\"2023-01-01\", end_date=\"2025-12-31\"\n",
    "        ),\n",
    "        \"earnings\": qj.fmp.get_earnings_surprises(symbol=\"AAPL\"),\n",
    "    }\n",
    "\n",
    "\n",
    "def analyze(raw):\n",
    "    returns = pd.DataFrame(\n",
    "        {symbol: change(history(raw[\"prices\"], symbol)[\"price\"]) for symbol in [\"AAPL\", \"SPY\", \"XLK\"]}\n",
    "    )\n",
    "    data = returns.join(factor_daily(raw[\"daily_factors\"]), how=\"inner\").dropna()\n",
    "    data[\"Market\"] = data[\"SPY\"] - data[\"RF\"]\n",
    "    data[\"Sector\"] = data[\"XLK\"] - data[\"SPY\"]\n",
    "    names = [\"Market\", \"Sector\", \"SMB\", \"HML\", \"RMW\", \"CMA\"]\n",
    "    events = frame(raw[\"earnings\"])\n",
    "    chosen = None\n",
    "    for event in events[\"date\"].sort_values(ascending=False):\n",
    "        train = data.loc[data.index < event].tail(120)\n",
    "        after = data.loc[data.index > event].head(5)\n",
    "        if len(train) == 120 and len(after) == 5:\n",
    "            chosen = event\n",
    "            break\n",
    "    if chosen is None:\n",
    "        raise ValueError(\"No event with a complete pre-event and post-event sample\")\n",
    "    design = np.column_stack([np.ones(len(train)), train[names]])\n",
    "    coefficients, _, rank, _ = np.linalg.lstsq(design, train[\"AAPL\"] - train[\"RF\"], rcond=None)\n",
    "    if rank != design.shape[1]:\n",
    "        raise ValueError(\"Attribution design is rank deficient\")\n",
    "    contribution = after[names].mul(coefficients[1:]).sum()\n",
    "    rows = [{\"Component\": name, \"Contribution pp\": 100 * contribution[name]} for name in names]\n",
    "    rows.append(\n",
    "        {\"Component\": \"Cash + intercept\", \"Contribution pp\": 100 * (after[\"RF\"].sum() + len(after) * coefficients[0])}\n",
    "    )\n",
    "    residual = after[\"AAPL\"].sum() - sum(row[\"Contribution pp\"] for row in rows) / 100\n",
    "    rows.append({\"Component\": \"Residual\", \"Contribution pp\": 100 * residual})\n",
    "    result = pd.DataFrame(rows)\n",
    "    if not np.isclose(result[\"Contribution pp\"].sum(), 100 * after[\"AAPL\"].sum()):\n",
    "        raise ValueError(\"Attribution does not reconcile\")\n",
    "    result[\"Event date\"] = chosen\n",
    "    result[\"Training end\"] = train.index[-1]\n",
    "    result[\"Response end\"] = after.index[-1]\n",
    "    fig, (ax,) = canvas(\"Post-event attribution \u00b7 market to residual\", \"Return contribution \u00b7 pp\")\n",
    "    start = 0.0\n",
    "    levels = [start]\n",
    "    for index, row in result.iterrows():\n",
    "        value = row[\"Contribution pp\"]\n",
    "        ax.bar(index, value, bottom=start, color=PINK if row[\"Component\"] == \"Residual\" else BLUE)\n",
    "        ax.text(index, start + value, f\"{value:+.2f}\", ha=\"center\", va=\"bottom\", fontsize=9)\n",
    "        start += value\n",
    "        levels.append(start)\n",
    "    span = max(max(levels) - min(levels), 0.05)\n",
    "    ax.set_ylim(min(levels) - 0.12 * span, max(levels) + 0.12 * span)\n",
    "    ax.axhline(0, color=\"#afc1dc\", lw=0.8)\n",
    "    ax.bar(len(result), start, color=\"#afc1dc\")\n",
    "    ax.set_xticks(range(len(result) + 1), list(result[\"Component\"]) + [\"Observed\"], rotation=35, ha=\"right\")\n",
    "    ax.set_title(f\"Earnings {chosen:%Y-%m-%d} \u00b7 next five sessions\", loc=\"left\", color=\"#e9eef8\", fontsize=11)\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 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 factor_daily(response):\n",
    "    factors = frame(response, date=None)\n",
    "    factors.index = pd.to_datetime(factors[\"Date\"], utc=True)\n",
    "    if factors.index.has_duplicates:\n",
    "        raise ValueError(\"Duplicate factor dates\")\n",
    "    return factors[[\"Mkt-RF\", \"SMB\", \"HML\", \"RMW\", \"CMA\", \"RF\"]].apply(pd.to_numeric) / 100\n",
    "\n",
    "\n",
    "OUTPUT = \"sdk_event_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
}