Setting Dynamic Volume Thresholds for Rebates

Static rebate tiers drift out of alignment the moment promotional velocity spikes, a supply constraint bites, or seasonal demand shifts — and a fixed breakpoint that no longer reflects commercial reality either leaks margin on overpaid accruals or rejects legitimate vendor claims. This page documents the exact procedure for replacing hard-coded breakpoints with dynamic thresholds: tier boundaries that scale against a rolling sales-velocity signal and a seasonal coefficient, are snapshotted to an immutable configuration version before each run, and resolve with decimal-exact arithmetic. It is the implementation-level companion to the volume threshold validation cluster, which governs how aggregated volume is tested against contractual bands inside the broader claim validation rule engine.

Data flow from velocity and seasonal inputs through scaling, snapshot, and tier evaluation A left-to-right pipeline. Three inputs — a rolling velocity signal (a 30-day sum turned into a multiplier), a seasonal coefficient, and the versioned base tier bands — feed a Scale and clamp stage. That stage multiplies the velocity multiplier by the seasonal coefficient, clamps it between 0.80 and 1.50, and shifts each base breakpoint up or down. The scaled bands pass into a Snapshot gate that hashes them into an immutable config_hash before any claim is read. The frozen configuration drives the Tier evaluation stage, which resolves attained volume against the frozen bands and branches two ways: a clean pass routes to Auto-accrue, where the posted accrual carries its config_hash, while below-floor or past-cap volume routes to Manual and anomaly review handled by the scoring and confidence models. clean pass below-floor / past-cap Rolling velocity 30-day Σ → multiplier Seasonal coefficient demand-curve factor Base tier bands versioned · contiguous Scale & clamp multiplier × seasonal clamp 0.80 ≤ m ≤ 1.50 shifts breakpoints ↑/↓ Snapshot gate config_hash · SHA-256 frozen before eval Tier evaluation resolve attained volume against frozen bands Auto-accrue accrual posted — carries config_hash Manual / anomaly review → scoring & confidence models

Prerequisites

Before wiring dynamic scaling into the engine, confirm the following are in place:

  • A versioned base-tier contract. The static bands originate in agreement schema design and carry an agreement_id, a monotonic agreement_version, and ordered, non-overlapping lower_bound / upper_bound pairs. Dynamic scaling adjusts these bands; it never invents them.
  • Normalized, deduplicated volume. Sales arrive from the data ingestion normalization pipelines already reduced to a single unit-of-measure, with external identifiers resolved by SKU mapping & deduplication so velocity is not inflated by double-counted lines.
  • An aligned promotional timeline. Date window alignment checks must have bounded the transaction set to the contracted window before any rolling metric is computed — an off-promo sale folded into the velocity signal systematically over-scales every band above the breakpoint it tips.
  • Python packages: pandas>=2.2, pydantic>=2.6, and the standard-library decimal module. Every monetary and boundary field uses decimal.Decimal; never float.
  • Access role: write access to the versioned configuration table and read access to the contract registry (typically the reconciliation_etl service role).

Step-by-step implementation

Step 1 — Model the dynamic threshold contract

Encode the base tiers plus the scaling parameters as a Pydantic v2 model so that an out-of-range multiplier or an inverted band fails at load time rather than during accrual.

python
from decimal import Decimal
from typing import Optional
from pydantic import BaseModel, field_validator, model_validator


class DynamicTier(BaseModel):
    tier_index: int
    lower_bound: int          # inclusive, in qualifying units
    upper_bound: Optional[int]  # exclusive; None = open top band
    rate: Decimal             # per-unit or pct, 4 dp

    @field_validator("rate")
    @classmethod
    def non_negative(cls, v: Decimal) -> Decimal:
        if v < 0:
            raise ValueError("rate must be >= 0")
        return v


class ThresholdConfig(BaseModel):
    agreement_id: str
    agreement_version: int
    base_tiers: list[DynamicTier]
    velocity_floor: Decimal = Decimal("0.80")   # clamp scaling down
    velocity_cap: Decimal = Decimal("1.50")     # clamp scaling up
    seasonal_coefficient: Decimal = Decimal("1.00")

    @model_validator(mode="after")
    def bands_contiguous(self) -> "ThresholdConfig":
        ordered = sorted(self.base_tiers, key=lambda t: t.lower_bound)
        for lo, hi in zip(ordered, ordered[1:]):
            if lo.upper_bound != hi.lower_bound:
                raise ValueError("tier bands must be contiguous, non-overlapping")
        return self

Validation check: load a fixture with an intentional gap between bands and assert that ThresholdConfig(**fixture) raises ValidationError. A config that parses must be contiguous before it can be scaled.

Step 2 — Compute the rolling velocity signal

Derive the velocity multiplier from a rolling window over qualifying daily volume. The pandas rolling window documentation covers the moving-average mechanics; the load-bearing detail is min_periods — if the window holds fewer observations than min_periods it returns NaN, and an unhandled NaN propagates silently into every downstream comparison.

python
import pandas as pd

WINDOW_DAYS = 30
BASELINE_UNITS = Decimal("10000")  # contracted prior-period reference

def velocity_multiplier(daily: pd.Series) -> Decimal:
    rolling = daily.rolling(window=WINDOW_DAYS, min_periods=WINDOW_DAYS).sum()
    latest = rolling.iloc[-1]
    if pd.isna(latest):                # insufficient history -> neutral scaling
        return Decimal("1.00")
    return (Decimal(str(latest)) / BASELINE_UNITS).quantize(Decimal("0.0001"))

