Real-Time Streaming Velocity Updates

Batch velocity is always yesterday’s velocity. A nightly aggregation is fine for the slow tail of your catalog, but it cannot see a promotional SKU tripling its pick rate at 10am, and it will keep that item in a reserve slot until the next run — a full shift of avoidable travel. Streaming closes that gap by treating every confirmed pick and putaway as an event the instant the WMS emits it, folding it into a windowed velocity counter that re-slotting can read seconds later. This guide is part of the Velocity Data Ingestion & WMS Sync Pipelines architecture, and it is the streaming complement to the scheduled pulls in WMS & ERP Polling Strategies: where polling asks the WMS “what changed since my cursor?” on an interval, streaming has the WMS push each change as it happens. The engineering job is to turn that firehose into a stable, incrementally-maintained velocity signal without double-counting, without drowning on a lag spike, and without letting a late event corrupt a window that already closed.

Stream vs Poll vs Batch

The three ingestion models are not competitors — they cover different freshness-versus-cost regimes, and a mature facility runs all three against the same velocity store.

• Batch recomputes velocity from scratch over a bounded window (90 days is the working default) on a fixed cadence. It is the source of truth: deterministic, easy to reconcile, and immune to the double-counting hazards of incremental math. Its weakness is latency — a SKU’s tier is only as fresh as the last run.
• Polling pulls deltas since a stored watermark cursor on a short interval (seconds to minutes). It is a good fit when the WMS exposes a change-feed endpoint but no message broker, and it self-throttles under load. Its weakness is the interval floor: you cannot react faster than you poll, and tightening the interval eventually contends with WMS transaction throughput.
• Streaming consumes an event log (Kafka, Redpanda) where the WMS publishes each pick and putaway as it commits. Latency drops to the low seconds, and back-pressure is handled by consumer lag rather than by hammering the source. Its weakness is operational: you now own exactly-once semantics, out-of-order handling, and window state, none of which a batch job has to think about.

The rule of thumb: stream the hot path, batch the truth, poll when you have no broker. Streaming keeps fast-movers’ counters live for intraday re-slotting; the nightly batch re-derives every counter from the immutable log so incremental drift never accumulates. Treat the streaming counter as a low-latency cache over the authoritative batch aggregate, not as a replacement for it.

Input Data Requirements

A streaming velocity pipeline consumes one event per confirmed inventory movement. Every event must be self-describing, carry a stable idempotency key, and carry an event-time distinct from processing-time so windowing and late-event handling work correctly. A payload missing event_id forces you back to at-least-once with no dedup, and a payload missing event_ts makes windowing depend on wall-clock arrival — both silently corrupt the counters.

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime


@dataclass(frozen=True)
class PickEvent:
    """One confirmed inventory movement consumed from the event log."""
    event_id: str
    sku_id: str
    event_type: str          # "PICK" | "PUTAWAY"
    event_ts: datetime       # event-time, not arrival-time
    qty: int
    location_code: str
    partition_offset: int

Keying the topic by sku_id is the single most important schema decision. It guarantees every event for a SKU is totally ordered on one partition, so a single consumer instance owns that SKU’s counter and never races another instance updating the same key. It also lets you scale horizontally by partition without splitting a SKU’s history across workers.

Step-by-Step Implementation

The pipeline is a consumer loop that reads events, gates them through idempotency, folds them into windowed counters, and commits offsets only after the fold is durable. Each stage is small and independently testable.

1. Model the Windowed Counter

Velocity needs two window shapes at once. A sliding window (a continuously-moving trailing interval, e.g. the last 60 minutes) gives a smooth “current rate” that re-slotting reads. A tumbling window (fixed, non-overlapping buckets, e.g. per hour) gives clean, comparable totals for the batch reconciliation to check against. The counter below keeps a deque of (event_ts, qty) samples for the slide and an integer accumulator per tumbling bucket.

from __future__ import annotations

import logging
from collections import defaultdict, deque
from dataclasses import dataclass, field
from datetime import datetime, timedelta

logger = logging.getLogger("velocity.stream")


@dataclass
class SkuVelocity:
    """Incrementally-maintained velocity counters for one SKU."""
    sku_id: str
    slide_samples: deque[tuple[datetime, int]] = field(default_factory=deque)
    tumbling_buckets: dict[datetime, int] = field(default_factory=lambda: defaultdict(int))
    slide_qty: int = 0

    def add(self, ts: datetime, qty: int, bucket_size: timedelta) -> None:
        """Fold one event into the sliding and tumbling counters."""
        self.slide_samples.append((ts, qty))
        self.slide_qty += qty
        bucket = ts - timedelta(
            seconds=ts.timestamp() % bucket_size.total_seconds()
        )
        self.tumbling_buckets[bucket] += qty

    def evict_before(self, horizon: datetime) -> None:
        """Drop sliding samples older than the trailing window horizon."""
        while self.slide_samples and self.slide_samples[0][0] < horizon:
            _, old_qty = self.slide_samples.popleft()
            self.slide_qty -= old_qty

