Personalize this lesson
Adapt explanations and teaching visuals to your background and preferred voice.
Two hours after pickup, order O-201 is still in transit. The product question isn't "what minute will it arrive?" It's whether the service may show a delay warning, stay on normal tracking, or send the case to review because the evidence is stale.
Design an Automated Support Agent ended on a controlled action: reply, refuse, or hand off with evidence an operator can replay. Keep that habit for a conventional prediction service.
The service doesn't issue refunds or change carrier routing. It returns a risk score at a pinned prediction time with one of three actions: normal_tracking, warn_customer, or manual_review.
Pin the decision before you fit a model
Use one decision moment: two hours after first carrier pickup. Use one label: delivery after the promised end-of-day. A stored late_risk without those two pins can't be replayed, compared, or gated 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 the threshold passes and the evidence is fresh |
| unavailable data action | route to manual_review, no narrow ETA claim |
This isn't a minute-accurate arrival clock. You score late-delivery risk at pickup plus two hours, then let a versioned policy gate decide whether a customer-facing warning is allowed.
The feature bundle includes route distance, service tier, origin backlog, scan age, weekday, and carrier code. Every field has to be rebuilt from data that both occurred and arrived by prediction time. Batch and Streaming Feature Pipelines already required event_time <= prediction_time and ingested_at <= prediction_time. Feast's historical retrieval does the same kind of point-in-time join: it scans backward from each entity timestamp up to the feature view's TTL, then joins the latest eligible row.[1]

A high score never overrides a stale scan. The gate checks evidence first, then the threshold.
The portfolio artifact can follow this layout. The cells in this lesson prove policy, freshness, and the release receipt with frozen scores. Fitting train_booster.py under a time split stays in the training lab; don't claim you shipped a trained booster from this harness alone.
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.pyProve the snapshot uses past events only
Pickup is 2026-05-01 10:00 UTC, so the pinned score time is noon. Order O-201 has three carrier scans. Work the two clocks by hand before writing a filter.
| Scan | Event time | Ingested at | Known at noon? |
|---|---|---|---|
in_transit | 11:00 | 11:03 | yes: both clocks are before 12:00 |
hub_scan | 11:30 | 12:30 | no: the event happened, but it arrived late |
carrier_delay_posted | 15:00 | 15:03 | no: the event is still in the future |

The feature builder keeps the latest scan that passes both clocks. Scan age is then for that admitted row: noon minus 11:00 is one hour.
The first receipt should fail if either leakage form sneaks in: a future event, or a pre-noon event that arrived after scoring.
1from dataclasses import dataclass
2from datetime import datetime, timedelta, timezone
3from math import isfinite
4
5BASE_TIME = datetime(2026, 5, 1, 10, tzinfo=timezone.utc)
6
7def at(hours: float) -> datetime:
8 return BASE_TIME + timedelta(hours=hours)
9
10@dataclass(frozen=True)
11class ScanEvent:
12 order_id: str
13 event_time: datetime
14 ingested_at: datetime
15 status: str
16
17@dataclass(frozen=True)
18class FeatureRow:
19 order_id: str
20 prediction_at: datetime
21 scan_age_hours: float | None
22 backlog_age_hours: float | None
23 tier: str
24 feature_contract_id: str = "eta-features-v1"
25
26@dataclass(frozen=True)
27class ScoredRow:
28 features: FeatureRow
29 late_risk: float
30
31SCAN_EVENTS = [
32 ScanEvent("O-201", at(1), at(1.05), "in_transit"),
33 ScanEvent("O-201", at(1.5), at(2.5), "hub_scan"),
34 ScanEvent("O-201", at(5), at(5.05), "carrier_delay_posted"),
35]
36
37print("scan_events:", len(SCAN_EVENTS))
38print("statuses:", [event.status for event in SCAN_EVENTS])1scan_events: 3
2statuses: ['in_transit', 'hub_scan', 'carrier_delay_posted']The filter is two inequalities, not "the latest row in the table." event_time asks whether the world had already changed. ingested_at asks whether this service had already learned it.
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)1prediction_at: 2026-05-01T12:00:00+00:00
2admitted: ['in_transit']
3selected: in_transit1feature_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 15:00 scan is useful when the true outcome arrives. It can't help a model that scores at noon. Neither can hub_scan: it occurred before noon but landed in the log afterward. Keep this test next to the feature builder so both leakage forms show up before anyone fits a tree.
Route scores through a versioned policy
A production late_risk comes from a trained artifact fit under a time split. Gradient Boosted Trees in Production is the right place to fit that candidate: boosted trees are a defensible choice for this kind of table, and XGBoost was designed as a scalable system for sparse tabular learning.[2] It still has to beat a rule baseline on the action policy, not on a model metric alone.
This harness keeps selected features separate from frozen scored output so you can grade policy and receipts without waiting on a booster fit. 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. The latest carrier scan and the origin-backlog snapshot both have to 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)Check evidence first, then the score. O-201 is fresh, late_risk is 0.62, and the threshold is 0.40, so it should warn. A stale or malformed row shouldn't.
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 the design result you want a reviewer to see. A high model score isn't authority to message a customer when the evidence is stale. O-205 through O-209 apply the same rule to a malformed score, a missing freshness field, an impossible age, and a contract mismatch: unreliable evidence reaches review, not a narrow ETA claim.
Publish evidence against the rule baseline
The release gate has to test the action policy, not only the fitted score. First fit a rule such as hours_since_last_scan >= 18. Then compare the candidate on the same time-ordered train, validation, and test splits. 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. Those are local teaching values, not universal business constants. Predict the ledger before you run it: the 18-hour rule never sees expedited late order E-301 (scan age 5), so it should pay 150. The candidate warns on that row, and it also warns on on-time E-304 (late_risk 0.58), so it should pay 8.

