Take an AI Model From Laptop to Production

numpy==2.5.3
pandas==3.0.6
scikit-learn==1.9.1
joblib==1.6.0
fastapi==0.142.4
pydantic==2.13.5
uvicorn==0.54.0
httpx==0.28.1
pytest==9.1.1
# Resolved transitive dependencies, pinned for repeatable local/CI/container installs.
annotated-doc==0.0.5
annotated-types==0.8.0
anyio==4.15.1
certifi==2026.7.22
click==8.5.0
cloudpickle==3.1.2
colorama==0.4.6
h11==0.16.0
httpcore==1.0.9
idna==3.20
iniconfig==2.3.1
narwhals==2.26.0
opentelemetry-api==1.45.1
packaging==26.3
pluggy==1.6.0
pydantic_core==2.46.5
Pygments==2.21.0
python-dateutil==2.9.0.post0
scipy==1.18.1
six==1.17.0
starlette==1.7.0
threadpoolctl==3.7.0
typing-inspection==0.4.4
typing_extensions==4.16.0
tzdata==2026.5

"""Offline-only training: python -m src.train --version v1."""
import argparse
from datetime import datetime, timezone
import hashlib
from io import BytesIO
from importlib.metadata import version as package_version
import json
from pathlib import Path
import platform
import tempfile

import joblib
import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, roc_auc_score, confusion_matrix
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

from src.contract import ROOT, SEED, NUMERIC, CATEGORIES, FEATURES, DATA_PATH, DATA_SHA256, DATA_URL, DATA_REVISION, validate_version


def read_dataset(path: Path = DATA_PATH) -> pd.DataFrame:
    content = path.read_bytes()
    if hashlib.sha256(content).hexdigest() != DATA_SHA256:
        raise ValueError("Dataset checksum mismatch; run python -m scripts.prepare_data.")
    frame = pd.read_csv(BytesIO(content))
    if frame.shape != (7043, 21) or not frame.customerID.is_unique:
        raise ValueError("Expected the original 7,043-row, 21-column IBM sample.")
    frame["TotalCharges"] = pd.to_numeric(frame.TotalCharges.replace(r"^\s*$", np.nan, regex=True), errors="raise")
    if set(frame.Churn.unique()) != {"No", "Yes"}:
        raise ValueError("Unexpected target values.")
    for field, categories in CATEGORIES.items():
        if not set(frame[field].unique()).issubset(categories):
            raise ValueError(f"Unexpected category in {field}.")
    return frame


def split_data(frame):
    return train_test_split(frame[FEATURES], frame.Churn.eq("Yes").astype(int),
                            test_size=0.2, stratify=frame.Churn, random_state=SEED)


def make_pipeline(c=1.0):
    numeric = Pipeline([("impute", SimpleImputer(strategy="median")), ("scale", StandardScaler())])
    categorical = Pipeline([("impute", SimpleImputer(strategy="most_frequent")),
                            ("encode", OneHotEncoder(handle_unknown="ignore"))])
    prepare = ColumnTransformer([("numeric", numeric, NUMERIC), ("categorical", categorical, list(CATEGORIES))])
    return Pipeline([("prepare", prepare), ("classifier", LogisticRegression(C=c, max_iter=2000, random_state=SEED))])


