Build Your First End-to-End AI Project: From Data to Deployed API

Build your first AI project as a complete loop: define a business problem, prepare data, train and evaluate a model, package it behind an API, deploy it, and monitor it. The example is an e-commerce API that predicts whether an order is likely to be returned, helping the business cut reverse-logistics costs and improve product-page guidance. All code below was run and tested; the numbers quoted come from those runs.

Problem statement

An online retailer wants to reduce avoidable returns, which create shipping costs, restocking work, inventory distortion, and customer-service load. It wants a real-time service that estimates, when an order is placed, the probability the item will be returned. Business goal: reduce return-related costs without harming the customer experience. ML goal: given order and product attributes known at checkout, predict P(return = 1 | order features). Returns are a good first project because historical outcomes provide the labels and the value is measurable.

Conceptual explanation

An end-to-end system has six stages. Data collection pulls orders, products, customers, deliveries, and return outcomes. Feature engineering turns them into model-ready features computed only from information available at order time. Training and evaluation compares a simple baseline with a stronger model on a chronological holdout and measures ROC-AUC, average precision, and calibration. Packaging saves the pipeline (preprocessing plus model) with metadata that makes it verifiable and reproducible. Serving exposes a validated, authenticated REST endpoint. Monitoring tracks errors, latency, input and score drift, and eventually the actual return outcomes, feeding retraining.

A production ML system is not a notebook with a model: it needs reproducible training, versioned and integrity-checked artifacts, strict input validation, access control, observability, and a feedback path from live outcomes.

Runnable Python code

Project structure and requirements

ecommerce-return-risk/
  requirements.txt
  train.py         # data -> model artifact + metadata (hash, versions, baseline)
  app.py           # FastAPI service: auth, validation, logging, health/ready
  monitor.py       # drift check over logged predictions
  models/          # return_risk_v1.joblib, return_risk_v1.json
  logs/            # predictions.jsonl (rotated)

The earlier version pinned late-2024 releases; the versions below are the ones this project was tested with. Pin the exact versions you deploy, and use the same scikit-learn version for training and serving, since a pickled pipeline is only reliable under the version that created it.

# Versions this project was tested with; pin what you deploy
pandas==3.0.5
numpy==2.5.3
scikit-learn==1.9.1
joblib==1.6.0
fastapi==0.162.0
uvicorn[standard]==0.54.0
pydantic==2.13.5

train.py

Training compares a logistic-regression baseline with gradient boosting and keeps whichever has the better average precision. It writes the pipeline plus a metadata file holding the artifact’s SHA-256 hash, the scikit-learn version, the feature order, metrics, business-set risk-tier cut-offs, and a reference score distribution for drift monitoring. First-time customers get a missing customer return rate instead of an invented one — the earlier generator gave every first-time buyer a return history, and its sample request sent is_first_purchase: 1 together with a 22% personal return rate.

"""Train the return-risk model. Run: python train.py"""
import hashlib
import json
from pathlib import Path

import joblib
import numpy as np
import pandas as pd
import sklearn
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import (average_precision_score, brier_score_loss,
                             roc_auc_score)
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

MODEL_VERSION = "return-risk-v1"
MODEL_DIR = Path("models")
MODEL_PATH = MODEL_DIR / f"{MODEL_VERSION}.joblib"
META_PATH = MODEL_DIR / f"{MODEL_VERSION}.json"
SEED = 42
CATEGORIES = ["apparel", "electronics", "home", "beauty", "footwear"]
NUMERIC = ["price", "discount_pct", "customer_return_rate",
           "product_return_rate", "estimated_delivery_days",
           "is_first_purchase", "review_rating"]
FEATURES = ["category", *NUMERIC]

def create_synthetic_data(n_rows=20_000):
    """Replace with a point-in-time query over orders, products,
    customers, shipments, and returns."""
    rng = np.random.default_rng(SEED)
    category = rng.choice(CATEGORIES, n_rows, p=[.35, .2, .2, .1, .15])
    first = rng.random(n_rows) < 0.25
    df = pd.DataFrame({
        "category": category,
        "price": np.round(rng.lognormal(6.2, 0.6, n_rows), 2),
        "discount_pct": rng.integers(0, 61, n_rows),
        # A first purchase has no return history: missing, not 0
        "customer_return_rate": np.where(
            first, np.nan, np.round(rng.beta(2, 8, n_rows), 3)),
        "product_return_rate": np.round(rng.beta(2, 8, n_rows), 3),
        "estimated_delivery_days": rng.integers(1, 11, n_rows),
        "is_first_purchase": first.astype(int),
        "review_rating": np.clip(np.round(rng.normal(4.1, 0.7, n_rows), 2),
                                 1, 5),
    })
    base = df["category"].map({"apparel": 0.6, "footwear": 0.4,
                               "electronics": -0.2, "home": -0.3,
                               "beauty": -0.1})
    logit = (-3.4 + base + 0.02 * df["discount_pct"]
             + 3.0 * df["customer_return_rate"].fillna(0.25)
             + 3.0 * df["product_return_rate"]
             + 0.06 * df["estimated_delivery_days"]
             - 0.4 * (df["review_rating"] - 4.0))
    df["was_returned"] = rng.binomial(1, 1 / (1 + np.exp(-logit)))
    return df

