Broadcast Traffic Scheduling Automation

Optimizing Rotation Logic for Prime Time Slots

Prime-time inventory is the highest-yield, most heavily constrained segment of a broadcast schedule, so the rotation engine that fills it has almost no margin for error. This guide solves one exact operational task: taking a raw prime-time spot queue and producing a deterministic, drift-aware rotation that honours frequency caps, competitive separation windows, and contractual priority tiers before a single line reaches the traffic log. It is the prime-time specialization of Building Rule Engines for Spot Rotation, itself a subsystem of the broader Spot Scheduling Validation & Rule Engines pipeline. Getting this right is a compliance requirement, not a refinement: when rotation logic silently over-saturates an advertiser or lands two competing brands back-to-back at 20:15, the result is a make-good obligation, a contractual breach, and an as-run log that no auditor can reconcile.

The failure modes are specific to the 20:00–23:00 window. Multiple high-CPM spots from one advertiser bunch inside a single hour and breach the contractual frequency cap; spots placed too closely violate competitive-separation clauses that the engine shares with time-slot conflict detection; and live sports overruns preempt a break, forcing a re-sequence that ignores downstream constraints and creates compliance debt across every following hour. The remedy is a stateless, auditable optimizer that validates each candidate placement against a composite constraint matrix, flags violations upstream of the log, and emits an optimized sequence with explicit compliance metadata for automated recovery.

Prerequisites

  • Python 3.11+ — required for the datetime.timezone and dataclass(frozen=True) semantics the constraint checks rely on for immutable, tz-aware inputs.
  • Pinned dependencies — the core optimizer is standard library only; add pydantic==2.7.1 for the ingestion-boundary validator that rejects malformed payloads before they reach the engine.
  • A normalized queue — candidates already resolved to the canonical spot schema with a billing code normalization pass applied, so advertiser_id, duration_sec, and priority are guaranteed present and typed.
  • Contract parameters — the per-advertiser hourly frequency cap and the minimum competitive-separation window (in seconds) for each category, sourced from the order record rather than hardcoded.
  • Traffic system access — read access to the prime-time avail grid and write access to a versioned, staged traffic-log endpoint (never a direct write to the active log).

Step-by-Step Implementation

The optimizer decouples business rules from scheduling state: every candidate is evaluated as a pure function over immutable inputs, normalized to UTC so daylight-saving transitions and market clock drift never corrupt adjacency maths. Guaranteed inventory is resolved first, then by descending CPM within each tier; a placement that fails a hard constraint backtracks to the next viable hour boundary, and every committed placement updates the running frequency and separation state.

Prime-time rotation placement loop with backtracking Spots are drawn in priority-tier then descending-CPM order and each is checked against the hourly frequency cap and competitive-separation window. A "fits slot?" decision routes fitting spots to placement and a committed running-state update; failing spots backtrack one hour boundary and re-enter the check. A spot that still fails after backtracking is flagged UNRESOLVED and handed to make-good routing rather than dropped. For each spot priority tier · CPM desc Frequency-cap & separation check UTC interval maths Fits slot? yes Place spot VALIDATED flag Commit state count + last-placed no Backtrack +1 hour boundary re-check caps & separation still fails UNRESOLVED → make-good flagged, never silently dropped

Figure — Each spot is checked against frequency-cap and separation rules; if it fits the slot it is placed and the running state committed, otherwise the optimizer backtracks one hour boundary and re-checks — and a spot that still fails is flagged UNRESOLVED for make-good routing rather than silently dropped.

Step 1 — Domain types and structured audit logging

Goal: define immutable request and slot records and emit machine-parseable audit lines in the traffic-ops timestamp | level | module | spot_id shape, so every placement decision is reconstructable after the fact.

python 
"""
prime_time_rotation_optimizer.py
Deterministic prime-time rotation: frequency caps, separation windows,
priority tiers -> optimized schedule with compliance telemetry.
"""
from __future__ import annotations

import datetime
import logging
from dataclasses import dataclass, field
from enum import Enum
from typing import Dict, List, Optional

