Detecting Seasonal Compliance Drift with Python

This walkthrough sits under Time-Series Compliance Drift Analysis and solves one precise task: separating a genuine decline in a fixture’s compliance_percentage from the weekly and seasonal rhythm that every live aisle carries. A single fixture’s compliance almost never sits still — it sags on high-traffic weekends, recovers after a Monday restock, and dips predictably through a holiday promo window. If a dashboard alert fires on every one of those dips, category managers stop trusting it inside a week. The job here is to model the repeating pattern, subtract it, and only flag the residual movement that a real problem — a broken planogram push, a store that stopped restocking, a mislabelled slot_id — would leave behind. This page builds that detector step by step against a clean per-fixture daily series keyed by capture_timestamp, and each step is independently verifiable.

Prerequisites & Context Jump to heading#

Before running this detector, confirm the inputs below exist. The method assumes you already have the typed compliance payload the Planogram Sync & SKU Mapping Strategies pillar emits — this stage only reads compliance_percentage over time; it does not recompute it.

• Runtime: Python 3.11+ with numpy and statsmodels (>=0.14) on the analytics host. No GPU required.
• A clean per-fixture daily series: one compliance_percentage value per fixture_id per day, keyed by capture_timestamp. If your captures arrive several times a day, aggregate to a daily grain first (median is more robust than mean for this).
• Enough history to see the cycle: at least two full weekly periods, and preferably 56 days, before decomposition is meaningful. A period of 7 days captures the weekend/weekday rhythm; fewer than two periods and the seasonal term is unidentifiable.
• Reset and revision metadata: the registry_revision in force on each day and any known store-reset windows. A reset legitimately resets compliance to a new baseline, and Step 4 masks those boundaries so they are never scored as drift.
• A timezone-normalised timestamp: every capture_timestamp resolved to the store’s local timezone before it becomes a daily index, so daylight-saving transitions do not silently duplicate or drop a day.

A note on terms: seasonal drift here means the expected, repeating movement — the weekly wave and any promo-window swings. Genuine drift means a change in the underlying trend or an outlier the seasonal model cannot explain. The whole point is to flag the second while forgiving the first. The reverse problem — a scoring model that has quietly recalibrated — is not drift at all; that belongs to Threshold Tuning for Compliance Accuracy, and Step 4 explains how to tell them apart.

Step 1 — Build the Series and Handle Gaps Jump to heading#

Decomposition assumes a regular index: one row per day, no missing dates, no duplicates. Real capture cadence is never that clean — a store misses a day, a camera reboots, a public holiday skips a scan. Reindex to a continuous daily range first, then fill short gaps conservatively. Never forward-fill across a long outage; a week of carried-forward values invents a flat stretch that flattens the seasonal estimate and hides real movement on the far side.

from dataclasses import dataclass
import numpy as np
import pandas as pd


@dataclass(frozen=True)
class DriftConfig:
    period_days: int = 7           # weekly cycle
    z_threshold: float = 3.0       # residual z-score to flag
    max_gap_fill_days: int = 2     # never interpolate longer runs
    min_history_days: int = 56     # two-plus full periods

    def __post_init__(self) -> None:
        if self.period_days < 2:
            raise ValueError("period_days must span at least 2 samples")
        if self.z_threshold <= 0:
            raise ValueError("z_threshold must be positive")


def build_daily_series(
    rows: pd.DataFrame, cfg: DriftConfig, tz: str
) -> pd.Series:
    """Return a gap-filled, timezone-correct daily compliance series.

    `rows` has columns capture_timestamp (UTC) and compliance_percentage.
    """
    if rows.empty:
        raise ValueError("no capture rows for fixture")
    ts = pd.to_datetime(rows["capture_timestamp"], utc=True).dt.tz_convert(tz)
    daily = (
        pd.Series(rows["compliance_percentage"].to_numpy(), index=ts)
        .resample("1D").median()          # collapse multi-capture days
    )
    full = pd.date_range(daily.index.min(), daily.index.max(), freq="1D", tz=tz)
    daily = daily.reindex(full)
    # interpolate only short gaps; leave long outages as NaN for masking
    short = daily.interpolate(limit=cfg.max_gap_fill_days, limit_area="inside")
    if short.dropna().shape[0] < cfg.min_history_days:
        raise ValueError("insufficient history for seasonal decomposition")
    return short

