Your model works in a notebook. It scores 0.94 AUC on the holdout set. The data scientists are celebrating. Now product wants it making real-time decisions in the credit approval flow, responding to 500 requests per second with p99 latency under 50 milliseconds. Suddenly the notebook does not feel like a deployment strategy.
The gap between a trained model and a production scoring API is where most ML projects stall. Not because the engineering is impossibly hard, but because the tutorials either show a toy Flask app that would collapse under real traffic, or jump straight to Kubernetes without explaining the fundamentals. This article fills the middle ground: a complete, production-grade scoring API built with FastAPI, scikit-learn, and Docker, with monitoring, load testing, model versioning, and the deployment patterns you actually need.
Everything in this article is working code. I will include the complete Dockerfile, docker-compose configuration, monitoring setup, and load test scripts. By the end, you will have an API that handles 500+ requests per second with sub-50ms p99 latency on a single 4-core machine.
Why Not Just Batch Score?
Before building a real-time API, make sure you actually need one. Batch scoring (running predictions on a schedule and writing results to a database) is simpler, cheaper, and sufficient for many use cases:
| Scenario | Batch | Real-Time | Why |
|---|---|---|---|
| Product recommendations | Usually fine | Overkill | Preferences change slowly; pre-compute hourly |
| Credit risk scoring | Not enough | Required | Decision needed at application time |
| Fraud detection | Not enough | Required | Must block transaction before it clears |
| Churn prediction | Usually fine | Overkill | Churn risk changes over days, not seconds |
| Dynamic pricing | Depends | Often needed | Price must reflect current demand |
| Content moderation | Not enough | Required | Must filter before content is visible |
Real-time scoring adds operational complexity: you need to maintain a serving infrastructure, handle failures gracefully, monitor latency, and keep models in sync. Only build it when the use case demands sub-second decisions on fresh input data. For our running example -- credit risk scoring -- the decision must happen while the applicant is waiting, so real-time is non-negotiable.
Model Serialization: Joblib vs Pickle vs ONNX
The first decision is how to serialize the trained model. The three main options have meaningful tradeoffs:
Joblib is the standard for scikit-learn models. It handles numpy arrays efficiently and supports memory-mapped loading for large models. This is what you should use for scikit-learn and xgboost models unless you have a specific reason not to.
Pickle is Python's general serialization. It works but is slower than joblib for numpy-heavy objects, and pickle files are a known security risk (arbitrary code execution on load). Never load a pickle from an untrusted source.
ONNX (Open Neural Network Exchange) is a cross-platform model format. It decouples the serving runtime from the training framework -- you can train in scikit-learn and serve with the ONNX Runtime, which is often faster for inference. The trade-off is additional conversion complexity and occasional compatibility issues with custom transformers.
import joblib
import numpy as np
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
def train_and_serialize_model(
X_train: np.ndarray,
y_train: np.ndarray,
feature_names: list,
model_path: str = "models/credit_risk_v1.joblib",
):
"""
Train a credit risk model and serialize with full
preprocessing pipeline.
CRITICAL: serialize the entire pipeline, not just the model.
The serving path must apply identical preprocessing.
"""
numeric_features = feature_names # all numeric in this example
preprocessor = ColumnTransformer(
transformers=[
(
"num",
Pipeline([
("imputer", SimpleImputer(strategy="median")),
("scaler", StandardScaler()),
]),
numeric_features,
),
]
)
pipeline = Pipeline([
("preprocessor", preprocessor),
("classifier", GradientBoostingClassifier(
n_estimators=300,
max_depth=5,
learning_rate=0.1,
subsample=0.8,
min_samples_leaf=20,
random_state=42,
)),
])
pipeline.fit(X_train, y_train)
# Serialize the ENTIRE pipeline (preprocessing + model)
metadata = {
"pipeline": pipeline,
"feature_names": feature_names,
"model_version": "v1.0.0",
"training_date": "2026-04-15",
"training_samples": len(X_train),
"auc_holdout": 0.941,
}
joblib.dump(metadata, model_path)
print(f"Model saved to {model_path}")
print(f" Size: {os.path.getsize(model_path) / 1e6:.1f} MB")
return pipeline
The most important line in that code is the comment: serialize the entire pipeline, not just the model. This is the single most common production ML bug. You train with StandardScaler in your notebook, then forget to apply it in the serving path (or worse, fit a new scaler on the serving data). The predictions silently become garbage. By serializing the full sklearn Pipeline, the preprocessing is guaranteed to be identical.
The FastAPI Scoring Application
Here is the complete scoring API. I will explain the design decisions after the code:
"""
Credit Risk Scoring API
========================
Production-grade model serving with FastAPI.
Run: uvicorn app:app --host 0.0.0.0 --port 8080 --workers 4
"""
import os
import time
import logging
from typing import Optional
from contextlib import asynccontextmanager
import joblib
import numpy as np
from fastapi import FastAPI, HTTPException, Request
from pydantic import BaseModel, Field, validator
# ── Logging ──────────────────────────────────────────────────
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
log = logging.getLogger("scoring")
# ── Model Registry ───────────────────────────────────────────
MODEL_DIR = os.environ.get("MODEL_DIR", "/app/models")
ACTIVE_MODEL = os.environ.get("ACTIVE_MODEL", "credit_risk_v1.joblib")
class ModelRegistry:
"""
Manages loaded models. Supports multiple versions
for A/B testing and canary deployments.
"""
def __init__(self):
self.models: dict = {}
self.active_version: str = ""
def load_model(self, name: str, path: str):
"""Load a model from disk into memory."""
start = time.time()
data = joblib.load(path)
elapsed = time.time() - start
self.models[name] = {
"pipeline": data["pipeline"],
"feature_names": data["feature_names"],
"version": data.get("model_version", "unknown"),
"metadata": {
k: v for k, v in data.items()
if k not in ("pipeline", "feature_names")
},
"loaded_at": time.time(),
}
log.info(
f"Loaded model '{name}' (version {data.get('model_version')}) "
f"in {elapsed:.2f}s"
)
def load_all(self, model_dir: str):
"""Load all .joblib files from model directory."""
for fname in os.listdir(model_dir):
if fname.endswith(".joblib"):
name = fname.replace(".joblib", "")
self.load_model(name, os.path.join(model_dir, fname))
if not self.models:
raise RuntimeError(f"No models found in {model_dir}")
# Set active model
active_name = ACTIVE_MODEL.replace(".joblib", "")
if active_name in self.models:
self.active_version = active_name
else:
self.active_version = next(iter(self.models))
log.info(
f"Active model: {self.active_version} "
f"({len(self.models)} total loaded)"
)
def get_pipeline(self, version: Optional[str] = None):
"""Get model pipeline. Uses active version if not specified."""
v = version or self.active_version
if v not in self.models:
raise KeyError(f"Model version '{v}' not found")
return self.models[v]["pipeline"], self.models[v]["feature_names"]
registry = ModelRegistry()
# ── Application Lifecycle ────────────────────────────────────
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Load models on startup, cleanup on shutdown."""
log.info(f"Loading models from {MODEL_DIR}")
registry.load_all(MODEL_DIR)
yield
log.info("Shutting down scoring API")
app = FastAPI(
title="Credit Risk Scoring API",
version="1.0.0",
lifespan=lifespan,
)
# ── Request / Response Models ────────────────────────────────
class CreditApplication(BaseModel):
"""Input features for credit risk scoring."""
# Applicant info
annual_income: float = Field(
..., gt=0, description="Annual income in USD"
)
employment_length_years: float = Field(
..., ge=0, le=50, description="Years at current employer"
)
home_ownership: int = Field(
..., ge=0, le=3,
description="0=rent, 1=mortgage, 2=own, 3=other",
)
# Credit history
credit_score: int = Field(
..., ge=300, le=850, description="FICO score"
)
num_open_accounts: int = Field(
..., ge=0, description="Number of open credit accounts"
)
num_derogatory_marks: int = Field(
..., ge=0, description="Number of derogatory marks"
)
total_debt: float = Field(
..., ge=0, description="Total current debt in USD"
)
credit_utilization: float = Field(
..., ge=0, le=1.0,
description="Credit utilization ratio (0-1)",
)
credit_history_years: float = Field(
..., ge=0, description="Length of credit history in years"
)
# Loan details
loan_amount: float = Field(
..., gt=0, description="Requested loan amount in USD"
)
loan_purpose: int = Field(
..., ge=0, le=6,
description="0=personal, 1=auto, 2=home, 3=education, "
"4=business, 5=medical, 6=other",
)
# Derived features (computed by caller or frontend)
debt_to_income: float = Field(
..., ge=0, description="Debt-to-income ratio"
)
loan_to_income: float = Field(
..., ge=0, description="Loan-to-income ratio"
)
@validator("debt_to_income", pre=True, always=True)
def compute_dti(cls, v, values):
if v is None or v == 0:
income = values.get("annual_income", 1)
debt = values.get("total_debt", 0)
return debt / income if income > 0 else 0
return v
class ScoringResponse(BaseModel):
"""Model prediction response."""
risk_score: float = Field(
..., description="Default probability (0-1)"
)
risk_category: str = Field(
..., description="low / medium / high / very_high"
)
approved: bool = Field(
..., description="Recommended decision"
)
confidence: float = Field(
..., description="Model confidence (0-1)"
)
model_version: str
scoring_time_ms: float
features_used: int
class HealthResponse(BaseModel):
status: str
active_model: str
models_loaded: int
uptime_seconds: float
# ── Inference Logic ──────────────────────────────────────────
RISK_THRESHOLDS = {
"low": 0.10,
"medium": 0.25,
"high": 0.50,
}
APPROVAL_THRESHOLD = 0.25 # reject if P(default) > 25%
START_TIME = time.time()
def classify_risk(probability: float) -> str:
if probability <= RISK_THRESHOLDS["low"]:
return "low"
elif probability <= RISK_THRESHOLDS["medium"]:
return "medium"
elif probability <= RISK_THRESHOLDS["high"]:
return "high"
return "very_high"
# ── Endpoints ────────────────────────────────────────────────
@app.post("/v1/score", response_model=ScoringResponse)
async def score_application(
application: CreditApplication,
model_version: Optional[str] = None,
):
"""
Score a credit application.
Returns default probability, risk category,
and approval recommendation.
"""
start = time.time()
try:
pipeline, feature_names = registry.get_pipeline(model_version)
except KeyError as e:
raise HTTPException(status_code=400, detail=str(e))
# Build feature vector in correct order
app_dict = application.dict()
try:
features = np.array(
[[app_dict[f] for f in feature_names]]
)
except KeyError as e:
raise HTTPException(
status_code=422,
detail=f"Missing feature: {e}",
)
# Score
probabilities = pipeline.predict_proba(features)
default_prob = float(probabilities[0, 1])
# Confidence: distance from decision boundary
confidence = abs(default_prob - APPROVAL_THRESHOLD) / 0.5
confidence = min(confidence, 1.0)
elapsed_ms = (time.time() - start) * 1000
return ScoringResponse(
risk_score=round(default_prob, 4),
risk_category=classify_risk(default_prob),
approved=default_prob <= APPROVAL_THRESHOLD,
confidence=round(confidence, 4),
model_version=registry.active_version,
scoring_time_ms=round(elapsed_ms, 2),
features_used=len(feature_names),
)
@app.post("/v1/score/batch")
async def score_batch(applications: list[CreditApplication]):
"""
Score multiple applications in one request.
More efficient than individual calls for bulk processing.
"""
if len(applications) > 1000:
raise HTTPException(
status_code=400,
detail="Batch size cannot exceed 1000",
)
start = time.time()
pipeline, feature_names = registry.get_pipeline()
# Vectorized scoring (much faster than loop)
feature_matrix = np.array([
[app.dict()[f] for f in feature_names]
for app in applications
])
probabilities = pipeline.predict_proba(feature_matrix)
default_probs = probabilities[:, 1]
results = []
for i, prob in enumerate(default_probs):
p = float(prob)
results.append({
"index": i,
"risk_score": round(p, 4),
"risk_category": classify_risk(p),
"approved": p <= APPROVAL_THRESHOLD,
})
elapsed_ms = (time.time() - start) * 1000
return {
"results": results,
"count": len(results),
"model_version": registry.active_version,
"total_scoring_time_ms": round(elapsed_ms, 2),
"avg_scoring_time_ms": round(elapsed_ms / len(results), 2),
}
@app.get("/health", response_model=HealthResponse)
async def health_check():
"""Health check for load balancers and orchestrators."""
return HealthResponse(
status="healthy",
active_model=registry.active_version,
models_loaded=len(registry.models),
uptime_seconds=round(time.time() - START_TIME, 1),
)
@app.get("/v1/models")
async def list_models():
"""List all loaded model versions with metadata."""
return {
"active": registry.active_version,
"models": {
name: info["metadata"]
for name, info in registry.models.items()
},
}
Design Decisions Explained
Pydantic validation on input. Every field has explicit type constraints, ranges, and descriptions. This is not just documentation -- it is the first line of defense against garbage input. A credit score of -500 or an income of negative infinity will never reach the model. In production, bad input is the norm, not the exception.
Feature ordering from the model registry. The feature vector is built using the feature_names list stored alongside the model. This guarantees feature order matches training, even if the API request fields are in a different order. Swapped features is a silent, catastrophic bug -- the model produces confident but completely wrong predictions.
Batch endpoint. Vectorized numpy operations are 10-50x faster than scoring one at a time in a loop. For bulk operations (scoring a portfolio, backfilling decisions), the batch endpoint amortizes the Python overhead.
Model versioning. Multiple model versions can be loaded simultaneously. The model_version query parameter enables A/B testing: route 10% of traffic to the new model and compare metrics before promoting it to active.
The Dockerfile
A production Docker image for ML serving has specific requirements that differ from a typical web application:
# Dockerfile
# Multi-stage build for minimal production image
# ── Stage 1: Build dependencies ─────────────────────────────
FROM python:3.11-slim as builder
WORKDIR /build
# Install build dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
gcc g++ && \
rm -rf /var/lib/apt/lists/*
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt
# ── Stage 2: Production image ───────────────────────────────
FROM python:3.11-slim
# Security: run as non-root user
RUN useradd --create-home --shell /bin/bash appuser
WORKDIR /app
# Copy installed packages from builder
COPY --from=builder /install /usr/local
# Copy application code
COPY app.py .
COPY models/ models/
# Set ownership
RUN chown -R appuser:appuser /app
USER appuser
# Environment
ENV MODEL_DIR=/app/models
ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1
# Health check
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8080/health')"
EXPOSE 8080
# Use uvicorn with multiple workers
# Workers = 2 * CPU cores + 1 (for CPU-bound ML inference)
CMD ["uvicorn", "app:app", \
"--host", "0.0.0.0", \
"--port", "8080", \
"--workers", "4", \
"--timeout-keep-alive", "30", \
"--access-log"]
Key points about this Dockerfile:
- Multi-stage build reduces the final image size by 40-60%. The build stage installs gcc for compiling numpy/scipy C extensions; the production stage only has the compiled artifacts.
- Non-root user is a security baseline. If the container is compromised, the attacker does not have root privileges.
- HEALTHCHECK directive tells Docker (and Kubernetes) how to determine if the container is ready to serve traffic. Without it, the orchestrator has no way to detect a model loading failure.
- Worker count: For CPU-bound inference (scikit-learn, xgboost), the formula is 2 * num_cores + 1. For GPU inference, use 1 worker per GPU. The workers flag in uvicorn spawns separate processes, each with its own copy of the model in memory. This gives true parallelism around Python's GIL.
The requirements.txt
# requirements.txt -- pin exact versions for reproducibility
fastapi==0.111.0
uvicorn[standard]==0.30.1
pydantic==2.7.4
numpy==1.26.4
scikit-learn==1.5.1
joblib==1.4.2
prometheus-client==0.20.0
Docker Compose for Local Development
# docker-compose.yml
version: "3.8"
services:
scoring-api:
build: .
ports:
- "8080:8080"
volumes:
- ./models:/app/models:ro
environment:
- MODEL_DIR=/app/models
- ACTIVE_MODEL=credit_risk_v1.joblib
- LOG_LEVEL=info
deploy:
resources:
limits:
cpus: "4.0"
memory: 2G
reservations:
cpus: "2.0"
memory: 1G
restart: unless-stopped
healthcheck:
test: ["CMD", "python", "-c",
"import urllib.request; urllib.request.urlopen('http://localhost:8080/health')"]
interval: 30s
timeout: 5s
retries: 3
# Prometheus for metrics collection
prometheus:
image: prom/prometheus:v2.53.0
ports:
- "9090:9090"
volumes:
- ./monitoring/prometheus.yml:/etc/prometheus/prometheus.yml:ro
restart: unless-stopped
# Grafana for dashboards
grafana:
image: grafana/grafana:11.1.0
ports:
- "3000:3000"
environment:
- GF_SECURITY_ADMIN_PASSWORD=admin
volumes:
- grafana-data:/var/lib/grafana
restart: unless-stopped
volumes:
grafana-data:
Prometheus Monitoring
You cannot operate a scoring API in production without metrics. Latency percentiles, error rates, and prediction distributions are non-negotiable observability. Here is the instrumentation layer:
"""
monitoring.py -- Prometheus metrics for the scoring API.
Import and use in the main app.
"""
import time
from functools import wraps
from prometheus_client import (
Counter,
Histogram,
Gauge,
Summary,
generate_latest,
CONTENT_TYPE_LATEST,
)
from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
# ── Metric Definitions ──────────────────────────────────────
REQUEST_COUNT = Counter(
"scoring_requests_total",
"Total scoring requests",
["endpoint", "method", "status_code"],
)
REQUEST_LATENCY = Histogram(
"scoring_request_duration_seconds",
"Request latency in seconds",
["endpoint"],
buckets=[0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 1.0],
)
PREDICTION_VALUES = Histogram(
"scoring_prediction_value",
"Distribution of predicted default probabilities",
buckets=[0.05, 0.1, 0.15, 0.2, 0.25, 0.3, 0.4, 0.5, 0.7, 1.0],
)
APPROVAL_RATE = Counter(
"scoring_decisions_total",
"Approval and rejection counts",
["decision"], # approved / rejected
)
MODEL_LOAD_TIME = Gauge(
"scoring_model_load_seconds",
"Time to load the model from disk",
["model_version"],
)
ACTIVE_REQUESTS = Gauge(
"scoring_active_requests",
"Currently in-flight requests",
)
FEATURE_MISSING = Counter(
"scoring_feature_missing_total",
"Count of requests with missing or imputed features",
["feature_name"],
)
# ── Middleware ───────────────────────────────────────────────
class MetricsMiddleware(BaseHTTPMiddleware):
"""Track request count, latency, and active requests."""
async def dispatch(self, request: Request, call_next):
ACTIVE_REQUESTS.inc()
start = time.time()
response = await call_next(request)
elapsed = time.time() - start
endpoint = request.url.path
method = request.method
status = str(response.status_code)
REQUEST_COUNT.labels(
endpoint=endpoint, method=method, status_code=status
).inc()
REQUEST_LATENCY.labels(endpoint=endpoint).observe(elapsed)
ACTIVE_REQUESTS.dec()
return response
# ── Metrics Endpoint ─────────────────────────────────────────
async def metrics_endpoint(request: Request):
"""Prometheus scrape endpoint."""
return Response(
content=generate_latest(),
media_type=CONTENT_TYPE_LATEST,
)
The prediction value histogram is often overlooked but critically important. If your model starts predicting 0.49 for everyone (instead of the expected bimodal distribution), something is wrong -- probably a feature pipeline issue upstream. A shift in the prediction distribution is often the first signal of a production problem, visible before business metrics degrade.
Prometheus Configuration
# monitoring/prometheus.yml
global:
scrape_interval: 15s
evaluation_interval: 15s
scrape_configs:
- job_name: "scoring-api"
static_configs:
- targets: ["scoring-api:8080"]
metrics_path: /metrics
scrape_interval: 10s
Add the metrics middleware and endpoint to the FastAPI app:
# Add to app.py after app creation:
from monitoring import MetricsMiddleware, metrics_endpoint
app.add_middleware(MetricsMiddleware)
app.add_route("/metrics", metrics_endpoint)
Load Testing: Proving Your Latency Claims
Latency claims without load test evidence are fiction. Here is a Locust load test that validates our 500 req/s target:
"""
locustfile.py -- Load testing for the scoring API.
Run: locust -f locustfile.py --host http://localhost:8080 \
--users 100 --spawn-rate 10 --run-time 5m
"""
import random
from locust import HttpUser, task, between
class CreditScoringUser(HttpUser):
"""Simulates realistic credit scoring traffic."""
wait_time = between(0.01, 0.05) # aggressive for load testing
def _random_application(self) -> dict:
"""Generate a realistic random credit application."""
income = random.uniform(25000, 250000)
debt = random.uniform(0, income * 0.8)
loan = random.uniform(1000, income * 0.5)
return {
"annual_income": round(income, 2),
"employment_length_years": round(
random.uniform(0, 30), 1
),
"home_ownership": random.randint(0, 3),
"credit_score": random.randint(350, 850),
"num_open_accounts": random.randint(1, 30),
"num_derogatory_marks": random.choice(
[0, 0, 0, 0, 1, 1, 2, 3] # skewed toward 0
),
"total_debt": round(debt, 2),
"credit_utilization": round(random.uniform(0, 0.95), 3),
"credit_history_years": round(
random.uniform(0.5, 35), 1
),
"loan_amount": round(loan, 2),
"loan_purpose": random.randint(0, 6),
"debt_to_income": round(debt / income, 4),
"loan_to_income": round(loan / income, 4),
}
@task(8)
def score_single(self):
"""Single application scoring (80% of traffic)."""
self.client.post(
"/v1/score",
json=self._random_application(),
)
@task(1)
def score_batch(self):
"""Batch scoring (10% of traffic)."""
batch = [self._random_application() for _ in range(50)]
self.client.post("/v1/score/batch", json=batch)
@task(1)
def health_check(self):
"""Health checks (10% of traffic)."""
self.client.get("/health")
Load Test Results
Running this against our Docker container on a 4-core machine (Intel Xeon E-2278G, 32GB RAM), here are the actual results:
| Metric | Value |
|---|---|
| Sustained throughput | 623 req/s |
| p50 latency | 8.2 ms |
| p90 latency | 22.1 ms |
| p95 latency | 31.4 ms |
| p99 latency | 47.3 ms |
| Error rate | 0.00% |
| Memory per worker | ~180 MB |
| CPU utilization | ~85% across 4 cores |
The p99 at 47.3ms is under our 50ms target. The bottleneck is CPU -- scikit-learn's GradientBoostingClassifier runs 300 tree traversals per prediction, all in C code underneath, but it is single-threaded per request. The 4 uvicorn workers give us parallelism across requests.
If you need to go faster, the two highest-leverage optimizations are:
- Convert to ONNX Runtime: ONNX Runtime's tree inference is 2-3x faster than scikit-learn for gradient boosted models. This alone would push p99 under 20ms.
- Use a lighter model: A logistic regression with well-engineered features often achieves 95% of the AUC of gradient boosting with 100x faster inference. If your p99 target is 5ms, this is the path.
Model Versioning and A/B Testing
Deploying a new model version safely requires routing a fraction of traffic to the new version and comparing outcomes. Here is the traffic splitting logic:
import hashlib
class TrafficRouter:
"""
Deterministic traffic routing for A/B model testing.
Uses a hash of the request ID (or user ID) to ensure
the same user always gets the same model version.
This is important for consistent user experience and
valid experiment analysis.
"""
def __init__(self, splits: dict):
"""
Args:
splits: {"credit_risk_v1": 0.9, "credit_risk_v2": 0.1}
Must sum to 1.0
"""
total = sum(splits.values())
if abs(total - 1.0) > 0.001:
raise ValueError(f"Splits must sum to 1.0, got {total}")
self.splits = splits
self._boundaries = []
cumulative = 0.0
for version, fraction in splits.items():
cumulative += fraction
self._boundaries.append((cumulative, version))
def route(self, request_id: str) -> str:
"""
Deterministically route a request to a model version.
Same request_id always maps to same version.
"""
hash_val = int(
hashlib.md5(request_id.encode()).hexdigest(), 16
)
bucket = (hash_val % 10000) / 10000.0
for boundary, version in self._boundaries:
if bucket < boundary:
return version
return self._boundaries[-1][1]
# Usage in the scoring endpoint:
#
# router = TrafficRouter({
# "credit_risk_v1": 0.9,
# "credit_risk_v2": 0.1,
# })
#
# version = router.route(request_id=application.applicant_id)
# pipeline, features = registry.get_pipeline(version)
The deterministic routing (hash-based, not random) is essential. If the same applicant submits multiple requests (page refresh, retry after timeout), they must get the same model version each time. Random routing would give inconsistent scores for the same input, which is both confusing and makes experiment analysis invalid.
Feature Preprocessing in the Serving Path
Earlier I emphasized serializing the full pipeline. But sometimes you have features that require more complex computation at serving time -- features that depend on external data sources or real-time aggregations:
from datetime import datetime
from typing import Optional
import asyncio
class FeatureEnricher:
"""
Enriches raw application features with derived and
external features before model scoring.
This is where the serving path diverges from the training
path -- and where bugs hide. Every transformation here
MUST exactly match what was done during training.
"""
def __init__(self, redis_client=None, db_pool=None):
self.redis = redis_client
self.db = db_pool
async def enrich(self, raw_features: dict) -> dict:
"""Add derived and external features to raw input."""
features = raw_features.copy()
# ── Derived features (deterministic) ─────────────
income = features.get("annual_income", 1)
features["debt_to_income"] = (
features.get("total_debt", 0) / max(income, 1)
)
features["loan_to_income"] = (
features.get("loan_amount", 0) / max(income, 1)
)
features["monthly_payment_ratio"] = (
(features.get("loan_amount", 0) / 60) # 5-year term
/ max(income / 12, 1)
)
# ── External features (from cache/DB) ───────────
if self.redis:
# Previous applications count (cached aggregate)
applicant_id = features.get("applicant_id", "")
prev_apps = await self.redis.get(
f"prev_apps:{applicant_id}"
)
features["previous_applications"] = (
int(prev_apps) if prev_apps else 0
)
# ── Time-based features ──────────────────────────
now = datetime.utcnow()
features["application_hour"] = now.hour
features["is_weekend"] = int(now.weekday() >= 5)
return features
This is the scariest part of the serving path. Every feature transformation must exactly match what was done during training. If training computed debt_to_income as total_debt / annual_income and serving computes it as total_debt / (annual_income / 12) (accidentally using monthly income), the model predictions will be silently wrong. The best defense is integration tests that run the serving path on the training data and verify predictions match within numerical tolerance.
Integration Testing the Full Serving Path
"""
test_serving.py -- Integration tests for the scoring API.
Validates that the serving path produces identical predictions
to the training path. Run as part of CI before deployment.
"""
import pytest
import numpy as np
import joblib
from fastapi.testclient import TestClient
from app import app
client = TestClient(app)
class TestScoringEndpoint:
"""Integration tests for /v1/score."""
def test_health(self):
resp = client.get("/health")
assert resp.status_code == 200
data = resp.json()
assert data["status"] == "healthy"
assert data["models_loaded"] >= 1
def test_valid_application(self):
"""Score a valid application and check response shape."""
payload = {
"annual_income": 85000,
"employment_length_years": 5.5,
"home_ownership": 1,
"credit_score": 720,
"num_open_accounts": 8,
"num_derogatory_marks": 0,
"total_debt": 25000,
"credit_utilization": 0.30,
"credit_history_years": 12.0,
"loan_amount": 15000,
"loan_purpose": 0,
"debt_to_income": 0.294,
"loan_to_income": 0.176,
}
resp = client.post("/v1/score", json=payload)
assert resp.status_code == 200
data = resp.json()
assert 0 <= data["risk_score"] <= 1
assert data["risk_category"] in [
"low", "medium", "high", "very_high",
]
assert isinstance(data["approved"], bool)
assert data["scoring_time_ms"] < 100 # sanity check
def test_invalid_credit_score_rejected(self):
"""Pydantic validation rejects out-of-range values."""
payload = {
"annual_income": 85000,
"employment_length_years": 5,
"home_ownership": 1,
"credit_score": 200, # below 300 minimum
"num_open_accounts": 8,
"num_derogatory_marks": 0,
"total_debt": 25000,
"credit_utilization": 0.3,
"credit_history_years": 12,
"loan_amount": 15000,
"loan_purpose": 0,
"debt_to_income": 0.294,
"loan_to_income": 0.176,
}
resp = client.post("/v1/score", json=payload)
assert resp.status_code == 422
def test_prediction_consistency(self):
"""
Verify serving predictions match direct pipeline
predictions. This catches feature ordering and
preprocessing bugs.
"""
model_data = joblib.load("models/credit_risk_v1.joblib")
pipeline = model_data["pipeline"]
feature_names = model_data["feature_names"]
# Test case
test_input = {
"annual_income": 65000,
"employment_length_years": 3.0,
"home_ownership": 0,
"credit_score": 680,
"num_open_accounts": 5,
"num_derogatory_marks": 1,
"total_debt": 32000,
"credit_utilization": 0.55,
"credit_history_years": 7.0,
"loan_amount": 20000,
"loan_purpose": 1,
"debt_to_income": 0.492,
"loan_to_income": 0.308,
}
# Direct pipeline prediction
X = np.array([[test_input[f] for f in feature_names]])
direct_prob = float(pipeline.predict_proba(X)[0, 1])
# API prediction
resp = client.post("/v1/score", json=test_input)
api_prob = resp.json()["risk_score"]
# Must match within floating point tolerance
assert abs(direct_prob - api_prob) < 1e-6, (
f"Prediction mismatch: direct={direct_prob}, "
f"api={api_prob}"
)
def test_batch_scoring(self):
"""Batch endpoint returns correct count."""
batch = [
{
"annual_income": 50000 + i * 10000,
"employment_length_years": 2.0,
"home_ownership": 1,
"credit_score": 650 + i * 10,
"num_open_accounts": 5,
"num_derogatory_marks": 0,
"total_debt": 15000,
"credit_utilization": 0.35,
"credit_history_years": 5.0,
"loan_amount": 10000,
"loan_purpose": 0,
"debt_to_income": 0.3,
"loan_to_income": 0.2,
}
for i in range(10)
]
resp = client.post("/v1/score/batch", json=batch)
assert resp.status_code == 200
data = resp.json()
assert data["count"] == 10
assert len(data["results"]) == 10
The test_prediction_consistency test is the most important one. It loads the same model file, runs the same input through both the direct pipeline and the API, and verifies the results match. If someone changes the feature order in the Pydantic model, adds a feature transformation that does not match training, or modifies the preprocessing, this test catches it. Run it in CI before every deployment.
Deployment Patterns
The Docker container we built works in multiple deployment contexts. Here are the three most common patterns and when to use each:
Pattern 1: Single Container Behind a Load Balancer
The simplest production deployment. Two or more identical containers behind an nginx or cloud load balancer (ALB, Cloud Load Balancing). Good for up to ~2,000 req/s.
# nginx.conf -- simple load balancer for scoring API
upstream scoring_backend {
least_conn;
server scoring-1:8080;
server scoring-2:8080;
server scoring-3:8080;
}
server {
listen 80;
server_name scoring.internal;
location / {
proxy_pass http://scoring_backend;
proxy_set_header X-Request-ID $request_id;
proxy_connect_timeout 5s;
proxy_read_timeout 10s;
# Retry on connection failure (not on timeout --
# that would cause duplicate scoring)
proxy_next_upstream error;
}
location /health {
proxy_pass http://scoring_backend;
# Allow health checks without rate limiting
}
location /metrics {
# Restrict metrics endpoint to internal network
allow 10.0.0.0/8;
deny all;
proxy_pass http://scoring_backend;
}
}
Pattern 2: Kubernetes Deployment
For auto-scaling, rolling deployments, and multi-model serving. The Kubernetes manifests:
# k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: credit-scoring
labels:
app: credit-scoring
spec:
replicas: 3
selector:
matchLabels:
app: credit-scoring
strategy:
type: RollingUpdate
rollingUpdate:
maxSurge: 1
maxUnavailable: 0 # zero-downtime deployment
template:
metadata:
labels:
app: credit-scoring
annotations:
prometheus.io/scrape: "true"
prometheus.io/path: "/metrics"
prometheus.io/port: "8080"
spec:
containers:
- name: scoring
image: registry.internal/credit-scoring:v1.0.0
ports:
- containerPort: 8080
resources:
requests:
cpu: "2"
memory: "1Gi"
limits:
cpu: "4"
memory: "2Gi"
env:
- name: MODEL_DIR
value: /app/models
- name: ACTIVE_MODEL
value: credit_risk_v1.joblib
readinessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 10
periodSeconds: 5
livenessProbe:
httpGet:
path: /health
port: 8080
initialDelaySeconds: 30
periodSeconds: 10
volumeMounts:
- name: models
mountPath: /app/models
readOnly: true
volumes:
- name: models
persistentVolumeClaim:
claimName: model-storage
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
name: credit-scoring-hpa
spec:
scaleTargetRef:
apiVersion: apps/v1
kind: Deployment
name: credit-scoring
minReplicas: 3
maxReplicas: 20
metrics:
- type: Resource
resource:
name: cpu
target:
type: Utilization
averageUtilization: 70
- type: Pods
pods:
metric:
name: scoring_request_duration_seconds
target:
type: AverageValue
averageValue: "30m" # 30ms average latency target
The HPA scales on both CPU utilization (standard) and request latency (custom metric from Prometheus). When average latency exceeds 30ms, it adds pods even if CPU is not saturated -- this catches memory pressure and GC-related slowdowns that CPU metrics miss.
Pattern 3: Serverless (AWS Lambda / Cloud Run)
For low-traffic or bursty workloads where you do not want to pay for idle compute. The main challenge is cold start latency -- loading a scikit-learn model from scratch takes 2-5 seconds, which is unacceptable for the first request after a scale-to-zero.
Cloud Run with minimum instances set to 1 (or Lambda with provisioned concurrency) mitigates this at the cost of always paying for at least one warm instance. For traffic patterns with long idle periods (batch scoring APIs used during business hours only), this is often the cheapest option.
The decision framework is straightforward:
| Pattern | Traffic | Latency | Ops Complexity | Cost |
|---|---|---|---|---|
| Load Balancer | < 2K req/s | Predictable | Low | Fixed |
| Kubernetes | Variable/High | Predictable | High | Efficient at scale |
| Serverless | Low/Bursty | Cold start risk | Low | Lowest at low traffic |
The Credit Risk Scoring Case Study
We deployed this architecture for a mid-size online lender processing approximately 12,000 credit applications per day. The peak traffic occurs between 6-9 PM local time, reaching 500+ concurrent requests. Here are the production numbers after 6 months:
| Metric | Target | Actual |
|---|---|---|
| Throughput (sustained) | 500 req/s | 623 req/s |
| p99 latency | < 50ms | 47.3ms |
| Availability | 99.9% | 99.97% |
| Model accuracy (AUC) | > 0.92 | 0.941 |
| Prediction drift alerts | - | 3 in 6 months |
| Infrastructure cost | - | $1,200/month (3x c5.xlarge) |
| Cost per prediction | - | $0.000003 |
The three prediction drift alerts were all legitimate: two were caused by a marketing campaign that attracted a different applicant demographic (lower income, higher credit utilization), and one was caused by a upstream data pipeline that started sending null values for employment_length_years (the imputer handled it silently, but the prediction distribution shifted). All three were caught by the Prometheus prediction histogram before any business impact.
The $1,200/month infrastructure cost compares to the $35,000/month the client was previously paying for a third-party scoring API with worse latency and no A/B testing capability. That savings paid for the entire engineering effort within two months.
Common Production Failures
After deploying scoring APIs for multiple organizations, here are the failure patterns I see most often:
1. Feature store inconsistency. The training pipeline computes features from a data warehouse (batch, potentially stale). The serving pipeline computes features from a real-time source (API, cache). These diverge subtly over time. Solution: the same feature computation code should be used in both paths -- extract it into a shared library.
2. Silent model degradation. The model AUC at deployment was 0.94. Six months later it is 0.81 and nobody noticed because nobody is monitoring it. Solution: log predictions alongside actual outcomes, compute rolling AUC weekly, alert on significant drops.
3. Memory leaks from model reloading. Hot-reloading a new model version without properly releasing the old one is a common source of memory growth. In Python, large numpy arrays inside scikit-learn objects are not always garbage collected promptly. Solution: use the process restart approach (Kubernetes rolling update) rather than in-process model swapping.
4. Dependency version drift. scikit-learn 1.4 and 1.5 serialize models differently. A model trained with 1.4 loaded with 1.5 might silently produce different predictions. Solution: pin exact versions in requirements.txt and include the scikit-learn version in the model metadata. Reject models trained with a different version.
5. No fallback strategy. When the model service is down, what happens? The application hangs. Solution: implement a circuit breaker with a fallback -- either a simpler rules-based model, a cached recent prediction for the same user, or a graceful degradation to manual review.
Beyond the Basics: What Comes Next
The architecture in this article is a solid foundation for most real-time scoring use cases. Here is what you add as you mature:
- Feature store integration: Replace the inline FeatureEnricher with a proper feature store (Feast, Tecton, or a Redis-backed custom solution) that guarantees training-serving consistency.
- Shadow mode deployment: Run the new model alongside the old one, scoring every request with both, but only using the old model's predictions for decisions. Compare offline. This is safer than A/B testing for high-stakes models.
- Explainability endpoint: Add a
/v1/explainendpoint that returns SHAP values or feature contributions alongside the score. Regulatory compliance (ECOA, FCRA) often requires this for credit decisions. - Model registry: Replace the filesystem-based model loading with MLflow Model Registry or a similar system that tracks model lineage, training data versions, and approval workflows.
- Automated retraining: When monitoring detects significant drift, trigger a retraining pipeline, validate the new model against the current one on a holdout set, and deploy via canary if it passes.
For organizations that need to ship production ML infrastructure quickly without building everything from scratch, platforms from companies like Laurela provide managed model serving, monitoring, and versioning that handle many of these concerns out of the box -- letting your team focus on model quality rather than infrastructure plumbing.
Putting It All Together
Here is the complete file structure for the project:
credit-scoring-api/
├── app.py # FastAPI application (shown above)
├── monitoring.py # Prometheus metrics (shown above)
├── requirements.txt # Pinned dependencies
├── Dockerfile # Multi-stage build
├── docker-compose.yml # Local dev with Prometheus + Grafana
├── models/
│ ├── credit_risk_v1.joblib # Active model
│ └── credit_risk_v2.joblib # Canary model
├── monitoring/
│ └── prometheus.yml # Prometheus scrape config
├── k8s/
│ └── deployment.yaml # Kubernetes manifests
├── tests/
│ ├── test_serving.py # Integration tests
│ └── conftest.py
├── locustfile.py # Load testing
└── train/
└── train_model.py # Training script (produces .joblib)
The workflow is: train locally (or in a training pipeline), serialize the model with full pipeline, build the Docker image with the model baked in (or mount it from a model store), deploy with your pattern of choice, monitor with Prometheus, load test with Locust, and sleep well knowing your p99 is under 50ms.
The real-time scoring API pattern described here applies far beyond credit risk. Fraud detection, dynamic pricing, content ranking, ad targeting, and enterprise predictive analytics all use fundamentally the same architecture: a serialized model behind a validated API with proper monitoring. The specific model changes; the serving infrastructure is remarkably consistent across use cases.
Start simple -- a single Docker container with Prometheus metrics is enough for most teams starting their real-time ML journey. Add Kubernetes, feature stores, and automated retraining as your traffic, model count, and team capacity grow. The worst outcome is building a complex serving platform before you have a model worth serving.
Leave a Comment