{ "cells": [ { "cell_type": "markdown", "id": "cell-000", "metadata": {}, "source": [ "# Lesson 8B — Capstone: An Enterprise Actuarial Rating Platform\n", "\n", "**Companion notebook** for the capstone. Prototype the platform end to end here, check\n", "each stage, then move the code into your `rating_platform` package:\n", "\n", "1. **Model**: the Course 3 champion, trained, validated, and saved with a manifest\n", "2. **Rating**: liability from the model (vectorized), collision and comprehensive from the manual\n", "3. **Service**: a FastAPI app that validates, prices, and stores quotes in a database\n", "4. **Monitoring**: drift of the incoming quotes against the in-force book\n", "\n", "The portal (Lesson 5), the containers (Lesson 7), and CI/CD (Lesson 8A) wrap around these." ] }, { "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", "import hashlib\n", "import json\n", "import secrets\n", "import tempfile\n", "import tomllib\n", "from datetime import datetime, timezone\n", "from decimal import ROUND_HALF_UP, Decimal\n", "from pathlib import Path\n", "from typing import Annotated, Any, Literal, Self\n", "\n", "import joblib\n", "import numpy as np\n", "import pandas as pd\n", "import sklearn\n", "from fastapi import Depends, FastAPI, Header, HTTPException, Request, status\n", "from fastapi.testclient import TestClient\n", "from pydantic import BaseModel, BeforeValidator, ConfigDict, Field, ValidationError, model_validator\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", "from sqlalchemy import JSON, String, create_engine, func, select\n", "from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column\n", "\n", "MOTOR = Path(\"../data/motor\")\n", "WORK = Path(tempfile.mkdtemp())\n", "EDGES = [16, 21, 25, 30, 40, 50, 65, 75, 121]\n", "LABELS = [\"17-20\", \"21-24\", \"25-29\", \"30-39\", \"40-49\", \"50-64\", \"65-74\", \"75+\"]\n", "TREND, TARGET_LR = 1.04 ** 1.5, 0.68\n", "\n", "\n", "def money(x):\n", " return float(Decimal(repr(float(x))).quantize(Decimal(\"0.01\"), rounding=ROUND_HALF_UP))" ] }, { "cell_type": "markdown", "id": "cell-002", "metadata": {}, "source": [ "## Stage 1 — the champion model and its manifest (from Lesson 8A)" ] }, { "cell_type": "code", "execution_count": null, "id": "cell-003", "metadata": {}, "outputs": [], "source": [ "def log_miles(m):\n", " return np.log(m / 12_000)\n", "\n", "\n", "def glm(columns, family):\n", " cats = [c for c in (\"vehicle_style\", \"age_band\") if c in columns]\n", " steps = [(\"cat\", OneHotEncoder(drop=[{\"vehicle_style\": \"Sedan\", \"age_band\": \"40-49\"}[c] for c in cats],\n", " handle_unknown=\"ignore\", sparse_output=False), cats),\n", " (\"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", "class PurePremiumModel:\n", " def _prep(self, frame):\n", " return frame.assign(age_band=pd.cut(frame[\"driver_age\"], EDGES, right=False, labels=LABELS).astype(str))\n", "\n", " def fit(self, frame):\n", " f = self._prep(frame)\n", " self.freq = glm([\"vehicle_style\", \"age_band\", \"urban\", \"annual_mileage\"], PoissonRegressor)\n", " self.freq.fit(f, f[\"n_liab\"] / f[\"exposure\"], glm__sample_weight=f[\"exposure\"])\n", " c = f[f[\"n_liab\"] > 0]\n", " self.sev = glm([\"vehicle_style\", \"urban\"], GammaRegressor)\n", " self.sev.fit(c, c[\"inc_liab\"] / c[\"n_liab\"], glm__sample_weight=c[\"n_liab\"])\n", " self.fs = f[\"n_liab\"].sum() / (self.freq.predict(f) * f[\"exposure\"]).sum()\n", " self.ss = c[\"inc_liab\"].sum() / (self.sev.predict(c) * c[\"n_liab\"]).sum()\n", " return self\n", "\n", " def pure_premium(self, frame):\n", " f = self._prep(frame)\n", " return self.freq.predict(f) * self.fs * self.sev.predict(f) * self.ss\n", "\n", "\n", "frame = pd.read_parquet(MOTOR / \"model_frame.parquet\")\n", "ARTIFACT = WORK / \"champion\"\n", "ARTIFACT.mkdir()\n", "joblib.dump(PurePremiumModel().fit(frame), ARTIFACT / \"model.joblib\")\n", "digest = hashlib.sha256((ARTIFACT / \"model.joblib\").read_bytes()).hexdigest()\n", "(ARTIFACT / \"manifest.json\").write_text(json.dumps({\"version\": f\"glm-{digest[:10]}\", \"sha256\": digest,\n", " \"sklearn_version\": sklearn.__version__}))\n", "print(\"artifact\", json.loads((ARTIFACT / \"manifest.json\").read_text())[\"version\"])" ] }, { "cell_type": "markdown", "id": "cell-004", "metadata": {}, "source": [ "## Stage 2 — the contract and the manual" ] }, { "cell_type": "code", "execution_count": null, "id": "cell-005", "metadata": {}, "outputs": [], "source": [ "Style = Literal[\"Sedan\", \"SUV\", \"Hatchback\", \"Pickup\", \"Minivan\", \"Sports\"]\n", "\n", "\n", "def _split(v):\n", " return v.split(\"|\") if isinstance(v, str) else v\n", "\n", "\n", "def _int(v):\n", " return int(float(v)) if isinstance(v, str) and v.replace(\".\", \"\", 1).isdigit() else v\n", "\n", "\n", "Deductible = Annotated[Literal[250, 500, 1000], BeforeValidator(_int)]\n", "\n", "\n", "class QuoteRequest(BaseModel):\n", " model_config = ConfigDict(frozen=True, extra=\"forbid\")\n", " driver_age: float = Field(ge=16, le=110)\n", " vehicle_style: Style\n", " vehicle_value: float = Field(gt=0, le=500_000)\n", " annual_mileage: float = Field(gt=0, le=150_000)\n", " urban: bool\n", " coverages: Annotated[frozenset[Literal[\"LIAB\", \"COLL\", \"COMP\"]], BeforeValidator(_split)]\n", " coll_deductible: Deductible | None = None\n", " comp_deductible: Deductible | None = None\n", "\n", " @model_validator(mode=\"after\")\n", " def _rules(self) -> Self:\n", " if \"LIAB\" not in self.coverages:\n", " raise ValueError(\"liability (LIAB) is compulsory\")\n", " for cov, ded in ((\"COLL\", self.coll_deductible), (\"COMP\", self.comp_deductible)):\n", " if (cov in self.coverages) != (ded is not None):\n", " raise ValueError(f\"{cov} deductible must be given if and only if {cov} is bought\")\n", " return self\n", "\n", "\n", "manual = tomllib.loads((MOTOR / \"rating_manual.toml\").read_text())\n", "\n", "\n", "def band(value, table):\n", " return next(f for u, f in zip(table[\"upper\"], table[\"factor\"]) if value < u)\n", "\n", "\n", "def coll_premium(r):\n", " rate = (manual[\"base_rate\"][\"COLL\"] * manual[\"style\"][r.vehicle_style] * band(r.driver_age, manual[\"driver_age\"])\n", " * band(r.annual_mileage, manual[\"annual_mileage\"]))\n", " return money(rate * r.vehicle_value / 30_000 * manual[\"deductible\"][str(r.coll_deductible)])\n", "\n", "\n", "def comp_premium(r):\n", " return money(manual[\"base_rate\"][\"COMP\"] * band(r.annual_mileage, manual[\"annual_mileage\"]) * r.vehicle_value\n", " / 30_000 * manual[\"deductible\"][str(r.comp_deductible)])" ] }, { "cell_type": "markdown", "id": "cell-006", "metadata": {}, "source": [ "---\n", "## Your turn 1 — rate many vehicles with one model call\n", "\n", "Write `quote_many(model, requests)`: build one DataFrame of the requests' rating\n", "variables, call `model.pure_premium` **once**, and turn each pure premium into a\n", "liability premium: pure × `TREND` ÷ `TARGET_LR`, rounded with `money`. Add COLL and COMP\n", "(from the functions above) for the vehicles that bought them. Return a list of\n", "`{coverage: premium}` dicts." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-007", "metadata": {}, "outputs": [], "source": [ "def quote_many(model, requests):\n", " ... # your code here\n", "\n", "\n", "base = QuoteRequest(driver_age=45, vehicle_style=\"Sedan\", vehicle_value=30_000, annual_mileage=12_000, urban=False,\n", " coverages=\"LIAB|COLL|COMP\", coll_deductible=500, comp_deductible=500)\n", "young = base.model_copy(update={\"driver_age\": 22, \"vehicle_style\": \"Sports\", \"vehicle_value\": 45_000,\n", " \"annual_mileage\": 16_000, \"urban\": True, \"coll_deductible\": 1000,\n", " \"comp_deductible\": 250})\n", "champion = joblib.load(ARTIFACT / \"model.joblib\")\n", "quotes = quote_many(champion, [base, young])\n", "print(quotes)\n", "check(\"base risk\", None if quotes is None else list(quotes[0].values()), [726.73, 576.85, 158.35])\n", "check(\"young urban sports car, LIAB\", None if quotes is None else quotes[1][\"LIAB\"], 4648.59)" ] }, { "cell_type": "markdown", "id": "cell-008", "metadata": {}, "source": [ "## Stage 3 — the quote log" ] }, { "cell_type": "code", "execution_count": null, "id": "cell-009", "metadata": {}, "outputs": [], "source": [ "class Base(DeclarativeBase):\n", " pass\n", "\n", "\n", "class QuoteRecord(Base):\n", " __tablename__ = \"quotes\"\n", " id: Mapped[int] = mapped_column(primary_key=True)\n", " created_at: Mapped[datetime] = mapped_column(default=lambda: datetime.now(timezone.utc))\n", " request: Mapped[dict[str, Any]] = mapped_column(JSON)\n", " premiums: Mapped[dict[str, float]] = mapped_column(JSON)\n", " total_premium: Mapped[float]\n", " model_version: Mapped[str] = mapped_column(String(40))" ] }, { "cell_type": "markdown", "id": "cell-010", "metadata": {}, "source": [ "## Your turn 2 — the service\n", "\n", "Complete `create_app(artifact, database_url, api_key)`. At startup it loads the model\n", "(checking the SHA-256 against the manifest) and creates the tables. Write:\n", "- `POST /quotes/batch` (API key required; 1 to 1,000 `QuoteRequest`s): price them with\n", " `quote_many`, store one `QuoteRecord` each **in one transaction**, and return\n", " `{\"quote_ids\": [...], \"total_premium\": ...}` with status 201\n", "- `GET /quotes/count`: `{\"quotes\": n}`" ] }, { "cell_type": "code", "execution_count": null, "id": "cell-011", "metadata": {}, "outputs": [], "source": [ "def create_app(artifact, database_url, api_key):\n", " manifest = json.loads((artifact / \"manifest.json\").read_text())\n", " if hashlib.sha256((artifact / \"model.joblib\").read_bytes()).hexdigest() != manifest[\"sha256\"]:\n", " raise RuntimeError(\"model file does not match its manifest\")\n", " model = joblib.load(artifact / \"model.joblib\")\n", " engine = create_engine(database_url)\n", " Base.metadata.create_all(engine)\n", " app = FastAPI(title=\"Motor Rating Platform (prototype)\")\n", " app.state.engine = engine\n", "\n", " def session():\n", " with Session(engine) as s:\n", " yield s\n", "\n", " def require_key(x_api_key: Annotated[str | None, Header()] = None):\n", " if x_api_key is None or not secrets.compare_digest(x_api_key, api_key):\n", " raise HTTPException(status.HTTP_401_UNAUTHORIZED, \"missing or invalid X-API-Key\")\n", "\n", " ... # your code here\n", "\n", " return app\n", "\n", "\n", "KEY = \"capstone-key\"\n", "client = TestClient(create_app(ARTIFACT, f\"sqlite:///{WORK / 'platform.db'}\", KEY))" ] }, { "cell_type": "markdown", "id": "cell-012", "metadata": {}, "source": [ "## Your turn 3 — a day of new business\n", "\n", "Validate every row of `quote_requests.csv` (drop the `quote_id` column; empty text →\n", "`None`), send the valid ones in one batch, and store the response. Then check that a\n", "request without the key is refused and that the database holds every quote." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-013", "metadata": {}, "outputs": [], "source": [ "raw = pd.read_csv(MOTOR / \"quote_requests.csv\", dtype=str, keep_default_na=False)\n", "valid, response, refused, stored = [], None, None, None\n", "... # your code here\n", "check(\"valid requests\", len(valid), 963)\n", "check(\"total premium quoted\", None if response is None else response[\"total_premium\"], 1577722.41)\n", "check(\"no key\", refused, 401)\n", "check(\"quotes stored\", stored, 963)" ] }, { "cell_type": "markdown", "id": "cell-014", "metadata": {}, "source": [ "## Your turn 4 — monitoring from the log\n", "\n", "Read the stored requests back **from the database** (`select(QuoteRecord.request)`),\n", "and compute the PSI of `driver_age` (manual bands 16, 25, 30, 65, 75, 121) against the\n", "2025 in-force book." ] }, { "cell_type": "code", "execution_count": null, "id": "cell-015", "metadata": {}, "outputs": [], "source": [ "book = pd.read_csv(MOTOR / \"book_2025.csv\")\n", "age_psi = None\n", "... # your code here\n", "check(\"driver age PSI\", None if age_psi is None else round(age_psi, 4), 0.0061)" ] }, { "cell_type": "markdown", "id": "cell-016", "metadata": {}, "source": [ "## Next: package it\n", "\n", "Move each stage into its module (`model.py`, `rating.py`, `schemas.py`, `db.py`,\n", "`api.py`, `monitoring.py`, `portfolio.py`, `dashboard.py`), read every setting from the\n", "environment (`settings.py`), write the tests, and wrap it:\n", "\n", "```bash\n", "uv run pytest --cov # tests, coverage gate\n", "docker compose up --build # PostgreSQL + API + portal\n", "```\n", "\n", "and push: the CI workflow from Lesson 8A builds and smoke-tests the images." ] } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python" } }, "nbformat": 4, "nbformat_minor": 5 }