Model serving — FastAPI, batch scoring আর ONNX

# বাজারঘর (কাল্পনিক দোকান): seed দিয়ে বানানো ডেলিভারির ডেটা আর একটি pipeline
import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import HistGradientBoostingRegressor
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder

DISTRICTS = ["Dhaka", "Chattogram", "Sylhet", "Rajshahi"]
FEATURES = ["km", "district", "rain", "weight_kg"]

def deliveries(n=3000, seed=68):
    rng = np.random.default_rng(seed)
    district = rng.choice(DISTRICTS, n, p=[0.5, 0.2, 0.15, 0.15])
    km = rng.uniform(1, 40, n).round(1)
    rain = rng.binomial(1, 0.3, n)
    weight = rng.gamma(2.0, 1.5, n).round(1)
    extra = pd.Series(district).map(dict(zip(DISTRICTS, [0, 3, 6, 5]))).to_numpy()
    hours = (6 + 0.55 * km + 4 * rain + extra + 0.8 * weight
             + (1 + 0.04 * km) * rng.standard_normal(n))
    weight[rng.random(n) < 0.08] = np.nan        # ৮% অর্ডারে ওজন লেখা হয়নি
    X = pd.DataFrame({"km": km, "district": district,
                      "rain": rain, "weight_kg": weight})
    return X, np.maximum(hours, 1).round(2)

def build_pipeline():
    pre = ColumnTransformer([
        ("district", OneHotEncoder(), ["district"]),
        ("weight", SimpleImputer(strategy="median"), ["weight_kg"]),
        ("num", "passthrough", ["km", "rain"])])
    return Pipeline([("pre", pre),
                     ("model", HistGradientBoostingRegressor(random_state=0))])

import pandas as pd, sklearn                 # sklearn: module-টির জন্য
from bazar_serve import DISTRICTS, FEATURES, deliveries, build_pipeline

X, y = deliveries()
pipe = build_pipeline().fit(X.iloc[:2400], y[:2400])
VERSION = "delivery-hgb-2026.11.1"
LIMITS = {"km": (1, 40), "weight_kg": (0.1, 30)}

def validate(p):
    errs = [f"{k}: unknown field" for k in p if k not in FEATURES]
    for k, (lo, hi) in LIMITS.items():
        v = p.get(k)
        if v is None and k == "weight_kg":
            continue                              # ওজন নেই: pipeline-এর imputer সামলাবে
        if isinstance(v, bool) or not isinstance(v, (int, float)):
            errs.append(f"{k}: number required, got {v!r}")
        elif not lo <= v <= hi:
            errs.append(f"{k}: {v} outside [{lo}, {hi}]")
    if p.get("district") not in DISTRICTS:
        errs.append(f"district: unknown {p.get('district')!r}")
    if p.get("rain") not in (0, 1):
        errs.append(f"rain: 0/1 required, got {p.get('rain')!r}")
    return errs
def predict(p):
    if errs := validate(p):
        return {"ok": False, "errors": errs, "model_version": VERSION}
    row = pd.DataFrame([{k: p.get(k) for k in FEATURES}]).astype({"weight_kg": float})
    return {"ok": True, "hours": round(float(pipe.predict(row)[0]), 2), "model_version": VERSION}

ok = {"km": 12.5, "district": "Dhaka", "rain": 0, "weight_kg": 2.0}
for p in [ok, {**ok, "weight_kg": None}, {**ok, "km": "12.5"}, {**ok, "km": 80},
          {**ok, "district": "Cox's Bazar"}, {**ok, "coupon": "EID10"}]:
    r = predict(p)
    print(r["model_version"], r["hours"] if r["ok"] else r["errors"][0])
print("no guard, km=80 ->", round(float(pipe.predict(pd.DataFrame([{**ok, "km": 80}]))[0]), 2))

import time, numpy as np
import pandas, sklearn                       # bazar_serve module-এর জন্য
import matplotlib.pyplot as plt
from bazar_serve import deliveries, build_pipeline

X, y = deliveries()
pipe = build_pipeline().fit(X, y)
night, _ = deliveries(n=20000, seed=7)          # আজ রাতে স্কোর করার ২০,০০০ অর্ডার

lat, single = [], []
for i in range(150):                           # একটি একটি করে ১৫০টি ডাক
    t0 = time.perf_counter()
    single.append(pipe.predict(night.iloc[[i]])[0])
    lat.append((time.perf_counter() - t0) * 1000)
p50, p95 = np.percentile(lat, [50, 95])
rate = {"1": 1000 / np.mean(lat)}
print(f"1 row/call : p50 {p50:.2f} ms, p95 {p95:.2f} ms, {rate['1']:,.0f} rows/s")
print(f"  20,000 rows this way ~ {20000 / rate['1']:.0f} s")

for chunk in [200, 2000, 20000]:
    t0 = time.perf_counter()
    out = np.concatenate([pipe.predict(night.iloc[s:s + chunk])
                          for s in range(0, len(night), chunk)])
    sec = time.perf_counter() - t0
    rate[str(chunk)] = len(night) / sec
    print(f"chunk {chunk:>5}: {sec:.2f} s for all, {rate[str(chunk)]:,.0f} rows/s")
