Capstone: Delivery ETA Prediction

Two hours after pickup, order O-201 is still in transit. The customer opens their tracking app asking "when will it arrive?", while operations has to decide whether a proactive delay warning is safe to show. Four completed historical deliveries can support a duration baseline, but they can't justify an arrival promise down to the exact minute. When thin evidence gets dressed up as a precise promise, a shiny model score won't rescue customer trust.

Design an Automated Support Agent ended on a disciplined operational rule: reply, refuse, or prepare a handoff with evidence an operator can inspect. Keep that discipline here: retain the inputs, source availability, and release configuration that explain a proposed action.

The project combines an arrival estimate with late-risk decisions at a pinned prediction time. We build the pieces separately: as-of scan selection and an authored-score router, fitted synthetic models, and a historical arrival baseline. The router proposes normal_tracking, warn_customer, or manual_review; it doesn't dispatch trucks, send alerts, or issue refunds.

The initial walkthrough uses hand-authored scores to exercise policy boundaries and fallback branches. A companion lab fits models, selects the cheapest of four thresholds on validation data, scores a later test split, and checks parameter serialization. Its warning model beats a scan-age rule but loses to warning everyone. That negative result is part of the lesson. Neither fixture establishes deployment readiness.

Pin the decision before you fit a model

Before fitting any weights, freeze the operational question. We score an order at a scheduled snapshot while it's actively in transit. For order O-201, that moment happens to be two hours after pickup. Other orders might have been on the road for thirty hours across cross-country freight corridors. The prediction target is binary: does delivery occur after the original promised deadline known when the customer placed the order? A rigid contract assuming every order is scored at pickup-plus-two-hours would break down across multi-day ground shipping networks where pickup occurred days ago.

Arrival estimation and delay warnings answer two distinct operational questions. An estimated arrival timestamp isn't a guaranteed deadline. The historical range describes these observed durations; it supplies no stated future prediction interval coverage. Held-out coverage estimates and a theorem under declared assumptions are different kinds of evidence.

The delay warning asks for the probability of missing the original promise, given the current snapshot. Its estimated late_risk feeds a versioned policy gate. A classifier's probability estimate can be poorly calibrated. A held-out reliability curve compares predicted probabilities with observed late fractions in buckets; include bucket counts and relevant slices. It estimates past calibration rather than proving future frequencies.^{[1]Reference 1Probability calibrationhttps://scikit-learn.org/stable/modules/calibration.html} Once an undelivered package's deadline has passed, use a separately defined post-breach incident policy.

Always store delivery promises as timezone-aware UTC instants. Converting a local customer deadline like "delivered by 8:00 PM Thursday" requires the promise's declared timezone (often the destination timezone) and daylight-saving rules; don't assume midnight UTC matches the customer's end of day. If operations updates an internal promise during transit, don't let that revised timestamp silently overwrite the original customer-facing benchmark. Unresolved shipments that haven't delivered yet aren't on-time deliveries, and canceled or returned orders require explicit exclusion rules. Always document and report those cohort filters.

Next, define what the model's feature store can inspect. Our lightweight lab uses five features: carrier scan age, backlog snapshot age, service tier, elapsed transit time since pickup, and remaining hours until the promise deadline. Carrier scan age measures how recently a physical barcode was scanned at a hub; backlog snapshot age measures how recently the origin warehouse queue was calculated. A scaled production service might add weather severity, route congestion, or depot processing queues, but only if each source maintains its own availability timestamps.

Every admitted observation must have occurred and become accessible to the scoring service by the prediction cutoff. Here, ingested_at means queryable online, not merely written to a data lake. Feast supports filter_by_created_timestamp=True on supported offline stores when the source maps created_timestamp_column. That column must represent the availability clock you need: the flag filters created timestamps at or before the entity timestamp and excludes nulls. Default created-time deduplication alone doesn't impose that cutoff.^{[2]Reference 2Point-in-time Joinshttps://github.com/feast-dev/feast/blob/master/docs/getting-started/concepts/point-in-time-joins.md} Neither event time nor TTL alone establishes online availability.

A high risk score should never override stale input evidence. The policy gate inspects data freshness before evaluating the decision threshold, because a model prediction is only as reliable as the snapshot that fed it.

The diagram outlines our target serving architecture. Our runnable scope consists of a local feature and policy test harness alongside the synthetic training lab. The download script writes a dataset snapshot, fitted model parameters, test predictions, and a release evaluation report. You'll still need live carrier telemetry, an authenticated API, load tests, and a monitored rollout before claiming a production-ready service.

Prove the snapshot uses past events only

Pickup occurred at 2026-05-01 10:00 UTC, and our scheduled scoring snapshot runs at noon (12:00 UTC). Carrier telemetry records three scans for O-201. Which scans can our feature builder inspect at noon? You can't answer that question from event timestamps alone. Track both clocks carefully before writing a query.

First, filter candidate scans through both temporal boundaries, then select the latest admitted record. The derived feature scan_age_hours equals $t_{\text{pred}} - t_{\text{event}}$ for that admitted event: noon minus 11:00 yields exactly one hour.

Your pipeline tests should fail loudly if either form of leakage slips through: a future scan that hasn't happened yet, or an earlier scan whose ingestion pipeline didn't deliver it to the online store before scoring.

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from math import isfinite

BASE_TIME = datetime(2026, 5, 1, 10, tzinfo=timezone.utc)

def finite_number(value: object) -> float | None:
    if type(value) not in (int, float):
        return None
    try:
        number = float(value)
    except (OverflowError, ValueError):
        return None
    return number if isfinite(number) else None

