Handle late data and delayed labels

Problem

When you need this. A user event happened before a prediction, but reached the event log an hour later. A retrospective query sees it. The original service could not. If training uses that event, it learns from information that was unavailable at the decision it is meant to reproduce.

Outcome labels have a second clock. A missing return event today does not necessarily mean the user failed to return: the observation window may still be open, or the outcome collector may be late. You will add both availability rules to the synthetic user-return example from split design.

The local experiment assembles two eligible training rows at the first cutoff and five at a later cutoff. Historical feature IDs stay unchanged. There is no model training or inference in this lesson.

Concepts

Event time says when an activity happened; arrival time says when it reached the system. Prediction time is the decision being reconstructed. A historical feature may use only events inside its feature window that were available before that prediction.

An outcome horizon is the future period whose result supplies a label. A cohort is the set of prediction observations selected for one dataset assembly; each eligible observation needs its historical features and a ready label. Finalization and assembly times determine whether that label is ready to use. A missing event in an unfinished window is not a negative outcome. The feature window and label horizon are independent settings: changing one must not silently move the other.

Architecture

The assembler takes observations, events and finalized labels. It first reconstructs each observation’s available feature events, then checks whether the prediction and its outcome window are eligible at the assembly cutoff. The report retains selected feature IDs and exclusion reasons, allowing a later cutoff to add ready labels without rewriting historical inputs.

Working code

Define when each piece of information becomes usable

The feature is a count of events during the 24 hours before predict_at. An event must also have arrived strictly before that prediction. The label says whether the user returned during the following 24 hours.

For a prediction at noon on January 8, features use events from noon on January 7 up to, but excluding, noon on January 8. The outcome covers noon on January 8 up to, but excluding, noon on January 9. These adjacent windows do not share their boundary event.

Download the complete lab and enter mlops-data-availability/:

python3.12 -m venv .venv
source .venv/bin/activate
python inspect_availability.py
python -m unittest -v test_availability.py

The ZIP includes an original synthetic scenario, implementation, tests and result record. It uses the standard library. No event service, API key, package installation or training job is required.

Reconstruct features using both event and arrival time

feature_events() selects the observation’s user and the half-open event window [predict_at − 24 hours, predict_at). It then requires arrived_at < predict_at. Advancing the dataset assembly time does not change either condition. FEATURE_WINDOW and OUTCOME_WINDOW are separate settings, so changing the label horizon does not move the feature window.

"""Reconstruct features at prediction time; admit only ready outcome labels."""
from collections import Counter
from datetime import datetime, timedelta

FEATURE_WINDOW = timedelta(days=1)
OUTCOME_WINDOW = timedelta(days=1)


def instant(value):
    try:
        at = datetime.fromisoformat(value.replace('Z', '+00:00'))
    except (AttributeError, TypeError, ValueError) as error:
        raise ValueError('time must be an ISO UTC timestamp') from error
    if at.tzinfo is None or at.utcoffset().total_seconds() != 0:
        raise ValueError('time must have a UTC offset')
    return at


def check_id(value):
    if not isinstance(value, str) or not value or len(value) > 100:
        raise ValueError('IDs must be bounded nonempty strings')


def validate(observations, events, labels):
    seen = set()
    known = {}
    for observation in observations:
        if not isinstance(observation, dict) or set(observation) != {'id', 'entity', 'predict_at'}:
            raise ValueError('observation needs id, entity and predict_at')
        check_id(observation['id']); check_id(observation['entity'])
        if observation['id'] in seen:
            raise ValueError('duplicate observation ID')
        seen.add(observation['id'])
        known[observation['id']] = instant(observation['predict_at'])
    seen = set()
    for event in events:
        if not isinstance(event, dict) or set(event) != {'id', 'entity', 'event_at', 'arrived_at'}:
            raise ValueError('event needs identity and both event/arrival times')
        check_id(event['id']); check_id(event['entity'])
        if event['id'] in seen:
            raise ValueError('duplicate event ID')
        seen.add(event['id'])
        if instant(event['arrived_at']) < instant(event['event_at']):
            raise ValueError('an event cannot arrive before it happens')
    seen = set()
    for label in labels:
        if not isinstance(label, dict) or set(label) != {'id', 'value', 'ready_at'}:
            raise ValueError('label needs observation ID, value and ready_at')
        check_id(label['id'])
        if label['id'] in seen or label['id'] not in known:
            raise ValueError('duplicate or unknown label observation ID')
        seen.add(label['id'])
        if type(label['value']) is not int or label['value'] not in (0, 1):
            raise ValueError('label value must be integer 0 or 1')
        if instant(label['ready_at']) < known[label['id']] + OUTCOME_WINDOW:
            raise ValueError('this cohort requires a finalized full outcome window')


