Client
-> HTTPS gateway or load balancer
-> authenticated FastAPI service
-> validated preprocessing + versioned model
-> structured logs, metrics, and traces
-> dashboards and alerts

Release path
Versioned source + dataset
-> train and evaluate
-> model registry or protected artifact store
-> container build and scan
-> staging tests
-> canary production release
-> monitor, promote, or roll back

house-price-mlops/
├── app/
│   ├── main.py
│   ├── inference.py
│   ├── schemas.py
│   └── settings.py
├── training/
│   ├── train.py
│   ├── evaluate.py
│   └── data_contract.py
├── tests/
│   ├── test_api.py
│   ├── test_model.py
│   └── test_contract.py
├── artifacts/              # local demo; production uses governed storage
│   ├── model.joblib
│   └── metadata.json
├── deployment/
├── .github/workflows/ci.yml
├── Dockerfile
├── requirements.txt
├── README.md
├── MODEL_CARD.md
└── RUNBOOK.md

dataset: house_prices_v1
features:
  bedrooms: {type: integer, minimum: 0, maximum: 20}
  size_sqft: {type: number, exclusive_minimum: 0, maximum: 50000}
  location_score: {type: number, minimum: 0, maximum: 10}
target:
  name: sale_price_usd
  type: number
split:
  strategy: time_aware
owner: housing-ml-team

import json
from pathlib import Path
import joblib
import pandas as pd
from sklearn.impute import SimpleImputer
from sklearn.linear_model import Ridge
from sklearn.metrics import mean_absolute_error
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler

FEATURES = ["bedrooms", "size_sqft", "location_score"]
train = pd.read_csv("data/train.csv")
test = pd.read_csv("data/test.csv")

model = Pipeline([
    ("impute", SimpleImputer(strategy="median")),
    ("scale", StandardScaler()),
    ("regressor", Ridge(alpha=1.0))
])
model.fit(train[FEATURES], train["sale_price_usd"])
mae = float(mean_absolute_error(
    test["sale_price_usd"], model.predict(test[FEATURES])
))

Path("artifacts").mkdir(exist_ok=True)
joblib.dump(model, "artifacts/model.joblib")
Path("artifacts/metadata.json").write_text(json.dumps({
    "model_version": "house-price-1.0.0",
    "data_version": "houses-2026-07-01",
    "features": FEATURES,
    "test_mae": mae
}, indent=2))

[ ] Data contract and leakage checks pass
[ ] Candidate improves on baseline
[ ] Important slices meet accepted error thresholds
[ ] Boundary and robustness tests pass
[ ] Artifact integrity and dependency checks pass
[ ] Latency, memory, image size, and cost are acceptable
[ ] Model card, lineage, owner, and approval evidence are complete

from pydantic import BaseModel, ConfigDict, Field

class HouseFeatures(BaseModel):
    model_config = ConfigDict(extra="forbid")
    bedrooms: int = Field(ge=0, le=20)
    size_sqft: float = Field(gt=0, le=50000)
    location_score: float = Field(ge=0, le=10)

class PredictionResponse(BaseModel):
    prediction_id: str
    estimated_price_usd: float
    model_version: str
    warning: str = "Demonstration estimate; not a professional valuation."

from contextlib import asynccontextmanager
from pathlib import Path
import json
import time
import uuid
import joblib
import pandas as pd
from fastapi import FastAPI, HTTPException
from app.schemas import HouseFeatures, PredictionResponse

model = None
metadata = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global model, metadata
    model = joblib.load("artifacts/model.joblib")
    metadata = json.loads(Path("artifacts/metadata.json").read_text())
    yield
    model = None

app = FastAPI(title="House Price API", version="1.0.0", lifespan=lifespan)

@app.get("/health/live")
def live():
    return {"status": "alive"}

@app.get("/health/ready")
def ready():
    if model is None:
        raise HTTPException(status_code=503, detail="Model is not ready")
    return {"status": "ready", "model_version": metadata["model_version"]}