def train(model_version="v1", models_dir: Path = ROOT / "models", c=1.0):
    validate_version(model_version)
    target = models_dir / model_version
    if target.exists():
        raise FileExistsError(f"{model_version} already exists. Versions are immutable; choose a new version.")
    X_train, X_test, y_train, y_test = split_data(read_dataset())
    pipeline = make_pipeline(c)
    pipeline.fit(X_train, y_train)
    probability = pipeline.predict_proba(X_test)[:, list(pipeline.classes_).index(1)]
    prediction = (probability >= 0.5).astype(int)
    metrics = {
        "accuracy": accuracy_score(y_test, prediction),
        "precision": precision_score(y_test, prediction, zero_division=0),
        "recall": recall_score(y_test, prediction, zero_division=0),
        "f1": f1_score(y_test, prediction, zero_division=0),
        "roc_auc": roc_auc_score(y_test, probability),
        "confusion_matrix": confusion_matrix(y_test, prediction, labels=[0, 1]).tolist(),
        "majority_accuracy": float((y_test == y_train.mode().iloc[0]).mean()),
    }
    metadata = {
        "schema_version": 1, "model_version": model_version,
        "trained_at": datetime.now(timezone.utc).isoformat(),
        "algorithm": "LogisticRegression", "parameters": {"C": c, "max_iter": 2000},
        "features": FEATURES, "categories": CATEGORIES, "threshold": 0.5,
        "training_rows": len(X_train), "test_rows": len(X_test), "seed": SEED,
        "dataset": {"id": "IBM Telco Customer Churn", "source": DATA_URL, "revision": DATA_REVISION, "sha256": DATA_SHA256},
        "metrics": metrics, "python_version": platform.python_version(),
        "packages": {name: package_version(name) for name in ["numpy", "pandas", "scikit-learn", "joblib"]},
    }
    models_dir.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory(prefix=".training-", dir=models_dir) as temporary:
        stage = Path(temporary)
        joblib.dump(pipeline, stage / "model.joblib")
        metadata["artifact_sha256"] = hashlib.sha256((stage / "model.joblib").read_bytes()).hexdigest()
        (stage / "metadata.json").write_text(json.dumps(metadata, indent=2) + "\n", encoding="utf-8")
        reloaded = joblib.load(stage / "model.joblib")  # Only this process's trusted output.
        np.testing.assert_allclose(reloaded.predict_proba(X_test), pipeline.predict_proba(X_test), rtol=0, atol=0)
        # Rename an already complete directory; readers never see a half-written version.
        stage.rename(target)
    print(json.dumps(metadata, indent=2))
    return metadata


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--version", default="v1")
    parser.add_argument("--c", type=float, default=1.0, help="Logistic Regression inverse regularization strength")
    arguments = parser.parse_args()
    if not np.isfinite(arguments.c) or arguments.c <= 0:
        parser.error("--c must be finite and positive")
    train(arguments.version, c=arguments.c)

"""Strict public contract: required fields, no silent string/bool coercion."""
from typing import Literal
from pydantic import BaseModel, ConfigDict, Field, model_validator


