Saving and loading a model

import time
import joblib
import numpy as np
import pandas as pd
import sklearn
from pathlib import Path
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

df = pd.read_csv(DATA / "uy_ijara.csv")
NUMERIC = ["maydon_m2", "xonalar", "qavat", "metro_km", "qurilgan_yil"]
TEXT = ["tuman"]
X = df[NUMERIC + TEXT]
y = df["ijaraga_berildi_30k"]
Xtr, Xte, ytr, yte = train_test_split(X, y, test_size=0.25, random_state=42, stratify=y)

prep = ColumnTransformer([
    ("num", Pipeline([("imp", SimpleImputer(strategy="median")),
                      ("sc", StandardScaler())]), NUMERIC),
    ("text", OneHotEncoder(handle_unknown="ignore"), TEXT),
])
pipe = Pipeline([("prep", prep),
                 ("model", LogisticRegression(max_iter=2000))]).fit(Xtr, ytr)
original_pred = pipe.predict(Xte)

print(f"sklearn version   : {sklearn.__version__}")
print(f"Pipeline accuracy : {accuracy_score(yte, original_pred):.4f}")
print(f"Pipeline steps    : {[name for name, _ in pipe.steps]}")

SAVE = ISH / "models"
SAVE.mkdir(parents=True, exist_ok=True)

joblib.dump(pipe, SAVE / "pipeline.joblib")
joblib.dump(pipe.named_steps["model"], SAVE / "model_only.joblib")

for name in ["pipeline.joblib", "model_only.joblib"]:
    print(f"  {name:<22} {(SAVE / name).stat().st_size:>7} bytes")

model_only = joblib.load(SAVE / "model_only.joblib")
print("We give the bare model RAW data:")
try:
    model_only.predict(Xte)
    print("  no error")
except Exception as x:
    print(f"  {type(x).__name__}: {str(x)[:95]}")
print()
print("Good news: this error came out LOUD: the text column 'tuman' isn't numeric.")

raw_num = Xte[NUMERIC].fillna(Xte[NUMERIC].median())          # NO scaling
expected = model_only.n_features_in_
extra = np.zeros((len(raw_num), expected - len(NUMERIC)))     # zeros instead of OneHot
fake_input = np.hstack([raw_num.values, extra])

fake_pred = model_only.predict(fake_input)
agree = float((fake_pred == original_pred).mean())
print(f"Columns the model expects   : {expected}")
print(f"Was there an error          : no")
print(f"Agreement with right answer : {agree:.4f}")
print(f"Actual accuracy             : {accuracy_score(yte, fake_pred):.4f}")
print(f"Original pipeline accuracy  : {accuracy_score(yte, original_pred):.4f}")
print()
print("The model ran, answered, nobody saw an error, and the answers are broken.")

loaded = joblib.load(SAVE / "pipeline.joblib")
tests = []

def attempt(name, frame):
    try:
        b = loaded.predict(frame)
        return {"case": name, "result": "worked",
                "agrees with original": f"{float((b == original_pred).mean()):.4f}"}
    except Exception as x:
        return {"case": name, "result": type(x).__name__,
                "agrees with original": "-"}

tests.append(attempt("original", Xte))
tests.append(attempt("columns shuffled", Xte[TEXT + NUMERIC]))
tests.append(attempt("numbers in reverse order", Xte[NUMERIC[::-1] + TEXT]))
tests.append(attempt("one column MISSING", Xte.drop(columns=["metro_km"])))
with_extra = Xte.copy()
with_extra["new_column"] = 1.0
tests.append(attempt("extra column added", with_extra))
renamed = Xte.rename(columns={"maydon_m2": "xonalar", "xonalar": "maydon_m2"})
tests.append(attempt("two column NAMES swapped", renamed))
show(pd.DataFrame(tests))

"""Information saved together with the model."""
import json
import platform
from datetime import date
from pathlib import Path


