Train, register, serve and monitor a model

Pinned dependencies, the dataset downloaded and verified by checksum, an MLflow server with a SQLite backend running locally, and a feature pipeline that both training and serving will import.

1. 1.1
   Create the repository and install the dependencies.

   bashCopy

   mkdir mlops-loop && cd mlops-loop && git init -b main
   uv venv --python 3.12 && source .venv/bin/activate
   mkdir -p bike tests deploy/grafana/provisioning/datasources deploy/grafana/provisioning/dashboards deploy/grafana/dashboards .github/workflows data
   touch bike/__init__.py
   cat > pyproject.toml <<'EOF'
   [project]
   name = "mlops-loop"
   version = "0.1.0"
   requires-python = ">=3.12"
   dependencies = [
     "mlflow>=3.1,<4",
     "scikit-learn>=1.5,<2",
     "pandas>=2.2,<3",
     "numpy>=2,<3",
     "fastapi>=0.115,<1",
     "uvicorn>=0.34,<1",
     "prometheus-client>=0.21,<1",
     "httpx>=0.27,<1",
   ]
   
   [project.optional-dependencies]
   dev = ["pytest>=8,<9", "ruff>=0.8,<1"]
   
   [tool.setuptools]
   packages = ["bike"]
   EOF
   uv pip install -e '.[dev]'
   printf '.venv/\ndata/\nmlruns/\nmlflow.db\nmodel/\npredictions.db\n__pycache__/\n.pytest_cache/\n.ruff_cache/\n' > .gitignore

2. 1.2
   Write the data module. The dataset is the UCI bike-sharing set: 17,379 hourly rows for 2011–2012 in Washington DC. The download is verified by SHA-256 so the data hash logged with each run means something.

   pythonCopy

   # bike/data.py
   import hashlib
   import io
   import zipfile
   from pathlib import Path
   
   import httpx
   import pandas as pd
   
   URL = "https://archive.ics.uci.edu/static/public/275/bike+sharing+dataset.zip"
   SHA256 = "b70182d0d0508e9abbb79306ce5c0cec34869000f8220175ac83d11dbe845401"
   DATA = Path("data/hour.csv")
   
   
   def download() -> Path:
       if DATA.exists():
           return DATA
       raw = httpx.get(URL, timeout=60, follow_redirects=True).content
       digest = hashlib.sha256(raw).hexdigest()
       if digest != SHA256:
           raise RuntimeError(f"checksum mismatch: {digest}")
       DATA.parent.mkdir(exist_ok=True)
       with zipfile.ZipFile(io.BytesIO(raw)) as z:
           DATA.write_bytes(z.read("hour.csv"))
       return DATA
   
   
   def data_hash() -> str:
       return hashlib.sha256(DATA.read_bytes()).hexdigest()[:16]
   
   
   def load() -> pd.DataFrame:
       df = pd.read_csv(download(), parse_dates=["dteday"])
       df["ts"] = df["dteday"] + pd.to_timedelta(df["hr"], unit="h")
       return df.sort_values("ts").reset_index(drop=True)

3. 1.3
   Write the feature pipeline — the single most important file for consistency. It builds the model inputs from raw rows, including level: yesterday's trailing 28-day daily ridership. That feature is what lets a tree model follow a growing city; without it, a model trained on 2011 cannot predict 2012's volume no matter how often you retrain. Training and serving both import this function, so there is no way for them to disagree.

   pythonCopy

   # bike/features.py
   import pandas as pd
   
   FEATURES = ["hr", "weekday", "workingday", "holiday", "weathersit", "temp", "atemp", "hum", "windspeed", "season", "level"]
   CATEGORICAL = ["hr", "weekday", "weathersit", "season"]
   TARGET = "cnt"
   NUMERIC = [f for f in FEATURES if f not in CATEGORICAL]
   
   
   def daily_level(df: pd.DataFrame) -> pd.Series:
       """Trailing 28-day mean of daily totals, shifted one day: what was known yesterday."""
       daily = df.groupby("dteday")[TARGET].sum().sort_index()
       return daily.rolling(28, min_periods=7).mean().shift(1)
   
   
   def add_level(df: pd.DataFrame, level: pd.Series | None = None) -> pd.DataFrame:
       level = daily_level(df) if level is None else level
       out = df.copy()
       out["level"] = out["dteday"].map(level)
       return out.dropna(subset=["level"])
   
   
   def to_matrix(df: pd.DataFrame) -> pd.DataFrame:
       x = df[FEATURES].copy()
       for c in CATEGORICAL:
           x[c] = x[c].astype("category")
       return x

4. 1.4
   Test that the level feature never sees the future — a leak here would make every offline metric a lie — then start MLflow.

   pythonCopy

   # tests/test_features.py
   import pandas as pd
   
   from bike.features import FEATURES, add_level, daily_level, to_matrix
   
   
   def make(days=40):
       rows = []
       for d in range(days):
           for h in range(24):
               rows.append({"dteday": pd.Timestamp("2011-01-01") + pd.Timedelta(days=d), "hr": h, "weekday": d % 7,
                            "workingday": 1, "holiday": 0, "weathersit": 1, "temp": 0.5, "atemp": 0.5, "hum": 0.5,
                            "windspeed": 0.1, "season": 1, "cnt": 100 + d})
       return pd.DataFrame(rows)
   
   
   def test_level_is_shifted_and_uses_only_past_days():
       df = make()
       level = daily_level(df)
       day10 = pd.Timestamp("2011-01-11")
       past = df[df.dteday < day10].groupby("dteday").cnt.sum()
       assert abs(level[day10] - past.tail(28).mean()) < 1e-9
       assert pd.isna(level.iloc[0])          # nothing known before day one
   
   
   def test_matrix_has_expected_columns_and_dtypes():
       x = to_matrix(add_level(make()))
       assert list(x.columns) == FEATURES
       assert str(x["hr"].dtype) == "category" and x["temp"].dtype.kind == "f"

   Start the tracking server in a second terminal and leave it running: mlflow server --backend-store-uri sqlite:///mlflow.db --artifacts-destination ./mlruns --port 5000. Then export MLFLOW_TRACKING_URI=http://localhost:5000 in the first.