def preprocessing(scale):
    numeric = (make_pipeline(SimpleImputer(add_indicator=True),
                             StandardScaler()) if scale else "passthrough")
    return ColumnTransformer([
        ("num", numeric, NUMERIC),
        ("cat", OneHotEncoder(handle_unknown="ignore",
                              sparse_output=False), ["category"])])

def main():
    df = create_synthetic_data()
    # Synthetic rows have no timestamps; real data needs a time split
    X_train, X_test, y_train, y_test = train_test_split(
        df[FEATURES], df["was_returned"], test_size=0.2,
        stratify=df["was_returned"], random_state=SEED)

    candidates = {
        "logistic baseline": make_pipeline(
            preprocessing(scale=True), LogisticRegression(max_iter=1000)),
        "gradient boosting": make_pipeline(  # handles NaN natively
            preprocessing(scale=False),
            HistGradientBoostingClassifier(random_state=SEED)),
    }
    results = {}
    for name, pipe in candidates.items():
        proba = pipe.fit(X_train, y_train).predict_proba(X_test)[:, 1]
        results[name] = {
            "roc_auc": round(roc_auc_score(y_test, proba), 4),
            "average_precision": round(
                average_precision_score(y_test, proba), 4),
            "brier": round(brier_score_loss(y_test, proba), 4)}
    best = max(results, key=lambda n: results[n]["average_precision"])
    model = candidates[best]
    test_proba = model.predict_proba(X_test)[:, 1]

    MODEL_DIR.mkdir(exist_ok=True)
    joblib.dump(model, MODEL_PATH)
    metadata = {
        "model_version": MODEL_VERSION,
        "algorithm": best,
        "sha256": hashlib.sha256(MODEL_PATH.read_bytes()).hexdigest(),
        "sklearn_version": sklearn.__version__,
        "features": FEATURES,
        "return_rate": round(float(df["was_returned"].mean()), 4),
        "metrics": results,
        # Reference distribution for drift monitoring
        "score_deciles": np.quantile(test_proba,
                                     np.linspace(0, 1, 11)).tolist(),
        "risk_tiers": {"high": 0.65, "medium": 0.35},  # set by the business
    }
    META_PATH.write_text(json.dumps(metadata, indent=2), encoding="utf-8")
    print(json.dumps({k: metadata[k] for k in
                      ("algorithm", "return_rate", "metrics")}, indent=2))

if __name__ == "__main__":
    main()

On the synthetic data (27% return rate) the logistic baseline won: ROC-AUC 0.699 and average precision 0.463, against 0.688 and 0.444 for gradient boosting — because the simulated relationship is close to linear. Establishing that baseline is why the comparison exists. The random split is used only because synthetic rows have no timestamps; real order data needs a chronological holdout, with customer and product return rates computed from windows ending before each order.

app.py

The earlier service could not start: it used Literal without importing it (the import sat in train.py), so app.py failed with a NameError. Beyond that fix, the version below replaces the deprecated @app.on_event("startup") with a lifespan handler; refuses to load an artifact whose hash or scikit-learn version does not match its metadata, because unpickling a tampered file can execute code; requires an API key, compared in constant time; and validates strictly. The earlier schema silently accepted the string "2499" as a price and ignored unknown fields, so a misspelled field name went unnoticed. It also adds separate liveness (/health) and readiness (/ready) endpoints, a request ID returned to clients and logged with errors, and a rotating, lock-protected prediction log. The earlier version appended to a file from concurrent worker threads with no rotation.

"""Serve the model. Run: API_KEYS=key1,key2 uvicorn app:app --port 8000"""
import hashlib
import json
import logging
import os
import secrets
import time
import uuid
from contextlib import asynccontextmanager
from logging.handlers import RotatingFileHandler
from pathlib import Path
from typing import Literal

import joblib
import pandas as pd
import sklearn
from fastapi import Depends, FastAPI, HTTPException, Request, Security
from fastapi.security import APIKeyHeader
from pydantic import BaseModel, ConfigDict, Field, model_validator

MODEL_PATH = Path("models/return-risk-v1.joblib")
META_PATH = Path("models/return-risk-v1.json")
LOG_PATH = Path("logs/predictions.jsonl")
log = logging.getLogger("return_risk")