def feature_events(observation, events, require_arrival=True):
    """The event-only variant is a retrospective diagnostic, not a training input."""
    at = instant(observation['predict_at'])
    return sorted(event['id'] for event in events
                  if event['entity'] == observation['entity']
                  and at - FEATURE_WINDOW <= instant(event['event_at']) < at
                  and (not require_arrival or instant(event['arrived_at']) < at))


def assemble(observations, events, labels, as_of):
    validate(observations, events, labels)
    assembled_at = instant(as_of)
    by_id = {label['id']: label for label in labels}
    rows, excluded = [], []
    for observation in sorted(observations, key=lambda row: row['id']):
        label = by_id.get(observation['id'])
        if instant(observation['predict_at']) > assembled_at:
            reason = 'prediction_in_future'
        elif label is None:
            reason = 'label_missing'
        elif instant(label['ready_at']) > assembled_at:
            reason = 'label_pending'
        else:
            ids = feature_events(observation, events)
            rows.append(dict(observation, feature_count=len(ids), feature_event_ids=ids,
                             label=label['value'], label_ready_at=label['ready_at']))
            continue
        excluded.append({'id': observation['id'], 'reason': reason})
    return {'as_of': as_of, 'rows': rows, 'excluded': excluded,
            'exclusion_counts': dict(sorted(Counter(row['reason'] for row in excluded).items()))}

For user-01-day-08, one event arrived before prediction. Another happened at 11:00 but arrived at 13:00, after the noon prediction. The event-time-only diagnostic counts both; the historical feature counts one. It is a deliberately incorrect comparison, not an alternative training configuration.

test_late_event_is_not_added_to_historical_features checks that exact event ID. The boundary tests also include an event at the feature window’s start, one at prediction time, and one arriving exactly at prediction time. This contract excludes the last two. If a real system wants arrival equality to count, it needs an ordering rule that makes availability at that instant unambiguous.

Event timestamps alone do not capture all historical availability. Feast’s point-in-time documentation distinguishes event-time matching from filtering when a value was created or made available. The lab implements its own explicit arrival condition; it does not install Feast or claim to reproduce a particular offline store.

Wait for a finalized outcome instead of inventing a negative

The label file contains a binary value and a readiness timestamp for each finalized outcome. In this cohort, ready_at must be at or after the full 24-hour outcome window. Assembly includes a label only when its readiness time is at or before as_of.

The lab applies the full-window readiness rule to both positive and negative labels. The fixture declares those labels separately from the partial feature-event log. It assumes an outcome collector has established completeness. The program validates timing and schema, but cannot prove that the collector observed every real event. In a service, that requires a documented coverage or watermark guarantee from the outcome source.

At the first assembly time, both a pending positive and a pending negative are excluded. A label that is entirely missing is also excluded. Their absence does not imply class zero. test_pending_negative_and_positive_are_both_excluded and test_missing_label_is_not_imputed_as_zero check those cases.

An observation whose prediction lies after as_of is excluded before label eligibility is considered. A label becomes eligible at exactly ready_at == as_of; the tests distinguish that inclusive boundary from the strict event-arrival boundary.

