Serving Causal Models with FastAPI

71.1 From estimates to a parameters file (40 min)

Concept. A model in production is a contract: inputs, outputs, and a version. For a causal simulator the "model" is small: 20 elasticities with standard errors, the prices and volumes they apply to, and a coupon curve. We store exactly that, plus where it came from.

First, the data the parameters are estimated from. Create data/make_delivery_data.py:

"""
Module 8 data: the inputs the simulator's parameters are estimated from. Provided.

    python data/make_delivery_data.py

Writes data/scenario/weekly_demand.csv (Module 6's capstone year) and data/scenario/coupon_experiment.csv
(Module 7's 40,000-customer randomised coupon test).
"""
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from causalkit.coupons import coupons  # noqa: E402
from causalkit.coupons import public as coupons_public  # noqa: E402
from causalkit.demand import demand, public  # noqa: E402

root = Path(__file__).parent
(root / "scenario").mkdir(exist_ok=True)
public(demand(seed=60)).write_csv(root / "scenario" / "weekly_demand.csv")
coupons_public(coupons(n=40_000, seed=640, randomized=True)).write_csv(root / "scenario" / "coupon_experiment.csv")
print("Wrote data/scenario/weekly_demand.csv and data/scenario/coupon_experiment.csv")

Then the builder. It reuses Module 6's DML and 2SLS and Module 7's DR-learner. Each pricing cell also records valid_change: the largest price move (95th percentile) that supplier costs actually caused in the data. That number comes back on Day 72. Create engine/parameters.py and engine/build.py:

"""
The causal parameters file: everything the simulator knows, in one versioned JSON file.

    {"version": "...", "created": "...", "sources": {...},
     "pricing": [{"segment", "category", "elasticity", "se", "margin", "price", "weekly_units",
                  "valid_change"}, ...],
     "coupons": {"customers", "campaigns_per_year", "margin", "shares": [...], "uplift": [...], "uplift_se": [...]},
     "assumptions": {"pull_forward": ...}}

elasticity: DML point estimate; se: the honest (2SLS) standard error (Module 6, Day 59).
valid_change: the size of price change that supplier costs actually produced in the data (95th percentile).
The 2SLS elasticity is learned from changes of about that size; beyond it the simulator extrapolates
and says so.
uplift: extra spend per customer in the base, per campaign, when the targeting model coupons the top
share of customers (Module 7, Day 65), with standard errors.
"""
import json
from pathlib import Path

REQUIRED_CELL = {"segment", "category", "elasticity", "se", "margin", "price", "weekly_units", "valid_change"}


def validate(p: dict) -> dict:
    """Fail loudly on a malformed file: a simulator should never run on half a parameter set."""
    for key in ["version", "pricing", "coupons", "assumptions"]:
        if key not in p:
            raise ValueError(f"parameters file has no '{key}'")
    for cell in p["pricing"]:
        missing = REQUIRED_CELL - cell.keys()
        if missing:
            raise ValueError(f"pricing cell {cell.get('segment')}/{cell.get('category')} is missing {sorted(missing)}")
        if cell["elasticity"] >= 0:
            raise ValueError(f"non-negative elasticity for {cell['segment']}/{cell['category']}: {cell['elasticity']}")
    c = p["coupons"]
    if not (len(c["shares"]) == len(c["uplift"]) == len(c["uplift_se"])):
        raise ValueError("coupon shares, uplift and uplift_se must have the same length")
    return p


def load_parameters(path: str | Path = "models/causal_parameters.json") -> dict:
    return validate(json.loads(Path(path).read_text()))


def save_parameters(p: dict, path: str | Path = "models/causal_parameters.json") -> None:
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    Path(path).write_text(json.dumps(validate(p), indent=2))

"""
Build the causal parameters file from data (Modules 6 and 7, packaged): run it whenever new data arrive.

    python -m engine.build data/scenario/weekly_demand.csv data/scenario/coupon_experiment.csv

Pricing: one elasticity per segment x category cell. DML gives the point estimate; 2SLS with the supplier
cost instrument gives the honest standard error (Module 6, Day 59).
Coupons: a DR-learner ranks customers (trained on half the experiment); doubly robust scores on the other
half give the extra spend per customer in the base when the top share is couponed (Module 7, Day 65).
"""
import hashlib
import sys
from datetime import date
from pathlib import Path

import doubleml as dml
import numpy as np
import pandas as pd
import pyfixest as pf
from econml.dr import DRLearner
from sklearn.model_selection import KFold, train_test_split

