Slotting Architecture · 20 min read

Velocity Data Ingestion & WMS Sync Pipelines: Production-Grade Architecture for Slotting Optimization

Warehouse slotting accuracy degrades within days when velocity metrics lag behind actual demand. The moment your pick-frequency data is 72 hours stale, the optimizer starts placing last month’s fast movers in golden-zone locations while this week’s demand spikes sit in the back of the building. A production-grade ingestion and synchronization pipeline is the connective tissue that keeps ERP transaction logs, historical sales, and live WMS location state moving in lock-step so that re-slotting decisions reflect reality. This is the data plane that feeds the Core Slotting Architecture & Velocity Taxonomies system — without deterministic, exactly-once data flow, every downstream scoring engine and assignment algorithm inherits silent corruption. This guide details the architecture, canonical data schemas, and runnable Python required to operationalize velocity tracking without disrupting floor operations, and it anchors the four component guides in this section: polling, history mapping, schema validation, and async batch processing.

Pipeline Architecture Overview

A velocity sync pipeline is a directed loop, not a one-way extract job. Inbound transactions (order lines, returns, adjustments, cycle counts) are pulled by watermark-tracked extractors, normalized against warehouse dimensions, validated against a strict schema contract, scored into velocity tiers by the batch engine, and pushed back to the WMS as slot directives. The WMS acknowledgment then becomes the next cycle’s input — completed relocations change future travel telemetry, which changes future velocity, which changes the next directive. Treating the acknowledgment as the closing edge of the loop is what separates a durable pipeline from a fragile nightly cron that quietly drifts out of sync.

The velocity sync loop — the WMS acknowledgment, not the extract, is what closes each cycle.

The remainder of this guide walks the loop clockwise: first the canonical records that flow along every edge, then each component as its own subsystem, then a runnable orchestrator that wires them together, and finally the operational parameters, failure modes, and go-live checklist that keep it healthy in production.

Core Data Model

Every stage in the pipeline reads and writes a small set of canonical records. Defining them as typed dataclasses up front prevents each component from inventing its own dictionary shape and makes the contract between extract, transform, score, and push explicit. Three records carry the pipeline: TransactionEvent (the raw inbound unit of work), SyncWatermark (the exactly-once bookmark), and VelocityProfile (the scored output that becomes a slot directive).

from __future__ import annotations
import logging
from dataclasses import dataclass, field
from datetime import datetime
from decimal import Decimal
from enum import Enum
from typing import Optional

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


class EventType(str, Enum):
    ORDER_LINE = "order_line"
    RETURN = "return"
    ADJUSTMENT = "adjustment"
    CYCLE_COUNT = "cycle_count"


@dataclass(frozen=True)
class TransactionEvent:
    """One immutable inbound record from ERP or WMS."""
    event_id: str
    sku_id: str
    event_type: EventType
    quantity: int
    location_code: Optional[str]
    occurred_at: datetime
    source: str  # e.g. "erp.oracle", "wms.manhattan"


@dataclass
class SyncWatermark:
    """Exactly-once bookmark persisted after downstream acknowledgment."""
    source: str
    last_ts: datetime
    last_event_id: str
    records_committed: int = 0


@dataclass
class VelocityProfile:
    """Scored output the assignment layer consumes."""
    sku_id: str
    demand_units_30d: Decimal
    cubic_velocity: Decimal        # units * case_cubic_ft / window_days
    velocity_class: str            # A/B/C composed with X/Y/Z predictability
    seasonality_index: float = 1.0
    computed_at: datetime = field(default_factory=datetime.utcnow)

    def is_promotable(self, threshold: Decimal) -> bool:
        promote = self.demand_units_30d >= threshold
        logger.debug("SKU %s demand=%s promote=%s", self.sku_id, self.demand_units_30d, promote)
        return promote

These records map cleanly onto the taxonomy defined in SKU Velocity Taxonomy Design: velocity_class is the composed ABC/XYZ label, and cubic_velocity is the space-normalized score the assignment engine ranks locations against.

Synchronization Cadence & Architecture Selection

