Monitoring ML Models in Production: Drift, Data Quality, Latency, and Business Metrics

Monitoring ML Models in Production: Drift, Data Quality, Latency, and Business Metrics

A production-oriented healthcare implementation using a 30-day readmission-risk model

Production ML monitoring is not simply watching model accuracy. A reliable system monitors whether incoming data remains valid, whether the population and model outputs are changing, whether the service is fast and available enough for its clinical workflow, and whether the model delivers the intended clinical and operational outcomes.

The running example is a hospital model that estimates the probability of an unplanned readmission within 30 days of an eligible discharge. Care teams may use the score as one input to discharge planning and follow-up prioritisation. It is not a diagnosis, a guarantee of readmission, or a substitute for clinical judgment.

Scope. The code is a tested reference implementation, not a turnkey hospital system or certified clinical product. Before clinical use the model and workflow need local clinical validation, an intended-use definition, safety analysis, privacy/security review, integration testing and any applicable regulatory assessment. A monitoring alert does not by itself prove a model is unsafe or that retraining is appropriate, and identifiable patient data must not reach general-purpose telemetry, logs or third-party monitoring services without approved safeguards.

1. The four monitoring layers

The layers run on different clocks. Data and service checks run continuously or per batch; drift checks run on a schedule; true performance waits until outcome labels have matured and been linked; clinical impact needs a longer window and a study design that accounts for interventions and other changes.

2. Data drift and prediction drift

Feature drift is a change in an input distribution relative to a chosen reference population; prediction drift is a change in the distribution of scores or decisions. Neither proves that quality has dropped: the cause may be a seasonal population shift, a new care pathway, a coding change or a broken upstream feed.

Use a fixed, versioned reference population and document its inclusion criteria, period, facility and feature schema.

Monitor numeric and categorical features, missingness, out-of-range rates and the score distribution, segmented by facility and clinically relevant cohorts where volumes allow; suppress small cells.

Treat PSI as a screening statistic. The 0.10 / 0.25 cutoffs are industry heuristics, not validated clinical limits; complement PSI with KS for numeric and chi-square or Jensen-Shannon for categorical features, and account for sample size and multiple comparisons.

PSI implementation

Bins come from reference quantiles with open outer edges, a constant reference gets an explicit at-baseline/different split, and empty bins are smoothed with a pseudocount in count units. A tiny epsilon (such as 1e-6) is not real smoothing: an empty bin then contributes a term whose size is set by the arbitrary epsilon (in testing, changing epsilon from 1e-3 to 1e-6 moved the same PSI from 11.2 to 16.7).

# drift_monitor.py
from __future__ import annotations
import numpy as np
def calculate_psi(reference, current, bins: int = 10, pseudocount: float = 0.5) -> float:
    """PSI over finite values of one numeric feature. Missingness is monitored separately."""
    if bins < 2:
        raise ValueError("bins must be >= 2")
    if pseudocount <= 0:
        raise ValueError("pseudocount must be positive")
    ref = np.asarray(reference, dtype=float)
    cur = np.asarray(current, dtype=float)
    ref, cur = ref[np.isfinite(ref)], cur[np.isfinite(cur)]
    if ref.size == 0 or cur.size == 0:
        raise ValueError("each sample needs at least one finite value")
    if np.unique(ref).size == 1:
        # Constant reference: compare "at baseline" vs "different" so shifts stay visible.
        baseline = ref[0]
        ref_counts = np.array([ref.size, 0.0])
        cur_counts = np.array([np.sum(cur == baseline), np.sum(cur != baseline)], dtype=float)
    else:
        edges = np.unique(np.quantile(ref, np.linspace(0, 1, bins + 1)))
        edges[0], edges[-1] = -np.inf, np.inf
        ref_counts = np.histogram(ref, bins=edges)[0].astype(float)
        cur_counts = np.histogram(cur, bins=edges)[0].astype(float)
    # Pseudocount in count units (not 1e-6): an empty bin then adds a bounded,
    # sample-size-aware penalty instead of a value set by an arbitrary epsilon.
    ref_p = (ref_counts + pseudocount) / (ref_counts.sum() + pseudocount * ref_counts.size)
    cur_p = (cur_counts + pseudocount) / (cur_counts.sum() + pseudocount * cur_counts.size)
    return float(np.sum((cur_p - ref_p) * np.log(cur_p / ref_p)))