Converting to the store’s local timezone before the resample("1D") is the load-bearing detail. Resampling a UTC index for a store several hours off UTC shears every day boundary, smears the weekly pattern, and — twice a year — produces a 23- or 25-hour day that decomposition reads as a spike.

Step 2 — Decompose Into Seasonal, Trend, and Residual Jump to heading#

With a regular series, split it into the three components the SVG shows: a slow trend, a repeating weekly seasonal wave, and the residual left over. STL (Seasonal-Trend decomposition using Loess) from statsmodels is the right tool — it is robust to outliers and lets the seasonal shape evolve slowly, unlike a rigid classical decomposition. Keep a rolling median/std fallback for series too short or too gappy for STL, so the detector degrades gracefully rather than throwing.

from statsmodels.tsa.seasonal import STL


@dataclass(frozen=True)
class Decomposition:
    trend: np.ndarray
    seasonal: np.ndarray
    residual: np.ndarray


def decompose(series: pd.Series, cfg: DriftConfig) -> Decomposition:
    """STL decomposition with a robust rolling fallback for short series."""
    values = series.to_numpy(dtype=float)
    n_valid = int(np.isfinite(values).sum())
    if n_valid >= 2 * cfg.period_days + 1 and not np.isnan(values).any():
        stl = STL(series, period=cfg.period_days, robust=True).fit()
        return Decomposition(
            trend=stl.trend.to_numpy(),
            seasonal=stl.seasonal.to_numpy(),
            residual=stl.resid.to_numpy(),
        )
    # fallback: rolling median as trend, day-of-week means as seasonal
    trend = series.rolling(cfg.period_days, center=True, min_periods=1).median()
    detr = series - trend
    seasonal = detr.groupby(series.index.dayofweek).transform("median")
    residual = series - trend - seasonal
    return Decomposition(trend.to_numpy(), seasonal.to_numpy(), residual.to_numpy())

The robust=True flag matters: a single genuine outlier — exactly the drift you want to catch — should not be allowed to bend the trend line toward itself and hide inside the fit. Robust STL down-weights outliers when estimating the trend and seasonal terms, so a real breakout stays in the residual where Step 3 can see it.

Step 3 — Flag Changepoints on the Residual Z-Score Jump to heading#

The residual is what remains after the expected weekly rhythm is removed. In a healthy fixture it is quiet noise centred on zero. Genuine drift shows up as a residual that steps outside the band the noise defines. Standardise the residual to a z-score and flag any day whose magnitude exceeds cfg.z_threshold (a value of 3.0 is a good default — roughly three standard deviations). Use a robust z-score built on the median and MAD (median absolute deviation) rather than mean and standard deviation, because the ordinary standard deviation is inflated by the very outliers you are hunting.

from typing import Literal

DriftState = Literal["stable", "drift_low", "drift_high"]


@dataclass(frozen=True)
class DriftFlag:
    index: pd.Timestamp
    residual: float
    z_score: float
    state: DriftState


def flag_drift(
    series: pd.Series, decomp: Decomposition, cfg: DriftConfig
) -> list[DriftFlag]:
    """Flag days whose residual z-score exceeds the threshold."""
    resid = decomp.residual
    finite = resid[np.isfinite(resid)]
    median = float(np.median(finite))
    mad = float(np.median(np.abs(finite - median))) or 1e-9
    # 1.4826 scales MAD to a standard-deviation-equivalent for normal noise
    robust_sigma = 1.4826 * mad

    flags: list[DriftFlag] = []
    for ts, r in zip(series.index, resid):
        if not np.isfinite(r):
            continue
        z = (r - median) / robust_sigma
        if abs(z) < cfg.z_threshold:
            continue
        state: DriftState = "drift_low" if z < 0 else "drift_high"
        flags.append(DriftFlag(ts, round(float(r), 3), round(float(z), 2), state))
    return flags

