Enforcing Competitive Separation Rules in Ad Scheduling

Two car dealers in the same break, or a fast-food chain airing ninety seconds after its direct rival, is not a cosmetic scheduling flaw — it is a breached advertiser contract and a credit the station will pay out. Competitive separation is the discipline of guaranteeing that spots from the same product category are held far enough apart in time that no advertiser is embarrassed by, or diluted against, a competitor. Enforce it by hand and it fails silently: a late make-good drops a pharma spot into a break already holding a pharma spot, a rotation reshuffle collapses the gap between two telco ads, and nobody notices until the advertiser’s agency does. This guide treats competitive separation as a deterministic gate inside Spot Scheduling Validation & Rule Engines: every candidate placement is scored against a category taxonomy and a set of daypart-aware separation windows before it is allowed onto the log. Traffic managers get the plain-language reasoning behind each rule; Python automation builders get a deployable module with strict typing, Pydantic v2 validation, and structured logging.

Separation enforcement sits alongside the other placement gates. It consumes the same normalized records that feed rule-based spot rotation, it runs against the collision findings produced by time slot conflict detection, and it must not be silently violated when make-good routing drops a recovered spot into an open avail. Every entity it touches is defined against the canonical spot schema, so the category a spot carries means the same thing to the scheduler, the rotation engine, and the billing system.

Concept & Data Model #

Competitive separation rests on two ideas: a category taxonomy that classifies every spot into a competitive group, and a separation window that says how much time — or how many intervening breaks — must sit between any two spots that share a group. The engine operates on five entities, each with a stable schema so that enforcement stays decoupled from playout and billing.

The category_code field is the pivot of the whole system. It draws from a controlled vocabulary — auto, qsr (quick-service restaurants), pharma, telco, insurance, retail_grocery, wireless_carrier — organised as a shallow tree. A wireless_carrier is a child of telco, so a rule written at telco can cascade to every carrier beneath it, while a rule written directly at wireless_carrier overrides the parent for that narrower group. This is what lets a station say “keep all telecoms four minutes apart, but keep the two national carriers six minutes apart” without duplicating policy.

Separation is measured three ways, encoded in the window scope. A same_break rule forbids two spots of the category in the same commercial break at all. A rolling rule requires a minimum elapsed gap between airings regardless of break boundaries — the common “N minutes apart” contract. A daypart rule forbids more than one airing of the category within the same daypart entirely, the strictest form, used for category-exclusive sponsorships. Layered on top, a ProductCategory.exclusive flag or a per-advertiser exclusivity clause promotes a category to sole-occupancy for a program or daypart, which the engine treats as a daypart-scope rule with an infinite gap.

Every candidate spot moves through a small state machine. It is PROPOSED when the scheduler offers it for a slot; it becomes CLEARED once it passes every applicable separation rule, CONFLICTED when a rule is breached, or EXEMPTED when a documented waiver in the order permits co-scheduling. A CONFLICTED spot is never written to the log — it is returned to the scheduler for re-slotting.

Figure — Competitive separation enforcement: a proposed spot is classified, matched to its daypart rule, and cleared if the gap holds; too small a gap is either waived by an exclusivity clause or returned to the scheduler with a ViolationRecord.

Implementation Approach #

Three design decisions shape a production separation engine, and each is a trade-off worth stating explicitly.

Taxonomy as data, not code. The competitive groups and their hierarchy change constantly — a new sponsor category is added, two brands merge, a client demands its own exclusivity tier. Hard-coding categories into the engine means a code deploy for every taxonomy edit. The engine instead loads the ProductCategory tree and its SeparationRule set from configuration owned by traffic managers, and resolves rules by walking the tree at runtime. The cost is that the config must be validated on load — a rule pointing at a non-existent category is a latent failure — so loading is a Pydantic pass, not a raw read.

Gap measurement over slot counting. Some systems approximate separation by counting intervening spots or breaks. That breaks the moment break lengths vary: three short breaks can be closer in wall-clock time than one long one. The engine measures separation in milliseconds of elapsed air time between the two candidate airings, so a rolling window of four minutes means four real minutes regardless of how the breaks fall. Break-scoped and daypart-scoped rules are then special cases layered on the same timeline.

Enforce at placement, verify on the whole log. Separation is checked twice. At placement time the engine validates one proposed spot against its neighbours before it lands, which is cheap and stops most conflicts at the source — the pattern shared with spot rotation rule engines so rotation and separation never diverge on category logic. A second, full-log sweep runs before publish to catch conflicts introduced by later edits or by a make-good drop, because a placement-time check on spot A cannot see spot B that arrives afterward. The product-category conflict scan documents that sweep in detail.

