Generating Morning Reset Briefings from Compliance Data

This walkthrough sits under Category Manager Briefing Automation and solves one precise task: turn the compliance data that landed overnight into a short, ranked briefing that a category manager can act on in the ninety minutes before the morning reset window opens. A raw dashboard is the wrong artifact at 6 a.m. — nobody reconciles two hundred fixtures across a chain before the doors open. What a manager needs is a one-page list of the handful of fixtures worth walking to today, each with the out_of_stock_flags and misplaced_sku_list that explain why, ordered so the top of the page is the highest-leverage fix. This page builds the scheduled job that produces that list: it gathers the overnight rollups, ranks every fixture by an impact score, renders a templated briefing per manager, and delivers it exactly once behind a completeness guard so it can never send before scoring finished. Each step is independently testable.

Prerequisites & Context Jump to heading#

This job is a consumer, not a producer. It runs after the compliance pipeline has settled for the night and assumes the following already exist. If any are missing, fix them upstream rather than papering over the gap in the briefing.

• Runtime: Python 3.11+ on a scheduled worker (cron, Airflow, or a Cloud Scheduler target). No GPU and no live camera access — this stage reads settled records only.
• Overnight scoring complete: every fixture in the reset window carries a finished compliance record with compliance_percentage, out_of_stock_flags, misplaced_sku_list, a capture_timestamp, and the registry_revision it was scored against. The score itself is produced upstream by the facings-vs-actuals stage documented in Automating Facings vs Actuals Validation — that is where the gap and misplacement signals originate.
• Rollups + drift flags available: per-fixture rollups for the window, plus the day-over-day drift score that flags fixtures trending the wrong way. Drift is what separates a fixture that is merely low today from one that is actively decaying.
• A recipient → category mapping: a table that maps each category to the category manager who owns it, with their delivery channel. A fixture with no owner is a data error, not a silent drop.
• A delivery ledger: a small persistent store keyed by recipient and reset date, so a re-run does not send twice. This is the backbone of the idempotency guarantee in Step 4.

A note on framing: this briefing is a digest, not an alert. Time-sensitive, single-event problems belong in the Real-Time Compliance Alerting & Webhooks path; the morning briefing is the calm, ranked, once-a-day counterpart that a manager reads with coffee, not a pager that fires at 2 a.m.

Step 1 — Gather the Window’s Rollups Jump to heading#

Start by loading every settled rollup for the reset window into a typed record. Keep the record close to the upstream payload so nothing is lost in translation, and carry the fields the ranking and template both need. Filter to the window explicitly — a stray record from yesterday’s late re-score must not leak in.

from dataclasses import dataclass, field
from datetime import datetime, timedelta


@dataclass(frozen=True)
class FixtureRollup:
    """One settled overnight compliance rollup for a single fixture."""
    fixture_id: str
    store_id: str
    category: str
    compliance_percentage: float          # 0-100, from the scoring stage
    sales_velocity: float                 # units/day, catalog-weighted
    drift_score: float                    # day-over-day decline, >= 0
    out_of_stock_flags: list[str] = field(default_factory=list)
    misplaced_sku_list: list[str] = field(default_factory=list)
    capture_timestamp: datetime | None = None
    registry_revision: int = 0
    scoring_complete: bool = False


def gather_window(
    rollups: list[FixtureRollup], window_start: datetime, window_end: datetime
) -> list[FixtureRollup]:
    """Keep only rollups captured inside the reset window."""
    kept: list[FixtureRollup] = []
    for r in rollups:
        if r.capture_timestamp is None:
            continue
        if window_start <= r.capture_timestamp < window_end:
            kept.append(r)
    return kept

Step 2 — Rank Fixtures by an Impact Score Jump to heading#

A briefing is only useful if the top of the page is the most valuable fix. Rank every fixture by an impact score that combines three signals: the compliance gap (how far below target the fixture sits), sales velocity (a gap on a fast mover costs more revenue than the same gap on a slow one), and drift (a fixture actively decaying deserves attention before one that is merely stable). Out-of-stock and misplacement counts add pressure on top. The weights are explicit and versioned so the ranking is auditable and reproducible.

Crucially, the sort is deterministic: ties break on fixture_id, so the same rollups always produce the same order. A briefing that reshuffles on re-run erodes trust immediately.

@dataclass(frozen=True)
class ScoreWeights:
    target_compliance: float = 95.0
    gap: float = 1.0
    velocity: float = 0.15
    drift: float = 2.0
    oos: float = 3.0
    misplaced: float = 1.5


def impact_score(r: FixtureRollup, w: ScoreWeights) -> float:
    """Higher means fix this sooner. A pure function of the rollup + weights."""
    gap = max(0.0, w.target_compliance - r.compliance_percentage)
    return round(
        w.gap * gap
        + w.velocity * r.sales_velocity
        + w.drift * r.drift_score
        + w.oos * len(r.out_of_stock_flags)
        + w.misplaced * len(r.misplaced_sku_list),
        3,
    )


def rank_fixtures(
    rollups: list[FixtureRollup], w: ScoreWeights
) -> list[tuple[FixtureRollup, float]]:
    """Sort by impact descending, breaking ties on fixture_id for determinism."""
    scored = [(r, impact_score(r, w)) for r in rollups]
    scored.sort(key=lambda pair: (-pair[1], pair[0].fixture_id))
    return scored

Step 3 — Render a Templated Briefing per Manager Jump to heading#

With a ranked list in hand, group fixtures by the manager who owns their category and render a compact briefing for each. Keep the template tight: the top N fixtures, each with its score, its gap, and the concrete out_of_stock_flags and misplaced_sku_list that tell the manager what they will find when they walk up. Markdown renders cleanly in email, Slack, and a dashboard, so it is the portable default; an HTML pass can wrap the same structure later.