@app.post("/v1/predict", response_model=PredictionResponse)
def predict(features: HouseFeatures):
    prediction_id = str(uuid.uuid4())
    started = time.perf_counter()
    frame = pd.DataFrame([features.model_dump()])
    value = max(0.0, float(model.predict(frame)[0]))
    # Emit a structured event with prediction ID, version, status,
    # and duration; avoid logging raw feature values by default.
    duration_ms = round((time.perf_counter() - started) * 1000, 2)
    return PredictionResponse(
        prediction_id=prediction_id,
        estimated_price_usd=round(value, 2),
        model_version=metadata["model_version"]
    )

from fastapi.testclient import TestClient
from app.main import app

def test_prediction_contract():
    with TestClient(app) as client:
        response = client.post("/v1/predict", json={
            "bedrooms": 3,
            "size_sqft": 1800,
            "location_score": 8.0
        })
    assert response.status_code == 200
    body = response.json()
    assert body["estimated_price_usd"] >= 0
    assert body["model_version"] == "house-price-1.0.0"
    assert body["prediction_id"]

def test_invalid_size_is_rejected():
    with TestClient(app) as client:
        response = client.post("/v1/predict", json={
            "bedrooms": 3, "size_sqft": -10, "location_score": 8
        })
    assert response.status_code == 422

FROM python:3.12-slim
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
RUN groupadd --system app && useradd --system --gid app app
WORKDIR /service
COPY requirements.txt .
RUN pip install --no-cache-dir --require-hashes -r requirements.txt
COPY app ./app
COPY artifacts ./artifacts
RUN chown -R app:app /service
USER app
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

docker build -t house-price-api:1.0.0 .
docker run --rm -p 8000:8000 house-price-api:1.0.0

curl http://localhost:8000/health/ready
curl -X POST http://localhost:8000/v1/predict \
  -H 'Content-Type: application/json' \
  -d '{"bedrooms":3,"size_sqft":1800,"location_score":8}'

name: ci
on:
  pull_request:
  push:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    permissions:
      contents: read
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
          cache: pip
      - run: pip install --require-hashes -r requirements-dev.txt
      - run: ruff check .
      - run: pytest -q
      - run: python training/evaluate.py --verify-artifacts
      - run: docker build -t house-price-api:${{ github.sha }} .
      # Add organization-approved secret, dependency, image,
      # infrastructure, provenance, and policy checks.

service:
  min_replicas: 2        # illustrative; choose from tests and objectives
  max_replicas: 10
  request_timeout_seconds: 5
  graceful_shutdown_seconds: 30
  concurrency_per_replica: 16
  readiness_path: /health/ready
  liveness_path: /health/live
  resources: {cpu: '1', memory: 1Gi}
  autoscaling_signal: concurrent_requests
  deployment_strategy: canary

Approved immutable release
-> staging smoke, contract, security, load, and resilience tests
-> synthetic prediction
-> shadow representative production requests
-> small canary
-> compare errors, latency, saturation, predictions, slices, outcomes, and cost
-> gradually promote or stop and roll back
-> record evidence and reconcile affected predictions

[ ] Intended use, prohibited use, owner, architecture, objectives, and risks documented
[ ] Data source, contract, split, quality, lineage, privacy, and version controlled
[ ] Model reproducible and compared across baseline, slices, robustness, latency, and cost
[ ] Model, preprocessing, schema, metadata, evaluation, and approval registered immutably
[ ] API typed, versioned, validated, tested, observable, and safely handles errors
[ ] Container non-root, minimal, locked, scanned, provenanced, verified, and benchmarked
[ ] CI/CD tests code, data, model, security, image, contracts, infrastructure, and deployment
[ ] Production uses identity, secrets, TLS, network, request, rate, concurrency, timeout, and cost controls
[ ] Capacity, scaling, shutdown, overload, dependencies, quotas, and unit cost tested
[ ] Service, data, model, outcome, security, and cost monitoring is actionable
[ ] Staging, synthetic, canary, promotion, rollback, restoration, and reconciliation demonstrated
[ ] README, model card, dashboards, alerts, runbook, feedback, maintenance, and retirement complete