2. Gate Events for Idempotency

At-least-once delivery is the default and the safe choice — a broker will redeliver on a consumer restart or rebalance. That means the same event_id can arrive twice, and a naive += qty double-counts. The fix is an idempotent update: keep a bounded set of recently-applied event_ids and skip any repeat. This turns at-least-once transport into effectively-once counting without the coordination cost of true exactly-once.

class IdempotencyGate:
    """Bounded LRU of applied event ids to make counter updates idempotent."""

    def __init__(self, capacity: int = 500_000) -> None:
        self._capacity = capacity
        self._seen: dict[str, None] = {}

    def is_new(self, event_id: str) -> bool:
        if event_id in self._seen:
            logger.debug("duplicate event_id %s dropped", event_id)
            return False
        self._seen[event_id] = None
        if len(self._seen) > self._capacity:
            # Evict oldest insertion (dict preserves insertion order in 3.7+).
            oldest = next(iter(self._seen))
            del self._seen[oldest]
        return True

3. Hold a Watermark for Late Events

Events do not arrive in perfect order — a slow WMS worker or a partition rebalance can deliver a 10:59 pick after an 11:01 pick. A watermark is the pipeline’s assertion that “no event older than max_ts - lateness_bound will still be accepted.” Events inside the bound update their window normally; events beyond it are too late to change an already-committed tumbling bucket and are routed to a dead-letter path for the batch reconciliation to absorb instead.

@dataclass
class Watermark:
    """Tracks the max seen event-time and admits late events up to a bound."""
    lateness_bound: timedelta
    max_ts: datetime | None = None

    def admit(self, event_ts: datetime) -> bool:
        """Advance the watermark; return False if the event is beyond the bound."""
        if self.max_ts is None or event_ts > self.max_ts:
            self.max_ts = event_ts
        too_late = event_ts < (self.max_ts - self.lateness_bound)
        if too_late:
            logger.warning(
                "late event at %s beyond bound (watermark %s); dead-lettering",
                event_ts.isoformat(), self.max_ts.isoformat(),
            )
        return not too_late

4. Run the Consumer Loop

The loop ties the pieces together: poll a batch of events, gate each for idempotency and lateness, fold survivors into their SKU counter, evict expired sliding samples, and commit offsets only after the batch is fully applied. Committing after the fold — never before — is what makes a crash replay-safe: an un-committed offset is redelivered and the idempotency gate absorbs the repeat.

from __future__ import annotations

from datetime import datetime, timedelta
from typing import Iterable


class VelocityStreamProcessor:
    """Consumes pick/putaway events and maintains windowed velocity counters."""

    def __init__(self, slide_window: timedelta, bucket_size: timedelta,
                 lateness_bound: timedelta) -> None:
        self.slide_window = slide_window
        self.bucket_size = bucket_size
        self.counters: dict[str, SkuVelocity] = {}
        self.gate = IdempotencyGate()
        self.watermark = Watermark(lateness_bound=lateness_bound)

    def process_batch(self, events: Iterable[PickEvent]) -> int:
        """Fold a polled batch into the counters; return events applied."""
        applied = 0
        for ev in events:
            if ev.event_type not in ("PICK", "PUTAWAY") or ev.qty <= 0:
                continue
            if not self.gate.is_new(ev.event_id):
                continue
            if not self.watermark.admit(ev.event_ts):
                continue
            counter = self.counters.setdefault(ev.sku_id, SkuVelocity(ev.sku_id))
            counter.add(ev.event_ts, ev.qty, self.bucket_size)
            horizon = ev.event_ts - self.slide_window
            counter.evict_before(horizon)
            applied += 1
        logger.info("batch applied: %d events, %d live SKUs",
                    applied, len(self.counters))
        return applied

    def current_rate(self, sku_id: str) -> int:
        """Return the trailing-window pick quantity for one SKU."""
        c = self.counters.get(sku_id)
        return c.slide_qty if c else 0

The focused, broker-specific version of this loop — with a real confluent-kafka consumer, a consumer-group subscription, and manual offset commits — is built end to end in How to Stream Pick Events with Kafka Consumers.