def missing_rate(values) -> float:
    """Share of NaN/None values. Infinite values are invalid, not missing."""
    arr = np.asarray(values, dtype=float)
    if arr.size == 0:
        raise ValueError("sample must be non-empty")
    return float(np.isnan(arr).mean())
if __name__ == "__main__":
    rng = np.random.default_rng(42)
    reference_age = rng.normal(58, 15, 10_000)
    current_age = rng.normal(67, 16, 2_000)
    print("age_psi:", round(calculate_psi(reference_age, current_age), 4))

Note. PSI compares finite values only. Monitor the missing-value rate separately, because a missingness shift can itself be clinically important, and count infinite values as invalid rather than missing. Set thresholds from historical variation, sample size, clinical review and alert-fatigue analysis; never retrain automatically on a PSI breach.

Production drift workflow

Read the versioned reference and current feature aggregates from approved storage; validate schema and units first (a unit error is a data-quality incident, not drift); compute drift and missingness per feature and approved cohort; persist values with sample counts, reference version, feature schema, job version and window; alert the feature and clinical/data-science owners with that context; and investigate data defects, workflow, population and label shifts before deciding to recalibrate, retrain, restrict use or roll back.

3. Data quality and clinical data contracts

A model can fail when the population has not changed at all. EHR upgrades, laboratory interfaces, unit changes, code-set updates, delayed messages and duplicate encounters corrupt inputs, and type checks alone miss them: a creatinine value in µmol/L parses as a number while the model expects mg/dL. Before scoring, check schema and types, completeness by facility and subgroup, validity (units, ranges, timestamp semantics), consistency (unique encounters, coherent admission/discharge times), freshness, and lineage (source system, transformation and feature-definition versions, batch ID).

Batch validation with row-level quarantine

The validator returns a batch report plus a reason code for every failing row, so one bad record quarantines that record rather than the whole batch. It treats present-but-unparseable values as invalid (with errors="coerce" alone, "abc" silently becomes a missing value), does not count null encounter IDs as duplicates, parses timestamps as ISO 8601 (with an inferred format, current pandas turns rows formatted differently from the first row into NaT), tolerates small clock skew, and measures feed staleness from the newest arrival, so a legitimate backfill of older records does not fail the feed. Clinical bounds are illustrative and must be approved locally.

# data_quality.py
from __future__ import annotations
import pandas as pd
REQUIRED = {
    "patient_token", "encounter_id", "age_years", "creatinine_mg_dl",
    "prior_admissions_12m", "discharge_datetime", "ingested_at",
}
NULLABLE = {"creatinine_mg_dl"}  # "not measured" is allowed; everything else is required
def _numeric(s: pd.Series) -> tuple[pd.Series, pd.Series]:
    """Return parsed values and a mask of present-but-unparseable entries."""
    parsed = pd.to_numeric(s, errors="coerce")
    return parsed, s.notna() & parsed.isna()