def render_briefing(
    manager: str,
    ranked: list[tuple[FixtureRollup, float]],
    reset_date: str,
    top_n: int = 5,
) -> str:
    """Render one manager's markdown briefing from their ranked fixtures."""
    lines = [
        f"# Morning reset briefing — {manager}",
        f"_Reset date: {reset_date} · top {top_n} of {len(ranked)} fixtures_",
        "",
    ]
    if not ranked:
        lines.append("No open compliance issues in your categories today.")
        return "\n".join(lines)

    for i, (r, score) in enumerate(ranked[:top_n], start=1):
        gap = max(0.0, 95.0 - r.compliance_percentage)
        lines.append(
            f"## {i}. `{r.fixture_id}` — {r.category} @ {r.store_id}  "
            f"(impact {score})"
        )
        lines.append(
            f"- Compliance `{r.compliance_percentage:.0f}%` · gap `{gap:.0f}%` "
            f"· drift `{r.drift_score:.1f}`"
        )
        if r.out_of_stock_flags:
            lines.append(f"- Out of stock: {', '.join(r.out_of_stock_flags)}")
        if r.misplaced_sku_list:
            lines.append(f"- Misplaced: {', '.join(r.misplaced_sku_list)}")
        lines.append("")
    return "\n".join(lines)

Step 4 — Schedule, Guard, and Deliver Idempotently Jump to heading#

The last step wires the pieces into a scheduled entrypoint with two guarantees that matter more than the ranking itself. First, a completeness guard: the job refuses to send if any expected fixture in the window is missing a finished score, because a briefing built from half the data is worse than no briefing — it quietly hides the fixtures that had not scored yet. Second, idempotent delivery: a per-recipient, per-reset-date key in the ledger means a retry or a double-trigger never sends twice. Dedupe against yesterday is layered on top — fixtures that were on yesterday’s briefing and have not worsened are suppressed, so the list stays focused on what is new or decaying.

class ScoringIncompleteError(RuntimeError):
    """Raised when the window has fixtures without a finished score."""


def run_morning_briefing(
    now: datetime,
    rollups: list[FixtureRollup],
    recipient_by_category: dict[str, str],
    expected_fixture_ids: set[str],
    sent_ledger: set[tuple[str, str]],
    yesterday_fixture_ids: set[str],
    w: ScoreWeights = ScoreWeights(),
) -> dict[str, str]:
    """Scheduled entrypoint. Returns {manager: briefing} for what was sent."""
    window_end = now.replace(minute=0, second=0, microsecond=0)
    window_start = window_end - timedelta(hours=10)
    reset_date = window_end.date().isoformat()

    window = gather_window(rollups, window_start, window_end)

    # Completeness guard: every expected fixture must be present AND finished.
    scored_ids = {r.fixture_id for r in window if r.scoring_complete}
    missing = expected_fixture_ids - scored_ids
    if missing:
        raise ScoringIncompleteError(
            f"{len(missing)} fixtures unscored at {now.isoformat()}: "
            f"{sorted(missing)[:5]}"
        )

    # Dedupe vs yesterday: drop unchanged carry-overs (drift == 0 and no gap).
    def is_actionable(r: FixtureRollup) -> bool:
        if r.fixture_id not in yesterday_fixture_ids:
            return True
        return r.drift_score > 0 or r.compliance_percentage < w.target_compliance - 5

    actionable = [r for r in window if is_actionable(r)]

    sent: dict[str, str] = {}
    for manager in set(recipient_by_category.values()):
        mine = [
            r for r in actionable
            if recipient_by_category.get(r.category) == manager
        ]
        key = (manager, reset_date)
        if key in sent_ledger:            # already delivered — skip
            continue
        briefing = render_briefing(manager, rank_fixtures(mine, w), reset_date)
        # deliver(manager, briefing)  # email / Slack transport goes here
        sent_ledger.add(key)
        sent[manager] = briefing
    return sent

Verification & Testing Jump to heading#

Prove each guarantee with an assertion, not a glance at the inbox:

1. Deterministic ranking. Rank the same rollups twice and assert the two fixture_id orders are identical; then shuffle the input list and assert the ranked order is unchanged, confirming the tie-break is stable.
2. No send on incomplete data. Mark one expected fixture scoring_complete=False (or drop it entirely) and assert run_morning_briefing raises ScoringIncompleteError and that sent_ledger is untouched.
3. Idempotent delivery. Call the entrypoint twice with the same sent_ledger; assert the second call returns an empty dict and the ledger has exactly one key per manager.
4. Dedupe vs yesterday. Put a fixture in yesterday_fixture_ids with drift_score of 0.0 and compliance above target-minus-five, and assert it is absent from today’s briefing; raise its drift above 0 and assert it reappears.
5. Impact ordering is sane. Assert a high-velocity fixture with a 20% gap outranks a slow mover with the same gap, and that a drifting fixture outranks a stable one when gaps match.

A healthy run logs one briefing per manager, a ledger write per send, and a top-of-list fixture whose impact score is dominated by a real gap or out-of-stock cluster rather than noise.

Troubleshooting Jump to heading#

Related Jump to heading#

• Category Manager Briefing Automation — the parent cluster that owns recipient mapping, cadence, and the briefing lifecycle this job plugs into
• Real-Time Compliance Alerting & Webhooks — the real-time counterpart for single-event problems, versus this once-a-day digest
• Automating Facings vs Actuals Validation — the upstream stage that produces the gap and misplacement signals the impact score ranks on