The first architectural decision is cadence, and it is almost always over-engineered. Real-time streaming introduces significant overhead: WMS APIs enforce strict rate limits, and continuous slotting recalculation triggers a flood of unnecessary relocation tasks for material handlers. Batch processing aligns naturally with operational rhythm — velocity scoring runs during off-peak windows and consolidated directives ship during planned maintenance or shift changes. A hybrid design typically delivers the best operational ROI: near-real-time delta ingestion for exception handling, paired with nightly batch aggregation for full taxonomy recalculation. Target delta latency for exception events under 500ms, while full batch reconciliation must complete inside a 4-hour maintenance window so it never overlaps a peak picking shift. The wrong cadence does not merely waste compute; sub-hourly full recalculation causes slotting thrash, where SKUs oscillate across tier boundaries and generate move tasks that cost more labor than they save.

Data Extraction & WMS/ERP Polling

ERP and legacy WMS systems rarely expose clean, pre-aggregated velocity feeds. Extractors must pull incremental deltas rather than full table scans, minimizing database lock contention and network payload. Polling intervals should adjust dynamically to transaction volume, and watermark tracking must guarantee exactly-once semantics so that order lines, returns, and inventory adjustments are captured without duplication or dropped high-velocity spikes. The WMS & ERP Polling Strategies guide covers cursor pagination, adaptive backoff, and clock-skew handling in depth; the essential rule is that a watermark is only advanced after the downstream stage acknowledges the batch, never on read. Persist watermarks in a low-latency store (Redis or a small Postgres table) and update them atomically. A watermark advanced on read instead of on commit is the single most common cause of silent data loss in these pipelines.

The watermark advances only after the WMS acknowledgment, so any mid-cycle crash replays the batch without loss.

import aiohttp
from typing import Any, Dict, List


async def poll_deltas(
    session: aiohttp.ClientSession,
    watermark: SyncWatermark,
    batch_size: int = 500,
) -> List[TransactionEvent]:
    """Pull events strictly newer than the persisted watermark."""
    params: Dict[str, Any] = {"since": watermark.last_ts.isoformat(), "limit": batch_size}
    async with session.get("https://api.erp.internal/v1/transactions", params=params) as resp:
        resp.raise_for_status()
        rows = await resp.json()
    events = [
        TransactionEvent(
            event_id=r["id"], sku_id=r["sku"], event_type=EventType(r["type"]),
            quantity=int(r["qty"]), location_code=r.get("loc"),
            occurred_at=datetime.fromisoformat(r["created_at"]), source=watermark.source,
        )
        for r in rows
    ]
    logger.info("polled %d events from %s since %s", len(events), watermark.source, watermark.last_ts)
    return events

Transformation & Sales History Mapping

Raw transactional data must be normalized before it can inform slotting logic. Sales history has to be mapped to warehouse-relevant dimensions — seasonality curves, promotional uplift, pack-size variation, and cross-docking frequency — because velocity taxonomies (ABC by unit volume, XYZ by demand predictability, cubic velocity by space utilization) depend on accurate historical baselines. The Sales History Data Mapping guide details how to align ERP SKU hierarchies with WMS location attributes so that velocity scores reflect actual storage and picking behavior rather than accounting-level abstractions. In practice this layer flattens nested order structures, resolves unit-of-measure conversions (each vs. case vs. pallet), and applies rolling 30/60/90-day demand windows with exponential decay so recent demand dominates without discarding seasonal signal. Get the unit-of-measure conversion wrong and a single pallet SKU will masquerade as a 48-unit hyper-mover, dragging it into a golden-zone slot it does not deserve.

def to_velocity_profile(sku_id: str, events: List[TransactionEvent], case_cubic_ft: Decimal) -> VelocityProfile:
    """Collapse a SKU's events into a scored profile over the 30-day window."""
    units = sum(e.quantity for e in events if e.event_type == EventType.ORDER_LINE)
    cubic = Decimal(units) * case_cubic_ft / Decimal(30)
    profile = VelocityProfile(
        sku_id=sku_id,
        demand_units_30d=Decimal(units),
        cubic_velocity=round(cubic, 4),
        velocity_class="UNSCORED",
    )
    logger.debug("mapped %s -> %d units, cubic=%s", sku_id, units, profile.cubic_velocity)
    return profile

