{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# What growth do analyst estimates imply?\n",
    "Compare filed and expected EPS, then check whether the latest reported short position increased.\n",
    "AAPL EPS actuals are matched to SEC filings. Future consensus is a snapshot, not revisions. The short-interest comparison uses FINRA\u2019s latest and prior reported positions, settlement 2026-10-15."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "\"\"\"Compare filed and expected EPS, then check whether the latest reported short position increased.\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 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",
    "        \"short\": qj.domains.call(\"equity.shorting.get_short_interest\", provider=\"finra\", symbol=\"AAPL\"),\n",
    "        \"income\": qj.domains.call(\n",
    "            \"equity.fundamentals.get_income_statement\", provider=\"fmp\", symbol=\"AAPL\", period=\"annual\", limit=5\n",
    "        ),\n",
    "        \"filings\": qj.domains.call(\n",
    "            \"regulatory.filings.get_company_filings\", provider=\"sec\", symbol=\"AAPL\", form_type=\"10-K\", limit=12\n",
    "        ),\n",
    "        \"estimates\": qj.fmp.get_analyst_estimates(symbol=\"AAPL\", period=\"annual\", limit=8),\n",
    "    }\n",
    "\n",
    "\n",
    "def analyze(raw):\n",
    "    actual = verified_income(raw)\n",
    "    actual = actual.loc[actual[\"accepted\"].dt.date <= AS_OF]\n",
    "    estimates = frame(raw[\"estimates\"])\n",
    "    future = estimates.loc[estimates[\"date\"].dt.date > AS_OF].copy()\n",
    "    if actual.empty or future.empty:\n",
    "        raise ValueError(\"Need verified actual EPS and future analyst estimates\")\n",
    "    baseline = decimal(actual.iloc[-1][\"eps\"])\n",
    "    actual[\"EPS index\"] = actual[\"eps\"].map(lambda x: 100 * ratio(x, baseline))\n",
    "    future[\"EPS index\"] = future[\"epsAvg\"].map(lambda x: 100 * ratio(x, baseline))\n",
    "    fig, (ax, bottom) = canvas(\n",
    "        \"AAPL \u00b7 expectations and short positioning\", \"Latest verified annual EPS = 100\", panels=2\n",
    "    )\n",
    "    ax.plot(actual[\"date\"], actual[\"EPS index\"], marker=\"o\", label=\"Filed annual EPS\")\n",
    "    ax.plot(future[\"date\"], future[\"EPS index\"], marker=\"o\", linestyle=\"--\", label=\"Future consensus EPS\")\n",
    "    ax.axhline(100, color=\"#afc1dc\", lw=0.8)\n",
    "    legend(ax)\n",
    "    result = pd.concat(\n",
    "        [\n",
    "            actual[[\"date\", \"EPS index\"]].assign(Observation=\"Filed\"),\n",
    "            future[[\"date\", \"EPS index\"]].assign(Observation=\"Estimate\"),\n",
    "        ]\n",
    "    )\n",
    "    short = frame(raw[\"short\"], \"settlement_date\").iloc[-1]\n",
    "    short_index = 100 * ratio(short[\"current_short_position\"], short[\"previous_short_position\"])\n",
    "    bottom.bar(\n",
    "        [\"Prior reported position\", \"Current reported position\"], [100, short_index], color=[BLUE, PINK], width=0.5\n",
    "    )\n",
    "    bottom.set_ylabel(\"Short-position index \u00b7 prior = 100\")\n",
    "    bottom.set_title(\"FINRA settlement \u00b7 \" + short[\"settlement_date\"].strftime(\"%Y-%m-%d\"), fontsize=10)\n",
    "    result[\"Latest / prior short position %\"] = short_index\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 verified_income(raw):\n",
    "    income = frame(raw[\"income\"])\n",
    "    filings = frame(raw[\"filings\"], \"reportDate\")\n",
    "    filings = filings[[\"reportDate\", \"acceptanceDateTime\", \"filingDate\"]].copy()\n",
    "    filings[\"accepted\"] = pd.to_datetime(filings[\"acceptanceDateTime\"], utc=True)\n",
    "    return income.merge(\n",
    "        filings[[\"reportDate\", \"accepted\"]],\n",
    "        left_on=\"date\",\n",
    "        right_on=\"reportDate\",\n",
    "        validate=\"one_to_one\",\n",
    "    ).sort_values(\"date\")\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",
    "OUTPUT = \"sdk_expectations.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
}