# One JSON object per line. Handlers serialize writes with a lock, so
# concurrent requests (sync endpoints run in a thread pool) do not
# interleave lines, and rotation bounds disk use.
LOG_PATH.parent.mkdir(exist_ok=True)
_audit = logging.getLogger("return_risk.audit")
_audit.propagate = False
_handler = RotatingFileHandler(LOG_PATH, maxBytes=50_000_000, backupCount=5,
                               encoding="utf-8")
_handler.setFormatter(logging.Formatter("%(message)s"))
_audit.addHandler(_handler)
_audit.setLevel(logging.INFO)

def load_artifact():
    """Refuse to unpickle anything that does not match its recorded hash;
    a tampered joblib file can execute arbitrary code when loaded."""
    meta = json.loads(META_PATH.read_text(encoding="utf-8"))
    digest = hashlib.sha256(MODEL_PATH.read_bytes()).hexdigest()
    if not secrets.compare_digest(digest, meta["sha256"]):
        raise RuntimeError("model artifact hash mismatch: not loading")
    if meta["sklearn_version"] != sklearn.__version__:
        raise RuntimeError(f"model trained with scikit-learn "
                           f"{meta['sklearn_version']}, running "
                           f"{sklearn.__version__}")
    return joblib.load(MODEL_PATH), meta

@asynccontextmanager
async def lifespan(app: FastAPI):
    keys = [k for k in os.environ.get("API_KEYS", "").split(",") if k]
    if not keys:
        raise RuntimeError("set API_KEYS before starting the service")
    app.state.api_keys = keys
    app.state.model, app.state.meta = load_artifact()  # once, at startup
    yield

app = FastAPI(title="E-commerce Return Risk API", version="1.0.0",
              lifespan=lifespan)
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)

def require_api_key(request: Request,
                    key: str | None = Security(api_key_header)) -> None:
    if key is None or not any(secrets.compare_digest(key, k)
                              for k in request.app.state.api_keys):
        raise HTTPException(status_code=401, detail="invalid API key")

class OrderFeatures(BaseModel):
    # Reject unknown fields and type coercion ("2499" is not a price)
    model_config = ConfigDict(extra="forbid", strict=True)

    category: Literal["apparel", "electronics", "home", "beauty",
                      "footwear"]
    price: float = Field(gt=0, le=1_000_000)
    discount_pct: int = Field(ge=0, le=100)
    customer_return_rate: float | None = Field(default=None, ge=0, le=1)
    product_return_rate: float = Field(ge=0, le=1)
    estimated_delivery_days: int = Field(ge=0, le=60)
    is_first_purchase: bool
    review_rating: float = Field(ge=1, le=5)

    @model_validator(mode="after")
    def history_matches_first_purchase(self):
        if self.is_first_purchase and self.customer_return_rate is not None:
            raise ValueError("a first purchase has no customer return rate")
        if not self.is_first_purchase and self.customer_return_rate is None:
            raise ValueError("customer_return_rate is required for "
                             "returning customers")
        return self

class PredictionResponse(BaseModel):
    request_id: str
    return_probability: float
    risk_tier: Literal["low", "medium", "high"]
    model_version: str
    latency_ms: float

@app.get("/health")  # liveness: the process is up
def health() -> dict:
    return {"status": "ok"}

@app.get("/ready")  # readiness: safe to route traffic here
def ready(request: Request) -> dict:
    return {"ready": True,
            "model_version": request.app.state.meta["model_version"]}

@app.post("/v1/predict-return-risk", response_model=PredictionResponse,
          dependencies=[Depends(require_api_key)])
def predict(order: OrderFeatures, request: Request) -> PredictionResponse:
    model, meta = request.app.state.model, request.app.state.meta
    request_id = str(uuid.uuid4())
    started = time.perf_counter()
    row = order.model_dump()
    row["is_first_purchase"] = int(row["is_first_purchase"])
    row["customer_return_rate"] = (float("nan") if row[
        "customer_return_rate"] is None else row["customer_return_rate"])
    try:
        frame = pd.DataFrame([row], columns=meta["features"])
        probability = float(model.predict_proba(frame)[0, 1])
    except Exception:
        log.exception("prediction failed request_id=%s", request_id)
        raise HTTPException(status_code=500,
                            detail=f"prediction failed ({request_id})")
    tiers = meta["risk_tiers"]
    tier = (
        "high" if probability >= tiers["high"]
        else "medium" if probability >= tiers["medium"] else "low")
    response = PredictionResponse(
        request_id=request_id, return_probability=round(probability, 4),
        risk_tier=tier, model_version=meta["model_version"],
        latency_ms=round((time.perf_counter() - started) * 1000, 2))
    # Features here contain no personal data; redact before logging if
    # yours do, and restrict access to the log files.
    _audit.info(json.dumps({"ts": time.time(), **response.model_dump(),
                            "features": order.model_dump()}))
    return response

monitor.py

