Building Rule Engines for Spot Rotation

Spot rotation is the deterministic constraint-satisfaction layer that turns a pool of contracted commercial orders into an executable playout sequence. Get it wrong and the failures are expensive and visible on air: the same advertiser airs three times in one break, two competing brands land back-to-back in violation of a contractual separation clause, a high-priority national spot is displaced by remnant inventory, or a retry silently double-books a slot. This guide sits within the Spot Scheduling Validation & Rule Engines system and describes how to build a rotation engine that is stateless, replayable, and auditable — one that guarantees the same output for the same inputs regardless of ingestion order or volume. It is written for traffic managers who need to reason about rotation rules in plain language, and for the Python developers who have to ship a deployable engine behind them. The core design tension is that traffic managers must be able to change rotation weights, daypart allocations, and separation thresholds without a scheduler redeploy, while developers need rigid type contracts and a single deterministic evaluation path so the pipeline stays predictable under audit.

Concept & Data Model #

Rotation operates on four entities. Keeping them as immutable, separately versioned records is what lets policy change without a code change, and what keeps every decision reconstructable after the fact.

• Spot Candidate: A discrete advertising unit competing for a slot. Carries the immutable metadata the engine evaluates — spot_id, advertiser, product category, duration, priority tier, and rotation weight. Field-level definitions live in Understanding Broadcast Spot Schemas and Metadata.
• Rotation Rule: A declarative constraint applied during evaluation — competitive separation, frequency cap, priority ordering, or daypart weighting. Rules are data, not code, so a traffic manager can tune a threshold without a redeploy.
• Slot: A time-bound, channel-specific position inside an avail into which one candidate is placed. The avail modeling that produces these windows is covered in Avails Mapping Strategies for Linear TV.
• Rotation Decision: The emitted record — accepted or rejected, with the conflicting spot IDs, the rule version applied, and an evaluation timestamp — appended to the audit ledger.

The engine reads a narrow, typed subset of the canonical spot schema. These are the only fields rotation actually evaluates; the ingestion boundary guarantees their types before a record reaches the engine.

Every rotation decision is a walk through the candidate pool: a candidate has its rotation rules applied, passes a conflict and separation check, and either enters the scheduled rotation or is reordered and retried with the next candidate.

Figure — Spots drawn from the pool have rotation rules (weight, separation, frequency cap) applied, then a conflict and separation check; passing candidates enter the scheduled rotation, while failures are reordered or replaced by the next candidate.

Implementation Approach #

Three design decisions determine whether a rotation engine stays deterministic and maintainable under production load.

Declarative rules over hardcoded logic. Rotation policy — separation minutes, frequency caps, tier weights — lives in a versioned rule set that the engine loads at evaluation time, not in if branches inside the solver. This is what lets a traffic manager raise prime-time separation from 15 to 30 minutes without a deploy, and it means every decision can name the exact rule_version_id it was evaluated under. The domain-specific tuning of those thresholds is the subject of Optimizing Rotation Logic for Prime Time Slots.

Hard constraints before soft. Evaluation order is not arbitrary. Hard constraints that can never be relaxed — temporal overlap, FCC competitive separation, network clearance — are checked first and short-circuit on the first violation. Only candidates that clear every hard rule proceed to soft scoring, where rotation weight and daypart preference rank the survivors. Collapsing the two into one score is the classic bug: a heavily weighted spot slips past a separation rule because its weight outscored the penalty.

Interval arithmetic over minute-bucketing. Temporal overlap and separation are computed with half-open interval comparison on tz-aware UTC datetimes, not by rounding placements into minute or frame buckets. Bucketing introduces edge errors at boundaries and hides sub-minute overlaps that surface as clipped spots at air. When placements do collide, isolating the exact collision boundary is delegated to Detecting Time Slot Conflicts in Traffic Logs, and any candidate the engine cannot place enters the compensatory pipeline described in Automating Make-Good Routing for Preemptions rather than being dropped.

Rotation is deliberately stateless between candidates: the engine takes a candidate plus the already-scheduled set and returns a decision, holding no mutable state of its own. Statelessness is what makes at-least-once broker delivery safe — a redelivered candidate produces the same spot_id-keyed decision and is deduplicated on commit.

Production Python Implementation #

The module below is a deployable rotation engine. It validates candidates with Pydantic, enforces UTC-only timestamps, applies hard constraints in a fixed order, and emits structured traffic-ops log lines (timestamp | level | module | message, with spot_id in every message so a decision can be traced from logs alone). It runs as-is against a Pydantic v2 environment.

from __future__ import annotations

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

from pydantic import BaseModel, Field, ValidationInfo, field_validator

# Traffic-ops structured logging: timestamp | level | module | message
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("rule_engine.rotation")


class ConflictType(str, Enum):
    TEMPORAL_OVERLAP = "temporal_overlap"
    SEPARATION_VIOLATION = "contractual_separation"
    FREQUENCY_CAP = "frequency_cap_exceeded"