def validate_batch(
    df: pd.DataFrame,
    now_utc: pd.Timestamp | None = None,
    max_feed_staleness: pd.Timedelta = pd.Timedelta(hours=24),
    clock_skew: pd.Timedelta = pd.Timedelta(minutes=5),
) -> tuple[dict, pd.DataFrame]:
    """Return (batch report, per-row reason codes). Rows with reasons go to quarantine."""
    now = pd.Timestamp.now(tz="UTC") if now_utc is None else pd.Timestamp(now_utc)
    if now.tzinfo is None:
        raise ValueError("now_utc must be timezone-aware")
    now = now.tz_convert("UTC")
    missing_columns = sorted(REQUIRED - set(df.columns))
    if missing_columns:
        return {"status": "REJECT_SCHEMA", "row_count": len(df),
                "missing_columns": missing_columns}, pd.DataFrame(index=df.index)
    age, age_bad = _numeric(df["age_years"])
    creat, creat_bad = _numeric(df["creatinine_mg_dl"])
    prior, prior_bad = _numeric(df["prior_admissions_12m"])
    # Explicit ISO 8601: with an inferred format, pandas turns rows that differ from the
    # first row's format into NaT. Naive strings are read as UTC, so the interface
    # contract must state that sources send offsets.
    discharge = pd.to_datetime(df["discharge_datetime"], errors="coerce", utc=True, format="ISO8601")
    ingested = pd.to_datetime(df["ingested_at"], errors="coerce", utc=True, format="ISO8601")
    # Illustrative bounds only; approve them with clinical informatics.
    reasons = pd.DataFrame({
        "null_required": df[sorted(REQUIRED - NULLABLE)].isna().any(axis=1),
        "unparseable_numeric": age_bad | creat_bad | prior_bad,
        "duplicate_encounter": df["encounter_id"].notna() & df["encounter_id"].duplicated(keep=False),
        "age_out_of_range": age.notna() & ~age.between(0, 120),
        "creatinine_out_of_range": creat.notna() & ~creat.between(0.1, 25.0),
        "prior_admissions_invalid": prior.notna() & ((prior < 0) | (prior % 1 != 0)),
        "bad_timestamp": (discharge.isna() & df["discharge_datetime"].notna())
                         | (ingested.isna() & df["ingested_at"].notna()),
        "future_ingestion": ingested > now + clock_skew,
        "discharge_after_ingestion": discharge > ingested + clock_skew,
    }, index=df.index)
    quarantined = reasons.any(axis=1)
    newest = ingested.max()
    staleness = None if pd.isna(newest) else now - newest
    stale = staleness is None or staleness > max_feed_staleness
    report = {
        "row_count": int(len(df)),
        "quarantined_rows": int(quarantined.sum()),
        "reason_counts": {k: int(v) for k, v in reasons.sum().items() if v},
        "null_rates": {c: float(df[c].isna().mean()) for c in sorted(REQUIRED)},
        "feed_staleness_minutes": None if staleness is None else staleness.total_seconds() / 60,
        "status": "REJECT_STALE_FEED" if stale else ("PARTIAL" if quarantined.any() else "PASS"),
    }
    return report, reasons[quarantined]

Note. Route quarantined rows to a protected workflow with reason code, source identifier, controlled access and a documented correction/replay process; never repair silently. Duplicate checks here are within-batch: also enforce uniqueness at the store with a constraint. Do not log names, record numbers, raw lab values or full records in application logs.

Units and clinical semantics

Store units with feature definitions and convert only through versioned, tested transformations. Define whether each feature is the latest value, a window aggregate or a value available at prediction time, and test that it existed at the prediction timestamp (no temporal leakage). Distinguish missing, not measured, not applicable and delayed: a missing lab can carry clinical meaning and should not be imputed as zero. Version code sets and terminology mappings, and watch for unmapped values after EHR changes.

4. Latency, availability, and operational health

A score is useful only when it arrives at the right point in the workflow. Monitor p50/p95/p99 latency, throughput, timeouts, errors, queue depth, dependency latency and saturation, and measure end-to-end latency separately from feature retrieval, preprocessing, inference and audit writes. Percentiles from a short sequential script say nothing about concurrency or queueing: emit histogram observations and let the metrics backend compute percentiles over defined windows.

Prometheus instrumentation

