{ "cells": [ { "cell_type": "markdown", "id": "cell-000", "metadata": {}, "source": [ "# Lesson 8B — Capstone: A Production Pricing Engine\n", "\n", "**Companion notebook** for the capstone. Prototype each stage here, check it against the\n", "checkpoints, then move the code into your `pricing_engine` package:\n", "\n", "1. **Extract** the raw records and validate their keys\n", "2. **Features**: earned exposure, driver factors, spatial features, liability claims\n", "3. **Model** the pure premium (Lesson 8A)\n", "4. **Indicate** rate changes by segment, capped\n", "\n", "The A/B test and the report are in Lesson 8A and the slides." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-001", "metadata": {}, "outputs": [], "source": [ "def check(label, got, expected, tol=1e-6):\n", " \"\"\"Compare your answer with the expected value; prints ✅ or ❌ (never raises).\"\"\"\n", " import math\n", " try:\n", " if isinstance(expected, float) or isinstance(got, float):\n", " ok = math.isclose(float(got), float(expected), rel_tol=tol, abs_tol=tol)\n", " elif isinstance(expected, (list, tuple)) and expected and isinstance(expected[0], str):\n", " ok = list(got) == list(expected)\n", " elif hasattr(expected, \"__len__\") and not isinstance(expected, str):\n", " import numpy as np\n", " ok = np.allclose(np.asarray(got, dtype=float), np.asarray(expected, dtype=float), rtol=tol, atol=tol)\n", " else:\n", " ok = got == expected\n", " except Exception:\n", " ok = False\n", " print((\"✅ \" if ok else \"❌ \") + f\"{label}: got {got!r}, expected {expected!r}\")\n", "from pathlib import Path\n", "\n", "import geopandas as gpd\n", "import numpy as np\n", "import pandas as pd\n", "from sklearn.compose import ColumnTransformer\n", "from sklearn.linear_model import GammaRegressor, PoissonRegressor\n", "from sklearn.pipeline import Pipeline\n", "from sklearn.preprocessing import FunctionTransformer, OneHotEncoder\n", "\n", "MOTOR = Path(\"../data/motor\")\n", "GEO = Path(\"../data/geo/nc_geo.gpkg\")\n", "EXPOSURE_END = pd.Timestamp(\"2025-12-31\")\n", "AGE_EDGES = [16, 21, 25, 30, 40, 50, 65, 75, 120]\n", "AGE_LABELS = [\"17-20\", \"21-24\", \"25-29\", \"30-39\", \"40-49\", \"50-64\", \"65-74\", \"75+\"]" ] }, { "cell_type": "markdown", "id": "cell-002", "metadata": {}, "source": [ "## Stage 1 — extract and validate\n", "\n", "A duplicate primary key or an orphan foreign key stops the engine before anything is\n", "calculated." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-003", "metadata": {}, "outputs": [], "source": [ "TABLES = {\n", " \"policies\": ([\"policy_id\"], [\"inception_date\", \"expiry_date\", \"cancellation_date\"]),\n", " \"vehicles\": ([\"vehicle_id\"], []),\n", " \"persons\": ([\"person_id\"], [\"date_of_birth\", \"licence_date\"]),\n", " \"policy_drivers\": ([\"policy_id\", \"person_id\"], []),\n", " \"coverages\": ([\"vehicle_id\", \"coverage\"], []),\n", " \"claims\": ([\"claim_id\"], [\"accident_date\", \"report_date\"]),\n", "}\n", "tables = {}\n", "for name, (key, dates) in TABLES.items():\n", " t = pd.read_csv(MOTOR / f\"{name}.csv\", parse_dates=dates)\n", " assert not t.duplicated(key).any(), f\"duplicate keys in {name}\"\n", " tables[name] = t\n", " print(f\"{name:15s} {len(t):8,d} rows\")\n", "for child, col, parent, pkey in [(\"vehicles\", \"policy_id\", \"policies\", \"policy_id\"),\n", " (\"claims\", \"vehicle_id\", \"vehicles\", \"vehicle_id\")]:\n", " assert tables[child][col].isin(tables[parent][pkey]).all(), f\"orphans in {child}.{col}\"" ] }, { "cell_type": "markdown", "id": "cell-004", "metadata": {}, "source": [ "---\n", "## Your turn 1 — earned exposure\n", "\n", "Car-years from inception to the **earliest** of expiry, cancellation, and\n", "`EXPOSURE_END`, counting both end days, divided by 365.25. Never negative." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-005", "metadata": {}, "outputs": [], "source": [ "def earned_exposure(start, expiry, cancel, end):\n", " ... # your code here\n", "\n", "\n", "test = earned_exposure(pd.Series(pd.to_datetime([\"2024-01-01\", \"2024-01-01\"])),\n", " pd.Series(pd.to_datetime([\"2024-12-31\", \"2024-12-31\"])),\n", " pd.Series(pd.to_datetime([None, \"2024-03-31\"])), EXPOSURE_END)\n", "check(\"exposure: full leap year, canceled after Q1\",\n", " None if test is None else [round(x, 4) for x in test], [round(366 / 365.25, 4), round(91 / 365.25, 4)])" ] }, { "cell_type": "markdown", "id": "cell-006", "metadata": {}, "source": [ "## Your turn 2 — spatial features\n", "\n", "For each policy's garaging point: the county (nearest county polygon), whether it lies\n", "inside an incorporated place (`urban`), and the distance to the coastline in km.\n", "Project to `EPSG:32119` (meters) first." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-007", "metadata": {}, "outputs": [], "source": [ "def spatial_features(policies, geo_path):\n", " ... # your code here\n", "\n", "\n", "raleigh = spatial_features(pd.DataFrame({\"policy_id\": [\"RAL\"], \"latitude\": [35.7796],\n", " \"longitude\": [-78.6382]}), GEO)\n", "check(\"downtown Raleigh county\",\n", " None if raleigh is None else raleigh[\"county_name\"].iloc[0], \"Wake\")\n", "check(\"downtown Raleigh is urban\",\n", " None if raleigh is None else bool(raleigh[\"urban\"].iloc[0]), True)" ] }, { "cell_type": "markdown", "id": "cell-008", "metadata": {}, "source": [ "## Stage 2 — the model frame\n", "\n", "One row per vehicle per term. This cell uses your two functions." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-009", "metadata": {}, "outputs": [], "source": [ "frame = None\n", "if test is not None and raleigh is not None:\n", " pol, veh = tables[\"policies\"].copy(), tables[\"vehicles\"]\n", " pol[\"exposure\"] = earned_exposure(pol[\"inception_date\"], pol[\"expiry_date\"],\n", " pol[\"cancellation_date\"], EXPOSURE_END)\n", " pol[\"earned_fraction\"] = pol[\"exposure\"] * 365.25 / ((pol[\"expiry_date\"] - pol[\"inception_date\"]).dt.days + 1)\n", " pol[\"policy_year\"] = pol[\"inception_date\"].dt.year\n", " dd = (tables[\"policy_drivers\"].merge(tables[\"persons\"], on=\"person_id\")\n", " .merge(pol[[\"policy_id\", \"inception_date\"]], on=\"policy_id\"))\n", " dd[\"age\"] = (dd[\"inception_date\"] - dd[\"date_of_birth\"]).dt.days / 365.25\n", " dd[\"licensed\"] = (dd[\"inception_date\"] - dd[\"licence_date\"]).dt.days / 365.25\n", " drivers = dd.groupby(\"policy_id\").agg(n_drivers=(\"person_id\", \"size\"), driver_age=(\"age\", \"min\"),\n", " years_licensed=(\"licensed\", \"min\"))\n", " cov = tables[\"coverages\"]\n", " liab = cov[cov[\"coverage\"] == \"LIAB\"].set_index(\"vehicle_id\")[\"annual_premium\"].rename(\"liab_premium\")\n", " cl = tables[\"claims\"]\n", " cl = cl[(cl[\"coverage\"] == \"LIAB\") & (cl[\"accident_date\"] <= EXPOSURE_END)].assign(\n", " incurred=lambda c: c[\"paid\"] + c[\"outstanding\"])\n", " claims = cl.groupby(\"vehicle_id\").agg(n_liab=(\"claim_id\", \"size\"), inc_liab=(\"incurred\", \"sum\"))\n", " frame = (veh.merge(pol[[\"policy_id\", \"policyholder_id\", \"policy_year\", \"exposure\", \"earned_fraction\",\n", " \"latitude\", \"longitude\"]], on=\"policy_id\", validate=\"many_to_one\")\n", " .merge(spatial_features(pol, GEO), on=\"policy_id\", validate=\"many_to_one\")\n", " .merge(drivers, on=\"policy_id\", validate=\"many_to_one\")\n", " .join(liab, on=\"vehicle_id\").join(claims, on=\"vehicle_id\"))\n", " frame[[\"n_liab\", \"inc_liab\"]] = frame[[\"n_liab\", \"inc_liab\"]].fillna(0)\n", " frame[\"ep_liab\"] = frame[\"liab_premium\"] * frame[\"earned_fraction\"]\n", " frame[\"age_band\"] = pd.cut(frame[\"driver_age\"], AGE_EDGES, right=False, labels=AGE_LABELS).astype(str)\n", " frame = frame[frame[\"exposure\"] > 0].reset_index(drop=True)\n", "check(\"vehicle-terms\", None if frame is None else len(frame), 84346)\n", "check(\"car-years\", None if frame is None else round(frame[\"exposure\"].sum()), 72781)\n", "check(\"liability claims\", None if frame is None else int(frame[\"n_liab\"].sum()), 4565)" ] }, { "cell_type": "markdown", "id": "cell-010", "metadata": {}, "source": [ "## Stage 3 — the champion model (from Lesson 8A)\n", "\n", "The GLM won the A/B test, so the production model is the GLM refit on **all** five years." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-011", "metadata": {}, "outputs": [], "source": [ "def log_miles(m):\n", " return np.log(m / 12_000)\n", "\n", "\n", "def glm_pipe(columns, family):\n", " cats = [c for c in (\"vehicle_style\", \"age_band\") if c in columns]\n", " base = {\"vehicle_style\": \"Sedan\", \"age_band\": \"40-49\"}\n", " steps = [(\"cat\", OneHotEncoder(drop=[base[c] for c in cats], handle_unknown=\"ignore\",\n", " sparse_output=False), cats), (\"urban\", \"passthrough\", [\"urban\"])]\n", " if \"annual_mileage\" in columns:\n", " steps.append((\"miles\", FunctionTransformer(log_miles), [\"annual_mileage\"]))\n", " return Pipeline([(\"prep\", ColumnTransformer(steps)),\n", " (\"glm\", family(alpha=1e-4, solver=\"newton-cholesky\", tol=1e-8, max_iter=1000))])\n", "\n", "\n", "freq = sev = None\n", "if frame is not None:\n", " freq = glm_pipe([\"vehicle_style\", \"age_band\", \"urban\", \"annual_mileage\"], PoissonRegressor)\n", " freq.fit(frame, frame[\"n_liab\"] / frame[\"exposure\"], glm__sample_weight=frame[\"exposure\"])\n", " c = frame[frame[\"n_liab\"] > 0]\n", " sev = glm_pipe([\"vehicle_style\", \"urban\"], GammaRegressor)\n", " sev.fit(c, c[\"inc_liab\"] / c[\"n_liab\"], glm__sample_weight=c[\"n_liab\"])\n", " sev_scale = c[\"inc_liab\"].sum() / (sev.predict(c) * c[\"n_liab\"]).sum()\n", " print(f\"severity balance factor {sev_scale:.4f}\")" ] }, { "cell_type": "markdown", "id": "cell-012", "metadata": {}, "source": [ "## Your turn 3 — rate indications\n", "\n", "For the 2025 in-force book (`frame[frame.policy_year == 2025]`):\n", "\n", "indicated annual premium = pure premium × 1.04^1.5 ÷ 0.68\n", "\n", "Compare with the current annual premium (`liab_premium`) overall and by `urban`,\n", "`vehicle_style`, and `age_band`; cap each segment's change at ±25%. Store the overall\n", "indicated change and the urban segment's indicated and proposed changes." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-013", "metadata": {}, "outputs": [], "source": [ "overall = urban_ind = urban_prop = None\n", "if freq is not None:\n", " book = frame[frame[\"policy_year\"] == 2025].copy()\n", " ... # your code here\n", "check(\"overall indicated change\", None if overall is None else round(overall, 3), 0.143)\n", "check(\"urban indicated change\", None if urban_ind is None else round(urban_ind, 3), 0.336)\n", "check(\"urban proposed change\", None if urban_prop is None else round(urban_prop, 3), 0.25)" ] }, { "cell_type": "markdown", "id": "cell-014", "metadata": {}, "source": [ "## Next: package it\n", "\n", "Move each stage into its own module (`extract.py`, `features.py`, `models.py`,\n", "`evaluate.py`, `indicate.py`, `report.py`), put every assumption in `config.toml`, add\n", "a `cli.py` entry point, and write tests like the ones on the slides. Then:\n", "\n", "```bash\n", "uv run pricing-engine --config config.toml\n", "uv run pytest\n", "```" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python" } }, "nbformat": 4, "nbformat_minor": 5 }