Detecting Product-Category Conflicts in Commercial Breaks

A placement-time separation check can only see the spots already on the log; it cannot see the spot that arrives an hour later and lands two positions away. That gap is where category collisions reach air — an auto spot cleared cleanly at 8:04, then a second auto spot dropped into the same break at 8:06 by a make-good, and neither check saw the other. This guide solves one exact task: sweeping a finished break — or a rolling wall-clock window across several breaks — to find every pair of spots that share a competitive group and sit closer than their separation rule allows. It is the verification pass that sits under Enforcing Competitive Separation Rules, itself a subsystem of Spot Scheduling Validation & Rule Engines. Getting it right is not cosmetic: an undetected conflict is a billable advertiser credit and a finding that surfaces in the annual reconciliation, so the scan must be deterministic enough that replaying an archived log reproduces the identical set of violations.

The scan runs against the same category taxonomy the placement-time engine uses, so a conflict it flags and a conflict the scheduler prevented are measured by the same rule. It also runs after make-good routing has settled, because a recovered spot is the most common way a late category collision enters an otherwise clean break.

Prerequisites

Python 3.11+ — required for the X | None union syntax, list[Spot] typing, and timezone-aware datetime arithmetic used throughout.
Pydantic v2 — pin pydantic==2.7.1; the models below rely on v2 field_validator semantics and reject naive datetimes at construction.
A loaded separation policy — the ProductCategory taxonomy and SeparationRule set defined in Configuring Competitive Separation Windows by Daypart, validated on load so no rule points at a missing category.
Normalised spot records — each spot classified to a canonical category_code per the spot schema, with starts_at in timezone-aware UTC. Vendor category labels must be mapped upstream, not inside the scan.
Read access to the finished log — the scan is read-only over the scheduled or as-run log; it emits violations, it never edits placements.

Step-by-Step Implementation

The scan is a pure function over a list of spots: sort by air time, group into competitive categories, then compare each spot against the ones near it in time. A break-scoped rule collapses to a same-break_id test; a rolling rule compares wall-clock gaps within a sliding horizon. Every breach becomes a ViolationRecord, and the full set is deterministic in the spots and the policy.

Step 1 — Structured logging and the scan models

Goal: emit machine-parseable audit lines in the traffic-ops timestamp | level | module | spot_id shape, and fix the typed models the scan carries. The models are deliberately the same shapes the placement engine uses, so a conflict reads identically whichever gate found it.

python

from __future__ import annotations

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

from pydantic import BaseModel, Field, field_validator

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


def _log(level: int, spot_id: str, msg: str) -> None:
    logger.log(level, "%s | %s", spot_id, msg)


class WindowScope(str, Enum):
    SAME_BREAK = "same_break"
    ROLLING = "rolling"
    DAYPART = "daypart"


class Spot(BaseModel):
    spot_id: str
    advertiser_id: str
    category_code: str          # Canonical competitive group, e.g. "qsr"
    starts_at: datetime         # Airing time, timezone-aware UTC
    duration_ms: int = Field(gt=0)
    daypart: str
    break_id: str | None = None

    @field_validator("starts_at")
    @classmethod
    def _utc_aware(cls, v: datetime) -> datetime:
        # A naive datetime distorts every rolling-gap comparison by the offset.
        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 SeparationRule(BaseModel):
    rule_id: str
    category_code: str
    daypart: str
    min_gap_ms: int = Field(ge=0)
    scope: WindowScope
    same_break_forbidden: bool = True


class ViolationRecord(BaseModel):
    violation_id: str
    spot_id: str
    conflicting_spot_id: str
    rule_id: str
    category_code: str
    gap_ms: int                 # Measured gap that fell short of the window
    scope: WindowScope
    detected_at: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc)
    )

Step 2 — Index the policy for lookup

Goal: turn the flat rule list into a (category_code, daypart) index so each spot resolves its rule in constant time during the scan. A missing rule for a category is not an error — it means that category is unrestricted — but a duplicate rule is, because it makes the outcome depend on ordering.

python

def index_rules(rules: list[SeparationRule]) -> dict[tuple[str, str], SeparationRule]:
    # Exactly one rule per (category, daypart); refuse ambiguous policy.
    index: dict[tuple[str, str], SeparationRule] = {}
    for rule in rules:
        key = (rule.category_code.lower(), rule.daypart)
        if key in index:
            raise ValueError(f"duplicate separation rule for {key}")
        index[key] = rule
    logger.info("indexed %d separation rules", len(index))
    return index

Expected log line on load: 2026-07-17T22:03:11+00:00 | INFO | traffic.separation.break_scan | indexed 14 separation rules.

Step 3 — Scan a single break for same-break conflicts

Goal: for one commercial break, find every pair of spots sharing a competitive group where the rule forbids co-occupancy. This is the cheapest and strictest check — two direct competitors in one break is a breach regardless of the measured gap between them.

python

