Broadcast Traffic Scheduling Automation

Best Practices for Avails Inventory Tracking

This guide solves one specific operational task: building an automated pipeline that reconciles a linear station’s planned avails inventory against its as-run broadcast log, then routes every discrepancy — preemptions, program overruns, late makegoods, and billing-code drift — to the correct resolution queue in near real time. It sits under Avails Mapping Strategies for Linear TV, part of the broader Broadcast Traffic Architecture & Taxonomy system design. Without disciplined tracking, avails drift silently between the traffic system and playout, and the gap surfaces only at month-end billing — as unattributed revenue, disputed affiliate credits, and audit findings you can no longer reconstruct. The workflow below is written for broadcast traffic managers who need to understand the control flow, and for the Python developers who have to ship the reconciliation job that enforces it.

The end state is a deterministic, idempotent reconciliation loop: it pulls planned inventory through a read-only connection, diffs each avail against the as-run log, classifies conflicts, and writes an immutable audit trail that a financial reviewer can replay. Every avail is treated as a discrete, time-bound inventory unit keyed to a program boundary, daypart, and clearance tier — the same canonical spot schema the rest of the traffic stack shares.

Prerequisites

Before running the reconciliation job, confirm the following are in place:

Step-by-Step Implementation

Step 1 — Normalize timecodes to UTC before anything else

Goal: eliminate the single biggest source of false conflicts — timecode drift. Playout logs are often stamped in local broadcast time while the traffic system stores UTC. Normalize both sides to UTC and clip to program boundaries before any diff runs, so a fractional-second overlap never trips a spurious alert.

python 
import logging
from datetime import datetime, timezone

# Traffic-ops log pattern: timestamp | level | module | spot_id | message
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("avails.normalize")