def write_manifest(path, model, columns, metrics, note=""):
    """Writes `manifest.json` next to the model.

    Why it's needed: a `.joblib` file says nothing about itself. If the versions
    don't match, loading can break silently; the manifest shows that.
    """
    import sklearn
    import numpy
    record = {
        "created": str(date.today()),
        "python": platform.python_version(),
        "sklearn": sklearn.__version__,
        "numpy": numpy.__version__,
        "model_type": type(model).__name__,
        "expected_columns": list(columns),
        "metrics": metrics,
        "note": note,
    }
    Path(path).write_text(json.dumps(record, ensure_ascii=False, indent=2), encoding="utf-8")
    return record


def check_versions(manifest_path):
    """Does the current environment match the manifest. If not, a list of warnings."""
    import sklearn
    import numpy
    m = json.loads(Path(manifest_path).read_text(encoding="utf-8"))
    warnings = []
    for name, now in [("sklearn", sklearn.__version__), ("numpy", numpy.__version__)]:
        if m.get(name) != now:
            warnings.append(f"{name}: model saved with {m.get(name)}, now {now}")
    return warnings

import sys
sys.path.insert(0, str(ISH))
from models.manifest import write_manifest, check_versions

record = write_manifest(
    SAVE / "manifest.json", pipe, NUMERIC + TEXT,
    {"accuracy": round(accuracy_score(yte, original_pred), 4),
     "test_size": int(len(yte))},
    note="uy_ijara.csv, target ijaraga_berildi_30k",
)
for k, v in record.items():
    print(f"  {k:<18} {v}")
print()
print(f"Version check: {check_versions(SAVE / 'manifest.json') or 'matches'}")

"""Checks that the save -> load -> predict chain isn't broken."""
import joblib
import numpy as np


def round_trip(model, X_sample, path):
    """Saves the model, loads it back, and checks the predictions match EXACTLY.

    This test runs before every deploy. It's cheap, but it catches every
    "I thought I saved it" kind of mistake.
    """
    before = model.predict(X_sample)
    joblib.dump(model, path)
    after = joblib.load(path).predict(X_sample)
    if not np.array_equal(before, after):
        diff = int((before != after).sum())
        raise AssertionError(f"save/load changed the predictions: {diff} rows differ")
    return len(X_sample)

from models.round_trip import round_trip

n = round_trip(pipe, Xte, SAVE / "pipeline.joblib")
print(f"Round trip: predictions identical on {n} rows: OK")

old_model = Pipeline([
    ("prep", prep),
    ("model", LogisticRegression(max_iter=2000, C=0.0005)),   # a different setting
]).fit(Xtr, ytr)
joblib.dump(old_model, SAVE / "old_artifact.joblib")

before = pipe.predict(Xte)
after = joblib.load(SAVE / "old_artifact.joblib").predict(Xte)
diff = int((before != after).sum())
print(f"Expected model accuracy : {accuracy_score(yte, before):.4f}")
print(f"Old model in the folder : {accuracy_score(yte, after):.4f}")
print(f"Rows that differ        : {diff} / {len(Xte)}")
print()
if diff:
    print("In this case the round trip raises AssertionError and the deploy stops.")
else:
    print("No difference: with this setting the two models gave the same answers.")

import hashlib

