Shelf Analytics

Integrating Legacy POS Data with Modern Vision APIs for Shelf Analytics

Within the Security Boundaries for Retail Image Data component of the Core Architecture for Shelf Analytics platform, this page solves one narrow but recurring task: how to reconcile batch-oriented legacy point-of-sale transaction logs with real-time, spatially aware vision detections so that planogram compliance scoring does not fire false violations. Legacy POS systems emit rigid, hourly-batched logs keyed by UPC and register; modern vision APIs return JSON detections keyed by bounding box, confidence, and shelf coordinate. The two streams disagree on identifier, cadence, and meaning of “empty.” Bridge them naively and an out-of-stock alert pages a category manager twenty minutes before the morning restock truck arrives. The job here is a deterministic join — SKU normalization, temporal windowing, and a velocity-weighted compliance score — that turns transactional truth and visual reality into a single signal without co-mingling raw imagery with sales data.

Reconciling batch legacy POS logs with real-time vision detections into a single compliance signal Two lanes feed a central reconciliation engine. The left lane carries a legacy POS feed (EDI 852/867, fixed-width and CSV, keyed by UPC in hourly batches) into a SKU normalization stage that maps UPC to vision class through a grace-period alias table and quarantines deprecated codes. The right lane carries vision API detections (JSON with bbox, confidence and facing_count, keyed by class_id) into a typed pydantic contract that rejects malformed records at the boundary. Both converge on a reconciliation engine that normalizes timestamps to a UTC sliding window and weights each sale by an exponential-decay curve — sales near the capture instant carry weight one, sales twenty-four hours earlier decay toward exp of minus one. The engine emits a central ShelfState compliance score between zero and one, plus two anomaly flags: shrink or misplacement when facings and velocity are both zero, and backstock overflow when the shelf reads full despite high velocity. LEGACY POS · BATCH VISION API · REAL-TIME Legacy POS feed EDI 852 / 867 · fixed-width · CSV keyed by UPC · hourly batch Vision API detections JSON · bbox · confidence facing_count · keyed by class_id SKU normalization · UPC ↔ vision class lookup · grace-period alias table · deprecated UPC → quarantine Typed contract · pydantic v2 validation · reject malformed at edge · stable class_id + facings quarantine manual review Reconciliation engine UTC sliding window · exponential-decay weighting weight ≈ exp(−1) weight = 1.0 −24h window start capture (t=0) ShelfState compliance_score 0.0 – 1.0 anomaly: shrink / misplacement facings = 0 · velocity = 0 not a clean stockout anomaly: backstock overflow shelf full · velocity > 5 post-replenishment capture

Prerequisites and Context

Before applying this page, you need these pieces in place:

  • A vision detection feed that already emits a stable class_id, a bbox_confidence in the 0.0–1.0 range, and a per-fixture facing_count. If your detections are still noisy, calibrate them first with the techniques in Bounding Box Extraction & SKU Localization — reconciliation amplifies upstream detection error rather than hiding it.
  • A POS export you can read on a schedule: fixed-width files, CSV dumps, or EDI 852/867 feeds carrying store id, register id, transaction timestamp, UPC/EAN, quantity sold, and a promotional flag.
  • Python 3.11+ (the zoneinfo module is used for timezone normalization), plus pydantic v2, pandas, and numpy.
  • A defined image-capture cadence. The reconciliation window must be at least as long as your replenishment cycle; a default lookback of 24 hours suits most grocery formats, with 48 hours for low-traffic stores.
  • The classification envelope from the security tier. This join consumes metadata_only fields and POS figures; it must never dereference a raw_shelf_photo object_key, so transactional data and imagery stay in separate trust zones.

Step-by-Step Implementation

Step 1 — Establish a canonical SKU mapping

POS keys products by UPC/EAN; the vision model keys them by class label. Build a version-controlled, bidirectional lookup so that every detection resolves to a POS SKU before any compliance math runs. Handle vendor packaging changes and private-label rotations with a grace-period alias table that routes deprecated UPCs to the current vision label for a configurable window of 30–90 days, after which the alias expires and the old code is quarantined for manual review.

# sku_mapping.yaml — version-controlled, reviewed quarterly
version: "2.1"
mappings:
  - vision_class_id: "bev_cola_500ml"
    legacy_pos_upc: "049000000123"
    status: "active"
    effective_from: "2023-01-01"
    aliases:
      - legacy_upc: "049000000119"
        expires_at: "2024-03-31"
        reason: "packaging_refresh_v2"

Verify the mapping loads and round-trips: a UPC resolved to a class and back must return the same UPC, and every active detection class must have exactly one live mapping.

Step 2 — Define typed contracts for both streams

Use pydantic so a malformed POS row or a detection missing its facing_count is rejected at the boundary instead of poisoning a compliance score three stages later.

from datetime import datetime
from typing import Optional

from pydantic import BaseModel, Field


class VisionDetection(BaseModel):
    image_id: str
    capture_utc: datetime
    class_id: str
    bbox_confidence: float = Field(ge=0.0, le=1.0)
    facing_count: int = Field(ge=0)


class POSTransaction(BaseModel):
    store_id: str
    register_id: str
    transaction_utc: datetime
    upc: str
    quantity_sold: int = Field(ge=0)
    promo_flag: bool = False


class ShelfState(BaseModel):
    image_id: str
    class_id: str
    detected_facings: int
    pos_velocity_24h: int
    compliance_score: float = Field(ge=0.0, le=1.0)
    anomaly_flag: Optional[str] = None

Step 3 — Normalize timestamps to UTC

POS batches arrive in local store time, frequently without a DST adjustment, while edge cameras stamp captures from whatever their NTP daemon believes. Normalize everything to UTC at ingestion using a store-specific timezone, then sort chronologically. Without this step, a one-hour DST drift silently shifts every transaction outside the reconciliation window.

