"""Verify filing dates, measure earnings-to-cash conversion and calculate inflation-adjusted free-cash-flow growth. Research example. Live observations can change. """ from datetime import UTC, datetime from decimal import Decimal from textwrap import fill import matplotlib.pyplot as plt import numpy as np import pandas as pd AS_OF = datetime.now(UTC).date() NAVY, BLUE, PINK = "#132244", "#8fb9ff", "#e7a2cc" def fetch(qj): return { "bls_cpi": qj.domains.call( "macro.us.get_cpi", provider="bls", start_year=2020, end_year=2025, category="all", seasonally_adjusted=True ), "income": qj.domains.call( "equity.fundamentals.get_income_statement", provider="fmp", symbol="AAPL", period="annual", limit=5 ), "cash": qj.domains.call( "equity.fundamentals.get_cash_flow_statement", provider="fmp", symbol="AAPL", period="annual", limit=5 ), "filings": qj.domains.call( "regulatory.filings.get_company_filings", provider="sec", symbol="AAPL", form_type="10-K", limit=12 ), } def analyze(raw): joined = verified_income(raw).merge( frame(raw["cash"])[["date", "operatingCashFlow", "freeCashFlow"]], on="date", validate="one_to_one", ) if joined.empty: raise ValueError("No matched cash-flow and filing observations") for source, name in [("operatingCashFlow", "Operating cash / earnings"), ("freeCashFlow", "Free cash / earnings")]: joined[name] = [ 100 * ratio(cash, earnings) for cash, earnings in zip(joined[source], joined["netIncome"], strict=True) ] labels = joined["accepted"].dt.strftime("%Y-%m-%d") fig, (ax, bottom) = canvas("AAPL ยท earnings, cash and purchasing power", "% of reported net income", panels=2) x = np.arange(len(joined)) ax.bar(x - 0.18, joined["Operating cash / earnings"], 0.36, label="Operating cash / earnings") ax.bar(x + 0.18, joined["Free cash / earnings"], 0.36, label="Free cash / earnings") ax.axhline(100, color="#afc1dc", lw=0.8, linestyle="--") ax.set_xticks(x, labels, rotation=20) ax.set_xlabel("SEC acceptance date") legend(ax) cpi = bls_cpi(raw["bls_cpi"]) def fiscal_cpi(end): points = cpi.loc[(cpi.index > end - pd.DateOffset(years=1)) & (cpi.index <= end)] if len(points) != 12: raise ValueError("Need twelve matching fiscal-year CPI observations") return decimal(str(points.mean())) real_cash = pd.Series( [decimal(cash) / fiscal_cpi(end) for cash, end in zip(joined["freeCashFlow"], joined["date"], strict=True)], index=joined["date"], ) growth = (100 * change(real_cash)).dropna() joined["Real FCF growth %"] = joined["date"].map(growth) bottom.bar(growth.index.year.astype(str), growth, color=PINK) bottom.set_ylabel("CPI-adjusted FCF growth %") bottom.axhline(0, color="#afc1dc", lw=0.8) return joined[["date", "accepted", "Operating cash / earnings", "Free cash / earnings", "Real FCF growth %"]], fig def payload(response): if hasattr(response, "to_dict"): response = response.to_dict() while isinstance(response, dict): if response.get("error") or response.get("error_code") or response.get("success") is False: raise ValueError("Provider request failed") if "data" in response: response = response["data"] elif "value" in response: response = response["value"] else: break return response def frame(response, date="date"): rows = payload(response) if not isinstance(rows, list) or not rows: raise ValueError("No provider observations returned") result = pd.DataFrame(rows) if date: result[date] = pd.to_datetime(result[date], utc=True) result = result.sort_values(date) return result def decimal(value): if pd.isna(value): raise ValueError("Required monetary value or quantity is missing") return Decimal(str(value)) def ratio(numerator, denominator): denominator = decimal(denominator) if denominator == 0: raise ValueError("Required denominator is zero") return float(decimal(numerator) / denominator) def change(values, periods=1): previous = values.shift(periods) return pd.Series( [ratio(x, y) - 1 if pd.notna(x) and pd.notna(y) else np.nan for x, y in zip(values, previous, strict=True)], index=values.index, ) def macro(response, column): values = frame(response).set_index("date")[column] values = pd.to_numeric(values, errors="coerce").dropna() if values.empty: raise ValueError(f"No numeric observations for {column}") return values def verified_income(raw): income = frame(raw["income"]) filings = frame(raw["filings"], "reportDate") filings = filings[["reportDate", "acceptanceDateTime", "filingDate"]].copy() filings["accepted"] = pd.to_datetime(filings["acceptanceDateTime"], utc=True) return income.merge( filings[["reportDate", "accepted"]], left_on="date", right_on="reportDate", validate="one_to_one", ).sort_values("date") def canvas(title, ylabel, panels=1): plt.rcParams.update( { "figure.facecolor": NAVY, "axes.facecolor": NAVY, "savefig.facecolor": NAVY, "text.color": "#e9eef8", "axes.labelcolor": "#afc1dc", "xtick.color": "#afc1dc", "ytick.color": "#afc1dc", "axes.edgecolor": "#34486b", "grid.color": "#34486b", "font.size": 10, "axes.prop_cycle": plt.cycler(color=[BLUE, PINK, "#ba9dea"]), } ) fig, axes = plt.subplots(panels, 1, figsize=(9, 6.1), squeeze=False, layout="constrained") fig.suptitle(fill(title, width=58), fontsize=17, x=0.04, ha="left", weight="medium") for ax in axes.flat: ax.grid(axis="y", alpha=0.45) ax.set_axisbelow(True) ax.spines[["top", "right"]].set_visible(False) axes.flat[0].set_ylabel(ylabel) return fig, list(axes.flat) def legend(ax): ax.legend(frameon=False, labelcolor="#e9eef8", fontsize=9) def run(): from quantjourney.sdk import QuantJourneyAPI qj = QuantJourneyAPI.from_env() raw = fetch(qj) table, fig = analyze(raw) print(table.to_string(index=False)) fig.savefig(OUTPUT, dpi=180) plt.show() def bls_cpi(response): return macro(response, "cpi_all") OUTPUT = "sdk_cash_conversion.png" if __name__ == "__main__": run()