class Customer(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid", allow_inf_nan=False)
    tenure: int = Field(ge=0, le=120, description="Months; training sample spans 0–72.")
    MonthlyCharges: float = Field(ge=0, le=500)
    TotalCharges: float | None = Field(ge=0, le=50000, description="Required key; null means unknown and uses the training median.")
    Contract: Literal["Month-to-month", "One year", "Two year"]
    PaymentMethod: Literal["Electronic check", "Mailed check", "Bank transfer (automatic)", "Credit card (automatic)"]
    InternetService: Literal["DSL", "Fiber optic", "No"]
    OnlineSecurity: Literal["Yes", "No", "No internet service"]
    TechSupport: Literal["Yes", "No", "No internet service"]

    @model_validator(mode="after")
    def consistent_services(self):
        for field in ("OnlineSecurity", "TechSupport"):
            if (self.InternetService == "No") != (getattr(self, field) == "No internet service"):
                raise ValueError(f"{field} must agree with InternetService.")
        return self


class Prediction(BaseModel):
    predicted_class: Literal["stay", "churn"]
    churn_probability: float = Field(ge=0, le=1)
    model_version: str
    inference_latency_ms: float = Field(ge=0)

"""python -m uvicorn api.main:app --host 127.0.0.1 --port 8000"""
from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import FastAPI
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from api.schemas import Customer, Prediction
from src.contract import ROOT
from src.model_loader import load_active
from src.predict import predict
from src.logging_utils import log_event


def create_app(config_path: Path = ROOT / "config" / "model.json", models_dir: Path = ROOT / "models"):
    @asynccontextmanager
    async def lifespan(app):
        # No artifact paths/versions are accepted from HTTP clients.
        try:
            app.state.model = load_active(config_path, models_dir)
        except Exception as error:
            log_event("model_load", "failed", error_type=type(error).__name__)
            raise
        log_event("model_load", "ready", app.state.model.metadata.model_version)
        yield
        app.state.model = None

    application = FastAPI(title="Customer Churn Prediction Service", version="1.0.0", lifespan=lifespan)

    @application.exception_handler(RequestValidationError)
    async def invalid_request(request, error):
        log_event("request_validation", "rejected", application.state.model.metadata.model_version, status_code=422)
        # Keep locations/messages useful without echoing customer values in errors.
        details = [{"loc": list(item["loc"]), "msg": item["msg"], "type": item["type"]}
                   for item in error.errors()]
        return JSONResponse(status_code=422, content={"detail": details})

    @application.get("/health")
    def health():
        return {"status": "ok", "model_loaded": True, "active_version": application.state.model.metadata.model_version}

    @application.get("/model-info")
    def model_info():
        metadata = application.state.model.metadata
        return {"active_version": metadata.model_version, "algorithm": metadata.algorithm,
                "trained_at": metadata.trained_at, "evaluation_metrics": metadata.metrics.model_dump(),
                "threshold": metadata.threshold, "feature_schema": Customer.model_json_schema()}

    @application.post("/predict", response_model=Prediction)
    def predict_customer(customer: Customer):
        loaded = application.state.model
        try:
            result = predict(customer, loaded)
        except Exception as error:
            log_event("inference", "failed", loaded.metadata.model_version, error_type=type(error).__name__, status_code=500)
            return JSONResponse(status_code=500, content={"detail": "Inference failed; consult operator logs."})
        log_event("inference", "ok", result.model_version, inference_latency_ms=result.inference_latency_ms, status_code=200)
        return result

    return application


app = create_app()

from fastapi.testclient import TestClient
import pytest
from api.main import create_app
from src.model_loader import ModelLoadError


def test_health_info_and_real_prediction(artifacts, configuration, customer):
    with TestClient(create_app(configuration, artifacts)) as client:
        assert client.get("/health").json() == {"status": "ok", "model_loaded": True, "active_version": "v1"}
        info = client.get("/model-info").json()
        assert info["algorithm"] == "LogisticRegression"
        assert "TotalCharges" in info["feature_schema"]["required"]
        response = client.post("/predict", json=customer)
        assert response.status_code == 200
        result = response.json()
        assert result["model_version"] == "v1"
        assert result["predicted_class"] in {"stay", "churn"}
        assert 0 <= result["churn_probability"] <= 1
        assert result["inference_latency_ms"] >= 0
        assert client.post("/predict", json={**customer, "TotalCharges": None}).status_code == 200


@pytest.mark.parametrize("field,value", [
    ("tenure", -1), ("tenure", 121), ("tenure", "12"), ("tenure", True),
    ("MonthlyCharges", -1), ("MonthlyCharges", 501), ("MonthlyCharges", "70"),
    ("TotalCharges", -1), ("Contract", "Forever"), ("PaymentMethod", "cash"),
    ("InternetService", "Unknown"), ("OnlineSecurity", "Maybe"), ("TechSupport", "No internet service"),
])
def test_invalid_values_rejected(artifacts, configuration, customer, field, value):
    with TestClient(create_app(configuration, artifacts)) as client:
        response = client.post("/predict", json={**customer, field: value})
        assert response.status_code == 422
        assert response.json()["detail"]


def test_missing_and_extra_fields_rejected(artifacts, configuration, customer):
    with TestClient(create_app(configuration, artifacts)) as client:
        del customer["tenure"]
        assert client.post("/predict", json=customer).status_code == 422
        assert client.post("/predict", json={**customer, "tenure": 12, "customer_id": "private"}).status_code == 422


def test_missing_model_fails_startup(configuration, tmp_path):
    with pytest.raises(ModelLoadError, match="missing"):
        with TestClient(create_app(configuration, tmp_path)):
            pass

"""Educational feature-shift checks, not a retraining or performance decision."""
import argparse
import json
from pathlib import Path
import numpy as np
import pandas as pd
from src.contract import NUMERIC, CATEGORIES, FEATURES, ROOT, validate_version


def reference_statistics(training: pd.DataFrame):
    return {
        "schema_version": 1, "training_rows": len(training),
        "numeric": {name: {"mean": float(training[name].mean()),
                           "std": float(training[name].std(ddof=0)),
                           "missing_fraction": float(training[name].isna().mean())} for name in NUMERIC},
        "categorical": {name: {str(k): float(v) for k, v in training[name].value_counts(normalize=True).items()}
                        for name in CATEGORIES},
    }


def check_drift(batch: pd.DataFrame, reference: dict):
    if len(batch) < 50:
        raise ValueError("Use at least 50 rows; tiny batches make distribution comparisons unstable.")
    if batch.columns.tolist() != FEATURES or reference.get("schema_version") != 1:
        raise ValueError("Batch/reference feature contract mismatch.")
    results = {}
    for name in NUMERIC:
        values = pd.to_numeric(batch[name], errors="raise")
        nonmissing = values.dropna().to_numpy(dtype=float)
        if len(nonmissing) == 0 or not np.isfinite(nonmissing).all():
            raise ValueError(f"{name} needs finite observed values.")
        baseline = reference["numeric"][name]
        change = abs(float(values.mean()) - baseline["mean"])
        # Constant-reference columns use an explicit equality test, not division by zero.
        scale = baseline["std"]
        shift = change / scale if scale > 0 else (0.0 if change == 0 else None)
        missing_change = abs(float(values.isna().mean()) - baseline["missing_fraction"])
        status = "DRIFT DETECTED" if shift is None or shift >= 0.5 or missing_change >= 0.1 else "OK"
        results[name] = {"status": status, "absolute_mean_shift_in_training_std": shift,
                         "missing_fraction_change": missing_change}
    for name, categories in CATEGORIES.items():
        if batch[name].isna().any() or not set(batch[name]).issubset(categories):
            raise ValueError(f"Unexpected or missing category in {name}.")
        observed = batch[name].value_counts(normalize=True).to_dict()
        expected = reference["categorical"][name]
        total_variation = 0.5 * sum(abs(observed.get(category, 0) - expected.get(category, 0)) for category in categories)
        results[name] = {"status": "WARNING" if total_variation >= 0.15 else "OK",
                         "total_variation_distance": total_variation}
    return {"rows": len(batch), "features": results,
            "action": "Investigate data quality and labeled performance; drift alone does not justify retraining."}


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--version", default="v1")
    parser.add_argument("--batch", type=Path, required=True)
    args = parser.parse_args()
    validate_version(args.version)
    reference = json.loads((ROOT / "models" / args.version / "reference.json").read_text())
    result = check_drift(pd.read_csv(args.batch)[FEATURES], reference)
    for name, item in result["features"].items():
        print(f"{name}: {item['status']}")
    print(result["action"])


if __name__ == "__main__":
    main()

"""Operator-only version selection; restart the API after a successful switch."""
import argparse
import json
import os
from pathlib import Path
import tempfile
from src.contract import ROOT, validate_version
from src.model_loader import load_version
from src.logging_utils import log_event


def switch(version: str, config_path: Path = ROOT / "config" / "model.json", models_dir: Path = ROOT / "models"):
    validate_version(version)
    load_version(version, models_dir)  # Reject missing/invalid targets before altering config.
    content = json.dumps({"active_version": version}, indent=2) + "\n"
    temporary = None
    try:
        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", dir=config_path.parent, delete=False) as file:
            temporary = Path(file.name)
            file.write(content)
            file.flush()
            os.fsync(file.fileno())
        os.replace(temporary, config_path)
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)
    log_event("model_configuration", "selected_restart_required", version)


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--version", required=True)
    switch(parser.parse_args().version)
    print("Configuration updated. Restart every API worker/container to load this version.")