The middleware labels requests with the router's route template rather than the raw path, which keeps cardinality bounded and stays correct as routes are added (a hand-maintained allowlist drifts: one containing /v1/score would silently label all /v1/readmission-risk traffic as "other"). A score histogram per model version provides a PHI-free prediction-drift signal. Never use patient, encounter or request IDs, raw URLs or exception messages as labels.

# observability.py
# Dependencies: fastapi, prometheus-client. With several worker processes, set
# PROMETHEUS_MULTIPROC_DIR and use a multiprocess collector (see prometheus_client docs).
from time import perf_counter
from fastapi import FastAPI, Request
from prometheus_client import Counter, Histogram, make_asgi_app
REQUESTS = Counter(
    "readmission_api_requests_total", "Requests handled by the scoring API",
    ["route", "method", "status_class"],
)
DURATION = Histogram(
    "readmission_api_request_duration_seconds", "End-to-end API request duration",
    ["route", "method"], buckets=(0.01, 0.025, 0.05, 0.1, 0.2, 0.5, 1, 2, 5),
)
# Prediction drift signal without PHI: the score distribution per model version.
SCORES = Histogram(
    "readmission_risk_score", "Distribution of returned risk scores",
    ["model_version"], buckets=tuple(i / 10 for i in range(1, 11)),
)
def install_metrics(app: FastAPI) -> None:
    app.mount("/internal/metrics", make_asgi_app())  # restrict by network policy
    @app.middleware("http")
    async def observe_requests(request: Request, call_next):
        started, status = perf_counter(), "5xx"
        try:
            response = await call_next(request)
            status = f"{response.status_code // 100}xx"
            return response
        finally:
            # Route *template* set by the router (bounded cardinality), never the raw path.
            route = getattr(request.scope.get("route"), "path", "unmatched")
            REQUESTS.labels(route, request.method, status).inc()
            DURATION.labels(route, request.method).observe(perf_counter() - started)

Note. Keep the metrics endpoint off the public network. Add tracing with W3C trace context, and keep only opaque correlation IDs and release metadata in logs and traces.

SLOs and alerting

Derive latency and availability targets from the clinical workflow rather than copying a generic 500 ms target. Alert on sustained error-budget burn, tail latency, missing telemetry, queue saturation and dependency failures. Separate liveness from readiness; bound timeouts, concurrency and retries on EHR, feature-store and model calls. If a score cannot be produced, show a clear unavailable state and invoke the approved fallback; never show a stale score as current.

5. Model performance and clinical/business outcomes

Performance can be measured only on mature, correctly defined labels. A discharge from fewer than 30 days ago is not yet a negative, and the outcome definition must say how to treat readmission elsewhere, death before readmission, planned readmissions and incomplete EHR or claims capture.

Discrimination: ROC-AUC and average precision, read against prevalence and intended use.

Operating point: sensitivity and PPV at the clinically approved threshold.

Calibration: Brier score, observed/expected ratio, calibration intercept and slope, and a calibration table.

Subgroups and workload: performance and calibration by facility and relevant cohorts (subject to lawful use and privacy safeguards); patients flagged versus follow-up capacity.

A readmission model is predictive, not causal: a lower readmission rate after deployment does not show that the model caused it, because staffing, pathways and case mix also change. Evaluate impact with a prospective design under clinical oversight.

Evaluation code for matured labels

The function reports how many rows were excluded instead of dropping them silently, returns AUC and calibration fits as unavailable for one-class windows, and uses fixed-width risk bands that remain defined for constant scores. Calibration intercept and slope, which the metric list calls for, are computed with a small logistic fit on the logit of the predicted risk.

# outcome_metrics.py
from __future__ import annotations
import numpy as np
import pandas as pd
from sklearn.metrics import (
    average_precision_score, brier_score_loss, confusion_matrix, roc_auc_score,
)
def _logit(p: np.ndarray) -> np.ndarray:
    p = np.clip(p, 1e-6, 1 - 1e-6)
    return np.log(p / (1 - p))