Schema Enforcement & Feed Validation

Ingested feeds must pass strict structural validation before entering the transformation layer. Unvalidated payloads introduce silent corruption that causes downstream algorithms to misclassify high-velocity SKUs or reserve the wrong amount of space. Enforcing explicit contracts via Pydantic or JSON Schema stops type-coercion errors and missing critical fields at the ingestion boundary rather than three stages later. The Schema Validation for Inventory Feeds guide covers the full contract, including how to quarantine malformed records to a dead-letter queue instead of failing the whole batch. The canonical validated record looks like this:

from pydantic import BaseModel, Field, field_validator
from typing import Literal


class ValidatedVelocityRecord(BaseModel):
    sku_id: str = Field(..., min_length=4, max_length=20)
    location_code: str = Field(..., pattern=r"^[A-Z]{2}-\d{3}$")
    velocity_class: Literal["A", "B", "C", "X", "Y", "Z"]
    demand_units_30d: Decimal = Field(..., ge=0)
    cubic_velocity: Decimal = Field(..., ge=0, description="units * case_cubic_ft / 30d")
    seasonality_index: float = Field(default=1.0, ge=0.1, le=5.0)
    last_sync_ts: datetime

    @field_validator("cubic_velocity")
    @classmethod
    def clamp_precision(cls, v: Decimal) -> Decimal:
        logger.debug("validated cubic_velocity=%s", v)
        return round(v, 4)

Async Batch Processing for Velocity

Python’s asyncio enables highly concurrent I/O-bound work without the memory overhead of thread pools. Processing velocity calculations in bounded parallel chunks cuts wall-clock time by 60–80% versus synchronous iteration, which matters when a full recalculation touches hundreds of thousands of SKUs inside a fixed maintenance window. The Async Batch Processing for Velocity guide shows how to scale ingestion throughput while keeping execution deterministic and connection pools from exhausting. The core pattern is a semaphore-bounded gather over fixed-size chunks, where each chunk is fully independent so a single failure never poisons the run:

import asyncio
from typing import Callable, List


async def run_batches(
    records: List[dict],
    handler: Callable,
    concurrency: int = 20,
    chunk_size: int = 250,
) -> List[Any]:
    sem = asyncio.Semaphore(concurrency)

    async def _guarded(chunk: List[dict]) -> List[Any]:
        async with sem:
            return await handler(chunk)

    chunks = [records[i:i + chunk_size] for i in range(0, len(records), chunk_size)]
    results = await asyncio.gather(*(_guarded(c) for c in chunks), return_exceptions=True)
    ok = [item for r in results if isinstance(r, list) for item in r]
    failed = sum(1 for r in results if isinstance(r, Exception))
    logger.info("batch complete: %d records, %d failed chunks", len(ok), failed)
    return ok

Production Implementation: End-to-End Sync Runner

The following runner wires the components into a single durable cycle: poll deltas from the watermark, validate, map to profiles, score in async batches, push directives to the WMS, and only then advance and persist the watermark. Retry with jittered exponential backoff wraps the network calls, and every stage emits structured logs so a failed run is diagnosable from the logs alone.

import asyncio
import random
from functools import wraps


def retry_async(max_retries: int = 3, base_delay: float = 0.5, backoff: float = 2.0):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            delay = base_delay
            for attempt in range(max_retries):
                try:
                    return await func(*args, **kwargs)
                except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
                    if attempt == max_retries - 1:
                        logger.error("giving up after %d attempts: %s", max_retries, exc)
                        raise
                    sleep = delay + random.uniform(0, delay * 0.5)
                    logger.warning("retry %d in %.2fs: %s", attempt + 1, sleep, exc)
                    await asyncio.sleep(sleep)
                    delay *= backoff
        return wrapper
    return decorator


