{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Do earnings turn into cash?\n", "Verify filing dates, measure earnings-to-cash conversion and calculate inflation-adjusted free-cash-flow growth.\n", "AAPL FY2021\u20132025. Cash conversion uses matching fiscal periods and SEC acceptance dates. Real free-cash-flow growth uses fiscal-year average BLS CPI; current-vintage observations." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "\"\"\"Verify filing dates, measure earnings-to-cash conversion and calculate inflation-adjusted free-cash-flow growth.\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", " \"bls_cpi\": qj.domains.call(\n", " \"macro.us.get_cpi\", provider=\"bls\", start_year=2020, end_year=2025, category=\"all\", seasonally_adjusted=True\n", " ),\n", " \"income\": qj.domains.call(\n", " \"equity.fundamentals.get_income_statement\", provider=\"fmp\", symbol=\"AAPL\", period=\"annual\", limit=5\n", " ),\n", " \"cash\": qj.domains.call(\n", " \"equity.fundamentals.get_cash_flow_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", " }\n", "\n", "\n", "def analyze(raw):\n", " joined = verified_income(raw).merge(\n", " frame(raw[\"cash\"])[[\"date\", \"operatingCashFlow\", \"freeCashFlow\"]],\n", " on=\"date\",\n", " validate=\"one_to_one\",\n", " )\n", " if joined.empty:\n", " raise ValueError(\"No matched cash-flow and filing observations\")\n", " for source, name in [(\"operatingCashFlow\", \"Operating cash / earnings\"), (\"freeCashFlow\", \"Free cash / earnings\")]:\n", " joined[name] = [\n", " 100 * ratio(cash, earnings) for cash, earnings in zip(joined[source], joined[\"netIncome\"], strict=True)\n", " ]\n", " labels = joined[\"accepted\"].dt.strftime(\"%Y-%m-%d\")\n", " fig, (ax, bottom) = canvas(\"AAPL \u00b7 earnings, cash and purchasing power\", \"% of reported net income\", panels=2)\n", " x = np.arange(len(joined))\n", " ax.bar(x - 0.18, joined[\"Operating cash / earnings\"], 0.36, label=\"Operating cash / earnings\")\n", " ax.bar(x + 0.18, joined[\"Free cash / earnings\"], 0.36, label=\"Free cash / earnings\")\n", " ax.axhline(100, color=\"#afc1dc\", lw=0.8, linestyle=\"--\")\n", " ax.set_xticks(x, labels, rotation=20)\n", " ax.set_xlabel(\"SEC acceptance date\")\n", " legend(ax)\n", " cpi = bls_cpi(raw[\"bls_cpi\"])\n", "\n", " def fiscal_cpi(end):\n", " points = cpi.loc[(cpi.index > end - pd.DateOffset(years=1)) & (cpi.index <= end)]\n", " if len(points) != 12:\n", " raise ValueError(\"Need twelve matching fiscal-year CPI observations\")\n", " return decimal(str(points.mean()))\n", "\n", " real_cash = pd.Series(\n", " [decimal(cash) / fiscal_cpi(end) for cash, end in zip(joined[\"freeCashFlow\"], joined[\"date\"], strict=True)],\n", " index=joined[\"date\"],\n", " )\n", " growth = (100 * change(real_cash)).dropna()\n", " joined[\"Real FCF growth %\"] = joined[\"date\"].map(growth)\n", " bottom.bar(growth.index.year.astype(str), growth, color=PINK)\n", " bottom.set_ylabel(\"CPI-adjusted FCF growth %\")\n", " bottom.axhline(0, color=\"#afc1dc\", lw=0.8)\n", " return joined[[\"date\", \"accepted\", \"Operating cash / earnings\", \"Free cash / earnings\", \"Real FCF growth %\"]], 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 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", "def bls_cpi(response):\n", " return macro(response, \"cpi_all\")\n", "\n", "\n", "OUTPUT = \"sdk_cash_conversion.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 }