def _irls(X: np.ndarray, y: np.ndarray, offset: np.ndarray, iters: int = 50) -> np.ndarray:
    """Unpenalised logistic regression with an offset (Newton-Raphson)."""
    beta = np.zeros(X.shape[1])
    for _ in range(iters):
        mu = 1 / (1 + np.exp(-(X @ beta + offset)))
        w = mu * (1 - mu)
        step = np.linalg.solve(X.T @ (X * w[:, None]), X.T @ (y - mu))
        beta += step
        if np.max(np.abs(step)) < 1e-8:
            break
    return beta
def evaluate_matured_outcomes(
    df: pd.DataFrame, threshold: float, min_rows: int = 100,
) -> tuple[dict, pd.DataFrame]:
    """Each row: one eligible discharge whose 30-day outcome has matured and been ascertained."""
    required = ["predicted_risk", "readmitted_30d"]
    if missing := set(required) - set(df.columns):
        raise ValueError(f"missing columns: {sorted(missing)}")
    if not 0.0 <= threshold <= 1.0:
        raise ValueError("threshold must be in [0, 1]")
    data = df[required].apply(pd.to_numeric, errors="coerce")
    excluded = int(data.isna().any(axis=1).sum())  # report, never silently drop
    data = data.dropna()
    if len(data) < min_rows:
        raise ValueError(f"insufficient matured labels: {len(data)} (excluded {excluded})")
    if not data["predicted_risk"].between(0, 1).all():
        raise ValueError("predicted_risk must be in [0, 1]")
    if not data["readmitted_30d"].isin([0, 1]).all():
        raise ValueError("readmitted_30d must contain only 0 or 1")
    y = data["readmitted_30d"].to_numpy(dtype=int)
    score = data["predicted_risk"].to_numpy(dtype=float)
    pred = (score >= threshold).astype(int)
    tn, fp, fn, tp = confusion_matrix(y, pred, labels=[0, 1]).ravel()
    both = np.unique(y).size == 2
    lp = _logit(score)
    ones = np.ones((len(y), 1))
    metrics = {
        "n": int(len(y)), "excluded_rows": excluded,
        "prevalence": float(y.mean()), "flag_rate": float(pred.mean()),
        "roc_auc": float(roc_auc_score(y, score)) if both else None,
        "average_precision": float(average_precision_score(y, score)) if both else None,
        "brier_score": float(brier_score_loss(y, score)),
        "observed_expected_ratio": float(y.sum() / score.sum()) if score.sum() else None,
        # Calibration-in-the-large (ideal 0) and calibration slope (ideal 1).
        "calibration_intercept": float(_irls(ones, y, lp)[0]) if both else None,
        "calibration_slope": float(_irls(np.column_stack([ones, lp]), y, 0 * lp)[1]) if both else None,
        "sensitivity": float(tp / (tp + fn)) if tp + fn else None,
        "ppv": float(tp / (tp + fp)) if tp + fp else None,
        "specificity": float(tn / (tn + fp)) if tn + fp else None,
        "threshold": float(threshold),
    }
    # Equal-width bands stay defined when scores are constant or have few unique values.
    data["risk_band"] = pd.cut(data["predicted_risk"], np.linspace(0, 1, 11), include_lowest=True)
    calibration = (
        data.groupby("risk_band", observed=False)
        .agg(mean_predicted=("predicted_risk", "mean"),
             observed_rate=("readmitted_30d", "mean"),
             patients=("readmitted_30d", "size"))
        .reset_index()
    )
    return metrics, calibration

Note. min_rows is an example guard, not a statistical rule. A formal design should add confidence intervals, temporal windows, clustering of repeat patients, site effects and multiple-comparison control; a noisy small-cohort metric is not a safety conclusion.

6. Clinical workflow and business monitoring