# Audit trail: timestamp | level | module | spot_id-bearing message
logger = logging.getLogger("prime_time_optimizer")
logger.setLevel(logging.INFO)
_handler = logging.StreamHandler()
_handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | prime_time_optimizer | %(message)s"))
logger.addHandler(_handler)


class SpotPriority(Enum):
    GUARANTEED = 1   # contracted, must clear
    PRIORITY = 2     # preferred placement
    REMNANT = 3      # opportunistic fill


@dataclass(frozen=True)
class SpotRequest:
    spot_id: str
    advertiser_id: str
    duration_sec: int
    cpm: float
    priority: SpotPriority
    requested_start_utc: Optional[datetime.datetime] = None
    frequency_cap: int = 3          # max placements per advertiser per rolling hour
    min_separation_sec: int = 900   # competitive separation, default 15 min


@dataclass
class ScheduledSlot:
    spot_id: str
    start_utc: datetime.datetime
    end_utc: datetime.datetime
    compliance_flags: List[str] = field(default_factory=list)


class ConstraintViolationError(Exception):
    """Raised when a spot placement breaches a hard business rule."""

Expected log line: 2026-07-03 20:00:04,102 | INFO | prime_time_optimizer | optimizer initialized | spot_id=- window=20:00-23:00

Step 2 — UTC normalization and the constraint checks

Goal: enforce the three prime-time hard constraints as pure predicates. Normalizing every timestamp to UTC on ingestion is what stops a market-local DST shift from silently collapsing a separation window.

python 
class PrimeTimeRotationOptimizer:
    def __init__(self, prime_time_start: datetime.time, prime_time_end: datetime.time) -> None:
        self.pt_start = prime_time_start
        self.pt_end = prime_time_end
        self.scheduled: List[ScheduledSlot] = []
        self.advertiser_hourly_counts: Dict[str, Dict[str, int]] = {}
        self.last_placement_utc: Dict[str, datetime.datetime] = {}
        self.audit_log: List[Dict] = []
        logger.info("optimizer initialized | spot_id=- window=%s-%s", prime_time_start, prime_time_end)

    def _log_audit(self, event: str, spot_id: str, status: str, details: str = "") -> None:
        entry = {
            "timestamp_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(),
            "event": event, "spot_id": spot_id, "status": status, "details": details,
        }
        self.audit_log.append(entry)
        logger.info("%s | spot_id=%s | %s | %s", event, spot_id, status, details)

    def _normalize_utc(self, dt: datetime.datetime) -> datetime.datetime:
        # Reject implicit local time by pinning naive stamps to UTC, convert aware ones.
        return dt.replace(tzinfo=datetime.timezone.utc) if dt.tzinfo is None else dt.astimezone(datetime.timezone.utc)

    def _check_frequency_cap(self, advertiser_id: str, slot_start: datetime.datetime, cap: int) -> bool:
        hour_key = slot_start.strftime("%Y-%m-%d-%H")
        counts = self.advertiser_hourly_counts.setdefault(advertiser_id, {})
        return counts.get(hour_key, 0) < cap   # enforce the per-spot contractual hourly cap

    def _check_separation(self, advertiser_id: str, slot_start: datetime.datetime, min_sep: int) -> bool:
        last = self.last_placement_utc.get(advertiser_id)
        return last is None or (slot_start - last).total_seconds() >= min_sep

Step 3 — Priority-ordered placement with deterministic backtracking

Goal: resolve guaranteed inventory first, place greedily inside prime time, and on a hard-constraint failure backtrack exactly one hour boundary rather than dropping the spot outright — a placement that still fails is flagged UNRESOLVED for downstream recovery, never silently lost.