One command trains a model on data up to a chosen date, evaluates it on the following two months, logs everything needed to reproduce it, saves a drift reference, and registers the model.

1. 2.1
   Write the drift helpers first, because training must save the reference they compare against: quantile bin edges from the training data for each numeric feature and for the predictions, and a PSI function used identically offline and online.

   pythonCopy

   # bike/drift.py
   import numpy as np
   
   
   def edges(values, bins: int = 10) -> list[float]:
       q = np.quantile(np.asarray(values, dtype=float), np.linspace(0, 1, bins + 1))
       q[0], q[-1] = -np.inf, np.inf
       return [float(v) for v in q]
   
   
   def histogram(values, bin_edges: list[float]) -> list[float]:
       counts, _ = np.histogram(np.asarray(values, dtype=float), bins=np.asarray(bin_edges))
       total = max(1, counts.sum())
       return [float(c) / total for c in counts]
   
   
   def psi(reference: list[float], current: list[float], eps: float = 1e-4) -> float:
       """Population stability index between two binned distributions. <0.1 stable, 0.1-0.25 shifting, >0.25 drifted."""
       r = np.clip(np.asarray(reference), eps, None)
       c = np.clip(np.asarray(current), eps, None)
       return float(np.sum((c - r) * np.log(c / r)))
   
   
   def build_reference(x, predictions, numeric: list[str]) -> dict:
       ref = {"prediction": {"edges": edges(predictions)}}
       ref["prediction"]["hist"] = histogram(predictions, ref["prediction"]["edges"])
       for col in numeric:
           e = edges(x[col])
           ref[col] = {"edges": e, "hist": histogram(x[col], e)}
       return ref

2. 2.2
   Write train.py. Everything that could change the result is either a logged parameter (seed, hyperparameters, --until cut-off, git commit, data hash) or derived from one. The target is log-transformed because hourly counts are skewed; the metric is reported back in real units. A naive baseline (same hour last week) is logged too, so "the model is better than nothing" is a number.

   pythonCopy

   # train.py
   import argparse
   import subprocess
   
   import mlflow
   import numpy as np
   import pandas as pd
   from mlflow import MlflowClient
   from sklearn.ensemble import HistGradientBoostingRegressor
   from sklearn.metrics import mean_absolute_error
   
   from bike.data import data_hash, load
   from bike.drift import build_reference
   from bike.features import CATEGORICAL, FEATURES, NUMERIC, TARGET, add_level, to_matrix
   
   MODEL_NAME = "bike-demand"
   
   
   def split(df: pd.DataFrame, until: str, holdout_days: int = 61):
       cutoff = pd.Timestamp(until)
       train = df[df.dteday <= cutoff - pd.Timedelta(days=holdout_days)]
       val = df[(df.dteday > cutoff - pd.Timedelta(days=holdout_days)) & (df.dteday <= cutoff)]
       return train, val
   
   
   def naive_baseline(val: pd.DataFrame, history: pd.DataFrame) -> float:
       both = pd.concat([history, val]).set_index("ts")[TARGET]
       last_week = both.shift(24 * 7).reindex(val["ts"]).fillna(both.mean())
       return float(mean_absolute_error(val[TARGET], last_week))
   
   
   def main() -> None:
       ap = argparse.ArgumentParser()
       ap.add_argument("--until", default="2011-12-31", help="last date of data to use (holdout = last 61 days)")
       ap.add_argument("--seed", type=int, default=42)
       ap.add_argument("--max-iter", type=int, default=400)
       ap.add_argument("--learning-rate", type=float, default=0.08)
       ap.add_argument("--max-depth", type=int, default=8)
       args = ap.parse_args()
   
       df = add_level(load())
       train, val = split(df, args.until)
       x_train, x_val = to_matrix(train), to_matrix(val)
   
       mlflow.set_experiment(MODEL_NAME)
       with mlflow.start_run(run_name=f"until-{args.until}") as run:
           mlflow.log_params({
               "until": args.until, "seed": args.seed, "max_iter": args.max_iter,
               "learning_rate": args.learning_rate, "max_depth": args.max_depth,
               "features": ",".join(FEATURES), "train_rows": len(train), "val_rows": len(val),
               "data_hash": data_hash(),
               "git_commit": subprocess.run(["git", "rev-parse", "--short", "HEAD"], capture_output=True, text=True).stdout.strip() or "unknown",
           })
           model = HistGradientBoostingRegressor(
               max_iter=args.max_iter, learning_rate=args.learning_rate, max_depth=args.max_depth,
               random_state=args.seed, categorical_features=[FEATURES.index(c) for c in CATEGORICAL],
           )
           model.fit(x_train, np.log1p(train[TARGET]))
           pred_val = np.expm1(model.predict(x_val))
           mae = float(mean_absolute_error(val[TARGET], pred_val))
           mlflow.log_metrics({
               "val_mae": mae,
               "val_mae_naive": naive_baseline(val, train),
               "val_mean_cnt": float(val[TARGET].mean()),
               "train_mae": float(mean_absolute_error(train[TARGET], np.expm1(model.predict(x_train)))),
           })
           reference = build_reference(x_train, np.expm1(model.predict(x_train)), NUMERIC)
           mlflow.log_dict(reference, "reference.json")
           info = mlflow.sklearn.log_model(
               sk_model=model, name="model", input_example=x_train.head(5), registered_model_name=MODEL_NAME,
           )
           MlflowClient().set_model_version_tag(MODEL_NAME, info.registered_model_version, "until", args.until)
           print(f"run {run.info.run_id} → {MODEL_NAME} v{info.registered_model_version}: val_mae={mae:.1f} "
                 f"(naive {mlflow.get_run(run.info.run_id).data.metrics['val_mae_naive']:.1f})")
   
   
   if __name__ == "__main__":
       main()