{
  "assumptions": "Original synthetic fixture. Labels are declared finalized records, not inferred from the feature-event list. Outcome window is [predict_at, predict_at + 24 hours).",
  "assemble_at": "2026-01-11T12:00:00Z",
  "later_assemble_at": "2026-01-15T12:00:00Z",
  "observations": [
    {
      "id": "user-01-day-08",
      "entity": "user-01",
      "predict_at": "2026-01-08T12:00:00Z"
    },
    {
      "id": "user-02-day-08",
      "entity": "user-02",
      "predict_at": "2026-01-08T12:00:00Z"
    },
    {
      "id": "user-03-day-09",
      "entity": "user-03",
      "predict_at": "2026-01-09T12:00:00Z"
    },
    {
      "id": "user-04-day-10",
      "entity": "user-04",
      "predict_at": "2026-01-10T12:00:00Z"
    },
    {
      "id": "user-05-day-12",
      "entity": "user-05",
      "predict_at": "2026-01-12T12:00:00Z"
    },
    {
      "id": "user-06-day-08",
      "entity": "user-06",
      "predict_at": "2026-01-08T12:00:00Z"
    }
  ],
  "events": [
    {
      "id": "u1-known",
      "entity": "user-01",
      "event_at": "2026-01-07T14:00:00Z",
      "arrived_at": "2026-01-07T15:00:00Z"
    },
    {
      "id": "u1-late",
      "entity": "user-01",
      "event_at": "2026-01-08T11:00:00Z",
      "arrived_at": "2026-01-08T13:00:00Z"
    },
    {
      "id": "u1-cutoff",
      "entity": "user-01",
      "event_at": "2026-01-08T12:00:00Z",
      "arrived_at": "2026-01-08T12:00:00Z"
    },
    {
      "id": "u1-after",
      "entity": "user-01",
      "event_at": "2026-01-08T13:00:00Z",
      "arrived_at": "2026-01-08T13:30:00Z"
    },
    {
      "id": "u2-start",
      "entity": "user-02",
      "event_at": "2026-01-07T12:00:00Z",
      "arrived_at": "2026-01-07T12:00:00Z"
    },
    {
      "id": "u2-old",
      "entity": "user-02",
      "event_at": "2026-01-07T11:00:00Z",
      "arrived_at": "2026-01-07T11:00:00Z"
    },
    {
      "id": "u2-arrival-cutoff",
      "entity": "user-02",
      "event_at": "2026-01-08T11:00:00Z",
      "arrived_at": "2026-01-08T12:00:00Z"
    },
    {
      "id": "u3-known",
      "entity": "user-03",
      "event_at": "2026-01-09T10:00:00Z",
      "arrived_at": "2026-01-09T10:00:00Z"
    }
  ],
  "labels": [
    {
      "id": "user-01-day-08",
      "value": 1,
      "ready_at": "2026-01-09T13:00:00Z"
    },
    {
      "id": "user-02-day-08",
      "value": 0,
      "ready_at": "2026-01-10T12:00:00Z"
    },
    {
      "id": "user-03-day-09",
      "value": 0,
      "ready_at": "2026-01-12T12:00:00Z"
    },
    {
      "id": "user-04-day-10",
      "value": 1,
      "ready_at": "2026-01-12T12:00:00Z"
    },
    {
      "id": "user-05-day-12",
      "value": 0,
      "ready_at": "2026-01-14T12:00:00Z"
    }
  ]
}

Inspect how the cohort changes as labels arrive

Synthetic user-return cohort; features use arrival time, labels use readiness time
2026-01-11T12:00:00Z: eligible=2; excluded=4; reasons={"label_missing": 1, "label_pending": 2, "prediction_in_future": 1}
2026-01-15T12:00:00Z: eligible=5; excluded=1; reasons={"label_missing": 1}
user-01-day-08: known-before-prediction=1; retrospective-event-only=2
user-02-day-08: known-before-prediction=1; retrospective-event-only=2
Original feature IDs unchanged when more labels become ready: True
Pending or missing labels are excluded, never converted to negative outcomes.

The later assembly admits three additional rows. It still excludes the row with no label. The two original observations keep the same feature-event IDs even though a retrospective query can now see late arrivals.

The result includes every exclusion reason, rather than only the eligible-row count. Inspect those reasons before interpreting the dataset. If positive and negative outcomes become available at different rates, an early eligible cohort can have a different class mix from the eventual population. This experiment illustrates exclusion behavior, not an estimator that corrects delayed-outcome bias.

Save the availability contract beside the split contract

python inspect_availability.py --check
python inspect_availability.py --save my-cohort.json

The result record binds the input checksum to both assembly cutoffs, feature IDs, labels and excluded IDs. Use it alongside the previous lesson’s split membership. A chronological split and an as-known feature query answer different questions; one cannot replace the other.