Before any of this, the engine must trust the category a spot carries. Classification leans on the canonical spot schema for the category_code field definition, and the separation windows themselves are declared per daypart using the pattern in Configuring Competitive Separation Windows by Daypart.

Production Python Implementation #

The module below is deployable as a library or a validation worker. It uses Pydantic v2 for schema enforcement, strict type hints throughout, a taxonomy walk for rule resolution, and structured logging in the traffic-ops format timestamp | level | module | spot_id.

from __future__ import annotations

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

from pydantic import BaseModel, Field, field_validator

# --- Structured logging: timestamp | level | module | spot_id ---------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("separation_engine")


def _log(level: int, spot_id: str, msg: str) -> None:
    # Prepend the spot_id so every audit line is greppable by placement.
    logger.log(level, "%s | %s", spot_id, msg)


class WindowScope(str, Enum):
    SAME_BREAK = "same_break"    # No two of the category in one break
    ROLLING = "rolling"          # Minimum elapsed gap between airings
    DAYPART = "daypart"          # At most one airing per daypart (exclusive)


class ClearanceStatus(str, Enum):
    CLEARED = "cleared"          # Passed every applicable rule
    CONFLICTED = "conflicted"    # Breached a rule, no waiver
    EXEMPTED = "exempted"        # Breach waived by an exclusivity clause


class ProductCategory(BaseModel):
    category_code: str           # e.g. "wireless_carrier"
    parent_code: Optional[str] = None   # e.g. "telco"; None at the tree root
    display_name: str
    exclusive: bool = False      # Sole-occupancy category (sponsorship tier)

    @field_validator("category_code", "parent_code")
    @classmethod
    def _lower(cls, v: Optional[str]) -> Optional[str]:
        # Codes are case-insensitive; normalise so "QSR" and "qsr" resolve alike.
        return v.lower() if v is not None else v


class SeparationWindow(BaseModel):
    min_gap_ms: int = Field(ge=0)       # Required elapsed gap in milliseconds
    scope: WindowScope

    @property
    def is_infinite(self) -> bool:
        # A daypart-scope window is effectively an infinite gap within the part.
        return self.scope is WindowScope.DAYPART


class SeparationRule(BaseModel):
    rule_id: str
    category_code: str
    daypart: str                        # e.g. "prime", "daytime", "overnight"
    window: SeparationWindow
    same_break_forbidden: bool = True

    @field_validator("category_code")
    @classmethod
    def _lower(cls, v: str) -> str:
        return v.lower()


class Spot(BaseModel):
    spot_id: str
    advertiser_id: str
    category_code: str
    starts_at: datetime                 # Scheduled airing, timezone-aware UTC
    duration_ms: int = Field(gt=0)
    daypart: str
    break_id: Optional[str] = None      # The break this spot lands in, if known

    @field_validator("starts_at")
    @classmethod
    def _utc_aware(cls, v: datetime) -> datetime:
        # Naive datetimes silently corrupt gap math across a DST boundary.
        if v.tzinfo is None:
            raise ValueError("starts_at must be timezone-aware (UTC)")
        return v.astimezone(timezone.utc)

    @field_validator("category_code")
    @classmethod
    def _lower(cls, v: str) -> str:
        return v.lower()


class ViolationRecord(BaseModel):
    violation_id: str
    spot_id: str
    conflicting_spot_id: str
    rule_id: str
    gap_ms: int                         # Measured gap that fell short
    scope: WindowScope
    severity: str                       # "hard" (contractual) or "soft" (advisory)
    detected_at: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc)
    )


class ClearanceResult(BaseModel):
    status: ClearanceStatus
    spot_id: str
    violations: list[ViolationRecord] = Field(default_factory=list)