The goal is not to flag more patients; it is to support an agreed workflow without overwhelming care teams or widening inequities. Track the share of eligible discharges scored; high-risk patients reviewed by an authorised team member; follow-up scheduling, completion and time to first contact; workload and alert burden per 100 discharges; the prespecified unplanned 30-day readmission rate; and safety events and subgroup disparities.

Do not change the threshold just because the team is overloaded. Reassess calibration, decision-curve or utility trade-offs, capacity and clinical policy with clinical leadership; a threshold change is a clinical workflow change to be versioned, approved, communicated and evaluated.

7. Production healthcare monitoring architecture

End-to-end flow. An authenticated source emits a discharge event with a traceable ID; ingestion validates and quarantines invalid rows; point-in-time features are built from information available at prediction time; the service checks model/feature schema compatibility and scores under a deadline; the score is stored with model, feature-schema and policy versions, prediction time and a pseudonymous encounter reference; authorised users see it with intended-use text and a fallback path; privacy-minimised telemetry and drift aggregates are emitted; matured outcomes are linked through a controlled evaluation pipeline; and alerts go to named data, platform, model and clinical owners who record investigation and disposition.

8. Reference implementation: scoring API with audit

The service boundary below uses Pydantic v2 validation, separate liveness and readiness, and a bounded correlation ID. Authentication fails closed: a bearer token must be a valid RS256 JWT from the configured issuer and audience, with exp, sub and the readmission:score scope. Merely checking that an Authorization header is present, as placeholder code often does, accepts any string. Encounter-level authorisation is an injected policy that denies by default until wired to the EHR/care-team service. A score is returned only after its audit event is written, and the event records who asked and when. Caller-supplied request IDs are accepted only in a bounded format, because they are logged and echoed back.

# clinical_scoring_api.py
# Dependencies: fastapi, uvicorn, pydantic>=2, pyjwt[crypto], prometheus-client
from __future__ import annotations
import logging
import os
import re
from contextlib import asynccontextmanager
from datetime import datetime, timezone
from typing import Annotated
from uuid import uuid4
import jwt
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from pydantic import BaseModel, ConfigDict, Field
from observability import SCORES, install_metrics
log = logging.getLogger("clinical_scoring")
REQUEST_ID = re.compile(r"^[A-Za-z0-9-]{8,64}$")
INTENDED_USE = "Clinical decision support only; not a diagnosis or a substitute for clinician judgment."
class ScoreInput(BaseModel):
    model_config = ConfigDict(extra="forbid")
    encounter_token: str = Field(min_length=8, max_length=128)
    age_years: float = Field(ge=0, le=120)
    prior_admissions_12m: int = Field(ge=0, le=100)
    creatinine_mg_dl: float | None = Field(default=None, ge=0.1, le=25.0)
class ScoreOutput(BaseModel):
    request_id: str
    model_version: str
    score: float = Field(ge=0, le=1)
    prediction_time_utc: datetime
    intended_use: str
class ApprovedModel:
    version = "readmission-risk-v1.0.0"
    def predict_probability(self, x: ScoreInput) -> float:
        raw = 0.015 * x.age_years + 0.12 * x.prior_admissions_12m  # placeholder, not a clinical model
        if x.creatinine_mg_dl is not None:
            raw += 0.025 * x.creatinine_mg_dl
        return max(0.0, min(1.0, raw / 10.0))
def deny_all(subject: str, encounter_token: str) -> bool:
    return False  # fail closed until the EHR/care-team authorization service is wired in
def write_audit_event(event: dict) -> None:
    # Replace with a durable, access-controlled audit store; no direct identifiers or raw values.
    log.info("score_audit %s", event)