python 
class PrimeTimeRotationOptimizer:  # ...continued
    def _commit_state(self, advertiser_id: str, slot_start: datetime.datetime) -> None:
        """Record a committed placement so later spots see it in cap/separation checks."""
        hour_key = slot_start.strftime("%Y-%m-%d-%H")
        counts = self.advertiser_hourly_counts.setdefault(advertiser_id, {})
        counts[hour_key] = counts.get(hour_key, 0) + 1
        self.last_placement_utc[advertiser_id] = slot_start

    def _place(self, spot: SpotRequest, start: datetime.datetime, flags: List[str]) -> ScheduledSlot:
        slot = ScheduledSlot(
            spot_id=spot.spot_id, start_utc=start,
            end_utc=start + datetime.timedelta(seconds=spot.duration_sec),
            compliance_flags=flags,
        )
        self.scheduled.append(slot)
        self._commit_state(spot.advertiser_id, start)
        return slot

    def validate_and_schedule(self, spots: List[SpotRequest], base_date: datetime.date) -> List[ScheduledSlot]:
        # Guaranteed tier first, then highest CPM within each tier — priority inversion is forbidden.
        for spot in sorted(spots, key=lambda s: (s.priority.value, -s.cpm)):
            if spot.requested_start_utc:
                candidate = self._normalize_utc(spot.requested_start_utc)
            else:
                candidate = datetime.datetime.combine(base_date, self.pt_start, tzinfo=datetime.timezone.utc)
            try:
                if not self._check_frequency_cap(spot.advertiser_id, candidate, spot.frequency_cap):
                    raise ConstraintViolationError("frequency cap exceeded")
                if not self._check_separation(spot.advertiser_id, candidate, spot.min_separation_sec):
                    raise ConstraintViolationError("separation window violated")
                self._place(spot, candidate, ["VALIDATED"])
                self._log_audit("PLACEMENT", spot.spot_id, "SUCCESS", candidate.isoformat())
            except ConstraintViolationError as exc:
                self._log_audit("REJECTION", spot.spot_id, "VIOLATION", str(exc))
                fallback = candidate + datetime.timedelta(hours=1)   # backtrack one hour boundary
                if (self._check_frequency_cap(spot.advertiser_id, fallback, spot.frequency_cap)
                        and self._check_separation(spot.advertiser_id, fallback, spot.min_separation_sec)):
                    self._place(spot, fallback, ["BACKTRACKED", "VALIDATED"])
                    self._log_audit("BACKTRACK", spot.spot_id, "RESOLVED", fallback.isoformat())
                else:
                    self._log_audit("DROP", spot.spot_id, "UNRESOLVED", "no viable slot within constraints")
        return self.scheduled

Expected log line: ... | INFO | prime_time_optimizer | BACKTRACK | spot_id=PT-4471 | RESOLVED | 2026-07-03T21:00:00+00:00

Step 4 — Emit the compliance report

Goal: serialize the schedule and its audit trail into a single artifact so a make-good router or SIEM can act on BACKTRACKED and UNRESOLVED flags without re-deriving state.

python 
class PrimeTimeRotationOptimizer:  # ...continued
    def export_compliance_report(self) -> Dict:
        return {
            "generated_utc": datetime.datetime.now(datetime.timezone.utc).isoformat(),
            "total_scheduled": len(self.scheduled),
            "audit_entries": self.audit_log,
            "schedule": [
                {"spot_id": s.spot_id, "start_utc": s.start_utc.isoformat(),
                 "end_utc": s.end_utc.isoformat(), "flags": s.compliance_flags}
                for s in self.scheduled
            ],
        }

Verification & Testing

Confirm three invariants before trusting a run: guaranteed inventory always clears ahead of remnant, no advertiser exceeds its hourly cap, and re-running the same queue yields an identical schedule. Drive the optimizer against a small fixture and assert on the report.

python 
import datetime

