{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Does the short-interest denominator matter?\n", "Compare FINRA\u2019s reported days to cover with a clearly defined volume denominator aligned to the settlement date.\n", "AAPL short-interest settlement: 2026-10-15. FINRA\u2019s published denominator and an EOD trailing 20-session denominator can differ. Missing coverage raises an error." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "\"\"\"Compare FINRA\u2019s reported days to cover with a clearly defined volume denominator aligned to the settlement date.\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", " short = qj.finra.get_short_interest(symbol=\"AAPL\")\n", " settlement = frame(short, \"settlement_date\").iloc[-1][\"settlement_date\"]\n", " return {\n", " \"short\": short,\n", " \"short_prices\": qj.eod.get_historical_prices(\n", " symbols=[\"AAPL\"],\n", " start_date=(settlement - pd.Timedelta(days=120)).date().isoformat(),\n", " end_date=settlement.date().isoformat(),\n", " frequency=\"1d\",\n", " ),\n", " }\n", "\n", "\n", "def analyze(raw):\n", " short = frame(raw[\"short\"], \"settlement_date\").iloc[-1]\n", " prices = history(raw[\"short_prices\"], \"AAPL\")\n", " volume = prices.loc[prices.index <= short[\"settlement_date\"], \"volume\"].tail(20)\n", " if len(volume) < 20:\n", " raise ValueError(\"Need 20 sessions before the short-interest settlement date\")\n", " average_volume = sum(volume, Decimal(0)) / Decimal(len(volume))\n", " calculated = ratio(short[\"current_short_position\"], average_volume)\n", " result = pd.DataFrame(\n", " {\n", " \"Denominator\": [\"FINRA reported\", \"EOD trailing 20 sessions\"],\n", " \"Days to cover\": [float(short[\"days_to_cover\"]), calculated],\n", " }\n", " )\n", " title = \"AAPL \u00b7 short interest as of \" + short[\"settlement_date\"].strftime(\"%Y-%m-%d\")\n", " fig, (ax,) = canvas(title, \"Short position / average daily volume\")\n", " bars = ax.bar(result[\"Denominator\"], result[\"Days to cover\"], color=[BLUE, PINK], width=0.55)\n", " ax.bar_label(bars, fmt=\"%.2f\", padding=6, color=\"#e9eef8\")\n", " ax.set_ylim(0, result[\"Days to cover\"].max() * 1.2)\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", "OUTPUT = \"sdk_short_liquidity.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 }