class SeparationEngine:
    """Deterministic competitive-separation enforcement for spot placement."""

    def __init__(
        self,
        categories: list[ProductCategory],
        rules: list[SeparationRule],
    ) -> None:
        # Index the taxonomy and rules for O(1) lookup during placement.
        self._categories: dict[str, ProductCategory] = {
            c.category_code: c for c in categories
        }
        # Exactly one rule per (category_code, daypart) pair — enforce on load.
        self._rules: dict[tuple[str, str], SeparationRule] = {}
        for rule in rules:
            key = (rule.category_code, rule.daypart)
            if key in self._rules:
                raise ValueError(f"duplicate rule for {key}")
            self._rules[key] = rule

    # -- Taxonomy walk -------------------------------------------------------
    def _competitive_group(self, code: str) -> str:
        # Two spots compete if they share a category or a common ancestor at
        # which a rule is defined. Walk up until a rule root or the tree top.
        current: Optional[str] = code
        while current is not None:
            cat = self._categories.get(current)
            if cat is None:
                break
            if cat.parent_code is None:
                return current
            current = cat.parent_code
        return code

    def _resolve_rule(self, code: str, daypart: str) -> Optional[SeparationRule]:
        # Prefer a rule at the exact category; fall back up the tree so a
        # broad "telco" rule still governs a "wireless_carrier" spot.
        current: Optional[str] = code
        while current is not None:
            rule = self._rules.get((current, daypart))
            if rule is not None:
                return rule
            cat = self._categories.get(current)
            current = cat.parent_code if cat else None
        return None

    def _same_group(self, a: Spot, b: Spot) -> bool:
        return self._competitive_group(a.category_code) == \
            self._competitive_group(b.category_code)

    # -- Placement-time clearance -------------------------------------------
    def clear(
        self, candidate: Spot, neighbours: list[Spot], has_waiver: bool = False
    ) -> ClearanceResult:
        sid = candidate.spot_id
        rule = self._resolve_rule(candidate.category_code, candidate.daypart)
        if rule is None:
            # No policy for this category/daypart means unrestricted placement.
            _log(logging.INFO, sid, "no separation rule | cleared")
            return ClearanceResult(status=ClearanceStatus.CLEARED, spot_id=sid)

        violations: list[ViolationRecord] = []
        for other in neighbours:
            if other.spot_id == sid or not self._same_group(candidate, other):
                continue
            breach = self._is_breach(candidate, other, rule)
            if breach is not None:
                violations.append(breach)

        if not violations:
            _log(logging.INFO, sid,
                 f"cleared under rule {rule.rule_id} | daypart={candidate.daypart}")
            return ClearanceResult(status=ClearanceStatus.CLEARED, spot_id=sid)

        if has_waiver:
            # A documented exclusivity waiver permits co-scheduling; still logged.
            _log(logging.WARNING, sid,
                 f"{len(violations)} breach(es) waived by exclusivity clause")
            return ClearanceResult(status=ClearanceStatus.EXEMPTED, spot_id=sid,
                                   violations=violations)

        _log(logging.WARNING, sid,
             f"conflicted | {len(violations)} breach(es) under {rule.rule_id}")
        return ClearanceResult(status=ClearanceStatus.CONFLICTED, spot_id=sid,
                               violations=violations)

    def _is_breach(
        self, a: Spot, b: Spot, rule: SeparationRule
    ) -> Optional[ViolationRecord]:
        # Same-break rule: any co-occurrence in one break is a breach outright.
        if rule.same_break_forbidden and a.break_id is not None \
                and a.break_id == b.break_id:
            return self._violation(a, b, rule, gap_ms=0)
        # Daypart-scope rule: any second airing in the same daypart is a breach.
        if rule.window.scope is WindowScope.DAYPART and a.daypart == b.daypart:
            return self._violation(a, b, rule, gap_ms=0)
        # Rolling rule: measure real elapsed gap between the two airings.
        gap_ms = abs(int((a.starts_at - b.starts_at).total_seconds() * 1000))
        if gap_ms < rule.window.min_gap_ms:
            return self._violation(a, b, rule, gap_ms=gap_ms)
        return None

    @staticmethod
    def _violation(
        a: Spot, b: Spot, rule: SeparationRule, gap_ms: int
    ) -> ViolationRecord:
        return ViolationRecord(
            violation_id=f"VIO-{a.spot_id}-{b.spot_id}",
            spot_id=a.spot_id,
            conflicting_spot_id=b.spot_id,
            rule_id=rule.rule_id,
            gap_ms=gap_ms,
            scope=rule.window.scope,
            # Same-break and daypart-exclusive breaches are contractual ("hard");
            # a rolling-window near-miss is advisory ("soft") until the desk rules.
            severity="soft" if rule.window.scope is WindowScope.ROLLING else "hard",
        )

The engine never mutates the log directly. It returns a ClearanceResult; the scheduler decides whether to place, re-slot, or escalate. That separation keeps clearance pure and testable, and it means the same candidate can be re-checked against a changed neighbour set without side effects.

Validation & Edge Cases #

Broadcast operations produce boundary conditions that a naive separation check mishandles. Each of the following is exercised by the implementation above.