Tuning & Calibration

Three parameters decide the freshness-versus-stability trade-off: the sliding window width (how much history the “current rate” sees), the lateness bound (how long you wait for stragglers), and the commit cadence (how often you checkpoint offsets). Widen the window and the signal is smoother but slower to react; widen the lateness bound and you catch more late events but hold window state longer. Commit too rarely and a crash replays a large batch; commit too often and you throttle throughput on broker round-trips.

# streaming_velocity.yaml — one profile per facility
stream:
  topic: "wms.inventory.movements"     # keyed by sku_id
  consumer_group: "velocity-aggregator"
  slide_window_s: 3600                  # trailing window for "current rate" (60 min)
  tumbling_bucket_s: 900               # fixed bucket width for reconciliation (15 min)
  lateness_bound_s: 120                # admit events up to 2 min late; beyond -> dead-letter
  commit_every_n: 500                  # manual offset commit cadence (events)
  commit_every_s: 5.0                  # or time-based commit floor, whichever first
  max_poll_records: 1000               # events pulled per poll
  reslot_min_rate_delta: 0.20          # rate change that flags a SKU for re-slotting

# Equivalent Python config dict consumed by the processor
STREAMING_VELOCITY = {
    "topic": "wms.inventory.movements",
    "consumer_group": "velocity-aggregator",
    "slide_window_s": 3600,
    "tumbling_bucket_s": 900,
    "lateness_bound_s": 120,
    "commit_every_n": 500,
    "commit_every_s": 5.0,
    "max_poll_records": 1000,
    "reslot_min_rate_delta": 0.20,
}

Parameter sensitivity is uneven. slide_window_s is a strategy lever — set it to the horizon your re-slotting cares about and leave it. lateness_bound_s is an operations lever — fit it to the observed 99th-percentile skew between event-time and arrival-time on your topic, not a guess. commit_every_n is purely a throughput/replay trade-off and has no effect on counter correctness because the idempotency gate covers replays.

Validation & Testing

Never wire a streaming counter into re-slotting without asserting its core invariants: an idempotent replay must not change the total, the sliding window must evict expired samples, and a late event beyond the bound must be rejected rather than mutate a closed bucket. These checks drive the processor directly with no live broker.

from datetime import datetime, timedelta


def _proc() -> VelocityStreamProcessor:
    return VelocityStreamProcessor(
        slide_window=timedelta(hours=1),
        bucket_size=timedelta(minutes=15),
        lateness_bound=timedelta(minutes=2),
    )


def test_idempotent_replay_is_stable() -> None:
    proc = _proc()
    t = datetime(2025, 5, 6, 11, 0, 0)
    ev = PickEvent("e1", "SKU1", "PICK", t, 5, "A-01", 100)
    proc.process_batch([ev])
    proc.process_batch([ev])  # redelivery
    assert proc.current_rate("SKU1") == 5  # applied once, not twice


def test_sliding_window_evicts() -> None:
    proc = _proc()
    base = datetime(2025, 5, 6, 11, 0, 0)
    proc.process_batch([PickEvent("e1", "SKU1", "PICK", base, 4, "A-01", 1)])
    later = base + timedelta(hours=2)  # pushes the first sample out of the window
    proc.process_batch([PickEvent("e2", "SKU1", "PICK", later, 3, "A-01", 2)])
    assert proc.current_rate("SKU1") == 3


def test_late_event_beyond_bound_rejected() -> None:
    proc = _proc()
    base = datetime(2025, 5, 6, 11, 0, 0)
    proc.process_batch([PickEvent("e1", "SKU1", "PICK", base, 6, "A-01", 1)])
    stale = base - timedelta(minutes=10)  # beyond the 2-min lateness bound
    applied = proc.process_batch([PickEvent("e2", "SKU1", "PICK", stale, 9, "A-01", 2)])
    assert applied == 0

A healthy run applies each event exactly once, logs batch applied: N events per poll, and reports zero late-event dead-letters outside of known WMS lag windows. A rising dead-letter count is the earliest signal that your lateness_bound_s is too tight for current broker skew.

Integration Points

Streaming velocity is an input to placement, not a placement decision — the counters it maintains fan out to the downstream slotting layers, and each imposes a contract the aggregator must honour.