@retry_async()
async def push_directive(session: aiohttp.ClientSession, profile: VelocityProfile) -> str:
    payload = {"sku": profile.sku_id, "class": profile.velocity_class, "cubic": str(profile.cubic_velocity)}
    async with session.post("https://api.wms.internal/v1/slot-directives", json=payload) as resp:
        resp.raise_for_status()
        ack = await resp.json()
    return ack["directive_id"]


async def run_sync_cycle(
    session: aiohttp.ClientSession,
    watermark: SyncWatermark,
    case_cubic: Dict[str, Decimal],
    promote_threshold: Decimal,
) -> SyncWatermark:
    events = await retry_async()(poll_deltas)(session, watermark)
    if not events:
        logger.info("no new events for %s; watermark unchanged", watermark.source)
        return watermark

    by_sku: Dict[str, List[TransactionEvent]] = {}
    for ev in events:
        by_sku.setdefault(ev.sku_id, []).append(ev)

    profiles = [
        to_velocity_profile(sku, evs, case_cubic.get(sku, Decimal("1.0")))
        for sku, evs in by_sku.items()
    ]
    for p in profiles:
        p.velocity_class = "A" if p.is_promotable(promote_threshold) else "C"

    pushed = 0
    for p in profiles:
        try:
            directive_id = await push_directive(session, p)
            logger.info("pushed %s -> %s (%s)", p.sku_id, directive_id, p.velocity_class)
            pushed += 1
        except aiohttp.ClientError:
            logger.error("directive push failed for %s; routed to DLQ", p.sku_id)

    # Advance watermark ONLY after successful downstream work.
    latest = max(events, key=lambda e: e.occurred_at)
    committed = SyncWatermark(
        source=watermark.source, last_ts=latest.occurred_at,
        last_event_id=latest.event_id, records_committed=pushed,
    )
    logger.info("cycle committed: %d/%d directives, watermark -> %s", pushed, len(profiles), committed.last_ts)
    return committed

Scored profiles leave this runner and enter the assignment layer described in Location Assignment & ABC Classification Algorithms, where each SKU is ranked against candidate locations under weight, volume, and zone constraints before a physical move is authorized.

Operational Parameters

Externalize every tunable so cadence, decay, and tier thresholds change without a code deploy. The configuration below drives the runner above; both the YAML and its Python dict equivalent are shown so it can be loaded from a file or embedded in a settings module.

sync:
  delta_latency_ms: 500          # p95 target for exception deltas
  batch_window_hours: 4          # hard ceiling for full reconciliation
  poll_batch_size: 500           # rows per extractor page
velocity:
  window_days: 30                # rolling demand window
  decay_half_life_days: 14       # exponential decay on recent picks
  promote_threshold_units: 900   # 30d units to enter A class
  demote_hysteresis_windows: 2   # sustained windows before a demotion
concurrency:
  max_inflight_chunks: 20        # asyncio semaphore bound
  chunk_size: 250                # SKUs per async chunk
retry:
  max_retries: 3
  base_delay_s: 0.5
  backoff_factor: 2.0

CONFIG = {
    "sync": {"delta_latency_ms": 500, "batch_window_hours": 4, "poll_batch_size": 500},
    "velocity": {
        "window_days": 30, "decay_half_life_days": 14,
        "promote_threshold_units": 900, "demote_hysteresis_windows": 2,
    },
    "concurrency": {"max_inflight_chunks": 20, "chunk_size": 250},
    "retry": {"max_retries": 3, "base_delay_s": 0.5, "backoff_factor": 2.0},
}

The demote_hysteresis_windows value is deliberately conservative: a SKU must fall below tier for two consecutive evaluation windows before it is demoted, which prevents a single promotional spike from ping-ponging a slot in and out of the golden zone.

Operational Metrics & Production Validation

A velocity pipeline is only as valuable as its measurable impact on floor operations. Track these KPIs to validate both pipeline health and slotting efficacy:

Metric	Target Threshold	Measurement Method
Ingestion Success Rate	≥ 99.95%	`(successful_records / total_records) * 100`
Delta Sync Latency	< 500ms (p95)	Timestamp diff between ERP commit and WMS acknowledgment
Batch Reconciliation Window	≤ 4 hours	Cron start to final velocity-table commit
Slotting Accuracy Improvement	15–22% travel reduction	Pre/post pick-path telemetry
DLQ Growth Rate	< 0.1% of daily volume	Dead-letter queue size / total processed records