3. 2.3
   Train the first model on 2011 and look at the run in the MLflow UI. Then run it again with the same arguments and confirm the metric is identical to the decimal — that is what reproducible means.

   bashCopy

   export MLFLOW_TRACKING_URI=http://localhost:5000
   git add . && git commit -qm "feat: data, features, drift reference, training"
   python train.py --until 2011-12-31
   python train.py --until 2011-12-31
   open http://localhost:5000     # Experiments → bike-demand: two runs, same val_mae

   Reference numbers from this exact setup: val_mae about 32 on Nov–Dec 2011 (mean hourly count 130), naive baseline about 60. Yours should match closely; a different pandas or scikit-learn version can move the last digit.
4. 2.4
   Train a worse variant on purpose so the promotion gate has something to reject, then a differently-tuned one. Three versions now sit in the registry, none of them the champion yet.

   bashCopy

   python train.py --until 2011-12-31 --max-iter 20          # underfit: worse val_mae
   python train.py --until 2011-12-31 --learning-rate 0.05 --max-iter 600
   mlflow models search 2>/dev/null || python -c "from mlflow import MlflowClient; [print(v.version, v.tags, v.run_id[:8]) for v in MlflowClient().search_model_versions(\"name='bike-demand'\")]"

5. 2.5
   Add a training smoke test that runs on a small slice so CI can exercise the whole path in under a minute without registering anything.

   pythonCopy

   # tests/test_train.py
   import numpy as np
   from sklearn.ensemble import HistGradientBoostingRegressor
   from sklearn.metrics import mean_absolute_error
   
   from bike.features import CATEGORICAL, FEATURES, TARGET, add_level, to_matrix
   from tests.test_features import make
   
   
   def test_model_beats_mean_on_synthetic_data():
       df = add_level(make(days=60))
       df[TARGET] = (df["hr"] * 10 + df["level"] / 100).astype(float)   # learnable signal
       train, val = df[df.dteday < "2011-02-15"], df[df.dteday >= "2011-02-15"]
       model = HistGradientBoostingRegressor(max_iter=100, random_state=0,
                                             categorical_features=[FEATURES.index(c) for c in CATEGORICAL])
       model.fit(to_matrix(train), np.log1p(train[TARGET]))
       mae = mean_absolute_error(val[TARGET], np.expm1(model.predict(to_matrix(val))))
       baseline = mean_absolute_error(val[TARGET], np.full(len(val), train[TARGET].mean()))
       assert mae < baseline / 3

A champion alias that only moves when a candidate is measurably better on the same holdout, with the decision recorded on the model version.

1. 3.1
   Write promote.py. It reads the candidate's and the current champion's val_mae from their runs, refuses if the holdout windows differ (an unfair comparison), and moves the alias only on improvement beyond a small margin. The old champion becomes previous, which is your one-command rollback.

   pythonCopy

   # promote.py
   import argparse
   import sys
   
   import mlflow
   from mlflow import MlflowClient
   
   MODEL_NAME = "bike-demand"
   
   
   def metrics_for(client: MlflowClient, version: str) -> tuple[float, str]:
       mv = client.get_model_version(MODEL_NAME, version)
       run = mlflow.get_run(mv.run_id)
       return run.data.metrics["val_mae"], run.data.params["until"]
   
   
   def main() -> int:
       ap = argparse.ArgumentParser()
       ap.add_argument("version", help="candidate model version number")
       ap.add_argument("--min-improvement", type=float, default=0.02, help="relative MAE improvement required")
       ap.add_argument("--allow-different-holdout", action="store_true")
       args = ap.parse_args()
       client = MlflowClient()
   
       cand_mae, cand_until = metrics_for(client, args.version)
       try:
           champ = client.get_model_version_by_alias(MODEL_NAME, "champion")
       except mlflow.exceptions.MlflowException:
           champ = None
   
       if champ is not None:
           champ_mae, champ_until = metrics_for(client, champ.version)
           if champ_until != cand_until and not args.allow_different_holdout:
               print(f"refusing: candidate holdout ends {cand_until}, champion's ends {champ_until}; "
                     f"retrain the champion's config with --until {cand_until} or pass --allow-different-holdout")
               return 2
           improvement = (champ_mae - cand_mae) / champ_mae
           print(f"champion v{champ.version} val_mae={champ_mae:.2f}  candidate v{args.version} val_mae={cand_mae:.2f}  improvement={improvement:+.1%}")
           if improvement < args.min_improvement:
               client.set_model_version_tag(MODEL_NAME, args.version, "promotion", f"rejected vs v{champ.version}")
               return 1
           client.set_registered_model_alias(MODEL_NAME, "previous", champ.version)
       else:
           print(f"no champion yet; candidate v{args.version} val_mae={cand_mae:.2f}")
   
       client.set_registered_model_alias(MODEL_NAME, "champion", args.version)
       client.set_model_version_tag(MODEL_NAME, args.version, "promotion", "champion")
       print(f"champion → v{args.version}")
       return 0
   
   
   if __name__ == "__main__":
       sys.exit(main())

2. 3.2
   Promote version 1, then try the underfit version 3 (rejected, exit 1), then version 4 (promoted only if it beat v1 by 2 %). Look at the aliases in the UI.

   bashCopy

   python promote.py 1
   python promote.py 3; echo "exit $?"
   python promote.py 4; echo "exit $?"
   python -c "from mlflow import MlflowClient; c=MlflowClient(); print({a: c.get_model_version_by_alias('bike-demand', a).version for a in ('champion','previous') if True})" 2>/dev/null || true

   If v4 was not better by 2 %, the champion stays at v1 and that is the correct outcome. The gate is the product here, not the promotion.