Validation check: feed a series shorter than WINDOW_DAYS and assert the function returns Decimal("1.00") rather than NaN or a partial sum. A short feed must degrade to neutral scaling, not a phantom multiplier.

Step 3 — Scale the tier boundaries

Apply the clamped velocity multiplier and the seasonal coefficient to each base boundary, keeping all arithmetic in Decimal. Clamping to velocity_floor / velocity_cap is what prevents a demand anomaly from rewriting the tier ladder entirely.

python
def scaled_tiers(cfg: ThresholdConfig, raw_mult: Decimal) -> list[DynamicTier]:
    mult = max(cfg.velocity_floor, min(cfg.velocity_cap, raw_mult))
    factor = mult * cfg.seasonal_coefficient
    out = []
    for t in cfg.base_tiers:
        lo = int((Decimal(t.lower_bound) * factor).to_integral_value())
        hi = None if t.upper_bound is None else int(
            (Decimal(t.upper_bound) * factor).to_integral_value())
        out.append(DynamicTier(tier_index=t.tier_index,
                               lower_bound=lo, upper_bound=hi, rate=t.rate))
    return out

Validation check: assert the scaled bands remain ordered and contiguous (ThresholdConfig(base_tiers=scaled, ...) must still validate). Scaling must preserve the band invariants from Step 1, not just shift numbers.

Step 4 — Snapshot the active configuration

Before evaluating a single claim, freeze the scaled configuration to a versioned, hashed record. Every accrual must reference the exact config_hash that produced it, so a SOX auditor can reconstruct why a tier triggered on a given day.

python
import hashlib, json

def snapshot(cfg: ThresholdConfig, scaled: list[DynamicTier]) -> dict:
    payload = {
        "agreement_id": cfg.agreement_id,
        "agreement_version": cfg.agreement_version,
        "tiers": [t.model_dump(mode="json") for t in scaled],
    }
    blob = json.dumps(payload, sort_keys=True, default=str)
    return {"config_hash": hashlib.sha256(blob.encode()).hexdigest(),
            **payload}

Validation check: snapshot the same scaled config twice and assert the two config_hash values are identical; mutate one rate and assert the hash changes. Determinism here is what makes a re-run reproducible.

Step 5 — Evaluate volume and route the outcome

Resolve attained volume against the snapshotted bands, then hand near-miss or anomalous results to scoring & confidence models rather than auto-accruing them. A run that exceeds the cap-clamped top band is flagged for manual review instead of paying unbounded.

python
def resolve_tier(units: int, snap: dict) -> dict:
    selected = None
    for t in snap["tiers"]:
        upper = t["upper_bound"]
        if units >= t["lower_bound"] and (upper is None or units < upper):
            selected = t
            break
    route = "auto_accrue"
    if selected is None:
        route = "manual_review"          # below floor band
    elif snap["tiers"][-1] is selected and units > BASELINE_UNITS * 2:
        route = "anomaly_review"         # runaway volume guard
    return {"tier_index": None if selected is None else selected["tier_index"],
            "config_hash": snap["config_hash"], "route": route}

Validation check: assert that a units value above twice the baseline lands on anomaly_review and that the returned config_hash matches the snapshot. Every decision must carry the hash that justified it.

Common failure modes and fixes

  1. Floating-point drift in tier comparisons. Computing boundaries with float lets rounding error compound across millions of line items, dropping claims into the wrong band. Keep every boundary and rate in decimal.Decimal and quantize explicitly (Decimal("0.0001")) — never mix a float multiplier into the chain.
  2. Silent NaN from min_periods. A rolling window with too little history returns NaN; an unguarded NaN compares falsely against every band and zeroes the accrual without error. Guard with pd.isna(...) and degrade to neutral (1.00) scaling, as in Step 2.
  3. Race condition on mid-cycle parameter edits. Reading live scaling parameters during each record evaluation means an analyst’s edit halfway through a job produces two different tier ladders in one accrual. Snapshot once at job start (Step 4) and evaluate every record against the frozen config_hash.
  4. Runaway payout on a demand spike. Without the velocity_cap clamp, a viral week scales the top band so high that no volume qualifies — or so low that everything over-accrues. Clamp the multiplier and route past-cap volume to anomaly_review rather than paying it.
  5. Threshold drift during data gaps. When a feed is late, the rolling sum silently shrinks and the ladder scales down. Detect the short window, hold the prior snapshot or apply a conservative floor through fallback routing logic, and log the substitution explicitly so the gap is auditable rather than invisible.

Operational checklist

Frequently asked questions

What happens when a retroactive tier is crossed mid-cycle? The snapshot taken at job start (Step 4) governs the run, so a mid-cycle crossing does not silently rewrite already-evaluated rows. The next scheduled run picks up the new rolling velocity, re-scales, and re-hashes — and because each accrual carries its config_hash, the retroactive recomputation is reconstructable rather than a black box.

Why scale the boundaries instead of scaling the rate? Scaling the rate changes how much each unit pays; scaling the boundary changes how much volume a vendor must move to qualify. Dynamic thresholds are about keeping attainment difficulty constant as demand shifts — so the boundary, not the rate, is the lever. Rate-side adjustments belong to payout structure modeling.

How do I keep two reconciliation runs reproducible? Freeze and hash the scaled configuration before evaluation, and pin the rolling window to a closed date range. Identical inputs then produce an identical config_hash and byte-identical accruals; any divergence points at an unsnapshotted live read.

Should the velocity signal use purchase or sell-through volume? Match the measure_basis declared on the contract tiers. Mixing a sell-through velocity signal against purchase-based bands double-counts channel lag and is a frequent source of over-scaling.

Back to