• Timezone and DST offsets. Rolling gaps are measured between two starts_at values. A naive local datetime crossing a spring-forward or fall-back boundary distorts the elapsed gap by an hour, which can hide a real conflict or invent a phantom one. The Pydantic validators reject naive datetimes outright and normalise everything to UTC before any subtraction.
• Taxonomy depth and rule cascade. A wireless_carrier spot with no carrier-specific rule must still inherit the telco rule. The tree walk in _resolve_rule climbs until it finds a rule or exhausts the ancestry, so adding a child category never silently drops it out of an existing parent policy.
• Same-break versus rolling collision. Two spots can share a break yet be far apart on the timeline if the break is long. A same_break rule treats co-occupancy as a breach regardless of measured gap, while a rolling rule ignores break boundaries entirely. Encoding the distinction in WindowScope stops one rule type from masquerading as the other.
• Make-good re-entry. A recovered spot dropped in by make-good routing is a new placement that must pass the same clearance as an original booking. Routing that skips separation to hit a pacing goal is the single most common way a category collision reaches air; the full-log sweep is the backstop.
• Zero-duration and malformed spots. duration_ms is constrained to be positive and starts_at must be timezone-aware, so a corrupt record fails validation at the boundary rather than passing a nonsensical gap into the comparison. A spot with no break_id still gets rolling and daypart checks; only the same-break test is skipped, and its absence is visible in the log.

Integration Points #

Upstream, the engine consumes normalised records. Spots arrive already classified into category_code by the ingestion adapters described in Avion & Avstar Ingestion Pipelines, so the engine works against a single canonical shape rather than vendor-specific category labels. The taxonomy and rule set are loaded once at start-up and refreshed on config change.

Downstream, a CONFLICTED result is returned to the scheduler and — where it must cross a service boundary — serialised as a small wire contract. The message carries the rule that was breached and the measured gap so the scheduler can decide between re-slotting and requesting a waiver:

{
  "message_type": "separation.violation.v1",
  "violation_id": "VIO-SP-2026-0717-0044-SP-2026-0717-0051",
  "spot_id": "SP-2026-0717-0044",
  "conflicting_spot_id": "SP-2026-0717-0051",
  "category_code": "wireless_carrier",
  "rule_id": "RULE-TELCO-PRIME",
  "daypart": "prime",
  "scope": "rolling",
  "required_gap_ms": 360000,
  "measured_gap_ms": 180000,
  "severity": "soft",
  "detected_at": "2026-07-17T23:14:02Z"
}

The required_gap_ms and measured_gap_ms pair is what a traffic manager needs to act: it says not only that a rule broke but by how much, which distinguishes a spot that missed by seconds from one dropped straight into a rival’s break. Consumers reconcile violations on violation_id, which is derived from the two spot IDs so a re-run over the same log produces the same identifier and duplicate reports collapse.

Compliance & Audit Considerations #

Competitive separation is contract-critical: a breach is a billable credit, and a pattern of breaches is the kind of finding that surfaces in an advertiser’s annual reconciliation. Three obligations apply directly.

Immutable violation trail. Every clearance decision — cleared, conflicted, or exempted — is emitted on the timestamp | level | module | spot_id log line and should be shipped to append-only storage. A ViolationRecord is never overwritten; a resolved conflict is a new placement decision, not an edited old one. When an advertiser disputes a placement, the trail reconstructs exactly which rule governed the daypart at air time and what gap actually ran, which is what a billing reconciliation or an as-run review needs.

Exclusivity as documented waiver, never silent override. An EXEMPTED result is only legitimate when the order carries a written co-scheduling clause. The engine records the exemption and the breach it waived, so an exemption is auditable rather than an untracked hole in the policy. An engine that silently downgrades a hard breach to a pass is indistinguishable, after the fact, from one that failed — the log must show the waiver.

Sponsorship identification carries through. Category-exclusive sponsorships often coincide with FCC sponsorship-identification obligations: a sole-sponsor daypart still needs its sponsor ID cleared. Separation clearance does not replace that clearance, and the two gates are kept independent so neither masks the other, consistent with the discipline in billing code normalization where the code a spot bills against is fixed before placement.

Troubleshooting & Common Errors #

The ERR_ORPHAN_RULE case deserves particular attention. It is silent by construction — an orphan rule loads without error yet never matches a spot — so the taxonomy and rule set must be cross-validated on load, the same fail-on-config-error discipline applied when tuning thresholds for scheduling accuracy. A rule the traffic desk wrote but the engine silently ignores is worse than a loud failure, because everyone believes separation is enforced when it is not.

Related #

• Spot Scheduling Validation & Rule Engines — the parent architecture that defines the taxonomy, contracts, and clearance guarantees this engine inherits.
• Detecting Product-Category Conflicts in Commercial Breaks — the full-log interval scan that catches conflicts a placement-time check cannot see.
• Configuring Competitive Separation Windows by Daypart — the declarative configuration that supplies the per-category, per-daypart windows this engine resolves.
• Building Rule Engines for Spot Rotation — the rotation logic that must share the same category model so rotation never undoes separation.