from causalkit.coupons import X_COLS
from causalkit.learners import clf, reg
from engine.parameters import save_parameters

DRIVERS = ["promo", "temperature", "holiday", "local_event", "competitor_index", "week"]
SHARES = [round(x, 1) for x in np.arange(0, 1.01, 0.1)]


def pricing_cells(demand: pd.DataFrame, recent_weeks: int = 13) -> list[dict]:
    g = demand.groupby(["store_id", "category"])
    demand = demand.assign(y=demand.log_units - g.log_units.transform("mean"),
                           d=demand.log_price - g.log_price.transform("mean"),
                           store=demand.store_id.str[1:].astype(int))
    last = demand.week.max()
    cells = []
    for (seg, cat), s in demand.groupby(["segment", "category"]):
        np.random.seed(58)
        data = dml.DoubleMLData(s[DRIVERS + ["y", "d", "store"]], "y", "d", DRIVERS, cluster_cols="store")
        learner = reg(n_estimators=500, num_leaves=31, min_child_samples=20)
        point = dml.DoubleMLPLR(data, ml_l=learner, ml_m=learner, n_folds=5).fit().coef[0]
        iv = pf.feols("log_units ~ promo | store_id + week | log_price ~ cost_index", s, vcov={"CRV1": "store_id"})
        first = pf.feols("log_price ~ cost_index + promo | store_id + week", s).coef()["cost_index"]
        cost_moves = first * (s.cost_index - s.groupby("store_id").cost_index.transform("mean"))
        recent = s[s.week > last - recent_weeks]
        weekly = recent.groupby("week").units.sum()
        cells.append({"segment": seg, "category": cat, "elasticity": round(float(point), 3),
                      "se": round(float(iv.se()["log_price"]), 3), "margin": float(s.margin.iloc[0]),
                      "price": round(float((recent.price * recent.units).sum() / recent.units.sum()), 3),
                      "weekly_units": round(float(weekly.mean()), 1),
                      "valid_change": round(float(np.expm1(np.percentile(np.abs(cost_moves), 95))), 3)})
    return cells


def coupon_curve(exp: pd.DataFrame, boot: int = 200) -> dict:
    train, test = train_test_split(exp, test_size=0.5, random_state=0)
    dr = DRLearner(model_propensity=clf(), model_regression=reg(),
                   model_final=reg(n_estimators=100, num_leaves=4, min_child_samples=300), cv=5,
                   min_propensity=0.02, random_state=0).fit(train.spend, train.coupon, X=train[X_COLS])
    rank = np.argsort(np.argsort(-dr.effect(test[X_COLS]))) / len(test)
    y, t = test.spend.to_numpy(), test.coupon.to_numpy()
    p = t.mean()
    mu0, mu1 = np.zeros(len(test)), np.zeros(len(test))
    for tr, te in KFold(5, shuffle=True, random_state=0).split(test):
        a = test.iloc[tr]
        for arm, mu in [(0, mu0), (1, mu1)]:
            mu[te] = reg().fit(a[a.coupon == arm][X_COLS], y[tr][t[tr] == arm]).predict(test.iloc[te][X_COLS])
    score = (mu1 + t * (y - mu1) / p) - (mu0 + (1 - t) * (y - mu0) / (1 - p))
    rng = np.random.default_rng(0)
    idx = [rng.integers(0, len(test), len(test)) for _ in range(boot)]
    uplift, se = [], []
    for s in SHARES:
        pol = rank < s
        uplift.append(round(float(np.mean(pol * score)), 3))
        se.append(round(float(np.std([np.mean(pol[i] * score[i]) for i in idx])), 3))
    return {"customers": 400_000, "campaigns_per_year": 12, "margin": 0.30, "shares": SHARES,
            "uplift": uplift, "uplift_se": se}


def build(demand_csv: str, coupons_csv: str, out: str = "models/causal_parameters.json") -> dict:
    demand = pd.read_csv(demand_csv)
    exp = pd.read_csv(coupons_csv)
    digest = hashlib.sha256(Path(demand_csv).read_bytes() + Path(coupons_csv).read_bytes()).hexdigest()[:8]
    params = {
        "version": f"{date.today().isoformat()}-{digest}",
        "created": date.today().isoformat(),
        "sources": {"demand": demand_csv, "demand_weeks": [int(demand.week.min()), int(demand.week.max())],
                    "coupons": coupons_csv, "coupon_customers": len(exp),
                    "pricing_method": "DML point estimate, 2SLS (supplier cost) standard error, per cell",
                    "coupon_method": "DR-learner ranking, doubly robust uplift on a held-out half"},
        "pricing": pricing_cells(demand),
        "coupons": coupon_curve(exp),
        "assumptions": {"pull_forward": 0.15},
    }
    save_parameters(params, out)
    return params