def at(hours: float) -> datetime:
    return BASE_TIME + timedelta(hours=hours)

@dataclass(frozen=True)
class ScanEvent:
    order_id: str
    event_time: datetime
    ingested_at: datetime
    status: str

@dataclass(frozen=True)
class FeatureRow:
    order_id: str
    prediction_at: datetime
    scan_age_hours: float | None
    backlog_age_hours: float | None
    tier: str
    feature_contract_id: str = "eta-features-v1"

@dataclass(frozen=True)
class ScoredRow:
    features: FeatureRow
    late_risk: float
    model_id: str = "frozen-score-fixture-v1"

SCAN_EVENTS = [
    ScanEvent("O-201", at(1), at(1.05), "in_transit"),
    ScanEvent("O-201", at(1.5), at(2.5), "hub_scan"),
    ScanEvent("O-201", at(5), at(5.05), "carrier_delay_posted"),
]

print("scan_events:", len(SCAN_EVENTS))
print("statuses:", [event.status for event in SCAN_EVENTS])

scan_events: 3
statuses: ['in_transit', 'hub_scan', 'carrier_delay_posted']

Our filter enforces two distinct inequalities rather than blindly grabbing the newest database row. Event timestamps ask whether physical reality had shifted. Ingestion timestamps verify whether this online service had actually received the update.

def scans_known_by(order_id: str, prediction_at: datetime) -> list[ScanEvent]:
    return [
        event for event in SCAN_EVENTS
        if (
            event.order_id == order_id
            and event.event_time <= prediction_at
            and event.ingested_at <= prediction_at
        )
    ]

def latest_scan_known_by(order_id: str, prediction_at: datetime) -> ScanEvent:
    admitted = scans_known_by(order_id, prediction_at)
    if not admitted:
        raise ValueError("no scan available at prediction time")
    return max(admitted, key=lambda event: (event.event_time, event.ingested_at))

prediction_at = at(2)
admitted_scan_statuses = {event.status for event in scans_known_by("O-201", prediction_at)}
selected_scan = latest_scan_known_by("O-201", prediction_at)
print("prediction_at:", prediction_at.isoformat())
print("admitted:", sorted(admitted_scan_statuses))
print("selected:", selected_scan.status)

prediction_at: 2026-05-01T12:00:00+00:00
admitted: ['in_transit']
selected: in_transit

feature_row = FeatureRow(
    order_id="O-201",
    prediction_at=prediction_at,
    scan_age_hours=(prediction_at - selected_scan.event_time).total_seconds() / 3600,
    backlog_age_hours=2,
    tier="expedited",
)
feature_score = ScoredRow(feature_row, late_risk=0.62)

assert selected_scan.status == "in_transit"
assert admitted_scan_statuses == {"in_transit"}
assert feature_row.scan_age_hours == 1
print("selected_scan:", selected_scan.status, selected_scan.event_time.isoformat(), selected_scan.ingested_at.isoformat())
print("ignored_late_arrival:", SCAN_EVENTS[1].status, SCAN_EVENTS[1].event_time.isoformat(), SCAN_EVENTS[1].ingested_at.isoformat())
print("ignored_future_scan:", SCAN_EVENTS[2].status, SCAN_EVENTS[2].event_time.isoformat(), SCAN_EVENTS[2].ingested_at.isoformat())

selected_scan: in_transit 2026-05-01T11:00:00+00:00 2026-05-01T11:03:00+00:00
ignored_late_arrival: hub_scan 2026-05-01T11:30:00+00:00 2026-05-01T12:30:00+00:00
ignored_future_scan: carrier_delay_posted 2026-05-01T15:00:00+00:00 2026-05-01T15:03:00+00:00

These filters protect what the scoring service could have known at noon. The 15:00 delay notice might inform a 16:00 snapshot, but it can't feed a model scoring at noon. The hub_scan event physically took place at 11:30, but its database ingestion completed at 12:30. Ingesting that scan into a noon feature row would create online-offline skew. Keep this dual-clock test next to your feature transformation logic so both leakage modes get caught before model training begins.

Route scores through a versioned policy