import json
from fastapi.testclient import TestClient
import pytest
from api.main import create_app
from scripts.switch_model import switch
from src.model_loader import ModelLoadError
from src.train import train


def test_switch_requires_restart_and_rollback_restores_predictions(artifacts, configuration, customer):
    train("v2", artifacts, c=0.5)
    with TestClient(create_app(configuration, artifacts)) as old_worker:
        first = old_worker.post("/predict", json=customer).json()
        switch("v2", configuration, artifacts)
        assert old_worker.get("/health").json()["active_version"] == "v1"
        with TestClient(create_app(configuration, artifacts)) as new_worker:
            assert new_worker.get("/model-info").json()["active_version"] == "v2"
            second = new_worker.post("/predict", json=customer).json()
            assert second["model_version"] == "v2"
            assert second["churn_probability"] != first["churn_probability"]
    switch("v1", configuration, artifacts)
    with TestClient(create_app(configuration, artifacts)) as rollback_worker:
        third = rollback_worker.post("/predict", json=customer).json()
        assert third["model_version"] == "v1"
        assert third["churn_probability"] == first["churn_probability"]
        assert third["predicted_class"] == first["predicted_class"]


def test_failed_switch_keeps_existing_configuration(artifacts, configuration):
    before = configuration.read_bytes()
    with pytest.raises(ModelLoadError):
        switch("v999", configuration, artifacts)
    assert configuration.read_bytes() == before
    assert not list(configuration.parent.glob("tmp*"))

FROM python:3.13.16-slim-bookworm@sha256:a1165e272e578941b84abc79e4ab38a0305cd12803a5c4247979ac7655f4d641

ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1 PIP_DISABLE_PIP_VERSION_CHECK=1
WORKDIR /app
COPY requirements.txt .
RUN python -m pip install --no-cache-dir --no-compile -r requirements.txt \
    && python -m pip check \
    && useradd --create-home --uid 10001 appuser

# No raw data, training scripts, test suite, credentials, or caches in the image.
COPY api/ ./api/
COPY src/__init__.py src/contract.py src/model_loader.py src/predict.py src/logging_utils.py ./src/
COPY --chown=10001:10001 config/ ./config/
# Training publishes a private temporary directory; give the runtime user access.
COPY --chown=10001:10001 models/ ./models/
USER 10001:10001
EXPOSE 8000
HEALTHCHECK --interval=10s --timeout=3s --start-period=20s --retries=3 \
    CMD python -c "import json,urllib.request; r=json.load(urllib.request.urlopen('http://127.0.0.1:8000/health',timeout=2)); assert r['status']=='ok' and r['model_loaded'] is True" || exit 1
CMD ["python", "-m", "uvicorn", "api.main:app", "--host", "0.0.0.0", "--port", "8000", "--no-access-log"]