if __name__ == "__main__":
    args = sys.argv[1:] or ["data/scenario/weekly_demand.csv", "data/scenario/coupon_experiment.csv"]
    p = build(*args)
    print(f"Wrote parameters version {p['version']}: {len(p['pricing'])} pricing cells, "
          f"coupon curve at {len(p['coupons']['shares'])} shares")

python data/make_delivery_data.py
python -m engine.build

Wrote parameters version 2026-11-05-e117a0ef: 20 pricing cells, coupon curve at 11 shares

(Your date will differ. The 8 characters after it are a hash of the input data, so the same data always gives the same suffix.) Open models/causal_parameters.json. It's readable, and every number has a source line.

71.2 The simulation engine (50 min)

Concept. For a price change d in a cell with elasticity e, constant-elasticity demand gives units × (1 + d)^e. Profit per unit rises from m·p to (m + d)·p. The counterfactual is d = 0. Uncertainty comes from drawing each elasticity from N(estimate, SE²) 2,000 times.

A scenario is validated before anything runs. pydantic rejects unknown keys and impossible values with a readable message. Create engine/scenario.py, engine/simulate.py and engine/__init__.py:

"""What a user can ask the simulator: a scenario. Pydantic validates it (FastAPI uses the same class)."""
from pydantic import BaseModel, Field, field_validator


class Scenario(BaseModel):
    """price_change: share changes in price, by key. Keys: "all", a segment ("student"), or a cell
    ("student/snacks"). The most specific key wins: {"all": 0.05, "student": 0.0} raises every price 5%
    except for students. coupon_share: share of customers the targeting model coupons (0 = no programme).
    elasticity_correlation: 0 = each cell's estimation error is independent; 1 = all cells are wrong in the
    same direction together (e.g. a bias shared by every estimate). Day 73."""
    price_change: dict[str, float] = Field(default_factory=dict)
    coupon_share: float = Field(0.0, ge=0.0, le=1.0)
    coupon_cost: float = Field(5.0, ge=0.0, le=50.0)
    pull_forward: float | None = Field(None, ge=0.0, le=0.9)
    elasticity_correlation: float = Field(0.0, ge=0.0, le=1.0)
    draws: int = Field(2000, ge=100, le=20000)
    seed: int = 0

    @field_validator("price_change")
    @classmethod
    def plausible_changes(cls, v):
        for key, change in v.items():
            if not -0.5 <= change <= 0.5:
                raise ValueError(f"price change for '{key}' is {change:.0%}; the simulator accepts -50% to +50%")
        return v

    def change_for(self, segment: str, category: str) -> float:
        for key in (f"{segment}/{category}", segment, "all"):
            if key in self.price_change:
                return self.price_change[key]
        return 0.0

"""
The simulator: for a scenario, the projected annual revenue and gross profit, against the counterfactual
(what happens if nothing changes), with uncertainty from the causal estimates.

Pricing, for each segment x category cell with elasticity e, margin m (share of price), price change d:
    units      -> units x (1 + d)^e
    revenue    -> price x (1 + d) x units x (1 + d)^e
    gross profit per unit -> (m + d) x price   (unit cost doesn't change)
Coupons: extra spend = customers x campaigns x uplift(share) x (1 - pull_forward);
         gross profit = margin x extra spend - cost x share x customers x campaigns.
Uncertainty: every draw takes each cell's elasticity from Normal(estimate, se) (independently across cells,
or partly shared: Scenario.elasticity_correlation) and the uplift from Normal(estimate, se). Assumed: cells and coupons don't interact (a price rise doesn't change who responds
to coupons); that's a modelling choice, and the result says so.
"""
import numpy as np

from engine.scenario import Scenario

WEEKS = 52


def _uplift(coupons: dict, share: float):
    """Interpolate the uplift curve (and its SE) at a coupon share."""
    s = np.asarray(coupons["shares"])
    return float(np.interp(share, s, coupons["uplift"])), float(np.interp(share, s, coupons["uplift_se"]))