Outcomes arrive weeks later, so the first signal of trouble is usually a shift in what the model sees or predicts. This script compares logged scores with the training-time score distribution using the population stability index (PSI).

"""Compare live prediction scores with the training-time distribution.
Run: python monitor.py   (e.g. hourly, from a scheduler)"""
import json
from pathlib import Path

import numpy as np

META_PATH = Path("models/return-risk-v1.json")
LOG_PATH = Path("logs/predictions.jsonl")
PSI_ALERT = 0.2  # common rule of thumb: <0.1 stable, >0.2 investigate

def psi(reference_edges, live_scores):
    """Population stability index over the reference decile bins."""
    edges = np.array(reference_edges, dtype=float)
    edges[0], edges[-1] = -np.inf, np.inf
    live = np.histogram(live_scores, bins=edges)[0] / len(live_scores)
    expected = np.full(len(edges) - 1, 1 / (len(edges) - 1))
    live = np.clip(live, 1e-6, None)
    return float(np.sum((live - expected) * np.log(live / expected)))

def main():
    meta = json.loads(META_PATH.read_text(encoding="utf-8"))
    lines = LOG_PATH.read_text(encoding="utf-8").splitlines()
    scores = [json.loads(line)["return_probability"] for line in lines]
    if len(scores) < 500:
        print(f"only {len(scores)} predictions logged; too few to judge")
        return
    value = psi(meta["score_deciles"], scores)
    status = "ALERT" if value > PSI_ALERT else "ok"
    print(f"{status}: score PSI {value:.3f} over {len(scores)} predictions")

if __name__ == "__main__":
    main()

Train, run, and call the API

python train.py
export API_KEYS="$(python -c 'import secrets; print(secrets.token_urlsafe(32))')"
uvicorn app:app --host 127.0.0.1 --port 8000   # behind a TLS proxy in prod

curl -s -X POST http://127.0.0.1:8000/v1/predict-return-risk \
  -H "Content-Type: application/json" -H "X-API-Key: $API_KEYS" \
  -d '{"category": "apparel", "price": 2499.0, "discount_pct": 40,
       "customer_return_rate": null, "product_return_rate": 0.31,
       "estimated_delivery_days": 5, "is_first_purchase": true,
       "review_rating": 3.8}'
# {"request_id": "3e85...", "return_probability": 0.5037,
#  "risk_tier": "medium", "model_version": "return-risk-v1",
#  "latency_ms": 25.72}

What the tests checked

Small real-world use case

A fashion retailer sees high apparel returns, especially for discounted items with size ambiguity, and calls the API at checkout. For high-risk items the storefront can show clearer size guidance or customer photos, recommend a better-fitting alternative, offer a virtual try-on, or flag the order for a packing check — without blanket restrictions that frustrate reliable customers. At a probability of 0.72 the product page might say: “Customers often exchange this item for a different size. Check the size guide before checkout.” The score guides help; it does not block the sale. Whether that guidance actually reduces returns is an experiment question, answered with a randomized holdout, not by the model’s metrics.

Common mistakes

Training on future or target information: actual delivery date, complaint text, return status, refund amount, or return reason inflate offline metrics and are unavailable at checkout.

Random splits for time-based data: behaviour, promotions, seasons, and catalogue change; validate on a later period.

Judging by accuracy alone: use precision, recall, average precision, and calibration, and choose actions and thresholds from business cost rather than 0.5.

Training–serving skew: package preprocessing with the model, keep the feature order in metadata, and serve with the same library versions.

Loose validation: coercion and ignored fields turn client bugs into silent wrong scores; forbid both and validate cross-field rules.

Deploying without monitoring or feedback: a healthy service can still produce drifting scores; treat the first model as the start of an iteration loop.

Production checklist

Data and features: governed sources with owners; a defined prediction time; feature definitions, transformations, and training snapshots stored; schemas, ranges, null rates, and categories validated. Model development: a baseline first; code, data version, parameters, environment, metrics, and artifact tracked together; discrimination, calibration, business impact, and fairness evaluated by segment; edge cases tested (new categories, extreme prices, missing histories). API and deployment: strict schemas; the model loaded once and verified at startup; versioned endpoints; liveness and readiness probes; authentication, TLS, and rate limiting at the gateway; secrets from the environment or a secret manager; containers with graceful shutdown. Monitoring and governance: request volume, error rate, latency percentiles, and resource use; access-restricted, redacted, retention-limited prediction logs; input and score drift plus actual outcomes once they arrive; alerts on errors, latency, schema violations, and drift; rollback for both model and service; scheduled performance reviews that trigger retraining.

Takeaway: Keep the first project small but production-minded: one business decision, one prediction target available at decision time, one reproducible training pipeline, one verified artifact, and one validated, authenticated endpoint — then validate on chronological real data before it touches checkout.

Back to all blogs