Ship a delivery-delay warning service with as-of features, versioned policy gates, baseline evidence, and monitored fallback.
The production ML lessons gave you each component in isolation. This capstone packages them into a service another engineer can evaluate: given an in-transit order at a defined timestamp, estimate late-delivery risk and decide whether the product may show a proactive delay warning.
The product contract is intentionally narrow. The model doesn't promise an exact arrival minute, issue refunds, or change carrier routing. It returns a risk score with a controlled action: normal_tracking, warn_customer, or manual_review when inputs are unreliable.
Use one decision moment: two hours after carrier pickup. Use one label: whether delivery occurred after the promised date. A prediction stored without those definitions can't be replayed later.
| Contract field | Pinned value |
|---|---|
| prediction event | two hours after first carrier pickup |
| label | delivered after promised end-of-day |
| score output | late_risk between zero and one |
| displayed action | warn only when threshold passes |
| unavailable data action | route to manual_review, no narrow ETA claim |
The feature bundle includes route distance, service tier, origin backlog, scan age, weekday, and carrier code. Every field must be reconstructed from data that both occurred and arrived by prediction time. The earlier pipeline lesson explained why a point-in-time replay needs event_time <= prediction_time and ingested_at <= prediction_time; Feast documents point-in-time correct historical retrieval for production feature data.[1]
Your repository artifact should contain this layout:
1eta-prediction/
2 data/
3 feature_contract.json
4 train_snapshot_manifest.json
5 training/
6 baseline.py
7 train_booster.py
8 evaluate_slices.py
9 artifacts/
10 delay_model_v1.json
11 threshold_policy_v1.json
12 metrics_v1.json
13 service/
14 api.py
15 schemas.py
16 monitoring/
17 drift_window.py
18 tests/
19 test_point_in_time_features.py
20 test_warning_gate.pyFirst fit a rule baseline such as hours_since_last_scan >= 18. Then fit the tree candidate using the same time-ordered train, validation, and test splits. XGBoost is a defensible implementation for structured features because its boosted-tree system is designed for sparse, scalable tabular learning.[2] It still must beat the baseline on the exact action policy, not a model metric alone.
Required release rows:
| Gate | Requirement |
|---|---|
| no feature leakage | replay test excludes post-prediction and late-arriving scans |
| expedited shipments | no missed warning in required validation slice |
| expected warning cost | better than rule baseline |
| feature freshness | stale scan/backlog returns fallback |
| API schema | model, feature, threshold, and freshness trace emitted |
The first portfolio receipt should prove the timestamp rule with actual events. Order O-201 has one scan that occurred and arrived before the prediction moment, one pre-prediction scan that arrived late, and one scan three hours later. The feature builder must select only the scan known at prediction time.
1from dataclasses import dataclass
2from datetime import datetime, timedelta, timezone
3from math import isfinite
4import json
5
6BASE_TIME = datetime(2026, 5, 1, 10, tzinfo=timezone.utc)
7
8def at(hours: float) -> datetime:
9 return BASE_TIME + timedelta(hours=hours)
10
11@dataclass(frozen=True)
12class ScanEvent:
13 order_id: str
14 event_time: datetime
15 ingested_at: datetime
16 status: str
17
18@dataclass(frozen=True)
19class FeatureRow:
20 order_id: str
21 prediction_at: datetime
22 scan_age_hours: float | None
23 backlog_age_hours: float | None
24 tier: str
25 feature_contract_id: str = "eta-features-v1"
26
27@dataclass(frozen=True)
28class ScoredRow:
29 features: FeatureRow
30 late_risk: float
31
32SCAN_EVENTS = [
33 ScanEvent("O-201", at(1), at(1.05), "in_transit"),
34 ScanEvent("O-201", at(1.5), at(2.5), "hub_scan"),
35 ScanEvent("O-201", at(5), at(5.05), "carrier_delay_posted"),
36]
37
38print("scan_events:", len(SCAN_EVENTS))
39print("statuses:", [event.status for event in SCAN_EVENTS])1scan_events: 3
2statuses: ['in_transit', 'hub_scan', 'carrier_delay_posted']1def scans_known_by(order_id: str, prediction_at: datetime) -> list[ScanEvent]:
2 return [
3 event for event in SCAN_EVENTS
4 if (
5 event.order_id == order_id
6 and event.event_time <= prediction_at
7 and event.ingested_at <= prediction_at
8 )
9 ]
10
11def latest_scan_known_by(order_id: str, prediction_at: datetime) -> ScanEvent:
12 admitted = scans_known_by(order_id, prediction_at)
13 if not admitted:
14 raise ValueError("no scan available at prediction time")
15 return max(admitted, key=lambda event: (event.event_time, event.ingested_at))
16
17prediction_at = at(2)
18admitted_scan_statuses = {event.status for event in scans_known_by("O-201", prediction_at)}
19selected_scan = latest_scan_known_by("O-201", prediction_at)
20print("prediction_at:", prediction_at.isoformat())
21print("admitted:", sorted(admitted_scan_statuses))
22print("selected:", selected_scan.status)1feature_row = FeatureRow(
2 order_id="O-201",
3 prediction_at=prediction_at,
4 scan_age_hours=(prediction_at - selected_scan.event_time).total_seconds() / 3600,
5 backlog_age_hours=2,
6 tier="expedited",
7)
8feature_score = ScoredRow(feature_row, late_risk=0.62)
9
10assert selected_scan.status == "in_transit"
11assert admitted_scan_statuses == {"in_transit"}
12assert feature_row.scan_age_hours == 1
13print("selected_scan:", selected_scan.status, selected_scan.event_time.isoformat(), selected_scan.ingested_at.isoformat())
14print("ignored_late_arrival:", SCAN_EVENTS[1].status, SCAN_EVENTS[1].event_time.isoformat(), SCAN_EVENTS[1].ingested_at.isoformat())
15print("ignored_future_scan:", SCAN_EVENTS[2].status, SCAN_EVENTS[2].event_time.isoformat(), SCAN_EVENTS[2].ingested_at.isoformat())1selected_scan: in_transit 2026-05-01T11:00:00+00:00 2026-05-01T11:03:00+00:00
2ignored_late_arrival: hub_scan 2026-05-01T11:30:00+00:00 2026-05-01T12:30:00+00:00
3ignored_future_scan: carrier_delay_posted 2026-05-01T15:00:00+00:00 2026-05-01T15:03:00+00:00The future scan is useful when the true outcome arrives, but it can't help a model that scores at noon. Neither can hub_scan: it occurred before noon but arrived afterward. Keeping this test near the feature builder makes both forms of leakage visible before model training begins.
A real late_risk comes from the trained model artifact. The boundary below focuses on what happens after scoring, so the harness keeps selected features separate from frozen scored output. The response always emits a versioned trace: model, observed and expected feature contracts, threshold policy, prediction time, and freshness fields. An operator can replay why a customer saw a warning or why a mismatched request fell back.
Freshness is part of the action contract. Both the latest carrier scan and the origin-backlog snapshot must be present, finite, non-negative, and recent enough. A missing contract version, invalid score, or invalid feature routes to review before the score can trigger a customer-facing message.
1@dataclass(frozen=True)
2class ReleasePolicy:
3 threshold: float
4 max_scan_age_hours: float
5 max_backlog_age_hours: float
6 model_id: str
7 feature_contract_id: str
8 threshold_policy_id: str
9
10POLICY = ReleasePolicy(
11 threshold=0.40,
12 max_scan_age_hours=24,
13 max_backlog_age_hours=8,
14 model_id="delay-model-v1",
15 feature_contract_id="eta-features-v1",
16 threshold_policy_id="eta-threshold-v1",
17)
18
19def score(
20 order_id: str,
21 scan_age_hours: float | None,
22 backlog_age_hours: float | None,
23 tier: str,
24 late_risk: float,
25 feature_contract_id: str = "eta-features-v1",
26) -> ScoredRow:
27 return ScoredRow(
28 FeatureRow(order_id, prediction_at, scan_age_hours, backlog_age_hours, tier, feature_contract_id),
29 late_risk,
30 )
31
32def response(scored: ScoredRow, action: str, reason: str) -> dict[str, object]:
33 row = scored.features
34 return {
35 "order_id": row.order_id,
36 "action": action,
37 "reason": reason,
38 "late_risk": scored.late_risk,
39 "prediction_at": row.prediction_at.isoformat(),
40 "scan_age_hours": row.scan_age_hours,
41 "backlog_age_hours": row.backlog_age_hours,
42 "model_id": POLICY.model_id,
43 "feature_contract_id": row.feature_contract_id,
44 "expected_feature_contract_id": POLICY.feature_contract_id,
45 "threshold_policy_id": POLICY.threshold_policy_id,
46 }
47
48print("policy:", POLICY.threshold_policy_id, "threshold=", POLICY.threshold)1def invalid_age(value: float | None) -> bool:
2 return value is None or not isfinite(value) or value < 0
3
4def input_issue(row: FeatureRow) -> str | None:
5 if row.feature_contract_id != POLICY.feature_contract_id:
6 return "feature_contract_mismatch"
7 if invalid_age(row.scan_age_hours):
8 return "invalid_scan_age"
9 if invalid_age(row.backlog_age_hours):
10 return "invalid_backlog_age"
11 if row.scan_age_hours > POLICY.max_scan_age_hours:
12 return "stale_scan_features"
13 if row.backlog_age_hours > POLICY.max_backlog_age_hours:
14 return "stale_backlog_features"
15 return None
16
17def route(scored: ScoredRow) -> dict[str, object]:
18 issue = input_issue(scored.features)
19 if issue is not None:
20 return response(scored, "manual_review", issue)
21 if not isfinite(scored.late_risk) or not 0 <= scored.late_risk <= 1:
22 return response(scored, "manual_review", "invalid_late_risk")
23 if scored.late_risk >= POLICY.threshold:
24 return response(scored, "warn_customer", "late_risk_threshold")
25 return response(scored, "normal_tracking", "below_threshold")
26
27print("O-201 route:", route(feature_score)["action"], route(feature_score)["reason"])1policy_cases = [
2 feature_score,
3 score("O-202", 3, 2, "standard", 0.25),
4 score("O-203", 31, 2, "standard", 0.81),
5 score("O-204", 4, 12, "standard", 0.75),
6 score("O-205", 4, 2, "standard", float("nan")),
7 score("O-206", 4, None, "standard", 0.75),
8 score("O-207", float("nan"), 2, "standard", 0.75),
9 score("O-208", -1, 2, "standard", 0.75),
10 score("O-209", 4, 2, "standard", 0.75, "eta-features-v0"),
11]
12
13for scored in policy_cases:
14 result = route(scored)
15 print(scored.features.order_id, result["action"], result["reason"])
16
17trace = route(feature_score)
18print("release_tuple:", trace.get("feature_contract_id"), trace.get("model_id"), trace.get("threshold_policy_id"))1O-201 warn_customer late_risk_threshold
2O-202 normal_tracking below_threshold
3O-203 manual_review stale_scan_features
4O-204 manual_review stale_backlog_features
5O-205 manual_review invalid_late_risk
6O-206 manual_review invalid_backlog_age
7O-207 manual_review invalid_scan_age
8O-208 manual_review invalid_scan_age
9O-209 manual_review feature_contract_mismatch
10release_tuple: eta-features-v1 delay-model-v1 eta-threshold-v1Orders O-203 and O-204 are key design results. High model scores aren't authority to message a customer when the evidence is stale. Orders O-205 through O-209 show the same rule for malformed output, missing freshness, impossible ages, and version mismatch: unreliable evidence reaches review, not a narrow ETA claim.
The release gate must test the action policy as well as the fitted score. The holdout below uses later shipments with known outcomes. A missed expedited warning costs 150 fixture units, a standard miss costs 60, and a false warning costs 8. These are local teaching values, not universal business constants.
1@dataclass(frozen=True)
2class HoldoutCase:
3 row: ScoredRow
4 delivered_late: bool
5
6holdout = [
7 HoldoutCase(score("E-301", 5, 1, "expedited", 0.78), True),
8 HoldoutCase(score("E-302", 20, 2, "standard", 0.64), True),
9 HoldoutCase(score("E-303", 4, 1, "standard", 0.12), False),
10 HoldoutCase(score("E-304", 3, 2, "standard", 0.58), False),
11]
12
13def warning_cost(case: HoldoutCase, warn: bool) -> int:
14 if warn and not case.delivered_late:
15 return 8
16 if not warn and case.delivered_late:
17 return 150 if case.row.features.tier == "expedited" else 60
18 return 0
19
20def baseline_warn(case: HoldoutCase) -> bool:
21 row = case.row.features
22 return input_issue(row) is None and row.scan_age_hours is not None and row.scan_age_hours >= 18
23
24def candidate_warn(case: HoldoutCase) -> bool:
25 return route(case.row)["action"] == "warn_customer"
26
27print("holdout rows:", len(holdout))
28print("baseline warns:", [case.row.features.order_id for case in holdout if baseline_warn(case)])1baseline_cost = sum(warning_cost(case, baseline_warn(case)) for case in holdout)
2candidate_cost = sum(warning_cost(case, candidate_warn(case)) for case in holdout)
3expedited_misses = sum(
4 case.row.features.tier == "expedited" and case.delivered_late and not candidate_warn(case)
5 for case in holdout
6)
7fallback_reasons = {row.features.order_id: route(row)["reason"] for row in policy_cases[2:]}
8print("baseline_cost:", baseline_cost, "candidate_cost:", candidate_cost, "expedited_misses:", expedited_misses)1trace_keys = {"feature_contract_id", "expected_feature_contract_id", "model_id", "threshold_policy_id"}
2freshness_trace_keys = {"prediction_at", "scan_age_hours", "backlog_age_hours"}
3
4release_gates = {
5 "replay_excludes_unavailable_scans": admitted_scan_statuses == {"in_transit"},
6 "lower_cost_than_rule_baseline": candidate_cost < baseline_cost,
7 "zero_expedited_misses": expedited_misses == 0,
8 "stale_scan_falls_back": fallback_reasons["O-203"] == "stale_scan_features",
9 "stale_backlog_falls_back": fallback_reasons["O-204"] == "stale_backlog_features",
10 "invalid_score_falls_back": fallback_reasons["O-205"] == "invalid_late_risk",
11 "missing_backlog_falls_back": fallback_reasons["O-206"] == "invalid_backlog_age",
12 "invalid_scan_age_falls_back": fallback_reasons["O-207"] == "invalid_scan_age",
13 "future_scan_age_falls_back": fallback_reasons["O-208"] == "invalid_scan_age",
14 "feature_contract_mismatch_falls_back": fallback_reasons["O-209"] == "feature_contract_mismatch",
15 "versioned_trace": all(trace.get(key) for key in trace_keys),
16 "freshness_trace": freshness_trace_keys.issubset(trace),
17}
18
19print("release_gates_pass:", all(release_gates.values()))1receipt = {
2 "bundle_id": "delivery-risk-v1",
3 "evaluation_snapshot": "eta-holdout-2026-05",
4 "previous_bundle": "delivery-risk-v0",
5 "baseline_cost": baseline_cost,
6 "candidate_cost": candidate_cost,
7 "expedited_misses": expedited_misses,
8 "release_gates": release_gates,
9 "candidate_decision": "candidate_for_shadow" if all(release_gates.values()) else "hold",
10}
11
12print(json.dumps(receipt, indent=2))1{
2 "bundle_id": "delivery-risk-v1",
3 "evaluation_snapshot": "eta-holdout-2026-05",
4 "previous_bundle": "delivery-risk-v0",
5 "baseline_cost": 150,
6 "candidate_cost": 8,
7 "expedited_misses": 0,
8 "release_gates": {
9 "replay_excludes_unavailable_scans": true,
10 "lower_cost_than_rule_baseline": true,
11 "zero_expedited_misses": true,
12 "stale_scan_falls_back": true,
13 "stale_backlog_falls_back": true,
14 "invalid_score_falls_back": true,
15 "missing_backlog_falls_back": true,
16 "invalid_scan_age_falls_back": true,
17 "future_scan_age_falls_back": true,
18 "feature_contract_mismatch_falls_back": true,
19 "versioned_trace": true,
20 "freshness_trace": true
21 },
22 "candidate_decision": "candidate_for_shadow"
23}The receipt doesn't claim broad production readiness. It proves one immutable candidate deserves shadow traffic: its feature snapshot excludes unavailable events, its policy tuple and freshness fields are visible, its cost beats the rule baseline on held-out fixtures, and its required fallback paths execute.
The deployment emits one row per score: request timestamp, feature version, model version, threshold version, feature freshness, score, action, and eventually the delivery label. Immediate monitoring catches nulls, stale scans, error rate, and score-distribution shift. Delayed monitoring computes missed-warning cost, calibration by score bucket, and slice performance.
Promotion should move a production alias from delivery-risk-v0 to separately evaluated delivery-risk-v1. Google Cloud's MLOps guidance describes this separation between validation, metadata, serving, monitoring, and continuous training stages.[3] A triggered retraining job creates evidence; it doesn't silently rewrite live behavior. Keep rollback available by retaining the prior alias target.
| Artifact | Reviewer should verify |
|---|---|
| feature contract | every field has type, timestamp boundary, and missing policy |
| training manifest | time split and dataset fingerprint exist |
| baseline comparison | candidate improves declared cost without required-slice misses |
| service API | stale inputs fail to a safer route |
| monitoring plan | input checks and delayed label metrics are distinct |
| rollback plan | prior artifact and threshold remain deployable |
Use the runnable examples as a small release harness. Change one input at a time, predict the result, then rerun the examples.
hub_scan ingestion from at(2.5) to at(1.75). Which scan should the as-of builder select?O-204 backlog age from 12 to 7. Which action replaces manual_review?float("nan") with 1.4 for O-205. Why should the service still refuse the score?0.40 to 0.80. Which expedited gate fails?threshold_policy_id from response(). Which release gate catches the incomplete trace?| Artifact | Strong submission demonstrates |
|---|---|
| model package | time-safe feature contract, baseline comparison, and calibrated warning threshold |
| service | versioned response trace and safe behavior for missing or stale scans |
| operations | input monitoring, delayed-label evaluation, candidate promotion, and rollback |
| Symptom | Cause | Fix |
|---|---|---|
| Warning appears accurate offline but misses live disruptions | future or late-arriving scans leaked into training | enforce replay tests on event and ingestion time |
| Customer receives unsupported ETA warning | service trusts score despite stale inputs | gate freshness before action |
| Missing or impossible freshness reaches threshold logic | comparisons assume valid numeric ages | reject null, non-finite, and negative ages |
| Customer receives warning from malformed score | serving boundary assumes output is finite and bounded | validate score before thresholding |
| Team can't reproduce a warning | artifact versions absent from response trace | log full release tuple |
Answer every question, then check your score. Score above 75% to mark this lesson complete.
8 questions remaining.
Questions and insights from fellow learners.