{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# How much revenue growth survives inflation?\n", "Deflate reported revenue with CPI averaged over the matching fiscal year, rather than comparing mismatched calendar dates.\n", "AAPL FY2022\u20132025. CPI is a broad US purchasing-power reference, not a company-specific price deflator. Current-vintage macro observations." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "\"\"\"Deflate reported revenue with CPI averaged over the matching fiscal year, rather than comparing mismatched calendar dates.\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", " \"income\": qj.fmp.get_income_statement(symbol=\"AAPL\", period=\"annual\", limit=5),\n", " \"bls_cpi\": qj.bls.get_cpi(start_year=2020, end_year=2025, category=\"all\", seasonally_adjusted=True),\n", " }\n", "\n", "\n", "def analyze(raw):\n", " income, cpi = frame(raw[\"income\"]), bls_cpi(raw[\"bls_cpi\"])\n", " rows = []\n", " for row in income.to_dict(\"records\"):\n", " end = row[\"date\"]\n", " start = end - pd.DateOffset(years=1)\n", " observations = cpi.loc[(cpi.index > start) & (cpi.index <= end)]\n", " if len(observations) < 12:\n", " continue\n", " rows.append(\n", " {\"Fiscal year\": end.year, \"Revenue\": decimal(row[\"revenue\"]), \"CPI\": decimal(str(observations.mean()))}\n", " )\n", " data = pd.DataFrame(rows).set_index(\"Fiscal year\")\n", " data[\"Deflated revenue\"] = data[\"Revenue\"] / data[\"CPI\"]\n", " data[\"Nominal growth %\"] = 100 * change(data[\"Revenue\"])\n", " data[\"CPI-adjusted growth %\"] = 100 * change(data[\"Deflated revenue\"])\n", " result = data[[\"Nominal growth %\", \"CPI-adjusted growth %\"]].dropna()\n", " if result.empty:\n", " raise ValueError(\"Insufficient matching fiscal-year CPI coverage\")\n", " fig, (ax,) = canvas(\"AAPL \u00b7 growth after inflation\", \"Annual revenue growth %\")\n", " x = np.arange(len(result))\n", " ax.bar(x - 0.18, result.iloc[:, 0], 0.36, label=\"Nominal\")\n", " ax.bar(x + 0.18, result.iloc[:, 1], 0.36, label=\"CPI-adjusted\")\n", " ax.set_xticks(x, result.index.astype(str))\n", " legend(ax)\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 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 bls_cpi(response):\n", " return macro(response, \"cpi_all\")\n", "\n", "\n", "OUTPUT = \"sdk_real_growth.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 }