Detecting As-Run vs Scheduled Discrepancies

This guide solves one exact task: given a scheduled log and an as-run log for a single channel and broadcast day, deterministically match each aired event back to the spot it was meant to be and classify the outcome — aired-as-scheduled, shifted, short, long, preempted, or missing — using explicit start-time and duration tolerances rather than exact equality. It is the detection procedure that sits under As-Run Reconciliation and Discrepancy Handling, itself a subsystem of Broadcast Traffic Architecture & Taxonomy. Getting the tolerances right is not cosmetic: set them too tight and every normal airing floods the traffic desk as a false discrepancy; set them too loose and a spot that genuinely ran short slips through and gets billed at full rate, which is exactly the kind of overstatement an FCC public-file or SOC 2 revenue audit is designed to catch.

The core requirement is that classification be reproducible. The same two logs, re-run tomorrow or replayed from cold storage during a billing dispute, must yield the identical set of discrepancies — so the matcher holds no hidden state and every decision is a pure function of the inputs and a declared tolerance policy. Those classifications become the credits and make-good candidates assembled in generating discrepancy reports for billing, so a mistake here propagates straight to the invoice.

Prerequisites #

• Python 3.11+ — required for the X | None union syntax, match statements, and timezone-aware datetime.now(timezone.utc) used below.
• Pydantic v2 — pin exactly (pydantic==2.7.1); the classifier relies on v2 field validators to reject naive datetimes at the boundary.
• Both logs normalized upstream — the scheduled and as-run streams must already carry canonical house_number, isci, and UTC timestamps per the canonical traffic field data dictionary, so this pass measures deviation, not field structure.
• A declared tolerance policy — start-time and duration tolerances held in config, owned by traffic managers, versioned so a change that re-classifies historical airings is auditable rather than silent.
• Read-only access to both logs — a mounted export or a SELECT-scoped DB role; detection must never mutate the schedule or the as-run telemetry it reads.
• Preemption context attached — as-run events displaced by an EAS emergency interrupt or a network break should carry a preemption_reason so the classifier can separate a deliberate preemption from a genuine miss.

Step-by-Step Implementation #

Detection runs as a pure pipeline over each scheduled spot: derive a match key, find the nearest unconsumed as-run event, measure start-time drift and duration delta, then apply the tolerance policy to land on exactly one DiscrepancyType. The state machine below is the whole decision surface.

Figure — The classification decision tree: match, then verify, then test duration delta against tolerance, then test start drift against tolerance, landing on exactly one discrepancy type per scheduled spot.

Step 1 — Model the tolerance policy and the inputs #

Goal: fix the typed records and the declared tolerance policy so classification is driven by configuration, not constants. The policy is versioned because changing it re-classifies historical airings, and that must be an auditable event.

from __future__ import annotations

import logging
from datetime import datetime, timezone
from enum import Enum

from pydantic import BaseModel, Field, field_validator

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("traffic.reconcile.detect")


class DiscrepancyType(str, Enum):
    AIRED_AS_SCHEDULED = "aired_as_scheduled"
    SHIFTED = "shifted"
    SHORT = "short"
    LONG = "long"
    PREEMPTED = "preempted"
    MISSING = "missing"


class TolerancePolicy(BaseModel):
    policy_version: str = "2026.07"          # bump when tolerances change
    start_tolerance_ms: int = 2_000          # drift within this is not a shift
    duration_tolerance_pct: float = Field(default=0.05, gt=0, lt=1)

    def duration_tolerance_ms(self, booked_ms: int) -> int:
        # A fraction of booked length: a :30 tolerates more absolute drift
        # than a :10, matching how encoders round longer creatives.
        return int(booked_ms * self.duration_tolerance_pct)