3. 3.3
   Write bake.py: it downloads the champion (or any alias) and its run's reference.json into ./model/, plus a VERSION file. The Docker image will contain exactly this folder, so a running container can always say which registry version it is.

   pythonCopy

   # bake.py
   import argparse
   import json
   import shutil
   from pathlib import Path
   
   import mlflow
   from mlflow import MlflowClient
   
   MODEL_NAME = "bike-demand"
   
   
   def main() -> None:
       ap = argparse.ArgumentParser()
       ap.add_argument("--alias", default="champion")
       ap.add_argument("--dst", default="model")
       args = ap.parse_args()
       dst = Path(args.dst)
       shutil.rmtree(dst, ignore_errors=True)
       dst.mkdir(parents=True)
   
       mv = MlflowClient().get_model_version_by_alias(MODEL_NAME, args.alias)
       mlflow.artifacts.download_artifacts(artifact_uri=f"models:/{MODEL_NAME}@{args.alias}", dst_path=str(dst / "model"))
       mlflow.artifacts.download_artifacts(artifact_uri=f"runs:/{mv.run_id}/reference.json", dst_path=str(dst))
       run = mlflow.get_run(mv.run_id)
       (dst / "VERSION").write_text(json.dumps({
           "name": MODEL_NAME, "version": mv.version, "alias": args.alias, "run_id": mv.run_id,
           "val_mae": run.data.metrics["val_mae"], "until": run.data.params["until"],
           "git_commit": run.data.params.get("git_commit"),
       }, indent=2))
       print((dst / "VERSION").read_text())
   
   
   if __name__ == "__main__":
       main()

A FastAPI service that loads the baked model, predicts for batches of hourly rows, stores each prediction so a label can be attached later, and exposes version, drift and quality metrics for Prometheus.

1. 4.1
   Write the service. Three ideas carry it: the feature code is imported, not re-implemented; every prediction is stored with its id so /labels can join actuals to it a day later; and the gauges are recomputed from a rolling window of the last seven days *of the data's clock* (ts), so a replay at 1,000× speed still produces meaningful windows.

   pythonCopy

   # bike/serve.py
   import json
   import os
   import sqlite3
   from collections import deque
   from pathlib import Path
   
   import mlflow.sklearn
   import numpy as np
   import pandas as pd
   from fastapi import FastAPI, HTTPException
   from prometheus_client import Counter, Gauge, Info, make_asgi_app
   from pydantic import BaseModel, Field
   
   from bike.drift import histogram, psi
   from bike.features import FEATURES, NUMERIC, to_matrix
   
   MODEL_DIR = Path(os.getenv("MODEL_DIR", "model"))
   DB = os.getenv("PREDICTIONS_DB", "predictions.db")
   WINDOW_HOURS = 24 * 7
   
   app = FastAPI(title="bike-demand")
   app.mount("/metrics", make_asgi_app())
   
   model = mlflow.sklearn.load_model(str(MODEL_DIR / "model"))
   reference = json.loads((MODEL_DIR / "reference.json").read_text())
   version = json.loads((MODEL_DIR / "VERSION").read_text())
   
   MODEL_INFO = Info("model", "Loaded model")
   MODEL_INFO.info({k: str(v) for k, v in version.items()})
   PREDICTIONS = Counter("model_predictions_total", "Predictions served")
   PSI = Gauge("model_psi", "PSI vs training reference over the last 7 data-days", ["signal"])
   MAE_7D = Gauge("model_mae_7d", "MAE over labelled predictions in the last 7 data-days")
   LABELLED_7D = Gauge("model_labelled_7d", "Labelled predictions in the last 7 data-days")
   VAL_MAE = Gauge("model_val_mae", "Holdout MAE of the loaded model at training time")
   VAL_MAE.set(version["val_mae"])
   
   window: deque[dict] = deque(maxlen=WINDOW_HOURS)
   
   
   def db() -> sqlite3.Connection:
       con = sqlite3.connect(DB)
       con.execute("create table if not exists predictions (id text primary key, ts text, prediction real, actual real)")
       return con
   
   
   class Row(BaseModel):
       id: str
       ts: str
       hr: int = Field(ge=0, le=23)
       weekday: int = Field(ge=0, le=6)
       workingday: int
       holiday: int
       weathersit: int = Field(ge=1, le=4)
       temp: float
       atemp: float
       hum: float
       windspeed: float
       season: int = Field(ge=1, le=4)
       level: float = Field(gt=0)
   
   
   class Label(BaseModel):
       id: str
       actual: float
   
   
   def refresh_gauges() -> None:
       if len(window) < 24:
           return
       frame = pd.DataFrame(window)
       PSI.labels("prediction").set(psi(reference["prediction"]["hist"], histogram(frame["prediction"], reference["prediction"]["edges"])))
       for col in NUMERIC:
           PSI.labels(col).set(psi(reference[col]["hist"], histogram(frame[col], reference[col]["edges"])))
       latest = pd.Timestamp(frame["ts"].max())
       since = (latest - pd.Timedelta(days=7)).isoformat()
       rows = db().execute("select prediction, actual from predictions where actual is not null and ts >= ?", (since,)).fetchall()
       LABELLED_7D.set(len(rows))
       if rows:
           MAE_7D.set(float(np.mean([abs(p - a) for p, a in rows])))
   
   
   @app.post("/predict")
   def predict(rows: list[Row]):
       if not 1 <= len(rows) <= 1000:
           raise HTTPException(422, "send 1-1000 rows")
       frame = pd.DataFrame([r.model_dump() for r in rows])
       preds = np.expm1(model.predict(to_matrix(frame))).clip(min=0)
       con = db()
       con.executemany("insert or replace into predictions (id, ts, prediction) values (?, ?, ?)",
                       [(r.id, r.ts, float(p)) for r, p in zip(rows, preds)])
       con.commit()
       for r, p in zip(rows, preds):
           window.append({**{f: getattr(r, f) for f in FEATURES}, "ts": r.ts, "prediction": float(p)})
       PREDICTIONS.inc(len(rows))
       refresh_gauges()
       return {"model_version": version["version"], "predictions": [{"id": r.id, "cnt": round(float(p), 1)} for r, p in zip(rows, preds)]}
   
   
   @app.post("/labels")
   def labels(items: list[Label]):
       con = db()
       n = sum(con.execute("update predictions set actual = ? where id = ?", (it.actual, it.id)).rowcount for it in items)
       con.commit()
       refresh_gauges()
       return {"matched": n}
   
   
   @app.get("/health")
   def health():
       return {"status": "ok", "model": version}