class SpotCandidate(BaseModel):
    """A spot competing for a rotation slot. Timestamps MUST be tz-aware UTC;
    the ingestion boundary guarantees this before the engine sees the record."""

    spot_id: str = Field(..., min_length=1)          # UUIDv5 idempotency key
    advertiser_id: str
    product_category: str                            # competitive-separation key
    start_time: datetime
    end_time: datetime
    priority_tier: int = Field(..., ge=0, le=9)      # 0 = highest priority
    weight: float = Field(1.0, gt=0)                 # rotation share within a tier

    @field_validator("start_time", "end_time")
    @classmethod
    def must_be_utc(cls, v: datetime) -> datetime:
        # Reject naive or non-UTC timestamps: DST-shifted local clocks are the
        # single most common cause of phantom separation violations.
        if v.tzinfo is None or v.utcoffset() != timedelta(0):
            raise ValueError("timestamps must be tz-aware UTC")
        return v

    @field_validator("end_time")
    @classmethod
    def positive_duration(cls, v: datetime, info: ValidationInfo) -> datetime:
        start = info.data.get("start_time")
        if start is not None and v <= start:
            raise ValueError("zero- or negative-duration spot is illegal")
        return v


class RotationFault(BaseModel):
    spot_id: str
    conflict_type: ConflictType
    conflicting_ids: list[str] = Field(default_factory=list)
    severity: str                                    # "critical" | "warning"
    rule_version_id: str
    evaluated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc))


class RotationRuleEngine:
    """Stateless, deterministic evaluator: same inputs -> same decision, so a
    run is byte-for-byte replayable during an audit."""

    def __init__(
        self,
        rule_version_id: str,
        min_separation_minutes: int = 15,
        freq_cap_per_hour: int = 4,
    ) -> None:
        self.rule_version_id = rule_version_id
        self.min_separation = timedelta(minutes=min_separation_minutes)
        self.freq_cap_per_hour = freq_cap_per_hour
        self.faults: list[RotationFault] = []

    @staticmethod
    def _overlaps(a: SpotCandidate, b: SpotCandidate) -> bool:
        # Half-open interval overlap on [start, end): no boundary double-count.
        return a.start_time < b.end_time and b.start_time < a.end_time

    def _advertiser_count_in_window(
        self, candidate: SpotCandidate, scheduled: list[SpotCandidate]
    ) -> int:
        window = timedelta(hours=1)
        return sum(
            1
            for s in scheduled
            if s.advertiser_id == candidate.advertiser_id
            and abs(s.start_time - candidate.start_time) < window
        )

    def evaluate(self, candidate: SpotCandidate, scheduled: list[SpotCandidate]) -> bool:
        # Deterministic order: sort by start time so fault ordering never
        # depends on list/set iteration order between runs.
        for spot in sorted(scheduled, key=lambda s: s.start_time):
            if self._overlaps(candidate, spot):
                self._reject(candidate, ConflictType.TEMPORAL_OVERLAP, [spot.spot_id], "critical")
                return False
            gap = abs((candidate.start_time - spot.end_time).total_seconds())
            if (spot.product_category == candidate.product_category
                    and gap < self.min_separation.total_seconds()):
                self._reject(candidate, ConflictType.SEPARATION_VIOLATION, [spot.spot_id], "warning")
                return False

        if self._advertiser_count_in_window(candidate, scheduled) >= self.freq_cap_per_hour:
            self._reject(candidate, ConflictType.FREQUENCY_CAP, [], "warning")
            return False

        logger.info(
            "spot_id=%s accepted tier=%s weight=%.2f rule=%s",
            candidate.spot_id, candidate.priority_tier, candidate.weight, self.rule_version_id,
        )
        return True

    def _reject(
        self,
        candidate: SpotCandidate,
        conflict_type: ConflictType,
        conflicting_ids: list[str],
        severity: str,
    ) -> None:
        self.faults.append(RotationFault(
            spot_id=candidate.spot_id,
            conflict_type=conflict_type,
            conflicting_ids=conflicting_ids,
            severity=severity,
            rule_version_id=self.rule_version_id,
        ))
        logger.warning(
            "spot_id=%s rejected reason=%s conflicts=%s rule=%s",
            candidate.spot_id, conflict_type.value,
            ",".join(conflicting_ids) or "none", self.rule_version_id,
        )

Soft scoring sits on top of this hard-constraint gate: candidates that return True are ranked within their priority_tier by weight, typically with a weighted round-robin so a spot with weight=2.0 earns twice the rotation share of a weight=1.0 sibling before the next tier is considered. Priority order is strict — tier 0 inventory is exhausted before any tier 1 candidate is scored — which is what prevents priority inversion between contracted and remnant buys.

Validation & Edge Cases #

Broadcast rotation breaks at boundaries, not in the common path. The engine has to handle each of these explicitly.