def file_hash(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()

digest = file_hash(SAVE / "pipeline.joblib")
print(f"pipeline.joblib SHA-256: {digest[:32]}...")
(SAVE / "pipeline.sha256").write_text(digest, encoding="utf-8")

expected = (SAVE / "pipeline.sha256").read_text(encoding="utf-8").strip()
print(f"Check before loading: {'matches' if file_hash(SAVE / 'pipeline.joblib') == expected else 'DOES NOT MATCH, DO NOT LOAD'}")

from skl2onnx import to_onnx
from skl2onnx.common.data_types import FloatTensorType, StringTensorType
import onnxruntime as ort

types = [(c, (StringTensorType if c in TEXT else FloatTensorType)([None, 1]))
         for c in NUMERIC + TEXT]
onx = to_onnx(pipe, initial_types=types,
              options={id(pipe.named_steps["model"]): {"zipmap": False}})
(SAVE / "pipeline.onnx").write_bytes(onx.SerializeToString())

session = ort.InferenceSession(str(SAVE / "pipeline.onnx"),
                               providers=["CPUExecutionProvider"])
inputs = {c: (Xte[c].values.reshape(-1, 1).astype(np.float32) if c in NUMERIC
              else Xte[c].values.reshape(-1, 1).astype(object))
          for c in NUMERIC + TEXT}
onnx_pred = session.run(None, inputs)[0].ravel()
print(f"ONNX and sklearn predictions agree: {float((onnx_pred == original_pred).mean()):.4f}")

one = {c: v[:1] for c, v in inputs.items()}
t0 = time.perf_counter()
for _ in range(300):
    session.run(None, one)
onnx_ms = (time.perf_counter() - t0) / 300 * 1000

one_df = Xte.iloc[:1]
t0 = time.perf_counter()
for _ in range(300):
    loaded.predict(one_df)
skl_ms = (time.perf_counter() - t0) / 300 * 1000

show(pd.DataFrame([
    {"format": "joblib (sklearn)", "size (bytes)": (SAVE / "pipeline.joblib").stat().st_size,
     "one prediction (ms)": round(skl_ms, 3), "runs code": "yes"},
    {"format": "ONNX", "size (bytes)": (SAVE / "pipeline.onnx").stat().st_size,
     "one prediction (ms)": round(onnx_ms, 3), "runs code": "no"},
]))
print(f"ONNX speed advantage: {skl_ms / onnx_ms:.0f}x")

show(pd.DataFrame([
    {"File": "pipeline.joblib", "What for": "the full Pipeline: preparation + model"},
    {"File": "manifest.json", "What for": "versions, columns, metrics, date"},
    {"File": "pipeline.sha256", "What for": "checking the file hasn't been replaced"},
    {"File": "pipeline.onnx", "What for": "a fast, code-free format for serving"},
    {"File": "DATASHEET.md", "What for": "Chapter 38: where the data came from"},
    {"File": "requirements.txt", "What for": "rebuilding the environment"},
]))
print()
print("All in one folder: ish_en/models/")
for p in sorted(SAVE.iterdir()):
    if p.is_file():
        print(f"  {p.name:<22} {p.stat().st_size:>7} bytes")

environment = f"""scikit-learn=={sklearn.__version__}
numpy=={np.__version__}
pandas=={pd.__version__}
joblib=={joblib.__version__}
"""
(SAVE / "requirements.txt").write_text(environment, encoding="utf-8")
print(environment)
print("The versions are pinned EXACTLY (==), because sklearn's internal format changes")
print("between versions, and an old model can break silently on a new version.")

show(pd.DataFrame([
    {"Step": "37. Data collected", "Status": "done"},
    {"Step": "38. Dataset built", "Status": "done"},
    {"Step": "39. Model adapted", "Status": "done"},
    {"Step": "40. Model saved", "Status": "pipeline + manifest + hash + ONNX"},
    {"Step": "41. Serving", "Status": "next chapter: Django API"},
]))

show(pd.DataFrame([
    {"Concept": "Save the whole Pipeline", "In one sentence": "Measured: the bare model silently gives wrong answers"},
    {"Concept": "Column order", "In one sentence": "Measured: it does NOT matter with a DataFrame, columns are picked by name"},
    {"Concept": "Missing column", "In one sentence": "A loud ValueError: safe"},
    {"Concept": "Swapped names", "In one sentence": "Measured: breaks SILENTLY, caught by the contract"},
    {"Concept": "Manifest", "In one sentence": "Version, columns, metrics sit next to the model"},
    {"Concept": "Round trip", "In one sentence": "Proves that saving worked instead of assuming it"},
    {"Concept": "pickle security", "In one sentence": "joblib.load runs code: never open a stranger's file"},
    {"Concept": "ONNX", "In one sentence": "Code-free, portable, much faster for one row"},
    {"Concept": "Pinning versions", "In one sentence": "With ==, because the internal format changes"},
]))