@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.model = ApprovedModel()           # load, verify checksum/signature, warm up
    app.state.jwt_key = os.environ["SCORING_JWT_PUBLIC_KEY"]  # or a PyJWKClient for JWKS
    app.state.jwt_issuer = os.environ["SCORING_JWT_ISSUER"]
    app.state.jwt_audience = os.environ["SCORING_JWT_AUDIENCE"]
    app.state.authorize_encounter = deny_all
    yield
app = FastAPI(lifespan=lifespan)
install_metrics(app)
bearer = HTTPBearer(auto_error=False)
@app.middleware("http")
async def correlation(request: Request, call_next):
    supplied = request.headers.get("x-request-id", "")
    # Caller-supplied IDs are logged and echoed: accept only a bounded, safe format.
    request.state.request_id = supplied if REQUEST_ID.fullmatch(supplied) else str(uuid4())
    response = await call_next(request)
    response.headers["x-request-id"] = request.state.request_id
    return response
def caller(request: Request,
           creds: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)]) -> dict:
    if creds is None:
        raise HTTPException(401, "authentication_required", {"WWW-Authenticate": "Bearer"})
    s = request.app.state
    try:
        claims = jwt.decode(
            creds.credentials, s.jwt_key, algorithms=["RS256"], audience=s.jwt_audience,
            issuer=s.jwt_issuer, options={"require": ["exp", "iss", "aud", "sub"]}, leeway=30,
        )
    except jwt.PyJWTError:
        raise HTTPException(401, "invalid_token", {"WWW-Authenticate": "Bearer"}) from None
    if "readmission:score" not in claims.get("scope", "").split():
        raise HTTPException(403, "insufficient_scope")
    return claims
@app.get("/health/live")
def live():
    return {"status": "alive"}
@app.get("/health/ready")
def ready(request: Request):
    model = getattr(request.app.state, "model", None)
    if model is None:
        raise HTTPException(503, "model_not_ready")
    return {"status": "ready", "model_version": model.version}
@app.post("/v1/readmission-risk", response_model=ScoreOutput)
def score(payload: ScoreInput, request: Request, claims: Annotated[dict, Depends(caller)]):
    state, request_id = request.app.state, request.state.request_id
    if not state.authorize_encounter(claims["sub"], payload.encounter_token):
        raise HTTPException(403, "not_authorized_for_encounter")
    model = state.model
    event = {"request_id": request_id, "subject": claims["sub"], "model_version": model.version,
             "at": datetime.now(timezone.utc).isoformat()}
    try:
        probability = model.predict_probability(payload)
        write_audit_event(event | {"outcome": "success"})  # no audit, no score
    except Exception:
        log.exception("scoring_failed request_id=%s", request_id)
        try:
            write_audit_event(event | {"outcome": "failure"})
        except Exception:
            log.exception("audit_failed request_id=%s", request_id)
        raise HTTPException(503, "scoring_temporarily_unavailable") from None
    SCORES.labels(model.version).observe(probability)
    return ScoreOutput(request_id=request_id, model_version=model.version, score=probability,
                       prediction_time_utc=datetime.now(timezone.utc), intended_use=INTENDED_USE)

Before deployment: replace the placeholder formula (never use it for care) with an approved, signature-verified artifact; add point-in-time feature retrieval with deadlines, clinical eligibility and policy checks, rate limits, a durable audit store, JWKS key rotation, secrets management, encryption and tested downtime behaviour.

9. Delayed labels and outcome maturity

Readmission outcomes are delayed by definition, so an unobserved outcome must never be treated as a negative. Evaluate a discharge only after the full 30-day window plus the agreed ingestion or claims lag, and classify incomplete follow-up, transfer ambiguity and missing capture separately. Store prediction, discharge and label-availability times with label provenance; define planned versus unplanned readmission, transfers, death as a competing event and cross-facility capture; report eligible, matured, ascertainable and excluded counts beside every metric; and recompute with a versioned evaluation snapshot when late labels arrive.

10. Privacy, security, safety, and governance