def to_utc(ts: datetime, tolerance_ms: int = 500) -> datetime:
    """Coerce a playout/as-run timestamp to UTC.

    Naive timestamps are assumed to already be broadcast-local and are
    treated as UTC only after the playout server's offset is applied
    upstream; here we reject naive values to force explicit tz handling.
    """
    if ts.tzinfo is None:
        raise ValueError("Refusing naive timestamp; playout offset must be resolved first.")
    normalized = ts.astimezone(timezone.utc)
    # Sub-tolerance drift is rounded away so it cannot register as an overlap.
    micros = (normalized.microsecond // (tolerance_ms * 1000)) * (tolerance_ms * 1000)
    return normalized.replace(microsecond=micros)

Expected log line when a naive value slips through:

text 
2026-07-03 14:02:11,204 | ERROR | avails.normalize | ValueError: Refusing naive timestamp; playout offset must be resolved first.

Step 2 — Pull planned inventory through a read-only, pooled connection

Goal: load the planned avails for a daypart without ever holding a write-capable connection. The manager validates the schema version on startup, enforces a hard statement timeout, and emits a structured audit record for every inventory pull.

python 
import json
from typing import List
from dataclasses import dataclass
from psycopg2 import pool, sql

audit = logging.getLogger("avails.db")


@dataclass
class AvailRecord:
    avail_id: str
    start_utc: datetime
    end_utc: datetime
    spot_length: int      # seconds
    clearance_code: str   # network clearance tier
    billing_code: str     # canonical NAT/LOC/AFF/MKG code
    status: str           # SCHEDULED | AIRED | PREEMPTED


class SecureTrafficDB:
    """Read-only access with connection pooling and strict timeouts."""

    def __init__(self, dsn: str, max_connections: int = 5, query_timeout_ms: int = 3000) -> None:
        self.pool = pool.SimpleConnectionPool(
            minconn=1,
            maxconn=max_connections,
            dsn=dsn,
            options=f"-c statement_timeout={query_timeout_ms} -c default_transaction_read_only=on",
        )
        self._validate_schema_version()

    def _validate_schema_version(self) -> None:
        conn = self.pool.getconn()
        try:
            with conn.cursor() as cur:
                cur.execute("SELECT version FROM traffic_schema_registry WHERE active = TRUE;")
                row = cur.fetchone()
                if not row or row[0] < 12:
                    raise RuntimeError("Traffic schema version mismatch. Aborting reconciliation.")
        finally:
            self.pool.putconn(conn)

    def fetch_planned_inventory(self, daypart: str, date: str) -> List[AvailRecord]:
        conn = self.pool.getconn()
        try:
            with conn.cursor() as cur:
                cur.execute(
                    sql.SQL(
                        """
                        SELECT avail_id, start_utc, end_utc, spot_length,
                               clearance_code, billing_code, status
                        FROM planned_inventory
                        WHERE daypart = %s AND broadcast_date = %s
                        ORDER BY start_utc ASC;
                        """
                    ),
                    (daypart, date),
                )
                records = [AvailRecord(*row) for row in cur.fetchall()]
        finally:
            self.pool.putconn(conn)
        audit.info(json.dumps({"event": "inventory_pull", "daypart": daypart, "count": len(records)}))
        return records

Expected log line:

text 
2026-07-03 14:02:12,880 | INFO | avails.db | {"event": "inventory_pull", "daypart": "PRIME", "count": 148}

Step 3 — Diff each avail and classify the conflict

Goal: compare planned inventory against the as-run log, then bucket every mismatch into a named queue — timecode_drift, billing_mismatch, or makegood_pending — so downstream automation can act without re-inspecting raw records. The HTTP session wraps the as-run API with retry and exponential backoff; a hard failure raises rather than silently dropping an avail. This classifier is the counterpart to standalone conflict detection in avails; here the goal is reconciliation and revenue attribution rather than pre-air validation.

python 
import requests
from typing import Dict
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

recon_log = logging.getLogger("avails.reconcile")
VALID_PREFIXES = {"NAT", "LOC", "AFF", "MKG"}


class ReconciliationEngine:
    def __init__(self, db: SecureTrafficDB, api_endpoint: str) -> None:
        self.db = db
        self.api = api_endpoint
        self.session = self._configure_retry_session()

    def _configure_retry_session(self) -> requests.Session:
        session = requests.Session()
        retry = Retry(
            total=3,
            backoff_factor=1.0,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["POST"],
        )
        session.mount("https://", HTTPAdapter(max_retries=retry))
        return session

    def reconcile_daypart(self, daypart: str, date: str) -> Dict[str, List[str]]:
        planned = self.db.fetch_planned_inventory(daypart, date)
        conflicts: Dict[str, List[str]] = {
            "timecode_drift": [], "billing_mismatch": [], "makegood_pending": [],
        }
        for avail in planned:
            try:
                if not self._matches_as_run_log(avail):
                    conflicts["timecode_drift"].append(avail.avail_id)
                    continue
                if avail.billing_code[:3] not in VALID_PREFIXES:
                    conflicts["billing_mismatch"].append(avail.avail_id)
                    continue
                if avail.status == "PREEMPTED":
                    conflicts["makegood_pending"].append(avail.avail_id)
            except Exception as exc:
                recon_log.error(json.dumps(
                    {"event": "reconciliation_failure", "spot_id": avail.avail_id, "error": str(exc)}
                ))
                raise RuntimeError(f"Unrecoverable validation failure for {avail.avail_id}") from exc

        recon_log.info(json.dumps(
            {"event": "reconciliation_complete", "daypart": daypart, "conflicts": conflicts}
        ))
        return conflicts

    def _matches_as_run_log(self, avail: AvailRecord) -> bool:
        # Placeholder for as-run comparison; returns True when planned == aired
        # within the Step 1 tolerance window.
        return True

Expected log line:

text 
2026-07-03 14:02:15,410 | INFO | avails.reconcile | {"event": "reconciliation_complete", "daypart": "PRIME", "conflicts": {"timecode_drift": [], "billing_mismatch": ["A-88213"], "makegood_pending": ["A-88240"]}}

Preempted avails land in makegood_pending, where they are picked up by makegood routing for preemptions to be re-slotted against equivalent inventory.

Avails reconciliation and conflict-classification flow Planned inventory pulled from the read-only database and the as-run log are UTC-normalized and diffed. If there is no discrepancy the avail is confirmed. If there is a discrepancy the conflict is classified into one of three named queues: timecode_drift, billing_mismatch, or makegood_pending, which hands preempted avails to make-good routing. Planned inventory read-only DB pull As-run log playout API Reconciliation diff UTC-normalized Discrepancy? no Confirmed avail audit trail written yes Classify route to queue timecode_drift billing_mismatch makegood_pending → make-good routing

Figure — Reconciliation flow: planned inventory and the as-run log are UTC-normalized and diffed; a confirmed avail is written to the audit trail, while every discrepancy is classified into the timecode_drift, billing_mismatch, or makegood_pending queue.

Verification & Testing

Reconciliation logic must be provable against fixture data before it touches a live daypart. Build a small fixture set that exercises each conflict bucket, then assert on the classified output. Because the engine is deterministic, the same fixtures must always yield the same buckets.

python 
def test_reconcile_classifies_each_conflict() -> None:
    fixtures = [
        AvailRecord("A-1001", _utc("18:00:00"), _utc("18:00:30"), 30, "P1", "NAT4471", "AIRED"),
        AvailRecord("A-1002", _utc("18:01:00"), _utc("18:01:30"), 30, "P1", "XYZ0001", "AIRED"),     # bad prefix
        AvailRecord("A-1003", _utc("18:02:00"), _utc("18:02:15"), 15, "P2", "AFF9920", "PREEMPTED"),  # makegood
    ]

    engine = ReconciliationEngine(db=_stub_db(fixtures), api_endpoint="https://asrun.local")
    result = engine.reconcile_daypart("PRIME", "2026-07-03")

    assert result["billing_mismatch"] == ["A-1002"]
    assert result["makegood_pending"] == ["A-1003"]
    assert result["timecode_drift"] == []


def _utc(hhmmss: str) -> datetime:
    return datetime.strptime(f"2026-07-03T{hhmmss}+0000", "%Y-%m-%dT%H:%M:%S%z")

The canonical expected output for the fixture set is a single confirmed avail (A-1001) and two classified conflicts. Run the suite in CI on every change to the classifier, and gate deployment on it: a regression that silently reclassifies billing_mismatch as confirmed is a direct revenue-attribution defect.

Edge Cases & Failure Handling

Three failure modes account for most production incidents. Handle each at the code level rather than by manual cleanup.

1. Timecode desynchronization. When a playout server’s clock drifts, start_utc and the as-run timestamp diverge past the Step 1 tolerance and every avail in the daypart flags as timecode_drift. Diagnose by comparing the delta; if it is uniform across the daypart the cause is the reference clock, not the data. Re-run normalization against the SMPTE-aligned clock, validate NTP sync on the playout node, and only then re-reconcile. Do not clear the drift queue by widening the tolerance — that masks genuine overlaps.

2. Connection-pool exhaustion. During high-volume windows a psycopg2.pool.PoolError (or statement timeouts above 3s) means every pooled connection is checked out. The pool bound is deliberate; the fix is to shorten the query, not to unbound the pool. Add a health check before getconn(), kill idle read transactions, and scale max_connections only as a temporary relief valve while the slow query is optimized.

3. Schema-version drift. A RuntimeError: Traffic schema version mismatch on startup means traffic_schema_registry.version fell below the pinned floor — usually an un-applied migration. Halt reconciliation (do not skip the check), apply the migration, confirm the registry row is active = TRUE, and restart with the version lock updated. Running against a stale schema silently attributes avails to retired billing buckets.

Failure mode Diagnostic indicator Recovery
Timecode desync uniform start_utc vs log_utc delta > 500 ms Re-run normalization on SMPTE clock; verify NTP; then reconcile
Billing divergence billing_mismatch queue spikes Reconcile against the canonical dictionary; patch the ingestion mapping
Pool exhaustion PoolError; timeouts > 3 s Health-check before getconn(); kill idle txns; optimize the query
Makegood stall makegood_pending not draining Verify affiliate clearance tier; trigger the makegood generator
Schema drift RuntimeError on startup Apply migration; confirm registry; restart with new version lock

Every recovery path must be idempotent: track execution state in a persistent ledger keyed by daypart and run ID so an interrupted job resumes from the last verified checkpoint instead of reprocessing — and double-attributing — an entire daypart.

FAQ

How is avails reconciliation different from pre-air conflict detection?

Pre-air validation asks will this schedule air cleanly? and runs before playout — that is the job of conflict detection in avails. Reconciliation asks what actually aired versus what we sold? and runs against the as-run log afterward. The first protects schedule integrity; the second protects revenue attribution and the audit trail.

Why must the reconciliation job use a read-only database role?

Reconciliation only reads planned inventory; it never mutates it. Binding the job to a read-only role (and default_transaction_read_only=on) removes an entire class of accidents — a bad diff can never overwrite committed inventory. The full role model is covered in Security Boundaries for Traffic Database Access.

A whole daypart flagged as billing_mismatch — where do I start?

That pattern almost always means the canonical billing dictionary and the ingestion mapping have diverged, not that 148 spots are individually wrong. Reconcile the dictionary first, following Standardizing Billing Codes Across Traffic Systems, then re-run the daypart. Only investigate individual avails once the mapping is confirmed correct.

What happens to preempted avails after they hit makegood_pending?

They are handed to makegood routing for preemptions, which finds equivalent inventory in the same daypart and clearance tier and re-slots the spot so the advertiser is made whole within the contracted window.

Can I safely re-run a reconciliation job that failed halfway?

Yes, provided you use the checkpoint ledger described in Edge Cases. Each run is keyed by daypart and run ID and skips already-verified avails, so a resumed job cannot double-attribute revenue. Without the ledger, a re-run would reprocess the full daypart and corrupt the audit trail.

Related content

Up one levelAvails Mapping Strategies for Linear TV