2. 4.2
   Test the service against a tiny model baked by the test itself — no MLflow server needed.

   pythonCopy

   # tests/test_serve.py
   import importlib
   import json
   
   import mlflow.sklearn
   import numpy as np
   from fastapi.testclient import TestClient
   from sklearn.ensemble import HistGradientBoostingRegressor
   
   from bike.drift import build_reference
   from bike.features import CATEGORICAL, FEATURES, NUMERIC, add_level, to_matrix
   from tests.test_features import make
   
   
   def bake_tiny_model(tmp_path):
       df = add_level(make(days=45))
       x = to_matrix(df)
       model = HistGradientBoostingRegressor(max_iter=30, random_state=0,
                                             categorical_features=[FEATURES.index(c) for c in CATEGORICAL]).fit(x, np.log1p(df["cnt"]))
       mlflow.sklearn.save_model(model, str(tmp_path / "model"))
       (tmp_path / "reference.json").write_text(json.dumps(build_reference(x, np.expm1(model.predict(x)), NUMERIC)))
       (tmp_path / "VERSION").write_text(json.dumps({"name": "bike-demand", "version": "0", "val_mae": 10.0}))
       return df
   
   
   def test_predict_then_label_updates_mae(tmp_path, monkeypatch):
       df = bake_tiny_model(tmp_path)
       monkeypatch.setenv("MODEL_DIR", str(tmp_path))
       monkeypatch.setenv("PREDICTIONS_DB", str(tmp_path / "p.db"))
       serve = importlib.reload(importlib.import_module("bike.serve"))
       client = TestClient(serve.app)
   
       rows = [{"id": str(i), "ts": r.dteday.isoformat(), **{f: (int(r[f]) if f in CATEGORICAL or f in ("workingday", "holiday") else float(r[f])) for f in FEATURES}}
               for i, r in df.tail(48).reset_index().iterrows()]
       out = client.post("/predict", json=rows).json()
       assert len(out["predictions"]) == 48 and out["model_version"] == "0"
   
       labels = [{"id": str(i), "actual": float(c)} for i, c in enumerate(df.tail(48)["cnt"])]
       assert client.post("/labels", json=labels).json()["matched"] == 48
       metrics = client.get("/metrics").text
       assert "model_mae_7d" in metrics and 'model_psi{signal="prediction"}' in metrics
       assert client.post("/predict", json=[]).status_code == 422

3. 4.3
   Bake the champion and run the service locally; check that /health names the registry version and that a prediction comes back.

   bashCopy

   pytest -q
   python bake.py
   uvicorn bike.serve:app --port 8000 &
   sleep 3
   curl -s localhost:8000/health | python3 -m json.tool
   curl -s localhost:8000/predict -H 'content-type: application/json' -d '[{"id":"x1","ts":"2012-01-01T08:00:00","hr":8,"weekday":0,"workingday":0,"holiday":0,"weathersit":1,"temp":0.3,"atemp":0.3,"hum":0.6,"windspeed":0.2,"season":1,"level":2300}]'
   kill %1

4. 4.4
   Write the Dockerfile. The model/ folder produced by bake.py is copied in; the image never talks to MLflow. The version label is stamped from VERSION at build time so docker inspect can answer "which model is this?".

   dockerfileCopy

   # syntax=docker/dockerfile:1
   FROM python:3.12-slim AS build
   COPY --from=ghcr.io/astral-sh/uv:0.5 /uv /bin/uv
   WORKDIR /app
   COPY pyproject.toml .
   COPY bike ./bike
   RUN uv venv /venv && VIRTUAL_ENV=/venv uv pip install --no-cache .
   
   FROM python:3.12-slim
   ARG MODEL_VERSION=unknown
   LABEL org.opencontainers.image.title="bike-demand" model.version="${MODEL_VERSION}"
   RUN useradd --create-home --uid 10001 app
   WORKDIR /app
   COPY --from=build /venv /venv
   COPY bike ./bike
   COPY model ./model
   RUN mkdir -p /data && chown app:app /data
   ENV PATH="/venv/bin:$PATH" PYTHONUNBUFFERED=1 MODEL_DIR=/app/model PREDICTIONS_DB=/data/predictions.db
   USER app
   EXPOSE 8000
   HEALTHCHECK --interval=30s --timeout=3s CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health')"
   CMD ["uvicorn", "bike.serve:app", "--host", "0.0.0.0", "--port", "8000"]

5. 4.5
   Build with the version from the baked VERSION file, run it, and commit.

   bashCopy

   V=$(python3 -c "import json; print(json.load(open('model/VERSION'))['version'])")
   docker build --build-arg MODEL_VERSION="$V" -t bike-demand:v"$V" .
   docker run -d --rm -p 8000:8000 --name bike bike-demand:v"$V"
   sleep 3 && curl -s localhost:8000/health && docker inspect bike --format '{{index .Config.Labels "model.version"}}'
   docker stop bike
   git add . && git commit -m "feat: promotion gate, bake, serving with drift and quality gauges, image" && git push