def scan_break(
    spots: list[Spot], rules: dict[tuple[str, str], SeparationRule]
) -> list[ViolationRecord]:
    # All spots here share one break_id; compare every same-category pair once.
    violations: list[ViolationRecord] = []
    ordered = sorted(spots, key=lambda s: s.starts_at)
    for i, a in enumerate(ordered):
        rule = rules.get((a.category_code, a.daypart))
        if rule is None or not rule.same_break_forbidden:
            continue
        for b in ordered[i + 1:]:
            if b.category_code != a.category_code:
                continue
            # Same category, same break, and the rule forbids co-occupancy.
            v = 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,
                category_code=a.category_code,
                gap_ms=0,
                scope=WindowScope.SAME_BREAK,
            )
            violations.append(v)
            _log(logging.WARNING, a.spot_id,
                 f"same-break conflict with {b.spot_id} | cat={a.category_code}")
    return violations

Step 4 — Scan a rolling wall-clock window across breaks

Goal: find rolling-window breaches that cross break boundaries. Sort the whole log by air time, then for each spot look back only as far as the largest gap any rule requires — spots older than that can never conflict, which keeps the scan near-linear instead of comparing every pair.

python

def scan_rolling(
    spots: list[Spot], rules: dict[tuple[str, str], SeparationRule]
) -> list[ViolationRecord]:
    violations: list[ViolationRecord] = []
    ordered = sorted(spots, key=lambda s: s.starts_at)
    # The look-back horizon is the widest rolling window in the policy.
    horizon_ms = max(
        (r.min_gap_ms for r in rules.values() if r.scope is WindowScope.ROLLING),
        default=0,
    )
    for i, a in enumerate(ordered):
        rule = rules.get((a.category_code, a.daypart))
        if rule is None or rule.scope is not WindowScope.ROLLING:
            continue
        # Walk backward only within the horizon; stop as soon as we exceed it.
        for j in range(i - 1, -1, -1):
            b = ordered[j]
            gap_ms = int((a.starts_at - b.starts_at).total_seconds() * 1000)
            if gap_ms > horizon_ms:
                break  # Everything earlier is even further away — prune.
            if b.category_code != a.category_code:
                continue
            if gap_ms < rule.min_gap_ms:
                # Report the earlier spot as spot_id, matching scan_break, so a
                # pair breaching both scopes collapses onto one violation_id.
                v = ViolationRecord(
                    violation_id=f"VIO-{b.spot_id}-{a.spot_id}",
                    spot_id=b.spot_id,
                    conflicting_spot_id=a.spot_id,
                    rule_id=rule.rule_id,
                    category_code=a.category_code,
                    gap_ms=gap_ms,
                    scope=WindowScope.ROLLING,
                )
                violations.append(v)
                _log(logging.WARNING, b.spot_id,
                     f"rolling conflict with {a.spot_id} | gap={gap_ms}ms "
                     f"< {rule.min_gap_ms}ms")
    return violations

Step 5 — Compose the full-log sweep

Goal: run both scans over a whole log and return a de-duplicated, deterministically ordered violation set. Same-break and rolling checks can both flag the same pair; collapse them on violation_id so a break-and-rolling double breach reports once, at the stricter scope.

python

def sweep_log(
    spots: list[Spot], rules: list[SeparationRule]
) -> list[ViolationRecord]:
    index = index_rules(rules)
    found: dict[str, ViolationRecord] = {}

    # Same-break pass, grouped by break_id.
    by_break: dict[str, list[Spot]] = {}
    for s in spots:
        if s.break_id is not None:
            by_break.setdefault(s.break_id, []).append(s)
    for break_spots in by_break.values():
        for v in scan_break(break_spots, index):
            found[v.violation_id] = v  # Same-break wins on collision.

    # Rolling pass across the whole log; do not overwrite a same-break hit.
    for v in scan_rolling(spots, index):
        found.setdefault(v.violation_id, v)

    ordered = sorted(found.values(), key=lambda x: (x.spot_id, x.conflicting_spot_id))
    logger.info("sweep complete | %d spots | %d violations",
                len(spots), len(ordered))
    return ordered

Figure — Rolling scan across breaks: spots A and B share the QSR group and sit inside the four-minute window, so a conflict is raised; B and C clear because their gap exceeds the horizon, and the unrelated auto spot is skipped.

Verification & Testing

Correct behaviour rests on two properties: determinism (the same log and policy always yield the same violation set, in the same order) and completeness (no cross-break rolling conflict is missed by the pruning). Both are assertable against a fixture drawn from a real break.

python

from datetime import datetime, timezone

RULES = [
    SeparationRule(rule_id="RULE-QSR-PRIME", category_code="qsr",
                   daypart="prime", min_gap_ms=240_000,   # 4 minutes
                   scope=WindowScope.ROLLING, same_break_forbidden=True),
]

def _spot(sid: str, cat: str, minute: int, brk: str) -> Spot:
    return Spot(spot_id=sid, advertiser_id=f"ADV-{cat}", category_code=cat,
                starts_at=datetime(2026, 7, 17, 20, minute, tzinfo=timezone.utc),
                duration_ms=30_000, daypart="prime", break_id=brk)