def simulate(params: dict, scenario: Scenario) -> dict:
    rng = np.random.default_rng(scenario.seed)
    n = scenario.draws
    cells, warnings = [], []
    total_rev = np.zeros(n)
    total_profit = np.zeros(n)
    base_rev = base_profit = 0.0
    rho = scenario.elasticity_correlation
    common = rng.normal(0, 1, n)                     # the part of the estimation error all cells share
    for c in params["pricing"]:
        d = scenario.change_for(c["segment"], c["category"])
        rev0 = c["price"] * c["weekly_units"] * WEEKS
        profit0 = c["margin"] * rev0
        e = c["elasticity"] + c["se"] * (np.sqrt(rho) * common + np.sqrt(1 - rho) * rng.normal(0, 1, n))
        units_ratio = (1 + d) ** e
        rev = rev0 * (1 + d) * units_ratio
        profit = (c["margin"] + d) * c["price"] * c["weekly_units"] * WEEKS * units_ratio
        base_rev += rev0
        base_profit += profit0
        total_rev += rev
        total_profit += profit
        if abs(d) > c["valid_change"] + 1e-9:
            warnings.append(f"{c['segment']}/{c['category']}: a {d:+.0%} change is outside the range seen in the "
                            f"data (±{c['valid_change']:.0%}); the projection extrapolates")
        cells.append({"segment": c["segment"], "category": c["category"], "price_change": d,
                      "revenue_base": rev0, "profit_base": profit0,
                      "profit_change_median": float(np.median(profit - profit0)),
                      "profit_change_p5": float(np.percentile(profit - profit0, 5)),
                      "profit_change_p95": float(np.percentile(profit - profit0, 95)),
                      "prob_profit_up": float(np.mean(profit > profit0)) if d != 0 else None})

    co = params["coupons"]
    pull = params["assumptions"]["pull_forward"] if scenario.pull_forward is None else scenario.pull_forward
    coupon_rev = np.zeros(n)
    coupon_profit = np.zeros(n)
    if scenario.coupon_share > 0:
        u, se = _uplift(co, scenario.coupon_share)
        extra = co["customers"] * co["campaigns_per_year"] * rng.normal(u, se, n) * (1 - pull)
        coupon_rev = extra
        coupon_profit = co["margin"] * extra - scenario.coupon_cost * scenario.coupon_share * co["customers"] * co["campaigns_per_year"]
    total_rev += coupon_rev
    total_profit += coupon_profit

    def band(x):
        return {"median": float(np.median(x)), "p5": float(np.percentile(x, 5)), "p95": float(np.percentile(x, 95))}

    return {
        "parameters_version": params["version"],
        "counterfactual": {"revenue": base_rev, "gross_profit": base_profit},
        "projected": {"revenue": band(total_rev), "gross_profit": band(total_profit)},
        "change": {"revenue": band(total_rev - base_rev), "gross_profit": band(total_profit - base_profit),
                   "prob_profit_up": (float(np.mean(total_profit > base_profit))
                                      if scenario.price_change or scenario.coupon_share > 0 else None),
                   "prob_loss_over_1m": float(np.mean(total_profit - base_profit < -1e6))},
        "coupons": {"share": scenario.coupon_share, "profit_change": band(coupon_profit),
                    "prob_profit_up": float(np.mean(coupon_profit > 0)) if scenario.coupon_share > 0 else None},
        "cells": cells,
        "draws": {"gross_profit_change": (total_profit - base_profit)[:500].round(0).tolist()},
        "warnings": warnings,
        "assumptions": {"pull_forward": pull, "cells_and_coupons_interact": False},
    }

"""NorthMart scenario engine (Module 8): causal parameters in, projected vs counterfactual outcomes out."""
from engine.parameters import load_parameters, save_parameters
from engine.scenario import Scenario
from engine.simulate import simulate

__all__ = ["Scenario", "load_parameters", "save_parameters", "simulate"]

Note what simulate returns: the counterfactual, the projection, the change (the causal quantity), the probability that profit goes up, per-cell results, warnings and the assumptions it used. A number without its assumptions can't be checked.

71.3 A scenario API with FastAPI (45 min)

Concept. Other teams' programs (the pricing system, a planning spreadsheet, tomorrow's app) shouldn't import our Python. They send JSON over HTTP and get JSON back. FastAPI turns the pydantic Scenario into request validation and documentation for free.

Create service/app.py:

"""
The NorthMart scenario API (Module 8, Day 71): the causal parameters and the simulator behind HTTP.

    uvicorn service.app:app --port 8001            # then open http://127.0.0.1:8001/docs

Endpoints
    GET  /health       is it up, and which parameters version is it serving?
    GET  /parameters   the causal estimates it uses (so anyone can audit them)
    POST /scenario     one scenario -> projected vs counterfactual, with uncertainty
    POST /compare      several named scenarios -> one comparison table
"""
import os
from functools import lru_cache

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from engine import Scenario, load_parameters, simulate

app = FastAPI(title="NorthMart scenario API", version="1.0.0",
              description="Projected vs counterfactual revenue and gross profit for pricing and coupon scenarios.")


@lru_cache
def params() -> dict:
    """Load the parameters file once (path from the PARAMS_PATH environment variable)."""
    return load_parameters(os.environ.get("PARAMS_PATH", "models/causal_parameters.json"))


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


@app.get("/parameters")
def get_parameters():
    p = params()
    return {"version": p["version"], "created": p["created"], "sources": p["sources"],
            "pricing": p["pricing"], "coupons": p["coupons"], "assumptions": p["assumptions"]}


@app.post("/scenario")
def run_scenario(scenario: Scenario, include_draws: bool = False):
    p = params()
    known = {c["segment"] for c in p["pricing"]} | {f"{c['segment']}/{c['category']}" for c in p["pricing"]} | {"all"}
    unknown = sorted(set(scenario.price_change) - known)
    if unknown:
        raise HTTPException(status_code=422, detail=f"unknown price_change keys: {unknown}; use 'all', a segment, "
                                                    "or 'segment/category'")
    result = simulate(p, scenario)
    if not include_draws:
        result.pop("draws")
    return result


class NamedScenario(BaseModel):
    name: str
    scenario: Scenario


@app.post("/compare")
def compare(scenarios: list[NamedScenario]):
    rows = []
    for s in scenarios:
        r = run_scenario(s.scenario)
        g = r["change"]["gross_profit"]
        rows.append({"name": s.name, "profit_change_median": round(g["median"]), "p5": round(g["p5"]),
                     "p95": round(g["p95"]), "prob_profit_up": r["change"]["prob_profit_up"],
                     "warnings": len(r["warnings"])})
    return {"parameters_version": params()["version"], "rows": rows}

Run it and open the interactive documentation:

uvicorn service.app:app --reload --port 8001

In Codespaces, open the forwarded port and go to /docs. Every endpoint is there, with a "Try it out" button. Stop the server with Ctrl+C.

71.4 Calling the API, and what it refuses (40 min)

Create day71/call_the_api.py:

# day71/call_the_api.py   (python -m day71.call_the_api)
"""
Call the scenario API the way another team's code would. FastAPI's TestClient runs the app in-process,
so this works without starting a server; swap in `requests` and the URL to call a running one.
"""
import json

from fastapi.testclient import TestClient

from service.app import app

client = TestClient(app)
print("GET /health ->", client.get("/health").json())

p = client.get("/parameters").json()
print(f"GET /parameters -> version {p['version']}, {len(p['pricing'])} cells; first cell:")
print("   ", p["pricing"][0])

r = client.post("/scenario", json={"price_change": {"all": 0.05}})
print("POST /scenario {'price_change': {'all': 0.05}} ->")
print("    gross profit change:", {k: round(v) for k, v in r.json()["change"]["gross_profit"].items()})

bad = client.post("/scenario", json={"price_change": {"students": 0.05}})
print("POST /scenario with a misspelt segment ->", bad.status_code, bad.json()["detail"])
bad = client.post("/scenario", json={"price_change": {"all": 0.8}})
print("POST /scenario with an 80% price rise ->", bad.status_code, bad.json()["detail"][0]["msg"])

cmp = client.post("/compare", json=[
    {"name": "5% everywhere", "scenario": {"price_change": {"all": 0.05}}},
    {"name": "coupons, top 50%", "scenario": {"coupon_share": 0.5}},
    {"name": "both", "scenario": {"price_change": {"all": 0.05}, "coupon_share": 0.5}},
]).json()
print("POST /compare ->")
print(json.dumps(cmp["rows"], indent=1))

python -m day71.call_the_api

GET /health -> {'status': 'ok', 'parameters_version': '2026-11-05-e117a0ef'}
GET /parameters -> version 2026-11-05-e117a0ef, 20 cells; first cell:
    {'segment': 'rural value', 'category': 'fresh', 'elasticity': -3.054, 'se': 0.101, 'margin': 0.28, 'price': 6.663, 'weekly_units': 5278.2, 'valid_change': 0.216}