import pandas as pd
from datetime import timezone
from zoneinfo import ZoneInfo


def normalize_timestamps(df: pd.DataFrame, store_tz: str) -> pd.DataFrame:
    """Convert local POS timestamps to UTC and sort chronologically."""
    tz = ZoneInfo(store_tz)
    df["transaction_utc"] = (
        pd.to_datetime(df["transaction_utc"])
        .dt.tz_localize(tz)
        .dt.tz_convert(timezone.utc)
    )
    return df.sort_values("transaction_utc")

Step 4 — Apply a velocity-weighted lookback window

For each capture event, query every transaction within the lookback window and weight it by recency. Sales that happened minutes before the photo describe the shelf you photographed far better than sales from 20 hours earlier, so apply an exponential decay to the transaction weights rather than a flat sum.

import numpy as np


def calculate_weighted_velocity(
    pos_df: pd.DataFrame, capture_utc: datetime, window_hours: int = 24
) -> float:
    """Exponentially decay recent sales toward the capture instant."""
    window_start = capture_utc - pd.Timedelta(hours=window_hours)
    recent = pos_df[pos_df["transaction_utc"] >= window_start]
    if recent.empty:
        return 0.0
    time_diffs = (capture_utc - recent["transaction_utc"]).dt.total_seconds()
    weights = np.exp(-time_diffs / (window_hours * 3600))
    return float(np.sum(recent["quantity_sold"] * weights))

Step 5 — Score compliance and flag anomalies

Compare detected facings against an expected baseline derived from recent velocity, then classify the two disagreements that matter operationally. An empty facing with zero sales is probably shrink or misplacement, not a stockout; a full shelf despite high velocity is usually a post-replenishment capture or backstock overflow. Both deserve different downstream handling — neither is a clean compliance pass.

def reconcile_shelf_state(
    vision: VisionDetection,
    pos_df: pd.DataFrame,
    store_tz: str,
    window_hrs: int = 24,
) -> ShelfState:
    """Deterministic SKU + temporal join into a single compliance signal."""
    pos_df = normalize_timestamps(pos_df, store_tz)
    velocity = calculate_weighted_velocity(pos_df, vision.capture_utc, window_hrs)

    expected_facings = max(1, int(velocity * 0.8))  # tunable baseline heuristic
    compliance = min(1.0, vision.facing_count / expected_facings)

    anomaly: Optional[str] = None
    if vision.facing_count == 0 and velocity == 0:
        anomaly = "potential_shrink_or_misplacement"
    elif vision.facing_count >= expected_facings and velocity > 5:
        anomaly = "post_replenishment_or_backstock_overflow"

    return ShelfState(
        image_id=vision.image_id,
        class_id=vision.class_id,
        detected_facings=vision.facing_count,
        pos_velocity_24h=int(velocity),
        compliance_score=round(compliance, 3),
        anomaly_flag=anomaly,
    )

The expected_facings baseline is the single most sensitive parameter here; calibrate it against ground-truth audits using the same methodology described in Threshold Tuning for Compliance Accuracy rather than hard-coding the 0.8 multiplier across every category.

Step 6 — Harden the ingestion path

Wrap external POS database queries in a circuit breaker so a slow query during a peak transaction window cannot stall the whole reconciliation fleet. Make message processing idempotent — replaying the same capture must produce the same ShelfState — and route malformed EDI 852 payloads or corrupted vision JSON to a dead-letter queue instead of dropping them. Every schema translation, window adjustment, and score mutation should emit an audit log line for regulatory traceability.

Verification and Testing

Confirm the join behaves before wiring it to alerts:

  • Round-trip the mapping. Assert resolve(resolve_upc(class_id)) == class_id for every active mapping, and that no detection class resolves to more than one live UPC.
  • Pin the decay. With a synthetic frame and one sale exactly at capture_utc, calculate_weighted_velocity must return that sale’s full quantity (weight 1.0); a sale at the window edge must return a weight near exp(-1).
  • Assert score bounds. ShelfState.compliance_score must always land in 0.0–1.0; a pydantic ValidationError here means your baseline produced a negative or non-finite expected count.
  • Replay for idempotency. Feed the same capture twice and assert both ShelfState objects are equal — proof the path carries no hidden mutable state.
  • Watch the false-positive rate. Run validate_compliance.py --dry-run over a week of historical captures and confirm the false stockout rate stays below 2% before promoting the change.

Troubleshooting

Symptom Root cause Remediation
Phantom stockouts Photo captured before the morning restock; POS shows zero movement but the engine expects full facings. Raise window_hours to 48 for low-traffic stores and set a min_facings_override in the YAML config for the affected fixtures.
SKU drift after a remodel Products rotated without updating the vision class dictionary, so detections resolve to nothing. Enable auto_alias_fallback and route unmapped detections to a quarantine queue; reconcile the quarantine log against weekly planogram change requests, then update sku_mapping.yaml.
Timestamp desync Edge cameras drift from NTP and POS batches use local time without a DST adjustment, pushing sales outside the window. Enforce chrony on edge devices, normalize all ingestion to UTC via zoneinfo, and apply a ±5s tolerance at the gateway.
High-velocity “false full” POS shows rapid sales but vision detects a full shelf because of stacked backstock or misplaced facings. Require bbox_overlap_ratio > 0.7 for the primary shelf plane and apply a depth filter so backstock behind the front facing is excluded.
Cascading latency at peak Synchronous POS queries pile up during high transaction volume and starve the reconciliation workers. Add a circuit breaker around the POS query, fall back to the last cached velocity, and reprocess from the dead-letter queue once the breaker closes.
Back to top