Validate determinism by running shadow-mode comparisons against historical WMS slotting logs: replay a known day of transactions and confirm the pipeline reproduces the same directives. Only after shadow mode matches should velocity-class transitions (for example B → A) be allowed to trigger real relocations.

Failure Modes & Remediation

Watermark advanced on read, not on commit. A crash between read and downstream write silently drops the in-flight batch. Remediation: persist the watermark only after the WMS acknowledgment, inside the same transaction that records committed counts.
Stale velocity windows. A frozen or slow extractor makes yesterday’s demand look like today’s, sending fast movers to the wrong zone. Remediation: alert when now - watermark.last_ts exceeds one cadence interval and fail the batch closed rather than scoring on stale data.
Unit-of-measure drift. Mixed each/case/pallet quantities inflate cubic velocity and over-promote SKUs. Remediation: resolve UoM in the mapping layer and assert non-null case_cubic_ft per SKU before scoring.
Slotting thrash from sub-window recalculation. Recalculating tiers too frequently oscillates SKUs across boundaries and floods handlers with move tasks. Remediation: enforce demote_hysteresis_windows and cap directive volume per cycle.
WMS sync lag and rate-limit throttling. Bursty pushes trip API rate limits and back up the directive queue. Remediation: bound concurrency with the semaphore, apply jittered backoff, and route persistent failures to the DLQ.
Unvalidated payload corruption. A single malformed field type coerces silently and misclassifies a SKU. Remediation: reject at the schema gate and quarantine the record rather than failing the whole batch.

Deployment Checklist

Provision the watermark store (Redis or Postgres) and seed each source’s initial SyncWatermark to the earliest replayable timestamp.
Load CONFIG from externalized YAML and verify promote_threshold_units against last quarter’s demand distribution.
Stand up the dead-letter queue and confirm quarantined records are queryable and replayable.
Run the pipeline in shadow mode for at least five business days, diffing generated directives against historical WMS logs.
Enable advisory mode: surface directives to planners for approval before any physical move.
Wire alerting on the five KPIs, especially watermark staleness and DLQ growth rate.
Cut over to automated push with a circuit breaker that halts execution if per-cycle relocation volume exceeds the configured cap.
Schedule the nightly full reconciliation inside the maintenance window and confirm it completes under the 4-hour ceiling for three consecutive nights before declaring go-live.

FAQ

How often should a velocity sync pipeline run?

Use a hybrid cadence: near-real-time delta ingestion (p95 under 500ms) for exception events, plus one nightly full recalculation inside a maintenance window that must finish under four hours. Sub-hourly full recalculation is almost always a mistake — it causes slotting thrash without improving accuracy.

How do I guarantee exactly-once processing?

Track a per-source watermark and advance it only after the downstream stage acknowledges the batch, never on read. Persist the watermark atomically alongside the committed record count so a crash mid-cycle replays cleanly instead of dropping in-flight events.

What causes SKUs to oscillate between velocity tiers?

Recalculating tiers too frequently or without hysteresis. Require a SKU to stay below tier for two consecutive evaluation windows (demote_hysteresis_windows) before demotion so a single promotional spike cannot ping-pong a slot in and out of the golden zone.

Should malformed records fail the whole batch?

No. Reject them at the schema-validation gate and route them to a dead-letter queue so the healthy majority of the batch still commits. Keep DLQ growth under 0.1% of daily volume and alert if it climbs.

WMS & ERP Polling Strategies — watermark cursors, adaptive backoff, and exactly-once extraction.
Sales History Data Mapping — aligning ERP SKU hierarchies with WMS location attributes.
Schema Validation for Inventory Feeds — contract enforcement and dead-letter quarantine.
Async Batch Processing for Velocity — bounded-concurrency recalculation at scale.
Core Slotting Architecture & Velocity Taxonomies — the scoring and assignment system this pipeline feeds.
Location Assignment & ABC Classification Algorithms — where scored profiles become physical slot assignments.