A fitted late_risk score must originate from a named model artifact trained strictly on past outcomes. Starting with logistic regression lets you audit coefficients and standardizations directly before testing more complex architectures like gradient boosted decision trees.^{[3]Reference 3LogisticRegressionhttps://scikit-learn.org/stable/modules/generated/sklearn.linear_model.LogisticRegression.html}

Release check: A trained model isn't a release decision on its own. Evaluate the downstream operational actions against declared baselines and business cost schedules, alongside continuous ETA errors.

Our harness keeps feature validation distinct from model inference. That separation lets you verify policy edge cases on synthetic fixtures without confusing test assertions with statistical model performance.

Our local trace dictionary declares model, feature, and threshold-policy IDs alongside the prediction time and freshness numbers. It isn't immutable storage or proof that those artifacts were loaded. Production replay also needs retained inputs, source records, transformation code, verified artifact bytes, and the exact policy configuration. This lab neither delivers notifications nor issues concessions.

Freshness checks form part of the action contract. Both supplied ages must be finite, nonnegative, and at most their configured limits. A schema mismatch, invalid score, or stale age returns a proposed review action. No notification or review delivery occurs in this harness.

@dataclass(frozen=True)
class ReleasePolicy:
    threshold: float
    max_scan_age_hours: float
    max_backlog_age_hours: float
    model_id: str
    feature_contract_id: str
    threshold_policy_id: str

    def __post_init__(self):
        threshold = finite_number(self.threshold)
        if threshold is None or not 0 <= threshold <= 1:
            raise ValueError("threshold must be finite and in [0, 1]")
        for age in (self.max_scan_age_hours, self.max_backlog_age_hours):
            number = finite_number(age)
            if number is None or number < 0:
                raise ValueError("freshness limits must be finite and nonnegative")
        for identifier in (self.model_id, self.feature_contract_id, self.threshold_policy_id):
            if type(identifier) is not str or not identifier.strip():
                raise ValueError("policy identifiers must be nonempty strings")

POLICY = ReleasePolicy(
    threshold=0.40,
    max_scan_age_hours=24,
    max_backlog_age_hours=8,
    model_id="frozen-score-fixture-v1",
    feature_contract_id="eta-features-v1",
    threshold_policy_id="eta-threshold-v1",
)

def score(
    order_id: str,
    scan_age_hours: float | None,
    backlog_age_hours: float | None,
    tier: str,
    late_risk: float,
    feature_contract_id: str = "eta-features-v1",
) -> ScoredRow:
    return ScoredRow(
        FeatureRow(order_id, prediction_at, scan_age_hours, backlog_age_hours, tier, feature_contract_id),
        late_risk,
    )

def response(scored: ScoredRow, action: str, reason: str) -> dict[str, object]:
    row = scored.features
    return {
        "order_id": row.order_id,
        "action": action,
        "reason": reason,
        "late_risk": json_number(scored.late_risk),
        "prediction_at": row.prediction_at.isoformat(),
        "scan_age_hours": json_number(row.scan_age_hours),
        "backlog_age_hours": json_number(row.backlog_age_hours),
        "model_id": scored.model_id,
        "expected_model_id": POLICY.model_id,
        "feature_contract_id": row.feature_contract_id,
        "expected_feature_contract_id": POLICY.feature_contract_id,
        "threshold_policy_id": POLICY.threshold_policy_id,
    }

print("policy:", POLICY.threshold_policy_id, "threshold=", POLICY.threshold)

Read the row in the same order as the gate. O-201 passes freshness checks, then its 0.62 score clears the 0.40 threshold and proposes warn_customer. Stale or malformed telemetry stops before that comparison.

def json_number(value: object) -> float | None:
    return finite_number(value)

def invalid_age(value: float | None) -> bool:
    number = json_number(value)
    return number is None or number < 0

def input_issue(row: FeatureRow) -> str | None:
    if row.feature_contract_id != POLICY.feature_contract_id:
        return "feature_contract_mismatch"
    if row.tier not in ("standard", "expedited"):
        return "unsupported_service_tier"
    if invalid_age(row.scan_age_hours):
        return "invalid_scan_age"
    if invalid_age(row.backlog_age_hours):
        return "invalid_backlog_age"
    if row.scan_age_hours > POLICY.max_scan_age_hours:
        return "stale_scan_features"
    if row.backlog_age_hours > POLICY.max_backlog_age_hours:
        return "stale_backlog_features"
    return None

def route(scored: ScoredRow, *, expected_prediction_at: datetime = prediction_at) -> dict[str, object]:
    # Validate the envelope before building a response. An API would map this to 400.
    if not isinstance(scored, ScoredRow) or not isinstance(scored.features, FeatureRow):
        raise ValueError("expected a scored feature row")
    row = scored.features
    for identifier in (row.order_id, row.feature_contract_id, scored.model_id):
        if type(identifier) is not str or not identifier.strip():
            raise ValueError("row identifiers must be nonempty strings")
    for moment in (row.prediction_at, expected_prediction_at):
        if type(moment) is not datetime or moment.tzinfo is None or moment.utcoffset() != timedelta(0):
            raise ValueError("prediction timestamps must be aware UTC instants")
    if row.prediction_at != expected_prediction_at:
        return response(scored, "manual_review", "prediction_snapshot_mismatch")
    if scored.model_id != POLICY.model_id:
        return response(scored, "manual_review", "model_mismatch")
    issue = input_issue(scored.features)
    if issue is not None:
        return response(scored, "manual_review", issue)
    if json_number(scored.late_risk) is None or not 0 <= scored.late_risk <= 1:
        return response(scored, "manual_review", "invalid_late_risk")
    if scored.late_risk >= POLICY.threshold:
        return response(scored, "warn_customer", "late_risk_threshold")
    return response(scored, "normal_tracking", "below_threshold")

print("O-201 route:", route(feature_score)["action"], route(feature_score)["reason"])

policy_cases = [
    feature_score,
    score("O-202", 3, 2, "standard", 0.25),
    score("O-203", 31, 2, "standard", 0.81),
    score("O-204", 4, 12, "standard", 0.75),
    score("O-205", 4, 2, "standard", float("nan")),
    score("O-206", 4, None, "standard", 0.75),
    score("O-207", float("nan"), 2, "standard", 0.75),
    score("O-208", -1, 2, "standard", 0.75),
    score("O-209", 4, 2, "standard", 0.75, "eta-features-v0"),
]

for scored in policy_cases:
    result = route(scored)
    print(scored.features.order_id, result["action"], result["reason"])

# Every fallback must remain valid JSON; NaN and infinity aren't JSON numbers.
import json
for scored in policy_cases:
    json.dumps(route(scored), allow_nan=False)
assert route(ScoredRow(feature_row, True))["reason"] == "invalid_late_risk"
assert route(ScoredRow(feature_row, "0.62"))["reason"] == "invalid_late_risk"
assert route(ScoredRow(feature_row, 0.62, "unexpected-model"))["reason"] == "model_mismatch"

trace = route(feature_score)
print("release_tuple:", trace.get("feature_contract_id"), trace.get("model_id"), trace.get("threshold_policy_id"))

O-201 warn_customer late_risk_threshold
O-202 normal_tracking below_threshold
O-203 manual_review stale_scan_features
O-204 manual_review stale_backlog_features
O-205 manual_review invalid_late_risk
O-206 manual_review invalid_backlog_age
O-207 manual_review invalid_scan_age
O-208 manual_review invalid_scan_age
O-209 manual_review feature_contract_mismatch
release_tuple: eta-features-v1 frozen-score-fixture-v1 eta-threshold-v1

Compare O-203 and O-204 with O-201: high scores can't override stale telemetry. These rows supply different scan ages at the same noon prediction time. Cases O-205 through O-209 exercise NaN, missing values, negative ages, and schema mismatch. Invalid numeric fields become JSON null. Malformed identifiers or timestamp envelopes raise ValueError; a real API would return a structured request error. The expected snapshot must come from trusted scheduling state. These checks don't authenticate caller-supplied ages or IDs, and the backlog age is authored rather than read from a source.

Check the cost calculation with four fixtures

Before evaluating trained models on large splits, test the cost accounting logic on four hand-authored fixtures. The heuristic rule warns whenever scan_age_hours >= 18; the model candidate warns when predicted late_risk >= 0.40. These rows don't claim statistical validity; they exist to verify that our cost ledger accounts for every operational penalty correctly.

Choose a fictional penalty schedule to make the action tradeoff visible. These are lesson units, not measured dollars, churn estimates, or an industry standard:

• False warning: Warn about a delay, then deliver on time. Charge $C_{\text{FP}} = 8$ units for the assumed inconvenience.
• Missed late delivery: Stay silent, then deliver late. Charge 60 units for standard service or 150 for expedited service.
• Correct decision: Charge zero for warning a late shipment or staying silent on an on-time shipment. This simplification omits notification cost on true positives, alert fatigue, and capacity limits. Add them to the cost model when they matter. A warning doesn't prevent lateness.

Deriving the optimal decision threshold

The classifier's usual $\tau = 0.50$ cutoff doesn't encode these costs. Suppose a standard shipment has a calibrated 20% late risk. Warning costs $0.8 \times 8 = 6.4$ expected units; silence costs $0.2 \times 60 = 12$. Which action would you choose?^{[4]Reference 4Tuning the decision threshold for class predictionhttps://scikit-learn.org/stable/modules/classification_threshold.html}

To see why, write the expected business cost of each action given a calibrated probability $p = P(\text{late} \mid x)$:

$\mathbb{E}[\text{Cost} \mid \text{warn}] = (1 - p) \cdot C_{\text{FP}} + p \cdot C_{\text{TP}}$

$\mathbb{E}[\text{Cost} \mid \text{silent}] = p \cdot C_{\text{FN}} + (1 - p) \cdot C_{\text{TN}}$

Setting the cost of correct outcomes to zero ($C_{\text{TP}} = 0$ and $C_{\text{TN}} = 0$), issuing a proactive warning is economically rational whenever its expected cost is less than or equal to the cost of staying silent:

$(1 - p) \cdot C_{\text{FP}} \le p \cdot C_{\text{FN}}$

$C_{\text{FP}} \le p \cdot (C_{\text{FN}} + C_{\text{FP}})$

Solving for $p$ yields the Bayes-optimal decision threshold $\tau^*$:

$\tau^* = \frac{C_{\text{FP}}}{C_{\text{FN}} + C_{\text{FP}}}$

Under this two-action, zero-correct-cost assumption, the tier thresholds are:

• Standard tier ($C_{\text{FP}} = 8$, $C_{\text{FN}} = 60$): $\tau^*_{\text{standard}} = \frac{8}{60 + 8} = \frac{8}{68} \approx 0.118$ A calibrated 12% late risk exceeds this threshold.
• Expedited tier ($C_{\text{FP}} = 8$, $C_{\text{FN}} = 150$): $\tau^*_{\text{expedited}} = \frac{8}{150 + 8} = \frac{8}{158} \approx 0.051$ A calibrated 5.1% late risk exceeds this threshold. Exactly 5.0% does not: rounding across a decision boundary changes the answer.

The first harness uses $0.40$ as an authored test setting. The fitted lab separately chooses the cheapest validation result among $\{0.2, 0.4, 0.6, 0.8\}$, breaking ties toward the larger threshold. It doesn't calibrate probabilities, search every threshold, evaluate the analytic tier thresholds, or measure alert capacity. Thus $0.40$ is the best of those four candidates on that validation cohort, not a globally optimal policy. Threshold tuning belongs on data separate from estimator fitting and final testing.^{[4]Reference 4Tuning the decision threshold for class predictionhttps://scikit-learn.org/stable/modules/classification_threshold.html}

Trace the four fixtures through the ledger before running the script. The 18-hour heuristic rule never flags expedited late shipment E-301 because its scan age is only 5 hours, incurring a 150-unit penalty. The candidate model catches E-301 and E-302, but issues an unnecessary warning on on-time shipment E-304 (late_risk = 0.58), incurring an 8-unit false alarm cost.

Separating unit tests for policy logic from statistical validation prevents confusing business arithmetic with generalization power:

@dataclass(frozen=True)
class CostFixture:
    row: ScoredRow
    delivered_late: bool

    def __post_init__(self):
        if type(self.delivered_late) is not bool:
            raise ValueError("delivered_late must be Boolean")
        if not isinstance(self.row, ScoredRow) or self.row.features.tier not in ("standard", "expedited"):
            raise ValueError("cost fixture requires a recognized service tier")

cost_fixtures = [
    CostFixture(score("E-301", 5, 1, "expedited", 0.78), True),
    CostFixture(score("E-302", 20, 2, "standard", 0.64), True),
    CostFixture(score("E-303", 4, 1, "standard", 0.12), False),
    CostFixture(score("E-304", 3, 2, "standard", 0.58), False),
]

def warning_cost(case: CostFixture, warn: bool) -> int:
    if type(warn) is not bool:
        raise ValueError("warn must be Boolean")
    if warn and not case.delivered_late:
        return 8
    if not warn and case.delivered_late:
        return 150 if case.row.features.tier == "expedited" else 60
    return 0

def baseline_warn(case: CostFixture) -> bool:
    row = case.row.features
    return input_issue(row) is None and row.scan_age_hours is not None and row.scan_age_hours >= 18

def candidate_warn(case: CostFixture) -> bool:
    return route(case.row)["action"] == "warn_customer"

print("cost fixture rows:", len(cost_fixtures))
print("baseline warns:", [case.row.features.order_id for case in cost_fixtures if baseline_warn(case)])

baseline_cost = sum(warning_cost(case, baseline_warn(case)) for case in cost_fixtures)
candidate_cost = sum(warning_cost(case, candidate_warn(case)) for case in cost_fixtures)
expedited_misses = sum(
    case.row.features.tier == "expedited" and case.delivered_late and not candidate_warn(case)
    for case in cost_fixtures
)
fallback_reasons = {row.features.order_id: route(row)["reason"] for row in policy_cases[2:]}
print("baseline_cost:", baseline_cost, "candidate_cost:", candidate_cost, "expedited_misses:", expedited_misses)

trace_keys = {"feature_contract_id", "expected_feature_contract_id", "model_id", "expected_model_id", "threshold_policy_id"}
freshness_trace_keys = {"prediction_at", "scan_age_hours", "backlog_age_hours"}

# This tests the admission function, not the provenance of a training dataset.
pit_ok = all(
    event.event_time <= prediction_at and event.ingested_at <= prediction_at
    for event in scans_known_by("O-201", prediction_at)
)

release_gates = {
    "replay_excludes_unavailable_scans": admitted_scan_statuses == {"in_transit"},
    "admitted_scan_clocks": pit_ok,
    "lower_cost_than_rule_baseline": candidate_cost < baseline_cost,
    "zero_expedited_misses": expedited_misses == 0,
    "stale_scan_falls_back": fallback_reasons["O-203"] == "stale_scan_features",
    "stale_backlog_falls_back": fallback_reasons["O-204"] == "stale_backlog_features",
    "invalid_score_falls_back": fallback_reasons["O-205"] == "invalid_late_risk",
    "missing_backlog_falls_back": fallback_reasons["O-206"] == "invalid_backlog_age",
    "invalid_scan_age_falls_back": fallback_reasons["O-207"] == "invalid_scan_age",
    "negative_scan_age_falls_back": fallback_reasons["O-208"] == "invalid_scan_age",
    "feature_contract_mismatch_falls_back": fallback_reasons["O-209"] == "feature_contract_mismatch",
    "versioned_trace": all(trace.get(key) for key in trace_keys),
    "freshness_trace": freshness_trace_keys.issubset(trace),
}

assert all(release_gates.values())
print("fixture_checks_pass:", all(release_gates.values()))

receipt = {
    "bundle_id": "frozen-score-fixture-v1",
    "evaluation_snapshot": "four-authored-policy-cases",
    "previous_bundle": None,
    "baseline_cost": baseline_cost,
    "candidate_cost": candidate_cost,
    "expedited_misses": expedited_misses,
    "fixture_checks": release_gates,
    "candidate_decision": "policy_unit_checks_passed" if all(release_gates.values()) else "hold",
}

print("bundle:", receipt["bundle_id"], "rollback:", receipt["previous_bundle"])
print("cost:", {"baseline": receipt["baseline_cost"], "candidate": receipt["candidate_cost"]})
print("fixture_checks_pass:", all(receipt["fixture_checks"].values()))
print("candidate_decision:", receipt["candidate_decision"])

bundle: frozen-score-fixture-v1 rollback: None
cost: {'baseline': 150, 'candidate': 8}
fixture_checks_pass: True
candidate_decision: policy_unit_checks_passed

This receipt records policy unit checks, not statistical generalization or permission to route traffic. Zero misses on one expedited fixture supplies no useful population reliability claim. Inspect slice denominators and empirical miss rates alongside total penalty units.

Fit models and evaluate their actual predictions

The downloadable lab generates 136 synthetic shipments with pickup, prediction, original promise, telemetry availability, delivery, and label-recording timestamps. Each shipment has one snapshot. With repeated scoring, define the evaluation target: performance on unseen shipments may require grouping and purging overlapping journeys; a service forecasting later snapshots of existing shipments needs its own chronological replay. Shared outcomes and future labels must never enter an earlier fit.

The lab trains a scaled logistic regression classifier for late risk and an ordinary least squares regressor for remaining transit hours. Feature transformations, including standard scaling, are fit strictly on training splits. The training pipeline freezes estimators, tunes the warning decision threshold across candidates [0.2, 0.4, 0.6, 0.8] against validation costs, and evaluates the final configuration once on a chronologically later test set. Selecting decision thresholds on test data invalidates the holdout split by turning it into a second tuning set.^{[5]Reference 5Common Pitfalls and Recommended Practices: Data Leakagehttps://scikit-learn.org/stable/common_pitfalls.html#data-leakage[4]Reference 4Tuning the decision threshold for class predictionhttps://scikit-learn.org/stable/modules/classification_threshold.html[6]Reference 6LinearRegressionhttps://scikit-learn.org/stable/modules/generated/sklearn.linear_model.LinearRegression.html}

Separate prediction time, label availability, and evaluation time

To evaluate forecasting on future shipping conditions, train before the evaluated prediction cohort. A shuffled split can put later conditions into training and answer a different question. Rolling-origin evaluation repeats this experiment at several cutoffs; our lab runs one train/validation/test split.^{[7]Reference 7Cross-validation: evaluating estimator performancehttps://scikit-learn.org/stable/modules/cross_validation.html}

Keep three boundaries separate:

1. Prediction Cutoff ($t_{\text{pred}}$): The exact scheduled snapshot time. Features may only incorporate events where both $t_{\text{event}} \le t_{\text{pred}}$ and $t_{\text{ingest}} \le t_{\text{pred}}$.
2. Label availability: Training labels must be recorded strictly before FIT_AT; threshold-selection labels before FREEZE_AT; final test labels before EVALUATE_AT. A delivered shipment whose confirmation arrives later is still unavailable at that cutoff. Unresolved isn't on time. Restricting evaluation to already matured outcomes can overrepresent faster shipments, so report exclusions and choose cohorts with an appropriate follow-up horizon.
3. Window gap: The lab leaves four days between prediction windows. This is an illustrative choice, not a universal transit bound. A gap doesn't ensure every journey or reconciliation finishes. Explicit label-availability checks and the shipment grouping contract establish eligibility; choose any gap for the actual overlap and dependence you need to control.

In our synthetic run, the four-day embargo gaps exclude eight shipments, and one additional shipment has a label that arrived after the fitting cutoff, yielding nine excluded records in total.

Download eta_lab.py, then run it from the directory containing the file:

uv run eta_lab.py --out eta-output

The same module executes inside the lesson validation harness. It prints split sizes, validation-tuned thresholds, and slice-level performance. Note that remaining transit errors use hours: an absolute error of 2.5 represents two and a half hours, not minutes.

import runpy

lab = runpy.run_path("assets/eta_lab.py")
result = lab["run_lab"]()
report = result["report"]
print("split rows:", report["rows"])
print("validation threshold:", report["threshold"])
print("validation costs:", report["validation_costs"])
for name, metrics in report["slices"].items():
    print(name, metrics)
print("fixture cost gate:", report["fixture_cost_gate_passed"])
print("decision:", report["decision"])

split rows: {'train': 79, 'validation': 24, 'test': 24, 'excluded': 9}
validation threshold: 0.4
validation costs: {'0.2': 32, '0.4': 16, '0.6': 368, '0.8': 548}
all {'n': 24, 'late_n': 13, 'missed_late_n': 1, 'miss_rate': 0.077, 'candidate_cost': 158, 'rule_cost': 1410, 'always_warn_cost': 88, 'always_silent_cost': 1410, 'eta_mae_hours': 2.479, 'median_mae_hours': 4.203}
expedited {'n': 11, 'late_n': 7, 'missed_late_n': 1, 'miss_rate': 0.143, 'candidate_cost': 150, 'rule_cost': 1050, 'always_warn_cost': 32, 'always_silent_cost': 1050, 'eta_mae_hours': 2.29, 'median_mae_hours': 4.089}
standard {'n': 13, 'late_n': 6, 'missed_late_n': 0, 'miss_rate': 0.0, 'candidate_cost': 8, 'rule_cost': 360, 'always_warn_cost': 56, 'always_silent_cost': 360, 'eta_mae_hours': 2.639, 'median_mae_hours': 4.3}
fixture cost gate: False
decision: hold_for_real_data_validation

These are actual fitted predictions on the synthetic test rows. Compare the warning model with the 18-hour rule and the constant always-warn and always-silent policies under identical penalties. The ETA comparison uses training-set median remaining duration per tier. These baselines answer different questions and need separate metrics.

The candidate costs 158 versus the rule's 1,410, but always-warning costs just $11 \times 8 = 88$ units. The model misses one of seven expedited late deliveries, costing 150; its standard false warning adds 8. The comparison gate therefore fails. This is a useful negative result, not a reason to retune on the revealed test outcomes. Design a new policy on separate development data and obtain a new final evaluation. Seven expedited positives are also too few for a strong reliability claim.

decision always holds because the data is synthetic; it isn't a computed production promotion decision. The separate fixture_cost_gate_passed checks only whether aggregate fixture cost beats all three warning baselines. The pipeline exports four files:

The lab checks schema, feature order, shapes, finite parameters, positive scales, and threshold range before scoring reloaded JSON; its outputs match the original fitted estimators. This format supports these two linear estimators without unpickling binaries.^{[8]Reference 8Model persistencehttps://scikit-learn.org/stable/model_persistence.html} Validation isn't artifact authentication. Fingerprints identify bytes, not trusted provenance. Files overwrite the same output names; retain an immutable run directory and lock dependencies for reproducibility. Recorded dependency versions alone don't pin the environment.

Explain an arrival estimate without promising its range

A late-risk classifier predicts whether a shipment will breach its promised delivery window. It doesn't predict when the package will arrive. Our trained regressor predicts remaining transit hours and adds them to prediction_at. A simpler historical duration baseline calculates total transit duration from completed deliveries and adds it to picked_up_at. Adding either duration to the wrong baseline clock shifts every arrival prediction across the network.

At noon, matching completed deliveries on route east-7 have durations of 8, 10, 12, and 14 hours: median 11. A 10:00 pickup gives an estimated 21:00 UTC arrival, with observed arrivals from 18:00 to midnight. A fifth shipment takes four hours, delivers at 13:00, and is recorded at 13:05. Both outcome clocks are after noon, so it is unavailable. Because our current package is still in transit after two hours, the matching rule also conditions history on total duration exceeding two hours. This small, matured cohort may still suffer selection bias.

What would a 90% interval actually require?

An accident or depot bottleneck can produce a long delay, but our four authored durations are symmetric around 11 hours. Their median isn't inherently optimistic. The problem is presenting an estimate as certainty. Three approaches make different claims:

Observed extremes. If the $n$ historical values and the next value are exchangeable with no ties (for example, IID draws from a continuous distribution), every rank among $n+1$ is equally likely. The next value lies inside the historical extremes at $n-1$ ranks, giving marginal coverage $(n-1)/(n+1)$. For four observations that's 60%. This averages over random histories and the next value, not a measured 60% guarantee conditional on our particular four durations. Independence alone doesn't supply these assumptions; drift or biased selection can violate them.

Estimated quantiles. Squared-error loss targets the conditional mean at the population optimum; a fitted linear regressor needn't recover it exactly. Pinball loss targets a conditional quantile:^{[9]Reference 9Linear Modelshttps://scikit-learn.org/stable/modules/linear_model.html}

$\mathcal{L}_q(y, \hat{y}) = \max\Big(q(y-\hat{y}),\;(1-q)(\hat{y}-y)\Big)$

Predicted 5th and 95th percentiles define a nominal 90% interval, not established 90% coverage. Separate fits can cross. Fixing their order doesn't establish calibration.

Split conformal intervals. Fit $\hat f$ independently of an untouched calibration cohort, then sort its absolute residuals $s_i=|y_i-\hat f(x_i)|$. Set

$k=\lceil(n+1)(1-\alpha)\rceil,\qquad \hat q=\begin{cases}s_{(k)},&k\le n\\+\infty,&k>n.\end{cases}$

Here $s_{(k)}$ is the $k$th smallest residual, not an interpolated sample quantile. Return $[\hat f(x)-\hat q,\hat f(x)+\hat q]$. Exchangeability of calibration and next-example scores gives marginal coverage at least $1-\alpha$. This doesn't promise that coverage for every route or individual shipment.^{[10]Reference 10Distribution-Free Predictive Inference for Regressionhttps://www.stat.cmu.edu/~ryantibs/papers/conformal.pdf[11]Reference 11A Gentle Introduction to Conformal Prediction and Distribution-Free Uncertainty Quantificationhttps://arxiv.org/abs/2107.07511}

Try $n=4$ and $\alpha=0.10$ before running the code. The required rank is 5, beyond the four residuals: the distribution-free construction returns an unbounded interval. Clamping the rank to 4 would change the guarantee. With nine residuals, rank 9 is available for a nominal 90% target. These residuals are an arithmetic exercise, not fitted ETA calibration data.

from math import ceil, inf, isfinite

def conformal_radius(residuals: list[float], alpha: float) -> float:
    if type(alpha) not in (int, float) or not 0 < alpha < 1 or not isfinite(alpha):
        raise ValueError("alpha must be finite and between zero and one")
    if type(residuals) is not list or not residuals:
        raise ValueError("residuals must be a nonempty list")
    values = []
    for residual in residuals:
        if type(residual) not in (int, float):
            raise ValueError("residuals must be numeric, excluding Booleans")
        try:
            number = float(residual)
        except OverflowError:
            raise ValueError("residual is outside the supported numeric range") from None
        if not isfinite(number) or number < 0:
            raise ValueError("residuals must be finite and nonnegative")
        values.append(number)
    rank = ceil((len(values) + 1) * (1 - alpha))
    return sorted(values)[rank - 1] if rank <= len(values) else inf

print("four residuals, 90% target:", conformal_radius([1, 2, 3, 4], 0.10))
print("nine residuals, 90% target:", conformal_radius(list(range(1, 10)), 0.10))

four residuals, 90% target: inf
nine residuals, 90% target: 9.0

An infinite radius means this construction supplies no informative finite window; don't serialize inf into an API promise. Under changing freight conditions, rolling windows may help empirical adaptation but don't restore exchangeability or guarantee 90% coverage by themselves. Methods for nonexchangeable data have different bounds and conditions; adaptive long-run error control also differs from a per-shipment probability claim. Monitor observed coverage, widths, counts, and delayed labels by relevant slices.^{[12]Reference 12Conformal Prediction Beyond Exchangeabilityhttps://arxiv.org/abs/2202.13415}

from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from statistics import median

@dataclass(frozen=True)
class CompletedDelivery:
    order_id: str
    route_id: str
    tier: str
    picked_up_at: datetime
    delivered_at: datetime
    recorded_at: datetime

def utc(month: int, day: int, hour: int, minute: int = 0) -> datetime:
    return datetime(2026, month, day, hour, minute, tzinfo=timezone.utc)

history = [
    CompletedDelivery("H-1", "east-7", "expedited", utc(4, 27, 8), utc(4, 27, 16), utc(4, 27, 16, 5)),
    CompletedDelivery("H-2", "east-7", "expedited", utc(4, 28, 8), utc(4, 28, 18), utc(4, 28, 18, 5)),
    CompletedDelivery("H-3", "east-7", "expedited", utc(4, 29, 8), utc(4, 29, 20), utc(4, 29, 20, 5)),
    CompletedDelivery("H-4", "east-7", "expedited", utc(4, 30, 8), utc(4, 30, 22), utc(4, 30, 22, 5)),
    CompletedDelivery("H-late", "east-7", "expedited", utc(5, 1, 9), utc(5, 1, 13), utc(5, 1, 13, 5)),
]
picked_up_at = utc(5, 1, 10)
prediction_at = utc(5, 1, 12)
elapsed = prediction_at - picked_up_at
known_matches = [
    row
    for row in history
    if row.route_id == "east-7"
    and row.tier == "expedited"
    and row.delivered_at <= prediction_at
    and row.recorded_at <= prediction_at
    and row.delivered_at > row.picked_up_at
    and row.delivered_at - row.picked_up_at > elapsed
]
if not known_matches:
    raise ValueError("no eligible history: omit ETA and request review")
durations = [row.delivered_at - row.picked_up_at for row in known_matches]
duration_hours = [duration.total_seconds() / 3600 for duration in durations]
estimated_arrival = picked_up_at + timedelta(hours=median(duration_hours))
observed_low = picked_up_at + min(durations)
observed_high = picked_up_at + max(durations)

print("historical matches:", len(known_matches))
print("arrival estimate:", estimated_arrival.isoformat())
print("observed duration range:", observed_low.isoformat(), "to", observed_high.isoformat())
print("late outcome admitted:", any(row.order_id == "H-late" for row in known_matches))
assert len(known_matches) == 4
assert estimated_arrival == utc(5, 1, 21)

historical matches: 4
arrival estimate: 2026-05-01T21:00:00+00:00
observed duration range: 2026-05-01T18:00:00+00:00 to 2026-05-02T00:00:00+00:00
late outcome admitted: False

Include sample count, cutoff, and estimation method alongside an ETA. The standalone example raises ValueError when no eligible history remains. An API adapter would translate that failure into an omitted ETA and a structured review disposition; it isn't implemented here. Define whether missing ETA also suppresses warnings. An absent estimate never establishes an on-time outcome.

Operate the service after release

Deploying a model artifact is the beginning of operational ownership, not the end. Production monitoring divides into two distinct operational horizons based on feedback latency:

1. Immediate serving-time telemetry: Each inference call emits a structured trace record containing request timestamp, model artifact fingerprint, feature schema version, threshold policy identifier, input freshness metrics, predicted risk score, and chosen action. Real-time dashboards monitor input null rates, scan age distributions, feature schema violations, scoring latency, and fallback frequencies. These checks detect upstream pipeline stalls and data corruption before bad predictions reach users.
2. Delayed cohort evaluation: After labels mature, compute realized warning penalties, reliability curves, Brier scores, and slice miss rates by route, tier, and shipment age. A Brier score measures overall probabilistic prediction quality, combining calibration with other components; it isn't a pure calibration metric. Counts and follow-up windows matter when outcomes are delayed.^{[1]Reference 1Probability calibrationhttps://scikit-learn.org/stable/modules/calibration.html}

Verify promotion and recovery at the serving layer

Retraining produces a candidate bundle. Registry aliases can name that bundle, but a pointer change doesn't load models, update cached workers, or route traffic. A serving rollout needs compatible preprocessing, schema, model, threshold policy, readiness checks, and verification of the release actually handling requests.

Google Cloud's MLOps architecture describes comparison with current/baseline models, performance across segments, infrastructure/API compatibility, and online validation. It doesn't mandate our monetary penalty schedule.^{[13]Reference 13MLOps: Continuous Delivery and Automation Pipelines in Machine Learning.https://docs.cloud.google.com/architecture/mlops-continuous-delivery-and-automation-pipelines-in-machine-learning}

For a real deployment, predeclare evidence and recovery criteria:

• Decision quality: Compare declared penalties against incumbent and simple constant policies on identical evaluation rows; include calibration and ETA errors where relevant. Our synthetic cost gate fails.
• Critical slices: Define denominators, follow-up windows, minimum sample sizes, uncertainty estimates, and acceptable errors before testing. Zero observed misses isn't a population guarantee.
• Serving recovery: Retain a complete compatible prior bundle, loading capacity, and a tested traffic-switch procedure. Measure recovery time and verify actual serving versions. Reverting an alias isn't instant recovery and doesn't undo notifications already sent or other completed effects.

Submission checklist

Treat this table as an evidence packet rather than a list of intentions. A reviewer should be able to trace each item from the pinned snapshot through the operational action to the recovery plan.

Practice: break the release contract

Use the runnable examples as a small release rehearsal. Change one input at a time, name the contract field it tests, predict the result, then rerun. The answer should explain which gate moved, not only which string changed.

1. Move hub_scan ingestion from at(2.5) to at(1.75). Which scan should the as-of builder select?
2. Change O-204 backlog age from 12 to 7. Which action replaces manual_review?
3. Replace float("nan") with 1.4 for O-205. Why should the service still refuse the score?
4. Raise threshold from 0.40 to 0.80. Which expedited gate fails?
5. Remove threshold_policy_id from response(). Which release gate catches the incomplete trace?
6. In the download, move a training row's label_recorded_at past FIT_AT. Should its original prediction timestamp keep it in training?
7. Hold the test predictions fixed and change the false-warning penalty from 8 to 20 units. What happens to the model-versus-always-warn comparison? Does a favorable answer authorize release?

Practice answer sketches