A drift_low flag — compliance falling faster than the weekly pattern predicts — is the alert a category manager cares about most. A drift_high flag is worth surfacing too: a sudden jump usually means a reset just landed or a scoring change slipped in, which Step 4 disambiguates. Tune z_threshold down toward 2.5 for high-value fixtures where you accept more false alarms, and up toward 3.5 for noisy bottom-shelf fixtures. This same sensitivity trade-off is why backfilled history must be consistent before you trust the residuals — see Backfilling Compliance History After a Schema Change.

Step 4 — Mask Reset Windows and Registry-Revision Boundaries Jump to heading#

A store reset legitimately moves compliance to a new baseline. So does a registry_revision bump that redefines what the planogram expects. Both produce a large, real residual on the boundary day — and both are not drift. Left unmasked, every reset generates a false drift_low on the day the old layout is torn down and a false drift_high when the new one lands. Mask a small window around each boundary so those transitions are excluded from flagging.

def mask_boundaries(
    flags: list[DriftFlag],
    reset_days: set[pd.Timestamp],
    revision_by_day: dict[pd.Timestamp, int],
    window_days: int = 1,
) -> list[DriftFlag]:
    """Drop flags within `window_days` of a reset or revision change."""
    revision_changes = {
        day for day, rev in revision_by_day.items()
        if revision_by_day.get(day - pd.Timedelta(days=1), rev) != rev
    }
    boundaries = reset_days | revision_changes
    window = pd.Timedelta(days=window_days)

    kept: list[DriftFlag] = []
    for f in flags:
        if any(abs(f.index - b) <= window for b in boundaries):
            continue  # boundary movement is a reset, not drift
        kept.append(f)
    return kept

Masking a boundary is not the same as ignoring it. The reset itself is a first-class event — it belongs in the versioned payload that Compliance Score APIs & Payload Contracts exposes, so downstream consumers know the baseline shifted deliberately. What Step 4 removes is only the drift interpretation of that shift. Keep window_days at 1 for a clean overnight reset; widen to 2 for chains whose resets roll across two nights.

Verification & Testing Jump to heading#

Confirm the detector deterministically on a synthetic series where you control the truth, rather than eyeballing a dashboard:

1. Pure seasonality is never flagged. Build 70 days of a 7-day sine wave plus small Gaussian noise, no trend change. Assert flag_drift returns an empty list — the weekly rhythm lives entirely in the seasonal term.
2. Injected drift is caught. Take the same series and subtract a ramp of 1.5 points/day over the last 10 days. Assert at least one drift_low flag appears in that window with abs(z_score) >= 3.0.
3. A single spike is isolated. Drop one day by 25 points and restore it. Assert exactly that day flags, and that its robust z-score is large while its neighbours stay under threshold — proof the MAD-based sigma was not inflated by the spike.
4. A reset is masked. Inject a +20-point step aligned to a reset_days entry. Assert flag_drift sees it but mask_boundaries removes it, so the final list is empty.
5. A DST day does not spike. Feed a UTC series across a spring-forward date for a non-UTC store and assert the daily index has no duplicated or missing day after build_daily_series.

A healthy production run shows flags clustering on genuinely troubled fixtures, with reset and revision days conspicuously absent from the flagged set.

Troubleshooting Jump to heading#

Related Jump to heading#

• Time-Series Compliance Drift Analysis — the parent cluster that frames drift detection across the compliance dashboard
• Backfilling Compliance History After a Schema Change — how to keep the historical series consistent so residuals stay trustworthy
• Threshold Tuning for Compliance Accuracy — telling a genuine drift apart from a scoring recalibration that shifted the whole baseline