print("chunked == one by one:", np.allclose(out[:150], single))

fig, ax = plt.subplots(figsize=(6, 3.2))
ax.bar(list(rate), list(rate.values()), color=["tab:red"] + ["tab:blue"] * 3)
ax.set_yscale("log")
ax.set_xlabel("rows per predict() call")
ax.set_ylabel("rows per second (log)")
ax.set_title("Same model, same 20,000 orders: call size matters")
plt.show()

import numpy as np
import pandas, sklearn                       # bazar_serve module-এর জন্য
from bazar_serve import deliveries, build_pipeline

X, y = deliveries()
pipe = build_pipeline().fit(X.iloc[:2400], y[:2400])
raw, yt = X.iloc[2400:].reset_index(drop=True), y[2400:]   # সার্ভিসে আসা কাঁচা অনুরোধ
miss = raw.weight_kg.isna()

def report(name, feats):
    err = np.abs(yt - pipe.predict(feats))
    print(f"{name:<24} MAE {err.mean():.2f} h | no-weight rows {err[miss].mean():.2f} h")

# training-এ: রাস্তার দূরত্ব (১ দশমিক), ওজন না থাকলে training median
report("same pipeline", raw)
# সার্ভিস-দল নিজে feature কোড আবার লিখল, তিনটি ছোট অমিলসহ:
report("km rounded", raw.assign(km=raw.km.round()))
report("missing weight -> 0", raw.assign(weight_kg=raw.weight_kg.fillna(0)))
report("straight-line km", raw.assign(km=(raw.km / 1.3).round(1)))
skew = raw.assign(km=(raw.km / 1.3).round(), weight_kg=raw.weight_kg.fillna(0))
report("all three", skew)

print("training median weight:", pipe.named_steps["pre"]
      .named_transformers_["weight"].statistics_)
# প্রতিকার: সার্ভিস শুধু কাঁচা field যাচাই করে একই pipeline-কে দেয়
offline = pipe.predict(raw)
online = np.array([pipe.predict(raw.iloc[[i]])[0] for i in range(50)])
print("max |offline - online| on 50 rows:", np.abs(offline[:50] - online).max())

import io, hashlib, platform
import joblib
import numpy as np
import pandas as pd
import sklearn
from bazar_serve import FEATURES, deliveries, build_pipeline

X, y = deliveries()
pipe = build_pipeline().fit(X.iloc[:2400], y[:2400])
golden = X.iloc[2400:2405]                      # ৫টি স্থির "golden" সারি
meta = {"model_version": "delivery-hgb-2026.11.1",
        "sklearn": sklearn.__version__, "python": platform.python_version(),
        "features": FEATURES, "train_rows": 2400,
        "data_hash": hashlib.sha256(pd.util.hash_pandas_object(X.iloc[:2400])
                                    .to_numpy().tobytes()).hexdigest()[:12],
        "golden_pred": pipe.predict(golden).round(4).tolist()}
buf = io.BytesIO()
joblib.dump({"pipe": pipe, "meta": meta}, buf)  # ডিস্কের বদলে memory-তে
print(f"saved {buf.tell() / 1024:.0f} KB;", {k: meta[k] for k in list(meta)[:3]})

def check(art):
    m, problems = art["meta"], []
    if m["sklearn"] != sklearn.__version__:
        problems.append(f"sklearn saved {m['sklearn']}, running {sklearn.__version__}")
    if m["features"] != FEATURES:
        problems.append(f"features differ: {m['features']}")
    diff = np.abs(art["pipe"].predict(golden) - m["golden_pred"]).max()
    problems += [f"golden diff {diff:.1e} h"] if diff > 1e-3 else []
    return problems or ["OK"]

buf.seek(0)
art = joblib.load(buf)                          # শুধু নিজের বিশ্বস্ত ফাইল!
print("reloaded :", check(art))
print("desktop  :", check({**art, "meta": {**meta, "sklearn": "1.9.1"}}))
print("old model:", check({**art, "meta": {**meta, "features": FEATURES[:3]}}))

import os
import joblib, numpy as np, skops
import skops.io as sio
from bazar_serve import deliveries