POST /scenario {'price_change': {'all': 0.05}} ->
    gross profit change: {'median': 458929, 'p5': 418879, 'p95': 501707}
POST /scenario with a misspelt segment -> 422 unknown price_change keys: ['students']; use 'all', a segment, or 'segment/category'
POST /scenario with an 80% price rise -> 422 Value error, price change for 'all' is 80%; the simulator accepts -50% to +50%
POST /compare ->
[ ... "5% everywhere" 458929 ... "coupons, top 50%" 5624836 ... "both" 6086041 ... ]

A misspelt segment is the most dangerous error a simulator can make silently: "students" matching nothing would return "no change", and someone would believe it. An API that refuses bad input loudly is a causal-inference safeguard, not just good engineering.

Try it. Ask the API what +8% in suburban family stores does, and check what happens if a caller sends coupon_share: 1.5. Create day71/try_suburban.py.

# day71/try_suburban.py   (python -m day71.try_suburban)
from fastapi.testclient import TestClient

from service.app import app

with TestClient(app) as client:
    r = client.post("/scenario", json={"price_change": {"suburban family": 0.08}}).json()
    g = r["change"]["gross_profit"]
    print(f"+8% suburban family: median {g['median']:,.0f} (90%: {g['p5']:,.0f} to {g['p95']:,.0f})")
    print(f"{len(r['warnings'])} warnings, e.g. {r['warnings'][0] if r['warnings'] else 'none'}")
    bad = client.post("/scenario", json={"price_change": {"suburban family": 0.08}, "coupon_share": 1.5})
    print("coupon_share 1.5 ->", bad.status_code)

+8% suburban family: median 186,698 (90%: 154,835 to 218,592)
0 warnings, e.g. none
coupon_share 1.5 -> 422

71.5 Day 71 mini-project: a real server, and how fast it is (75 min)

A slider app calls the simulator every time a slider moves. Above about 200 ms people notice the lag. Create day71/latency.py, which starts uvicorn in a background process, calls it over real HTTP, and times it:

# day71/latency.py   (python -m day71.latency)
"""
Mini-project: run the API as a real server (uvicorn, in a background process), call it over HTTP like
another team's app would, and measure how long a scenario takes. A slider-driven app needs answers in well
under a second.
"""
import statistics
import subprocess
import sys
import time

import requests

URL = "http://127.0.0.1:8001"
server = subprocess.Popen([sys.executable, "-m", "uvicorn", "service.app:app", "--port", "8001"],
                          stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
try:
    for _ in range(50):                                   # wait for it to start
        try:
            requests.get(f"{URL}/health", timeout=1)
            break
        except requests.ConnectionError:
            time.sleep(0.2)
    print("health:", requests.get(f"{URL}/health").json())
    for draws in [500, 2000, 10000]:
        times = []
        for i in range(20):
            t0 = time.perf_counter()
            r = requests.post(f"{URL}/scenario", json={"price_change": {"all": 0.05}, "coupon_share": 0.5,
                                                       "draws": draws, "seed": i})
            r.raise_for_status()
            times.append((time.perf_counter() - t0) * 1000)
        print(f"{draws:>6} draws: median {statistics.median(times):5.1f} ms, slowest {max(times):5.1f} ms")
finally:
    server.terminate()

python -m day71.latency

health: {'status': 'ok', 'parameters_version': '2026-11-05-e117a0ef'}
   500 draws: median   6.6 ms, slowest  26.6 ms
  2000 draws: median  11.0 ms, slowest  17.9 ms
 10000 draws: median  21.9 ms, slowest  24.2 ms

Around 10 ms for 2,000 draws: fast enough to call on every slider move, because the expensive work (estimation) happened once, in engine.build. Had we re-fitted DoubleML per request, each answer would take minutes. Separating estimation from simulation is what makes interactive causal tools possible.

Your timings will differ by machine; the ratio between them is the point.

71.6 Save your work with Git (20 min)

git add engine service data/make_delivery_data.py models/causal_parameters.json day71 requirements.txt causalkit/demand.py
git commit -m "Day 71: parameters file, simulation engine and scenario API"
git push

Day 71 checklist

• I built a versioned parameters file from the Module 6 and 7 estimators
• I can explain why estimation and simulation are separate steps
• My API validates scenarios and rejects misspelt segments and impossible values
• I called the API in-process (TestClient) and over HTTP (uvicorn)
• I measured how long a scenario takes, and know why it's fast