Pull requests run the tests and a smoke training; pushes to main (and a manual trigger with a --until date) train against a shared MLflow, run the promotion gate, and build and push the image only when the champion changed.

1. 5.1
   CI needs an MLflow server it can reach. A free DagsHub repository provides a hosted MLflow tracking server and registry; alternatively point MLFLOW_TRACKING_URI at any server you run. Set the two secrets, then make your laptop use the same server so the registry is one place.

   bashCopy

   # DagsHub: create a repo named mlops-loop, then Settings → Tokens
   gh secret set MLFLOW_TRACKING_URI --body "https://dagshub.com/YOUR_USER/mlops-loop.mlflow"
   gh secret set MLFLOW_TRACKING_USERNAME --body "YOUR_USER"
   gh secret set MLFLOW_TRACKING_PASSWORD --body "YOUR_DAGSHUB_TOKEN"
   # locally, same target:
   export MLFLOW_TRACKING_URI=https://dagshub.com/YOUR_USER/mlops-loop.mlflow MLFLOW_TRACKING_USERNAME=YOUR_USER MLFLOW_TRACKING_PASSWORD=YOUR_DAGSHUB_TOKEN
   python train.py --until 2011-12-31 && python promote.py 1

   Keeping the local SQLite server is fine for learning; the workflow below simply will not be able to register. The shared server is what makes "CI retrains and promotes" real.
2. 5.2
   The PR workflow: lint, tests, and a smoke training that uses only three months and never registers — it just proves the path runs.

   yamlCopy

   # .github/workflows/ci.yml
   name: ci
   on:
     pull_request:
   jobs:
     test:
       runs-on: ubuntu-latest
       steps:
         - uses: actions/checkout@v4
         - uses: astral-sh/setup-uv@v5
         - run: uv venv --python 3.12 && uv pip install -e '.[dev]'
         - run: .venv/bin/ruff check .
         - run: .venv/bin/pytest -q
         - name: Smoke training (local file store, no registry)
           env:
             MLFLOW_TRACKING_URI: file:./mlruns-ci
           run: |
             .venv/bin/python -c "from bike.data import download; download()"
             .venv/bin/python train.py --until 2011-04-30 --max-iter 50 || exit 1

3. 5.3
   The train workflow: on push to main or on demand with an until input, train, promote, and — only if the gate passed — bake, build and push the image tagged with the model version. The gate's exit code is the branch point.

   yamlCopy

   # .github/workflows/train.yml
   name: train
   on:
     push:
       branches: [main]
       paths: ["bike/**", "train.py", "promote.py", "bake.py", "pyproject.toml"]
     workflow_dispatch:
       inputs:
         until:
           description: "Last date of training data (YYYY-MM-DD)"
           default: "2011-12-31"
   
   env:
     MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_TRACKING_URI }}
     MLFLOW_TRACKING_USERNAME: ${{ secrets.MLFLOW_TRACKING_USERNAME }}
     MLFLOW_TRACKING_PASSWORD: ${{ secrets.MLFLOW_TRACKING_PASSWORD }}
   
   jobs:
     train:
       runs-on: ubuntu-latest
       permissions:
         contents: read
         packages: write
       outputs:
         promoted: ${{ steps.gate.outputs.promoted }}
       steps:
         - uses: actions/checkout@v4
         - uses: astral-sh/setup-uv@v5
         - run: uv venv --python 3.12 && uv pip install -e .
         - id: train
           run: |
             UNTIL="${{ github.event.inputs.until || '2011-12-31' }}"
             .venv/bin/python train.py --until "$UNTIL" | tee train.log
             echo "version=$(grep -o 'v[0-9]*:' train.log | tr -d 'v:')" >> "$GITHUB_OUTPUT"
         - id: gate
           run: |
             if .venv/bin/python promote.py "${{ steps.train.outputs.version }}" --allow-different-holdout; then
               echo "promoted=true" >> "$GITHUB_OUTPUT"
             else
               echo "promoted=false" >> "$GITHUB_OUTPUT"
             fi
         - if: steps.gate.outputs.promoted == 'true'
           run: .venv/bin/python bake.py
         - if: steps.gate.outputs.promoted == 'true'
           uses: docker/login-action@v3
           with:
             registry: ghcr.io
             username: ${{ github.actor }}
             password: ${{ secrets.GITHUB_TOKEN }}
         - if: steps.gate.outputs.promoted == 'true'
           uses: docker/build-push-action@v6
           with:
             context: .
             push: true
             build-args: MODEL_VERSION=${{ steps.train.outputs.version }}
             tags: |
               ghcr.io/${{ github.repository }}:latest
               ghcr.io/${{ github.repository }}:v${{ steps.train.outputs.version }}

   --allow-different-holdout is set in CI because a retrain with a later --until necessarily has a different holdout; the comparison is then "is the new model better on *its* recent holdout than the old one was on *its* holdout", which is the honest question when time has moved on. Locally, without the flag, promote.py protects you from comparing apples to oranges by accident.
4. 5.4
   Push, open a PR with a trivial change to see ci run, merge, and watch train train, gate, and (on the first run) promote and push the image.

   bashCopy

   git add .github && git commit -m "ci: tests + smoke train on PR; train-promote-build on main" && git push
   git switch -c ci/first-run && printf '\n' >> README.md && git commit -am "chore: trigger ci" && git push -u origin ci/first-run
   gh pr create --fill && gh pr checks --watch && gh pr merge --squash --delete-branch
   git switch main && git pull && gh run watch
   gh api "/user/packages/container/mlops-loop/versions" -q '.[].metadata.container.tags'

