Lesson 15.1: From Notebook to Decision Engine: Serving a Causal Model with FastAPI

Written lesson

Why deploy a causal model?

A causal estimate in a notebook informs one report. A causal model behind an API informs every decision: a caseworker deciding whether to offer a retraining subsidy, a credit officer choosing a loan offer, a ministry dashboard simulating coverage. The capstone architecture has three layers:

Text

data/ (Polars + DuckDB)  ->  notebooks/ (DoubleML, causal forest)  ->  backend/ (FastAPI)  ->  frontend/ (Next.js)

This lesson builds the middle: a trained model saved to disk and a small, tested web service that serves it. Work in a GitHub Codespace (or locally) for the files; the training cell also runs in Colab.

Python · Colab cell

!pip install -q numpy pandas scikit-learn joblib fastapi httpx pytest
import json, os, numpy as np, pandas as pd, joblib
from sklearn.ensemble import GradientBoostingRegressor, GradientBoostingClassifier
rng = np.random.default_rng(15)
os.makedirs("backend/models", exist_ok=True)

Train and save the model

We use the retraining-subsidy world from Module 8 with three inputs a caseworker can actually supply. The model is a DR-learner: compute doubly robust scores (Module 7) with cross-fitting, then regress them on the features. It is simple, transparent and works with any regression library.

Python · Colab cell

n = 12_000
X = pd.DataFrame({
    "poverty_index": rng.normal(0, 1, n),
    "years_education": rng.integers(8, 21, n),
    "unemployment_duration_months": rng.gamma(2, 4, n).round(),
})
tau = 300 + 120 * np.minimum(X.unemployment_duration_months, 24) - 80 * (X.years_education - 12) + 300 * X.poverty_index
p = 1 / (1 + np.exp(-(-0.3 + 0.05 * X.unemployment_duration_months + 0.3 * X.poverty_index)))
d = rng.binomial(1, p)
y = 28_000 - 2_000 * X.poverty_index + 900 * (X.years_education - 12) + d * tau + rng.normal(0, 2_000, n)

# Doubly robust scores with 2-fold cross-fitting
folds = rng.integers(0, 2, n); psi = np.zeros(n)
for k in (0, 1):
    tr, te = folds != k, folds == k
    mu1 = GradientBoostingRegressor().fit(X[tr & (d == 1)], y[tr & (d == 1)])
    mu0 = GradientBoostingRegressor().fit(X[tr & (d == 0)], y[tr & (d == 0)])
    e = np.clip(GradientBoostingClassifier().fit(X[tr], d[tr]).predict_proba(X[te])[:, 1], 0.02, 0.98)
    m1, m0 = mu1.predict(X[te]), mu0.predict(X[te])
    psi[te] = m1 - m0 + d[te] * (y[te] - m1) / e - (1 - d[te]) * (y[te] - m0) / (1 - e)

cate_model = GradientBoostingRegressor(n_estimators=200, max_depth=2, learning_rate=0.05).fit(X, psi)
print(f"Estimated ATE {psi.mean():,.0f} (true {tau.mean():,.0f}); "
      f"corr(predicted CATE, true CATE) = {np.corrcoef(cate_model.predict(X), tau)[0, 1]:.2f}")

artifact = {"model": cate_model, "features": list(X.columns), "version": "2026.11-dr-gbm",
            "ate": float(psi.mean()), "ate_se": float(psi.std() / np.sqrt(n))}
joblib.dump(artifact, "backend/models/cate_model.joblib")
print("Saved backend/models/cate_model.joblib")

The why

Save metadata with the model: feature order, a version string, the training ATE and its standard error. When a decision is questioned six months later, you must be able to say exactly which model made it.

The FastAPI decision engine

FastAPI turns Python functions into a REST API with automatic input validation (via Pydantic) and interactive documentation at /docs. The decision rule is explicit and configurable: recommend the subsidy when the expected gain exceeds its cost.

Python · Colab cell

%%writefile backend/app.py
from pathlib import Path
import joblib
import pandas as pd
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field

ARTIFACT = joblib.load(Path(__file__).parent / "models" / "cate_model.joblib")
SUBSIDY_COST_CAD = 1_200.0

app = FastAPI(title="Causal Decision Engine", version=ARTIFACT["version"])
app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"])


class CitizenProfile(BaseModel):
    poverty_index: float = Field(..., ge=-4, le=4, description="Standardised poverty index")
    years_education: int = Field(..., ge=0, le=25)
    unemployment_duration_months: int = Field(..., ge=0, le=120)


@app.get("/health")
def health():
    return {"status": "ok", "model_version": ARTIFACT["version"]}


@app.post("/predict_impact")
def predict_impact(profile: CitizenProfile):
    row = pd.DataFrame([profile.model_dump()])[ARTIFACT["features"]]
    lift = float(ARTIFACT["model"].predict(row)[0])
    return {
        "expected_wage_lift_cad": round(lift, 2),
        "subsidy_cost_cad": SUBSIDY_COST_CAD,
        "net_benefit_cad": round(lift - SUBSIDY_COST_CAD, 2),
        "recommend_intervention": bool(lift > SUBSIDY_COST_CAD),
        "average_effect_cad": round(ARTIFACT["ate"], 2),
        "model_version": ARTIFACT["version"],
    }

Run it from a terminal in the repository root, then open http://localhost:8000/docs:

Terminal

uvicorn backend.app:app --reload --port 8000

Automated tests

Tests protect the engine when data, models or code change. FastAPI's TestClient calls the API in memory, without starting a server.

Python · Colab cell

%%writefile tests/test_api.py
from fastapi.testclient import TestClient
from backend.app import app

client = TestClient(app)


def test_health():
    assert client.get("/health").json()["status"] == "ok"


def test_predict_impact_returns_valid_recommendation():
    payload = {"poverty_index": 1.5, "years_education": 10, "unemployment_duration_months": 18}
    r = client.post("/predict_impact", json=payload)
    assert r.status_code == 200
    data = r.json()
    assert isinstance(data["recommend_intervention"], bool)
    assert -10_000 < data["expected_wage_lift_cad"] < 50_000      # no impossible effects


def test_long_term_unemployed_gain_more():
    base = {"poverty_index": 0.0, "years_education": 12}
    short = client.post("/predict_impact", json=base | {"unemployment_duration_months": 1}).json()
    long = client.post("/predict_impact", json=base | {"unemployment_duration_months": 20}).json()
    assert long["expected_wage_lift_cad"] > short["expected_wage_lift_cad"]


def test_rejects_invalid_input():
    r = client.post("/predict_impact", json={"poverty_index": 0, "years_education": -3, "unemployment_duration_months": 5})
    assert r.status_code == 422

Python · Colab cell

import pytest
exit_code = pytest.main(["-q", "tests/test_api.py"])
print("All tests passed" if exit_code == 0 else f"pytest exit code {exit_code}")
assert exit_code == 0

Exercise

Add a /predict_batch endpoint that accepts a list of profiles and a budget_share (for example 0.3) and returns the profiles ranked by expected net benefit, with recommend_intervention set only for the top share. Write one test for it.

Checklist and recap

My model is saved with its feature list, version and training ATE.
The API validates inputs and makes the decision rule explicit.
Tests check structure, plausible ranges, an expected direction and invalid inputs.