Required release rows:
| Gate | Requirement |
|---|---|
| no feature leakage | replay test excludes post-prediction and late-arriving scans |
| expedited shipments | no missed warning in the required validation slice |
| expected warning cost | better than the rule baseline |
| feature freshness | stale scan or backlog returns fallback |
| API schema | model, feature, threshold, and freshness trace emitted |
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
4# Multi-row point-in-time audit: every admitted train feature must land before prediction_at.
5train_feature_events = [
6 {"order_id": "O-201", "ingested_at": at(1.05), "prediction_at": at(2), "split": "train"},
7 {"order_id": "O-210", "ingested_at": at(0.5), "prediction_at": at(2), "split": "train"},
8 {"order_id": "O-211", "ingested_at": at(3), "prediction_at": at(4), "split": "holdout"},
9]
10train_max_event = max(
11 event["ingested_at"] for event in train_feature_events if event["split"] == "train"
12)
13holdout_prediction_times = [
14 event["prediction_at"] for event in train_feature_events if event["split"] == "holdout"
15]
16pit_ok = all(
17 event["ingested_at"] <= event["prediction_at"] for event in train_feature_events
18) and all(train_max_event < prediction_at for prediction_at in holdout_prediction_times)
19
20release_gates = {
21 "replay_excludes_unavailable_scans": admitted_scan_statuses == {"in_transit"},
22 "multi_row_point_in_time_snapshot": pit_ok,
23 "lower_cost_than_rule_baseline": candidate_cost < baseline_cost,
24 "zero_expedited_misses": expedited_misses == 0,
25 "stale_scan_falls_back": fallback_reasons["O-203"] == "stale_scan_features",
26 "stale_backlog_falls_back": fallback_reasons["O-204"] == "stale_backlog_features",
27 "invalid_score_falls_back": fallback_reasons["O-205"] == "invalid_late_risk",
28 "missing_backlog_falls_back": fallback_reasons["O-206"] == "invalid_backlog_age",
29 "invalid_scan_age_falls_back": fallback_reasons["O-207"] == "invalid_scan_age",
30 "negative_scan_age_falls_back": fallback_reasons["O-208"] == "invalid_scan_age",
31 "feature_contract_mismatch_falls_back": fallback_reasons["O-209"] == "feature_contract_mismatch",
32 "versioned_trace": all(trace.get(key) for key in trace_keys),
33 "freshness_trace": freshness_trace_keys.issubset(trace),
34}
35
36print("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("bundle:", receipt["bundle_id"], "rollback:", receipt["previous_bundle"])
13print("cost:", {"baseline": receipt["baseline_cost"], "candidate": receipt["candidate_cost"]})
14print("release_gates_pass:", all(receipt["release_gates"].values()))
15print("candidate_decision:", receipt["candidate_decision"])1bundle: delivery-risk-v1 rollback: delivery-risk-v0
2cost: {'baseline': 150, 'candidate': 8}
3release_gates_pass: True
4candidate_decision: candidate_for_shadowThat receipt isn't a production blessing. It says this frozen candidate is worth shadow traffic: no leaked events, visible versions, cheaper than the 18-hour rule on the holdout fixtures, and every stale or invalid path falls back.
The 0.78 / 0.64 / 0.12 / 0.58 scores stay frozen so the policy math doesn't move under you. You still need a trainer that can produce scores from a time-split table. Fit a small logistic model on the same three fields, then lock the probabilities. Don't retune after you open the holdout.
1import numpy as np
2from sklearn.linear_model import LogisticRegression
3
4train_X = np.array(
5 [
6 [4.0, 1.0, 0.0],
7 [20.0, 2.0, 0.0],
8 [5.0, 1.0, 1.0],
9 [3.0, 2.0, 0.0],
10 [22.0, 1.0, 1.0],
11 [2.0, 1.0, 0.0],
12 [18.0, 3.0, 0.0],
13 [6.0, 1.0, 1.0],
14 ]
15)
16train_y = np.array([0, 1, 1, 0, 1, 0, 1, 1])
17model = LogisticRegression(random_state=0, max_iter=200).fit(train_X, train_y)
18holdout_X = np.array(
19 [
20 [5.0, 1.0, 1.0],
21 [20.0, 2.0, 0.0],
22 [4.0, 1.0, 0.0],
23 [3.0, 2.0, 0.0],
24 ]
25)
26frozen = model.predict_proba(holdout_X)[:, 1]
27print("frozen holdout scores:", [round(float(score), 3) for score in frozen])1frozen holdout scores: [0.659, 1.0, 0.311, 0.134]This cell only shows you can fit scores. The earlier receipt still owns 0.78 / 0.64 / 0.12 / 0.58, so a solver upgrade can't rewrite the cost ledger. In the portfolio repo, write fitted probabilities into delay_model_v1.json and keep train_booster.py as the path that created them.
Operate the service after release
Each score should emit one row immediately: request timestamp, feature version, model version, threshold version, feature freshness, score, action, and later the delivery label. Immediate monitoring catches nulls, stale scans, error rate, and score-distribution shift. Delayed monitoring waits for labels, then computes missed-warning cost, calibration by score bucket, and slice performance. Monitoring Predictive Models separates those two clocks; this capstone should keep them separate in the portfolio artifact too.
Promotion should move a production alias from delivery-risk-v0 to a separately evaluated delivery-risk-v1. Google Cloud's MLOps guidance treats model validation as a comparison against the current or baseline model before promotion, and treats retraining as a pipeline that produces a new candidate rather than a silent overwrite of live serving.[3] A triggered retraining job creates evidence. It doesn't rewrite the alias. Keep rollback available by retaining the prior alias target.
Submission checklist
| 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 |
Practice: break the release contract
Use the runnable examples as a small release harness. Change one input at a time, predict the result, then rerun.
- Move
hub_scaningestion fromat(2.5)toat(1.75). Which scan should the as-of builder select? - Change
O-204backlog age from12to7. Which action replacesmanual_review? - Replace
float("nan")with1.4forO-205. Why should the service still refuse the score? - Raise threshold from
0.40to0.80. Which expedited gate fails? - Remove
threshold_policy_idfromresponse(). Which release gate catches the incomplete trace?
Practice answer sketches
What changes when hub_scan ingestion moves from at(2.5) to at(1.75)?
Answer
hub_scan now occurred and arrived before noon, so it becomes the selected scan.
What changes when O-204 backlog age falls from 12 to 7?
Answer
O-204 becomes warn_customer because its 0.75 score exceeds threshold.
Why does score 1.4 still route O-205 to review?
Answer
1.4 is outside probability range, so route remains manual_review with invalid_late_risk.
Which gate fails when threshold moves from 0.40 to 0.80?
Answer
Candidate misses expedited late order E-301, so zero_expedited_misses fails.
Which gate fails when response trace drops threshold_policy_id?
Answer
versioned_trace fails because an operator can't replay the score-to-action policy.