log = [
    _spot("SP-A", "qsr", 0, "BRK-1"),      # 20:00
    _spot("SP-X", "auto", 3, "BRK-2"),     # 20:03 — different group, ignored
    _spot("SP-B", "qsr", 3, "BRK-2"),      # 20:03 — 3 min after A -> conflict
    _spot("SP-C", "qsr", 8, "BRK-3"),      # 20:08 — 5 min after B -> cleared
]

violations = sweep_log(log, RULES)

# 1. Exactly one conflict: A vs B, three minutes apart, under a four-minute rule.
assert len(violations) == 1
assert violations[0].spot_id == "SP-A"
assert violations[0].conflicting_spot_id == "SP-B"
assert violations[0].gap_ms == 180_000
assert violations[0].scope is WindowScope.ROLLING

# 2. Determinism: a re-run over the same log reproduces the identical set.
assert [v.violation_id for v in sweep_log(log, RULES)] == \
       [v.violation_id for v in violations]

Because sweep_log is pure in its inputs, the violation set is stable across machines and runs — replaying an archived log during an audit reproduces byte-identical violation_id values. Run the suite in CI against a golden fixture so a change to the pruning or the gap math that would silently drop a conflict fails the build instead of shipping to air.

Edge Cases & Failure Handling

Sub-second and identical timestamps. Two spots with the same starts_at produce a gap_ms of zero, which is always below any positive window — a correct hard conflict. The scan sorts by starts_at alone, so ties are ordered by list position; deriving violation_id from both spot IDs keeps the identifier stable regardless of which tied spot the sort placed first. Never break ties on arrival order, which is not reproducible across exports.
A spot with no break_id. As-run records sometimes lose the break grouping. Such a spot is skipped by the same-break pass but still fully covered by the rolling pass, so a missing break_id degrades gracefully to gap-based detection rather than dropping the spot from the scan. The skipped same-break check is visible in the absence of a same_break violation, not hidden.
Horizon shorter than a daypart rule. If the policy mixes rolling and daypart scopes, the rolling horizon computed in Step 4 does not bound daypart-scope checks. Run daypart-exclusive categories through the placement engine in Enforcing Competitive Separation Rules rather than the rolling scan, because a daypart rule needs a whole-daypart comparison, not a windowed one — using the rolling horizon for it would silently miss a second airing late in the part.

FAQ

Why run a full-log scan if the scheduler already checks separation at placement?

Because a placement-time check only sees the spots already on the log when a candidate lands. A spot booked later, or a recovered spot dropped in by make-good routing, can create a conflict that no earlier check could have seen. The sweep is the backstop that runs after the log is settled and before publish, catching exactly the collisions that placement-time enforcement structurally cannot.

How do I keep the scan fast on a full broadcast day of spots?

Sort once by air time and prune the look-back at the widest rolling window, as Step 4 does. Because spots older than the horizon can never conflict, the inner loop breaks early and the scan stays near-linear rather than comparing every pair. Grouping the same-break pass by break_id bounds that check to spots within a single break. If the policy is large, index the rules once as in Step 2 and reuse the index, exactly as the placement-time engine does.

What is the difference between a same-break conflict and a rolling conflict?

A same-break conflict is two spots of one competitive group in a single commercial break — a breach regardless of how many seconds separate them. A rolling conflict is two spots of the group that sit closer than the required minutes apart, even across different breaks. The daypart window configuration sets which scope applies to each category, and the sweep collapses a pair that breaches both onto the stricter same-break record.

Should the scan run on the scheduled log or the as-run log?

Both, at different times. Run it on the scheduled log before publish to stop a conflict reaching air, and run it again on the as-run log to confirm what actually aired stayed separated after live preemptions and swaps. The as-run pass feeds the same evidence trail that as-run reconciliation and billing depend on, because an advertiser credit turns on what aired, not what was planned.

Enforcing Competitive Separation Rules — the parent guide whose taxonomy, rules, and clearance model this scan verifies against the whole log.
Configuring Competitive Separation Windows by Daypart — the declarative windows that tell the scan how close is too close in each daypart.
Detecting Time Slot Conflicts in Traffic Logs — the sibling collision scan for overlapping placements, sharing the same interval-comparison discipline.

Detecting Product-Category Conflicts in Commercial Breaks

Prerequisites #

Step-by-Step Implementation #

Step 1 — Structured logging and the scan models #

Step 2 — Index the policy for lookup #

Step 3 — Scan a single break for same-break conflicts #

Step 4 — Scan a rolling wall-clock window across breaks #

Step 5 — Compose the full-log sweep #

Verification & Testing #

Edge Cases & Failure Handling #

FAQ #

Related #

Related content

Prerequisites

Step-by-Step Implementation

Step 1 — Structured logging and the scan models

Step 2 — Index the policy for lookup

Step 3 — Scan a single break for same-break conflicts

Step 4 — Scan a rolling wall-clock window across breaks

Step 5 — Compose the full-log sweep

Verification & Testing

Edge Cases & Failure Handling

FAQ

Related