• Timezone offsets and DST. Every timestamp is validated as tz-aware UTC at the model boundary. A market that ingests local wall-clock time across a daylight-saving transition will otherwise produce a candidate whose separation gap silently gains or loses an hour, flagging or hiding violations.
• Zero-duration avails. A spot with end_time <= start_time is rejected by the Pydantic validator before evaluation, never treated as an instantaneous placement that overlaps nothing.
• Sports overruns and preemption tiers. A live overrun pushes every downstream slot. The engine does not attempt to compress inventory into a shrunken window; displaced candidates fail placement and are handed to make-good routing, which re-queues them by priority tier rather than dropping the lowest-value spots first.
• Competitive separation across category, not advertiser. Two different advertisers in the same product_category (two auto dealers, two fast-food chains) must honor separation even though their advertiser_id differs. The engine groups the separation check on product_category, and the frequency cap on advertiser_id — conflating the two is a frequent source of missed violations.
• Frequency-cap window edges. The rolling one-hour cap is computed on the actual start_time delta, so a spot exactly 60 minutes after the fourth airing is legal while one at 59 minutes is not. Bucketing into clock hours would incorrectly clear a run of airings that straddles the top of the hour.

Integration Points #

Rotation is one stage in a longer pipeline. Upstream, it consumes candidates already normalized and typed by ingestion — the Pydantic validation contract that produces them is documented in Schema Validation with Pydantic for Traffic Data, part of the Avion & Avstar Ingestion Pipelines stack. Downstream, each accepted candidate becomes a placement that conflict detection, threshold tuning, and billing reconciliation all read.

The engine does not write another system’s tables directly. It emits a rotation decision to the message broker, partitioned by avail_id so that every candidate competing for one inventory window is evaluated in order on a single consumer. The emitted message carries enough lineage for downstream reconciliation to join back without re-running the engine:

{
  "message_type": "rotation.decision.v1",
  "spot_id": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
  "avail_id": "A-0912",
  "advertiser_id": "ADV-4471",
  "product_category": "quick_service_restaurant",
  "accepted": true,
  "priority_tier": 0,
  "weight": 2.0,
  "rule_version_id": "rot-2026.07-r3",
  "evaluated_at": "2026-07-03T14:22:07+00:00",
  "conflicting_ids": []
}

Contract stability matters more than convenience here: the message_type carries an explicit v1 so a schema change is additive and never silently breaks a consumer. Threshold and buffer calibration on the downstream side is handled by Tuning Thresholds for Scheduling Accuracy, which reads the same decision stream.

Compliance & Audit Considerations #

Rotation decisions are regulated artifacts, not just scheduling hints, so audit properties are part of the engine’s contract.

Competitive separation is a hard constraint. FCC guidance and network clearance agreements both treat category adjacency as a compliance obligation. A separation violation is recorded as a fault with the conflicting spot_id, never resolved by quietly relaxing the threshold to fit a candidate.

Political inventory. Any candidate flagged political inherits lowest-unit-charge and public-file disclosure obligations. Rotation must not place a political spot whose billing lineage cannot be resolved; the billing-code normalization that makes that lineage resolvable is covered in Standardizing Billing Codes Across Traffic Systems.

Immutable, versioned decision log. Every accept and every reject is appended to a write-once ledger, each entry stamped with the rule_version_id it was evaluated under and content-hash-chained to its predecessor. A correction is a new entry that references the original, never an in-place edit. Because the engine is a pure function of its inputs and rule version, an auditor can replay any historical run and obtain a byte-identical result — the property SOC 2 evidence collection depends on. Access to the ledger and the rule store itself is governed by the controls in Security Boundaries for Traffic Database Access.

Troubleshooting & Common Errors #

Separation flapping

Adjacent candidates alternately accepted and rejected across runs. Root cause is almost always non-deterministic iteration order over the scheduled set. Remediation: sort the scheduled spots by start_time before evaluation (as the reference engine does) so fault ordering is stable.

Priority inversion

A remnant tier-3 spot lands in a slot a contracted tier-0 spot should hold. Root cause is soft scoring folded into the hard-constraint pass so weight outranks tier. Remediation: exhaust each priority tier fully before scoring the next; never compare weights across tiers.

Phantom separation violation

A spot is rejected for separation that operators can see is clearly spaced. Root cause is a naive or non-UTC timestamp shifted by a DST transition. Remediation: enforce the tz-aware UTC validator at ingestion so no local wall-clock time ever reaches the engine.

Frequency-cap drift at the hour boundary

A fifth airing clears because the cap was bucketed into clock hours. Remediation: compute the cap on the rolling start_time delta, not on a truncated hour.

Quarantine backlog

Rejected candidates accumulate without being rebooked. Root cause is treating a rejection as a drop instead of a routing signal. Remediation: forward every unplaceable candidate to make-good routing, and open a circuit breaker if the quarantine rate exceeds its threshold rather than emitting a partial rotation.

Related #

• Optimizing Rotation Logic for Prime Time Slots — tuning separation windows, frequency caps, and priority tiers for the highest-yield dayparts.
• Detecting Time Slot Conflicts in Traffic Logs — isolating the exact collision boundary when two placements overlap.
• Automating Make-Good Routing for Preemptions — rebooking spots this engine cannot place into compliant windows.
• Tuning Thresholds for Scheduling Accuracy — calibrating drift tolerances and circuit-breaker limits against historical playout.
• Spot Scheduling Validation & Rule Engines — the parent system defining the taxonomy and boundaries this engine plugs into.