"""Run an availability experiment on an authored synthetic user-return scenario."""
import argparse
import hashlib
import json
from pathlib import Path
from availability import assemble, feature_events

ROOT = Path(__file__).resolve().parent


def record():
    raw = (ROOT / 'scenario.json').read_bytes()
    scenario = json.loads(raw)
    observations, events, labels = (scenario[k] for k in ('observations', 'events', 'labels'))
    current = assemble(observations, events, labels, scenario['assemble_at'])
    later = assemble(observations, events, labels, scenario['later_assemble_at'])
    diagnostics = [{'id': row['id'], 'as_known': feature_events(row, events),
                    'event_time_only': feature_events(row, events, require_arrival=False)}
                   for row in observations if row['id'] in {r['id'] for r in current['rows']}]
    before = {row['id']: row['feature_event_ids'] for row in current['rows']}
    after = {row['id']: row['feature_event_ids'] for row in later['rows']}
    return {'mode': 'synthetic data-availability inspection; no training or predictions',
            'input_sha256': hashlib.sha256(raw).hexdigest(), 'current': current, 'later': later,
            'diagnostics': diagnostics,
            'original_feature_ids_unchanged': all(ids == after[id] for id, ids in before.items())}


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--save', type=Path)
    parser.add_argument('--check', action='store_true')
    args = parser.parse_args()
    result = record()
    if args.check and result != json.loads((ROOT / 'result.json').read_text()):
        raise SystemExit('availability record differs from the bundled result')
    if args.save:
        args.save.write_text(json.dumps(result, indent=2) + '\n')
    print('Synthetic user-return cohort; features use arrival time, labels use readiness time')
    for name in ('current', 'later'):
        cohort = result[name]
        print(f"{cohort['as_of']}: eligible={len(cohort['rows'])}; "
              f"excluded={len(cohort['excluded'])}; reasons={json.dumps(cohort['exclusion_counts'], sort_keys=True)}")
    for row in result['diagnostics']:
        print(f"{row['id']}: known-before-prediction={len(row['as_known'])}; "
              f"retrospective-event-only={len(row['event_time_only'])}")
    print('Original feature IDs unchanged when more labels become ready:', result['original_feature_ids_unchanged'])
    print('Pending or missing labels are excluded, never converted to negative outcomes.')


if __name__ == '__main__':
    main()

Duplicate IDs, unknown label references, invalid binary values, naive timestamps and arrivals before an event happened are refused. A label finalized before its full outcome window is also refused under this cohort’s rule. These checks assume the recorded event and arrival times remain intact. Overwriting them would defeat historical reconstruction. The lab also keeps one label per observation; corrected labels need a version history and an additional as-known selection rule.

Design exercise: increase the outcome window from one day to seven. Identify which label readiness constraints must change, which historical features should remain unchanged, and why the one-day split gaps from the previous lesson no longer establish a sufficient temporal boundary by themselves.

Failure cases

The late-event case is the central leakage test: later arrival must not change what an earlier prediction saw. Pending and missing labels are separate failures of readiness; neither may be filled with zero to create extra training rows.

Deployment

Keep event, arrival, prediction and outcome-readiness fields when collecting a real training dataset. Agree with the outcome producer on what finalization means and how late corrections are represented. This assembler consumes trusted fixture times; production timestamp capture and correction policy must be implemented at their actual data boundaries.

Scaling

Historical reconstruction needs access to the relevant event and arrival ranges. Indexing and incremental materialization can reduce repeated scans, provided the same eligibility rule is preserved. Monitor label readiness by cohort so delayed labels do not silently change the training population.

Key takeaways

• An event that happened earlier may still have been unavailable to the prediction.
• Unfinished outcome windows cannot supply negative labels.
• Later assembly can change label eligibility without changing historical feature membership.

Best practices

• Retain availability timestamps instead of keeping only event time.
• Test equality at feature and arrival boundaries.
• Report excluded and pending observations beside eligible rows.

References

• Feast, point-in-time joins: event-time retrieval and availability-time filtering are distinct.
• Google, Rules of ML: preserve information available at serving time when constructing training examples.

Sources checked October 24, 2026. The displayed rows come from local execution of the authored synthetic scenario.