art = joblib.load("model.joblib")
pipe = art["pipe"]
sio.dump(pipe, "model.skops")
print("skops", skops.__version__, "| file", os.path.getsize("model.skops") // 1024, "KB")
untrusted = sio.get_untrusted_types(file="model.skops")
print("untrusted types:", untrusted)
try:
    sio.load("model.skops")                          # কিছু trust না করে
except Exception as e:
    print(type(e).__name__, ":", str(e)[:90])
loaded = sio.load("model.skops", trusted=untrusted)  # তালিকা নিজে পড়ে দেখার পরে
X, _ = deliveries()
print("max |diff| vs joblib:", np.max(np.abs(loaded.predict(X) - pipe.predict(X))))

from contextlib import asynccontextmanager
from typing import Literal

import joblib
import pandas as pd
from fastapi import FastAPI
from pydantic import BaseModel, ConfigDict, Field

from bazar_serve import FEATURES

class Order(BaseModel):
    model_config = ConfigDict(extra="forbid", strict=True)
    km: float = Field(ge=1, le=40)
    district: Literal["Dhaka", "Chattogram", "Sylhet", "Rajshahi"]
    rain: bool
    weight_kg: float | None = Field(default=None, gt=0, le=30)

class Estimate(BaseModel):
    hours: float
    model_version: str

class Batch(BaseModel):
    orders: list[Order] = Field(min_length=1, max_length=1000)

ART = {}

@asynccontextmanager
async def lifespan(app):
    ART.update(joblib.load("model.joblib"))    # পুরো pipeline + meta, একবারই
    yield

app = FastAPI(title="Bazarghar delivery ETA", lifespan=lifespan)

def score(orders):
    df = pd.DataFrame([o.model_dump() for o in orders], columns=FEATURES)
    df = df.astype({"rain": int, "weight_kg": float})
    return ART["pipe"].predict(df).round(2).tolist()

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

@app.post("/predict", response_model=Estimate)
def predict(order: Order):
    return Estimate(hours=score([order])[0],
                    model_version=ART["meta"]["model_version"])

@app.post("/predict/batch")
def predict_batch(batch: Batch):
    return {"model_version": ART["meta"]["model_version"],
            "hours": score(batch.orders)}

import fastapi, pydantic, sklearn
from fastapi.testclient import TestClient
from app import app

print("fastapi", fastapi.__version__, "pydantic", pydantic.VERSION, "sklearn", sklearn.__version__)
ok = {"km": 12.5, "district": "Dhaka", "rain": False, "weight_kg": 2.0}
cases = {
    "valid": ok,
    "no weight": {**ok, "weight_kg": None},
    "km as text": {**ok, "km": "12.5"},
    "km=80": {**ok, "km": 80},
    "unknown district": {**ok, "district": "Cox's Bazar"},
    "extra field": {**ok, "coupon": "EID10"},
}
with TestClient(app) as client:                 # with: lifespan চলে, model load হয়
    print("GET /health ->", client.get("/health").json())
    for name, body in cases.items():
        r = client.post("/predict", json=body)
        out = r.json() if r.status_code == 200 else [
            (e["loc"][-1], e["type"]) for e in r.json()["detail"]]
        print(f"{name:<17} {r.status_code} {out}")
    r = client.post("/predict/batch", json={"orders": [ok, {**ok, "rain": True}, {**ok, "km": 30.0}]})
    print("batch of 3        ", r.status_code, r.json())
    r = client.post("/predict/batch", json={"orders": [ok] * 1001})
    print("batch of 1001     ", r.status_code, r.json()["detail"][0]["type"])

import time
import numpy as np, onnxruntime as ort, skl2onnx, sklearn
from skl2onnx import to_onnx
from skl2onnx.common.data_types import FloatTensorType, StringTensorType
from bazar_serve import deliveries, build_pipeline

X, y = deliveries()
pipe = build_pipeline().fit(X.iloc[:2400], y[:2400])
types = [("km", FloatTensorType([None, 1])), ("district", StringTensorType([None, 1])),
         ("rain", FloatTensorType([None, 1])), ("weight_kg", FloatTensorType([None, 1]))]
onx = to_onnx(pipe, initial_types=types, target_opset=18)
open("model.onnx", "wb").write(onx.SerializeToString())
sess = ort.InferenceSession("model.onnx", providers=["CPUExecutionProvider"])
print("sklearn", sklearn.__version__, "| skl2onnx", skl2onnx.__version__, "| onnxruntime", ort.__version__)
print("inputs:", [(i.name, i.type) for i in sess.get_inputs()])

big, _ = deliveries(n=20000, seed=7)
def feeds(df):
    return {c: (df[[c]].to_numpy(dtype=np.float32) if c != "district"
                else df[[c]].to_numpy(dtype=object)) for c in df.columns}
p_sk = pipe.predict(big)
p_ox = sess.run(None, feeds(big))[0].ravel()
d = np.abs(p_sk - p_ox)
print(f"max |diff| = {d.max():.2e} h, mean = {d.mean():.2e} h, rows > 0.01 h: {(d > 0.01).sum()}")

def p50(fn, n):
    t = []
    for _ in range(n):
        t0 = time.perf_counter(); fn(); t.append((time.perf_counter() - t0) * 1000)
    return np.percentile(t, 50)
one = big.iloc[[0]]; f1 = feeds(one); fa = feeds(big)
print(f"1 row   p50: sklearn {p50(lambda: pipe.predict(one), 300):.3f} ms | onnxruntime {p50(lambda: sess.run(None, f1), 300):.3f} ms")
print(f"20,000  p50: sklearn {p50(lambda: pipe.predict(big), 10):.1f} ms | onnxruntime {p50(lambda: sess.run(None, fa), 10):.1f} ms")