def test_priority_and_cap_are_enforced() -> None:
    base = datetime.date(2026, 7, 3)
    start = datetime.datetime(2026, 7, 3, 20, 0, tzinfo=datetime.timezone.utc)
    spots = [
        SpotRequest("PT-1", "ACME", 30, 45.0, SpotPriority.REMNANT, start),
        SpotRequest("PT-2", "ACME", 30, 90.0, SpotPriority.GUARANTEED, start),
        SpotRequest("PT-3", "ACME", 30, 80.0, SpotPriority.GUARANTEED, start, frequency_cap=1),
    ]
    opt = PrimeTimeRotationOptimizer(datetime.time(20, 0), datetime.time(23, 0))
    schedule = opt.validate_and_schedule(spots, base)

    # Guaranteed, higher-CPM PT-2 takes the 20:00 slot; PT-3 hits the cap and backtracks to 21:00.
    assert schedule[0].spot_id == "PT-2"
    assert "BACKTRACKED" in next(s for s in schedule if s.spot_id == "PT-3").compliance_flags
    # Determinism: identical inputs reproduce an identical placement order.
    replay = PrimeTimeRotationOptimizer(datetime.time(20, 0), datetime.time(23, 0)).validate_and_schedule(spots, base)
    assert [s.spot_id for s in schedule] == [s.spot_id for s in replay]

A stable placement order across replays is exactly what lets you prove to an auditor that two scheduling attempts saw the same queue and produced the same log — the property scheduling-accuracy threshold tuning depends on when it measures drift.

Edge Cases & Failure Handling

Clock drift and DST shift. A market-local timestamp without an explicit offset makes a 15-minute separation window collapse or stretch across the spring/autumn transition. Enforce strict UTC normalization on ingestion (Step 2) and reject any payload missing tzinfo or ISO-8601 compliance at a pre-flight gate, before the record ever reaches validate_and_schedule.

Constraint deadlock. Overlapping frequency caps and separation windows can leave no viable slot for guaranteed inventory. Run the optimizer in dry-run telemetry mode first: it logs UNRESOLVED without committing, so traffic ops can temporarily relax a competitive-separation threshold or hand the deficit to make-good routing for preemptions for an off-peak fill rather than force an out-of-compliance placement.

Preemption cascade. A live overrun displaces 30+ minutes of scheduled breaks and invalidates every downstream adjacency. Snapshot the optimizer state before each commit; on preemption, revert to the last valid checkpoint, recompute separation windows from the new live end-time, and re-run the greedy resolver. Never patch the active log in place — regenerate and stage it, then swap atomically so playout never reads a half-written schedule.

Malformed payload injection. Third-party feeds deliver missing advertiser_id, negative durations, or invalid priority enums. Wrap ingestion in a Pydantic validator, quarantine the bad batch, and continue processing valid inventory so one poison record cannot stall the whole prime-time run.

FAQ

Why normalize everything to UTC instead of scheduling in market-local time?

Prime time is defined in local terms (20:00–23:00), but separation and frequency maths must run on a monotonic clock. If you subtract two local timestamps across a DST boundary you get a gap that is an hour wrong, which either fabricates a separation violation or hides a real one. Normalize to UTC at ingestion, do all interval arithmetic there, and render back to local only for display. The canonical field types this relies on are defined in the spot schema.

What happens to a guaranteed spot that can't be placed without breaking a rule?

It is never silently dropped. The optimizer backtracks one hour boundary and re-checks; if that still fails it emits a DROP/UNRESOLVED audit entry rather than committing an out-of-compliance placement. That flag is the trigger for make-good routing for preemptions, which resolves the contractual deficit into approved compensatory inventory.

How is this different from generic time-slot conflict detection?

Conflict detection answers “do these two placements overlap or violate separation?” — a pairwise, read-only check covered in detecting time-slot conflicts in traffic logs. Rotation optimization is the write side: it decides which spot goes where, ordering by priority tier and CPM and resolving the conflicts the detector would otherwise raise. In production the detector runs as a post-commit gate over the optimizer’s output.

Can traffic managers tune the caps and separation windows without a redeploy?

Yes — frequency_cap and min_separation_sec live on the SpotRequest, sourced from the order record, not baked into the engine. Externalize the defaults to a version-controlled policy store and reload them atomically. The accuracy trade-offs of moving those thresholds are covered in tuning thresholds for scheduling accuracy.

Related content

Up one levelBuilding Rule Engines for Spot Rotation