{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Which names have rising EPS estimates but falling prices?\n", "Compare next-quarter EPS revisions with 30-calendar-day price returns and retain the latest reported earnings date.\n", "AAPL, MSFT and NVDA. Revisions compare estimates for the same future fiscal period using EOD\u2019s current and 30-days-ago fields. Price returns use the last session on or before each boundary. FMP supplies reported earnings dates. A snapshot screen is not a point-in-time backtest." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "\"\"\"Compare next-quarter EPS revisions with 30-calendar-day price returns and retain the latest reported earnings date.\n", "\n", "Research example. Live observations can change.\n", "\"\"\"\n", "\n", "from datetime import UTC, datetime, timedelta\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", " raw = {\n", " \"trends\": qj.eod.get_trends_calendar(symbols=[\"AAPL.US\", \"MSFT.US\", \"NVDA.US\"]),\n", " \"revision_prices\": qj.eod.get_historical_prices(\n", " symbols=[\"AAPL\", \"MSFT\", \"NVDA\"],\n", " start_date=str(AS_OF - timedelta(days=90)),\n", " end_date=str(AS_OF),\n", " frequency=\"1d\",\n", " ),\n", " }\n", " for symbol, key in [(\"AAPL\", \"earnings\"), (\"MSFT\", \"earnings_msft\"), (\"NVDA\", \"earnings_nvda\")]:\n", " raw[key] = qj.fmp.get_earnings_surprises(symbol=symbol)\n", " return raw\n", "\n", "\n", "def analyze(raw):\n", " result = current_revision_rows(raw)\n", " fig, (ax,) = canvas(\"EPS revisions \u00b7 price disagreement\", \"30-day EPS estimate revision %\")\n", " for row in result.to_dict(\"records\"):\n", " ax.scatter(\n", " row[\"Price return %\"],\n", " row[\"EPS revision %\"],\n", " s=100,\n", " color=PINK if row[\"Rising EPS / falling price\"] else BLUE,\n", " )\n", " ax.annotate(\n", " row[\"Symbol\"], (row[\"Price return %\"], row[\"EPS revision %\"]), xytext=(8, 7), textcoords=\"offset points\"\n", " )\n", " ax.axhline(0, color=\"#afc1dc\", lw=0.8)\n", " ax.axvline(0, color=\"#afc1dc\", lw=0.8)\n", " ax.set_xlabel(\"30-calendar-day adjusted price return %\")\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 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 current_revision_rows(raw):\n", " estimates = frame(raw[\"trends\"])\n", " estimates = estimates.loc[(estimates[\"period\"] == \"+1q\") & (estimates[\"date\"].dt.date > AS_OF)]\n", " rows = []\n", " for symbol in [\"AAPL\", \"MSFT\", \"NVDA\"]:\n", " eligible = estimates.loc[estimates[\"code\"] == symbol + \".US\"].sort_values(\"date\")\n", " if eligible.empty:\n", " raise ValueError(\"Missing next-quarter estimate: \" + symbol)\n", " row = eligible.iloc[0]\n", " price = history(raw[\"revision_prices\"], symbol)[\"price\"]\n", " end = price.index[-1]\n", " beginning = price.loc[price.index <= end - pd.Timedelta(days=30)]\n", " if beginning.empty:\n", " raise ValueError(\"No 30-day price reference\")\n", " earnings = frame(raw[\"earnings\" if symbol == \"AAPL\" else \"earnings_\" + symbol.lower()])\n", " reported = earnings.loc[earnings[\"date\"] <= end]\n", " if reported.empty:\n", " raise ValueError(\"Missing reported earnings date\")\n", " previous = decimal(row[\"epsTrend30daysAgo\"])\n", " if previous <= 0:\n", " raise ValueError(\"A non-positive EPS baseline needs a different revision measure\")\n", " rows.append(\n", " {\n", " \"Symbol\": symbol,\n", " \"Fiscal period\": row[\"date\"],\n", " \"EPS revision %\": 100 * ratio(decimal(row[\"epsTrendCurrent\"]) - previous, previous),\n", " \"Price return %\": 100 * (ratio(price.iloc[-1], beginning.iloc[-1]) - 1),\n", " \"Last reported earnings\": reported[\"date\"].max(),\n", " \"Price date\": end,\n", " }\n", " )\n", " result = pd.DataFrame(rows)\n", " result[\"Rising EPS / falling price\"] = (result[\"EPS revision %\"] > 0) & (result[\"Price return %\"] < 0)\n", " return result\n", "\n", "\n", "OUTPUT = \"sdk_earnings_revisions.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 }