• Slot assignment. The live per-SKU rate feeds the solver so a SKU whose current-window velocity crosses a tier boundary becomes a candidate for a better bin. The optimization that consumes it is covered in Slot Assignment Optimization with Solvers; the streaming layer supplies fresh scores, the solver decides feasibility.
• Re-slotting triggers. A move is only worth executing when a rate change is decisive and sustained, not a momentary spike. The streaming counter emits a candidate when reslot_min_rate_delta is crossed, but the break-even and hysteresis logic in Threshold Optimization for Re-slotting decides whether the move actually fires this cycle.
• Reconciliation. Incremental counters drift — a dropped event, a gate eviction, a clock skew. The authoritative path is the nightly re-derivation in Async Batch Processing for Velocity, which recomputes every counter from the immutable log and overwrites the streaming cache, so drift never compounds across days.

Upstream, a poisoned event stream would corrupt every counter it touches, so the same distribution and freshness guards described in Data-Quality Monitoring for Inventory Feeds should sit on the topic before the aggregator, quarantining a bad producer batch instead of scoring it.

Failure Modes & Edge Cases

• Double-counting on rebalance. A consumer-group rebalance replays uncommitted events; without the idempotency gate every replayed pick inflates velocity. Remediation: keep the event_id gate in the hot path and commit offsets only after the fold, never before.
• Unbounded window state. Keeping every sliding sample forever leaks memory as the catalog grows. Remediation: evict on every add using the window horizon, and cap the idempotency set with LRU eviction.
• Late events mutating closed buckets. Accepting a straggler after its tumbling bucket was reported produces a total that disagrees with what re-slotting already acted on. Remediation: gate on the watermark and dead-letter beyond-bound events for the batch to absorb.
• Processing-time windows. Windowing on arrival-time instead of event_ts makes velocity depend on consumer lag — a lag spike looks like a demand spike. Remediation: window strictly on event-time and treat arrival-time only for commit bookkeeping.
• Streaming treated as the source of truth. Letting incremental counters run for weeks without reconciliation lets small drifts accumulate into wrong tiers. Remediation: overwrite the cache from the nightly batch and alert when streaming and batch totals diverge beyond a tolerance.

FAQ

When should I stream velocity instead of polling or batching?

Stream when intraday freshness changes a slotting decision and your WMS can publish to a broker. If a promotional or seasonal SKU can triple its pick rate within a shift and you want re-slotting to react the same day, the low-seconds latency of a stream pays for its operational cost. If your freshness need is measured in hours, polling on a short interval is simpler; if it is measured in a day, the nightly batch alone is enough. Most facilities run all three: stream the hot path, poll where there is no broker, and batch the authoritative truth.

Do I need exactly-once, or is at-least-once enough?

At-least-once transport plus an idempotent counter update gives you effectively-once counting at a fraction of the cost of true exactly-once. Keep a bounded set of applied event_ids and skip repeats; a redelivered event is then a no-op. Reserve broker-level exactly-once (transactional produce-consume) for cases where a downstream side effect — not just a counter — must never repeat. For velocity aggregation, idempotent updates are the pragmatic and correct choice.

How do I handle late and out-of-order events?

Window on event-time, not arrival-time, and hold a watermark that admits events up to a lateness bound fitted to your topic’s observed skew. Events inside the bound update their window normally; events beyond it are dead-lettered and absorbed by the nightly reconciliation instead of mutating a bucket that already drove a slotting decision. Setting the bound to the 99th-percentile event-time-to-arrival delay catches almost all stragglers without holding window state open indefinitely.

How does the streaming counter stay consistent with the batch aggregate?

Treat streaming as a low-latency cache, not the system of record. The nightly batch re-derives every counter from the immutable event log and overwrites the streaming store, so any incremental drift from dropped events or gate evictions is reset every day. Alert when the streaming total and the freshly-batched total for a SKU diverge beyond a tolerance — a persistent gap points to a lateness bound that is too tight or a producer emitting malformed events.

What is the right sliding window width for re-slotting?

Match it to the horizon your re-slotting break-even actually cares about. A 60-minute trailing window reacts fast enough to catch an intraday surge but is long enough that a single busy cart does not flip a tier. Pair it with a shorter tumbling bucket (15 minutes) for clean reconciliation totals. If you see re-slotting candidates thrashing, the window is too short or the reslot_min_rate_delta is too low — widen the window before touching the delta.

Related

• How to Stream Pick Events with Kafka Consumers — the focused consumer with manual offset commits and idempotent per-SKU counters.
• WMS & ERP Polling Strategies — the interval-based alternative when the WMS has no broker to stream from.
• Async Batch Processing for Velocity — the authoritative reconciliation path that re-derives counters from the log.
• Data-Quality Monitoring for Inventory Feeds — the freshness and distribution guards that sit in front of the topic.
• Velocity Data Ingestion & WMS Sync Pipelines — the parent architecture this streaming layer feeds.