{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# What does a put spread protect at expiry?\n", "Select strikes from an observed option chain, use executable sides of the quote and calculate the protected payoff per underlying share.\n", "SPY observed chain, expiry 2026-11-02; reference daily close 2026-10-30. Strike targets are 97% and 90% of the reference. Entry uses long ask minus short bid. The chart is an expiry scenario calculation, before fees, financing and assignment effects." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "\"\"\"Select strikes from an observed option chain, use executable sides of the quote and calculate the protected payoff per underlying share.\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 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", " dates = payload(qj.domains.call(\"derivatives.options.get_options_expirations\", provider=\"cboe\", symbol=\"SPY\"))\n", " dates = dates[\"expirations\"] if isinstance(dates, dict) else dates\n", " available = sorted((str(day)[:10] for day in dates if str(day)[:10] >= AS_OF.isoformat()))\n", " if not available:\n", " raise ValueError(\"No nonexpired SPY option expiration\")\n", " return {\n", " \"chain\": qj.domains.call(\n", " \"derivatives.options.get_options_chain\", provider=\"cboe\", symbol=\"SPY\", expiration_date=available[0]\n", " ),\n", " \"spy_spot\": qj.domains.call(\n", " \"equity.pricing.get_historical_prices\",\n", " provider=\"eod\",\n", " symbols=[\"SPY\"],\n", " start_date=(AS_OF - timedelta(days=14)).isoformat(),\n", " end_date=(AS_OF - timedelta(days=1)).isoformat(),\n", " frequency=\"1d\",\n", " ),\n", " }\n", "\n", "\n", "def analyze(raw):\n", " reference = history(raw[\"spy_spot\"], \"SPY\", adjusted=False)[\"price\"].iloc[-1]\n", " options = payload(raw[\"chain\"])[\"options\"]\n", " puts = [\n", " row\n", " for row in options\n", " if row[\"option_type\"] == \"put\"\n", " and all(pd.notna(row.get(key)) for key in [\"strike\", \"bid\", \"ask\"])\n", " and decimal(row[\"ask\"]) > 0\n", " and Decimal(0) <= decimal(row[\"bid\"]) <= decimal(row[\"ask\"])\n", " ]\n", " if not puts:\n", " raise ValueError(\"No usable observed put quotes\")\n", " long = min(puts, key=lambda row: abs(decimal(row[\"strike\"]) - reference * Decimal(\".97\")))\n", " lower = [row for row in puts if decimal(row[\"strike\"]) < decimal(long[\"strike\"])]\n", " if not lower:\n", " raise ValueError(\"No lower strike for the put spread\")\n", " short = min(lower, key=lambda row: abs(decimal(row[\"strike\"]) - reference * Decimal(\".90\")))\n", " cost = decimal(long[\"ask\"]) - decimal(short[\"bid\"])\n", " width = decimal(long[\"strike\"]) - decimal(short[\"strike\"])\n", " if not Decimal(0) <= cost <= width:\n", " raise ValueError(\"Observed quotes imply an invalid put-spread entry cost\")\n", " rows = []\n", " for scenario in np.linspace(-0.15, 0.15, 121):\n", " terminal = reference * (Decimal(1) + Decimal(str(scenario)))\n", " payoff = max(decimal(long[\"strike\"]) - terminal, Decimal(0)) - max(\n", " decimal(short[\"strike\"]) - terminal, Decimal(0)\n", " )\n", " rows.append(\n", " {\n", " \"Underlying move %\": 100 * scenario,\n", " \"Unprotected return %\": 100 * ratio(terminal - reference, reference),\n", " \"With put spread %\": 100 * ratio(terminal - reference + payoff - cost, reference),\n", " }\n", " )\n", " result = pd.DataFrame(rows)\n", " fig, (ax,) = canvas(\"SPY \u00b7 protection has a price and a limit\", \"Expiry return % \u00b7 per underlying share\")\n", " for column in [\"Unprotected return %\", \"With put spread %\"]:\n", " ax.plot(result[\"Underlying move %\"], result[column], label=column)\n", " ax.axhline(0, color=\"#afc1dc\", lw=0.8)\n", " ax.set_xlabel(\"Underlying price scenario at option expiry %\")\n", " legend(ax)\n", " result[\"Entry cost / reference %\"] = 100 * ratio(cost, reference)\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 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", "OUTPUT = \"sdk_option_overlay.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 }