class ScheduledSpot(BaseModel):
    spot_id: str
    house_number: str
    scheduled_at: datetime
    expected_duration_ms: int = Field(gt=0)
    preemptible: bool = True

    @field_validator("scheduled_at")
    @classmethod
    def _utc(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("scheduled_at must be timezone-aware (UTC)")
        return v.astimezone(timezone.utc)


class AsRunEntry(BaseModel):
    event_id: str
    house_number: str
    aired_at: datetime
    actual_duration_ms: int = Field(ge=0)
    verified: bool = True
    preemption_reason: str | None = None

    @field_validator("aired_at")
    @classmethod
    def _utc(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("aired_at must be timezone-aware (UTC)")
        return v.astimezone(timezone.utc)

Expected result: importing the module logs nothing yet, but constructing an AsRunEntry with a naive aired_at raises ValueError: aired_at must be timezone-aware (UTC) — the fail-closed guard that keeps DST math off naive datetimes.

Step 2 — Match a scheduled spot to its as-run event #

Goal: derive the deterministic key and select the nearest unconsumed as-run event. Matching keys on house_number and broadcast day, then breaks ties by proximity, so a uniformly shifted pod still matches spot-for-spot.

def broadcast_key(house_number: str, when: datetime) -> str:
    # House number scoped to the UTC broadcast date the traffic system assigned.
    return f"{house_number}:{when.date().isoformat()}"


def find_match(
    spot: ScheduledSpot, pool: list[AsRunEntry]
) -> AsRunEntry | None:
    key = broadcast_key(spot.house_number, spot.scheduled_at)
    candidates = [e for e in pool
                  if broadcast_key(e.house_number, e.aired_at) == key]
    if not candidates:
        return None
    # Nearest airing to the scheduled time wins; identical distances -> earliest.
    return min(candidates, key=lambda e: (
        abs((e.aired_at - spot.scheduled_at).total_seconds()), e.aired_at))

Expected result: for a spot scheduled at 18:37:00Z with two same-house airings at 18:37:01Z and 18:52:10Z, find_match returns the 18:37:01Z event — the four-second-drift airing, not the later reairing.

Step 3 — Measure drift and delta, then classify against tolerance #

Goal: compute signed start drift and duration delta and land on exactly one DiscrepancyType. Duration is tested before start time, so a spot that is both late and truncated bills as SHORT, not merely SHIFTED.

def classify(
    spot: ScheduledSpot, entry: AsRunEntry, policy: TolerancePolicy
) -> DiscrepancyType:
    start_drift_ms = int(
        (entry.aired_at - spot.scheduled_at).total_seconds() * 1000)
    duration_delta_ms = entry.actual_duration_ms - spot.expected_duration_ms
    dur_tol = policy.duration_tolerance_ms(spot.expected_duration_ms)

    if entry.actual_duration_ms == 0 or not entry.verified:
        result = DiscrepancyType.MISSING          # logged frames absent
    elif duration_delta_ms < -dur_tol:
        result = DiscrepancyType.SHORT            # pro-rata credit downstream
    elif duration_delta_ms > dur_tol:
        result = DiscrepancyType.LONG             # bill at booked length
    elif abs(start_drift_ms) > policy.start_tolerance_ms:
        result = DiscrepancyType.SHIFTED
    else:
        result = DiscrepancyType.AIRED_AS_SCHEDULED

    logger.info("%s | %s | drift=%dms delta=%dms",
                spot.spot_id, result.value, start_drift_ms, duration_delta_ms)
    return result

Expected log line: 2026-07-17T05:12:44+00:00 | INFO | traffic.reconcile.detect | SP-2026-0716-0442 | short | drift=1400ms delta=-4000ms — a spot that aired 1.4s late and 4s under its booked thirty, classified SHORT.

Step 4 — Classify the unmatched spot as preempted or missing #

Goal: distinguish a spot displaced by a higher-priority event from one that simply never aired. A preemption carries different contractual weight even though both route to make-good.

def classify_unmatched(
    spot: ScheduledSpot, as_run: list[AsRunEntry]
) -> DiscrepancyType:
    # PREEMPTED only when a sibling event that day carried a preemption reason,
    # evidence the pod was displaced rather than dropped by a trafficking error.
    day = spot.scheduled_at.date()
    displaced = any(e.preemption_reason and e.aired_at.date() == day
                    for e in as_run)
    result = (DiscrepancyType.PREEMPTED
              if spot.preemptible and displaced
              else DiscrepancyType.MISSING)
    logger.warning("%s | %s | no as-run match | make-good candidate",
                   spot.spot_id, result.value)
    return result

Expected log line: 2026-07-17T05:12:45+00:00 | WARNING | traffic.reconcile.detect | SP-2026-0716-0451 | preempted | no as-run match | make-good candidate when an EAS event sits in the same day, otherwise missing.

Verification & Testing #

Correct behaviour rests on two properties: tolerance sensitivity (a drift on the boundary lands on the right side of it) and determinism (the same inputs always produce the same class). Both are assertable against fixture data drawn from a real broadcast day.

from datetime import timedelta

POLICY = TolerancePolicy(start_tolerance_ms=2_000, duration_tolerance_pct=0.05)
BASE = datetime(2026, 7, 16, 18, 37, tzinfo=timezone.utc)

spot = ScheduledSpot(spot_id="SP-1", house_number="H7-QSR-030",
                     scheduled_at=BASE, expected_duration_ms=30_000)

# 1. Within both tolerances -> AIRED_AS_SCHEDULED (1.5s drift, +500ms delta).
ok = AsRunEntry(event_id="AR-1", house_number="H7-QSR-030",
                aired_at=BASE + timedelta(milliseconds=1_500),
                actual_duration_ms=30_500)
assert classify(spot, ok, POLICY) is DiscrepancyType.AIRED_AS_SCHEDULED

# 2. Duration short beyond the 5% (1500ms) band -> SHORT, not SHIFTED.
short = AsRunEntry(event_id="AR-2", house_number="H7-QSR-030",
                   aired_at=BASE + timedelta(seconds=9),
                   actual_duration_ms=26_000)
assert classify(spot, short, POLICY) is DiscrepancyType.SHORT

# 3. Start drift past 2s but full duration -> SHIFTED.
late = AsRunEntry(event_id="AR-3", house_number="H7-QSR-030",
                  aired_at=BASE + timedelta(seconds=6),
                  actual_duration_ms=30_000)
assert classify(spot, late, POLICY) is DiscrepancyType.SHIFTED

# 4. No airing, EAS sibling present -> PREEMPTED.
eas = AsRunEntry(event_id="AR-EAS", house_number="EAS-CH7",
                 aired_at=BASE, actual_duration_ms=45_000,
                 preemption_reason="eas_activation")
assert classify_unmatched(spot, [eas]) is DiscrepancyType.PREEMPTED

Because classify is pure, its output for a fixed policy is stable across machines and runs — replaying an archived day during an audit reproduces the identical discrepancy set. Run the suite in CI against a golden fixture of scheduled/as-run pairs so a change to a tolerance that would silently re-classify last quarter’s airings fails the build instead of shipping.

Edge Cases & Failure Handling #

• Systematic clock skew. If one source system’s clock is a few seconds off UTC, every spot in the day drifts by the same amount and a whole daypart classifies as SHIFTED. Detect it by watching for a near-constant start_drift_ms across unrelated spots; the fix is to re-sync NTP and re-run, never to widen start_tolerance_ms, which would also swallow genuine shifts. The broadcast_key join still holds under skew, so no spot is lost — only mis-labelled — which is why re-running after the correction is safe.
• Duplicate house numbers in one day. A creative booked several times per day produces several as-run events under one key. Consume each matched event exactly once (remove it from the pool after a match) so a second scheduled airing claims the second airing rather than double-matching the first. If the as-run log is genuinely missing one airing, the surplus scheduled spot correctly falls through to MISSING.
• Verified-but-zero airing. Playout can log an event with verified=True yet actual_duration_ms == 0 when it cued a spot that never rolled frames. The classifier treats zero duration as MISSING regardless of the verified flag, because a spot that emitted no video did not air — billing it as SHORT would invoice nothing-for-something. Route these to the same make-good queue used for make-good routing for preemptions.

FAQ #

Related #

• As-Run Reconciliation and Discrepancy Handling — the parent guide that defines the entities, matching key, and audit guarantees this detection pass implements.
• Generating Discrepancy Reports for Billing — how the classifications produced here aggregate into credits, make-good candidates, and audit rows.
• Automating Make-Good Routing for Preemptions — the routing engine that a MISSING or PREEMPTED classification hands an unfulfilled obligation.