The 2012 data replays through the service at high speed with labels arriving one day late; Prometheus scrapes the gauges; a Grafana dashboard shows the feature drift, the prediction drift and the MAE rising; and an alert fires when quality degrades.

1. 6.1
   Write the replay client. It walks 2012 hour by hour, computes level from the actuals it has already *released* (so it is exactly what production would know), sends each day's 24 rows to /predict, and sends the previous day's labels — the one-day delay of real ground truth.

   pythonCopy

   # replay.py
   import argparse
   import time
   
   import httpx
   import pandas as pd
   
   from bike.data import load
   from bike.features import FEATURES, TARGET, add_level, daily_level
   
   
   def main() -> None:
       ap = argparse.ArgumentParser()
       ap.add_argument("--url", default="http://localhost:8000")
       ap.add_argument("--start", default="2012-01-01")
       ap.add_argument("--end", default="2012-12-31")
       ap.add_argument("--seconds-per-day", type=float, default=1.0)
       args = ap.parse_args()
   
       df = load()
       level = daily_level(df)               # uses only days before each date, so no leakage into the replay
       df = add_level(df, level)
       days = df[(df.dteday >= args.start) & (df.dteday <= args.end)].groupby("dteday")
       previous = None
       with httpx.Client(base_url=args.url, timeout=30) as c:
           for day, rows in days:
               payload = [{"id": str(r.instant), "ts": r.ts.isoformat(),
                           **{f: (float(r[f]) if f in ("temp", "atemp", "hum", "windspeed", "level") else int(r[f])) for f in FEATURES}}
                          for r in rows.itertuples()]
               out = c.post("/predict", json=payload).json()
               if previous is not None:
                   c.post("/labels", json=[{"id": str(r.instant), "actual": float(getattr(r, TARGET))} for r in previous.itertuples()])
               mae = (pd.Series([p["cnt"] for p in out["predictions"]]) - rows[TARGET].to_numpy()).abs().mean()
               print(f"{day.date()}  rows={len(rows)}  day-MAE={mae:5.1f}  model=v{out['model_version']}")
               previous = rows
               time.sleep(args.seconds_per_day)
   
   
   if __name__ == "__main__":
       main()

   One second per day means 2012 takes six minutes. Prometheus's 15-second scrape sees roughly every second week, which is enough for the dashboard; use --seconds-per-day 3 if you want smoother lines.