Minimise PHI in logs, metrics, traces and drift datasets; prefer pseudonymous tokens and aggregate statistics, and suppress small subgroup counts.

Encrypt in transit and at rest; use managed keys, secrets management, network segmentation and least-privilege service identities; apply role-based access, access reviews, audit logging, retention policies and BAAs where applicable.

Maintain a model inventory, intended-use statement, model card, lineage, validation report, known limitations and change history; require clinical sign-off for threshold, feature, recalibration and retraining changes.

Define a safety escalation path for harmful recommendations, data corruption, subgroup underperformance and unavailability, and exercise backup/restore, disaster-recovery and incident-response tests.

HIPAA compliance is an organisational, system-wide responsibility, not a property of this code or of encryption alone. In the US, the HHS proposal to update the HIPAA Security Rule (published January 2025) is not yet final, and FDA's revised Clinical Decision Support Software guidance (January 2026) now addresses risk scores and single recommendations directly. Whether a readmission score is non-device CDS depends on its function and intended use, so obtain qualified regulatory, privacy and security review for your jurisdiction.

11. Alerting and response playbooks

Every alert needs an owner, severity, evidence window, runbook, acknowledgement target, escalation path and closure reason. Alert on sustained or clinically meaningful conditions rather than every fluctuation.

12. CI/CD for monitoring code

Monitoring code is production software: test its calculations, schema compatibility, missingness, timestamp handling, failure paths, access control and alert routing against de-identified or synthetic data. The workflow uses current major versions of the GitHub actions, and points the pip cache at requirements.lock; without cache-dependency-path, setup-python looks for requirements.txt and the cache step fails.

# .github/workflows/monitoring.yml
name: healthcare-ml-monitoring-ci
on:
  pull_request:
  push:
    branches: [main]
permissions:
  contents: read
jobs:
  test:
    runs-on: ubuntu-latest
    timeout-minutes: 20
    steps:
      # Pin to reviewed commit SHAs in production; tags shown for readability.
      - uses: actions/checkout@v7
        with:
          persist-credentials: false
      - uses: actions/setup-python@v7
        with:
          python-version: "3.13"
          cache: pip
          cache-dependency-path: requirements.lock  # default only looks for requirements.txt
      - name: Install locked dependencies
        run: python -m pip install --require-hashes -r requirements.lock
      - name: Static checks
        run: |
          ruff check .
          mypy .
      - name: Unit, contract and failure-path tests
        run: pytest -q tests/
      - name: Dependency audit
        run: pip-audit --require-hashes -r requirements.lock
      # Add image scanning, SBOM, artifact signing, protected deployment approval.

Note. Pin third-party actions to reviewed commit SHAs, use protected deployment environments, and prefer short-lived federated (OIDC) credentials over long-lived cloud keys.

13. Putting it together

An effective programme starts with data contracts and point-in-time correctness, continuously monitors feature and prediction drift and service health, and evaluates performance only once reliable outcomes mature, while tracking whether the clinical workflow can act on the score safely.

Monitoring should trigger investigation and a governed response, not automatic model replacement. The right response may be to fix an interface, update a mapping, recalibrate under validation, restrict use, add workflow capacity or retrain after review. The score remains one input to an authorised clinical decision, and every threshold, range, SLO and outcome definition here must be set and validated locally.

Further reading

NIST AI Risk Management Framework: https://www.nist.gov/itl/ai-risk-management-framework

FDA, Clinical Decision Support Software guidance (rev. Jan 2026): https://www.fda.gov/regulatory-information/search-fda-guidance-documents/clinical-decision-support-software

HHS, HIPAA Security Rule: https://www.hhs.gov/hipaa/for-professionals/security/index.html

Prometheus, Histograms and summaries: https://prometheus.io/docs/practices/histograms/

Evidently AI documentation: https://docs.evidentlyai.com/ · MLflow Model Registry: https://mlflow.org/docs/latest/ml/model-registry/

Back to all blogs