2. 6.2
   Write the monitoring stack: the service from the image, Prometheus scraping it, Grafana provisioned. If you already run the SRE track's stack, add the scrape job there instead.

   yamlCopy

   # deploy/compose.yml
   services:
     model:
       image: ${MODEL_IMAGE:-bike-demand:v1}
       ports: ["8000:8000"]
       volumes: ["predictions:/data"]
       restart: unless-stopped
   
     prometheus:
       image: prom/prometheus:v3.1.0
       volumes:
         - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
         - ./alerts.yml:/etc/prometheus/alerts.yml:ro
       ports: ["9090:9090"]
   
     grafana:
       image: grafana/grafana:11.4.0
       environment:
         GF_SECURITY_ADMIN_PASSWORD: admin
         GF_AUTH_ANONYMOUS_ENABLED: "true"
       volumes:
         - ./grafana/provisioning:/etc/grafana/provisioning:ro
         - ./grafana/dashboards:/var/lib/grafana/dashboards:ro
       ports: ["3000:3000"]
   
   volumes:
     predictions:
   
   # deploy/prometheus.yml
   global:
     scrape_interval: 5s
     evaluation_interval: 5s
   rule_files: [/etc/prometheus/alerts.yml]
   scrape_configs:
     - job_name: model
       static_configs:
         - targets: ["model:8000"]

   Copy the Grafana provisioning files from the SLO project (datasource pointing at http://prometheus:9090, and a dashboards provider reading /var/lib/grafana/dashboards).
3. 6.3
   Write the alert rules. Quality is the alert that matters — MAE over the last seven data-days more than 1.6× what the model achieved on its holdout, with at least 100 labelled rows so a quiet day cannot trigger it. Drift is a warning: it tells you *why* before quality tells you *that*.

   yamlCopy

   # deploy/alerts.yml
   groups:
     - name: model
       rules:
         - alert: ModelQualityDegraded
           expr: model_mae_7d > 1.6 * model_val_mae and model_labelled_7d >= 100
           for: 30s
           labels: {severity: page}
           annotations:
             summary: "7-day MAE {{ $value | printf \"%.0f\" }} vs holdout MAE — retrain"
         - alert: FeatureDrift
           expr: max without (instance) (model_psi{signal!="prediction"}) > 0.25
           for: 1m
           labels: {severity: ticket}
           annotations:
             summary: "Feature {{ $labels.signal }} PSI {{ $value | printf \"%.2f\" }} vs training reference"
         - alert: PredictionDrift
           expr: model_psi{signal="prediction"} > 0.25
           for: 1m
           labels: {severity: ticket}
           annotations:
             summary: "Prediction distribution PSI {{ $value | printf \"%.2f\" }}"

4. 6.4
   Build the dashboard in Grafana with these panels, then export it to deploy/grafana/dashboards/model.json: the loaded version as a stat from model_info; PSI per feature (model_psi, one line per signal); MAE vs holdout (model_mae_7d and model_val_mae on one panel); labelled count; predictions per minute.

   promqlCopy

   # Version stat panel (show the 'version' label)
   model_info
   
   # Drift, one line per signal
   model_psi
   
   # Quality: rolling MAE against what the model achieved offline
   model_mae_7d
   model_val_mae
   1.6 * model_val_mae
   
   # Throughput
   rate(model_predictions_total[1m]) * 60

5. 6.5
   Start the stack with the v1 image and replay the year. Watch the dashboard: level drifts off the chart within weeks (PSI in the single digits — 2012 volumes are outside anything the model saw), prediction PSI climbs through spring, and the MAE line crosses the alert threshold around April–May. In the reference run, Q1 MAE was about 43, Q2 about 71 against a holdout MAE of 32.

   bashCopy

   cd deploy && MODEL_IMAGE=bike-demand:v1 docker compose up -d && cd ..
   python replay.py --seconds-per-day 2
   # meanwhile: http://localhost:3000 (dashboard), http://localhost:9090/alerts (ModelQualityDegraded should fire in Q2)

   Notice which signal moved first. The level feature drifted in January; quality only degraded in April. Feature drift is an early warning; label-based quality is the truth — and it arrives late, which is why both exist.

The alert leads to a retrain on data through the end of Q1 2012, the gate promotes it, the new image ships, and the replay of Q2 shows the error falling — then write down what you would automate next.

1. 7.1
   Retrain with the data the model would have had at the end of March 2012, through the workflow so the whole path is exercised. The gate compares the candidate's holdout MAE (Feb–Mar 2012) with the champion's (Nov–Dec 2011); both are "the most recent two months the model had".

   bashCopy

   gh workflow run train -f until=2012-03-31
   gh run watch
   python -c "from mlflow import MlflowClient; c=MlflowClient(); v=c.get_model_version_by_alias('bike-demand','champion'); print('champion', v.version, v.tags)"

   Running locally instead: python train.py --until 2012-03-31 && python promote.py <version> --allow-different-holdout && python bake.py && docker build --build-arg MODEL_VERSION=<version> -t bike-demand:v<version> .
2. 7.2
   Deploy the new image and replay Q2 again. The predictions volume keeps the old predictions, so the first week of MAE still reflects v1 — watch it fall as new labelled predictions replace them.

   bashCopy

   V=$(python -c "from mlflow import MlflowClient; print(MlflowClient().get_model_version_by_alias('bike-demand','champion').version)")
   docker pull "ghcr.io/$(gh api user -q .login)/mlops-loop:v$V" 2>/dev/null && IMG="ghcr.io/$(gh api user -q .login)/mlops-loop:v$V" || IMG="bike-demand:v$V"
   cd deploy && MODEL_IMAGE="$IMG" docker compose up -d model && cd ..
   curl -s localhost:8000/health | python3 -c "import sys,json; print(json.load(sys.stdin)['model'])"
   python replay.py --start 2012-04-01 --end 2012-06-30 --seconds-per-day 2

   Reference numbers: Q2 2012 MAE about 71 with the 2011 model, about 50 after retraining through March. The alert clears once the 7-day window is dominated by the new model's predictions.
3. 7.3
   Write the README as a runbook for the loop: how to train, what the gate checks, how to bake and ship, what each alert means and what to do (retrain with --until = today, check the diff in the registry, deploy). Then commit the dashboard and the alert rules.

   bashCopy

   git add deploy README.md
   git commit -m "monitoring: prometheus + grafana stack, drift and quality alerts, replay client; README runbook"
   git push

4. 7.4
   Finish with the honest list of what is still manual: the alert does not trigger the workflow, the deploy is a docker compose up, and there is no shadow evaluation before promotion. Each is a next item below; write in the README which one you would do first and why.

   bashCopy

   # Optional: let the alert start the retrain. Alertmanager → a webhook that runs:
   #   gh workflow run train -f until=$(date +%F)
   # The on-call copilot project shows how to receive Alertmanager webhooks safely.

train.py fails with RESOURCE_DOES_NOT_EXIST or cannot register the model

The tracking URI points at a file store (mlruns/), which has no registry. Start mlflow server --backend-store-uri sqlite:///mlflow.db --artifacts-destination ./mlruns and set MLFLOW_TRACKING_URI=http://localhost:5000, or use the hosted server from the CI phase.

promote.py says refusing: candidate holdout ends … champion's ends …

That is the gate doing its job: the two models were evaluated on different windows. For a retrain that legitimately moves time forward, pass --allow-different-holdout; for a hyperparameter experiment, retrain with the same --until so the comparison is fair.

The service starts but /predict returns 500 with a category error

The categorical columns must have the same categories the model saw. to_matrix casts to category from the values present in the batch; if a batch contains a weathersit value of 4 that never appeared in training, scikit-learn rejects it. Add the full category lists to features.py (pd.Categorical(values, categories=[...])) so encoding is fixed, not inferred.

model_mae_7d never appears in /metrics

The gauge is only set once at least one labelled row is in the last 7 data-days. Confirm /labels returns matched > 0 — ids must match the ids used in /predict exactly (strings). In the replay, labels are sent for the previous day, so the gauge appears on day two.

PSI for temp is high even in January, before anything went wrong

The reference is the whole training period; comparing one winter week to a full year's temperatures always looks like drift. That is seasonality, not a bug. Options: a seasonal reference (same month last year), or alert only on prediction PSI and on quality, and treat feature PSI as a diagnostic — which is what the alert rules here do with severity: ticket.

The image build fails with model/: not found

bake.py has not been run in this checkout, or it ran against a different tracking server. Run it right before docker build; in CI it is a step of the same job for exactly this reason.

• Wire the ModelQualityDegraded alert to a webhook that runs gh workflow run train -f until=<today> — the on-call copilot project's receiver is a safe template — and make the deploy a Git change the GitOps project's Argo CD applies.
• Add shadow evaluation: before promotion, replay the champion's last 30 days of stored predictions through the candidate and compare MAE on real labels, not just the offline holdout.
• Log the replay's per-day MAE back to MLflow as a time series on the deployed version, so the registry page shows how a model aged.
• Swap the tabular model for a text classifier with the same loop — the serving, drift (embedding PSI) and label-delay mechanics are identical, which is the point of building the loop once.
• Add a data-quality gate in train.py (row counts, null rates, value ranges against the previous run) so a broken upstream extract cannot become a promoted model.