Routing Kitchen Waste to Cost Variances

This page shows a food-tech developer or automation engineer how to take a raw stream of kitchen waste events and route each one into the correct food-cost variance bucket, so a culinary manager sees whether margin leaked to prep yield, spoilage, portioning, or service comps rather than a single unactionable number. It is the concrete implementation companion to Waste Tracking & Routing Systems — read that first for the event-driven ingestion and BOM-resolution rationale, then follow the numbered steps here to stand up a router you can run against a reconciled waste feed today.

The reconciliation gap in any theoretical vs actual food cost calculation pipeline is rarely one failure; it is cumulative drift from unallocated trim, spoilage, plate returns, and remakes. Routing attributes each discarded gram to a specific variance sub-account so the signed delta produced downstream by the variance mapping methodologies layer carries a cause, not just a magnitude.

Prerequisites and Data Contract

Pin these versions and provision the two inputs before the steps apply. The router is only deterministic if the SKUs on incoming events are already canonical and the standard-cost matrix is period-aligned.

• Runtime: Python 3.11+, pydantic==2.*, pandas==2.2.*, numpy==1.26.*. Physical quantities are floats; every currency value is decimal.Decimal in Python and PostgreSQL NUMERIC at rest, so the money path never inherits binary-float drift.
• Environment: read access to the raw waste store (written by POS void hooks, prep-scale telemetry, and manual entry) and to the standard-cost matrix refreshed post-close. Routing runs after ingestion, not inline with a live POS pull — that scheduling belongs to the upstream async batch processing workflow.
• Assumption: source_sku is already resolved to a canonical component. Vendor and menu SKUs must be normalized via your POS taxonomy mapping before they reach this router; and the yield_factor on each event comes from the yield factor calculation frameworks, not from a guess at the routing layer.

The waste event input contract — one row per discard:

The variance ledger output contract — one immutable row per routed event:

The output guarantee: every event leaves the router in exactly one variance bucket with either a posted cost impact or an explicit flagged_for_review marker — never a silent zero handed to the ledger.

Step-by-Step Implementation

Each step is a self-contained block. Compose them in order inside one batch worker, partitioned by location and run once per reconciled day.

Step 1 — Validate and normalize each raw event

Reject impossible data at the boundary so the routing engine only ever sees clean rows. A Pydantic v2 model enforces the enum, coerces every timestamp to UTC, and refuses negative quantities, which almost always signal POS sync drift rather than real waste.

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

WasteType = Literal["prep_trim", "spoilage", "plate_return", "comp_remake"]

class WasteEvent(BaseModel):
    event_id: str
    location_id: str
    timestamp_utc: datetime
    source_sku: str
    waste_type: WasteType
    raw_quantity: float
    uom: str
    yield_factor: float = Field(default=1.0, gt=0.0, le=1.0)

    @field_validator("raw_quantity")
    @classmethod
    def reject_negative(cls, v: float) -> float:
        if v < 0:
            raise ValueError("Negative waste implies POS sync drift or manual entry error")
        return v

    @field_validator("timestamp_utc")
    @classmethod
    def enforce_utc(cls, v: datetime) -> datetime:
        return v.replace(tzinfo=timezone.utc) if v.tzinfo is None else v.astimezone(timezone.utc)

Note yield_factor is constrained to (0, 1] with gt=0.0, which closes the divide-by-zero door before any arithmetic runs.

Step 2 — Canonicalize units of measure to a base unit

Multi-unit fleets report the same discard as lbs, kg, oz, or g depending on the scale firmware. Resolve every reported unit to one base unit per SKU family before routing, or cross-location totals silently mix scales. This is the same discipline that standardizing portion sizes across locations applies on the portioning side.

UOM_TO_KG = {
    "kg": 1.0, "g": 0.001,
    "lb": 0.453592, "lbs": 0.453592, "oz": 0.0283495,
}

def to_base_kg(raw_quantity: float, uom: str) -> float:
    """Convert a reported weight to canonical kilograms."""
    factor = UOM_TO_KG.get(uom.strip().lower())
    if factor is None:
        raise KeyError(f"Unmapped unit alias: {uom!r} — extend UOM_TO_KG")
    return raw_quantity * factor

Step 3 — Map each waste type to a variance bucket

The accounting category is a pure function of waste_type. Keep the map explicit and total so an unknown type raises rather than defaulting into the wrong sub-account.

VARIANCE_MAP = {
    "prep_trim":    "yield_variance",
    "spoilage":     "inventory_shrinkage",
    "plate_return": "portion_execution_variance",
    "comp_remake":  "service_variance",
}

def bucket_for(waste_type: str) -> str:
    try:
        return VARIANCE_MAP[waste_type]
    except KeyError as exc:
        raise ValueError(f"Unroutable waste_type: {waste_type!r}") from exc

Step 4 — Apply the yield adjustment and compute a decimal-safe cost impact

Yield scaling applies strictly to prep_trim: purchased weight overstates usable loss, so trim is scaled by its yield_factor; every other category discards at 1.0. Money arithmetic runs in Decimal and quantizes once, so a month of daily postings never accumulates sub-cent float error.

from decimal import Decimal, ROUND_HALF_UP

def route_event(event: WasteEvent, standard_cost_per_kg: Decimal) -> dict:
    base_qty = to_base_kg(event.raw_quantity, event.uom)
    yield_scalar = event.yield_factor if event.waste_type == "prep_trim" else 1.0
    effective_qty = base_qty * yield_scalar

    cost_impact = (Decimal(str(effective_qty)) * standard_cost_per_kg).quantize(
        Decimal("0.0001"), rounding=ROUND_HALF_UP
    )
    return {
        "event_id": event.event_id,
        "location_id": event.location_id,
        "timestamp_utc": event.timestamp_utc.isoformat(),
        "variance_bucket": bucket_for(event.waste_type),
        "effective_quantity": effective_qty,
        "cost_impact": cost_impact,
        "routing_status": "posted",
    }

Identical inputs always produce identical outputs — the property that makes the ledger auditable.

Step 5 — Route high-volume batches without row iteration

At fleet scale, events arrive in daily batches of tens of thousands. Route them with a vectorized merge and column assignment instead of a Python loop. Quantities stay in NumPy; the money multiply is deferred to the NUMERIC ledger in Step 6, so the batch never touches float currency.

import pandas as pd
import numpy as np

def batch_route(df_events: pd.DataFrame, df_costs: pd.DataFrame) -> pd.DataFrame:
    """Vectorized routing. df_costs columns: source_sku, standard_cost_per_kg."""
    routed = df_events.merge(df_costs, on="source_sku", how="left")

    # Unmapped costs are flagged for review, never zeroed into the totals.
    routed["cost_missing"] = routed["standard_cost_per_kg"].isna()

    routed["effective_quantity"] = np.where(
        routed["waste_type"] == "prep_trim",
        routed["base_quantity"] * routed["yield_factor"],
        routed["base_quantity"],
    )
    routed["variance_bucket"] = routed["waste_type"].map(VARIANCE_MAP)
    routed["routing_status"] = np.where(
        routed["cost_missing"], "flagged_for_review", "posted"
    )
    return routed

Step 6 — Append to the variance ledger idempotently

Routed rows are immutable financial artifacts. Insert them into an append-only ledger keyed on (event_id, location_id) so a retried device transmission cannot double-post, and let PostgreSQL NUMERIC perform the money multiply exactly.

INSERT INTO variance_ledger (
    event_id, location_id, variance_bucket,
    effective_quantity, cost_impact, routing_status
)
SELECT
    s.event_id, s.location_id, s.variance_bucket,
    s.effective_quantity,
    ROUND(s.effective_quantity::numeric * s.standard_cost_per_kg, 4),
    s.routing_status
FROM staging_routed AS s
ON CONFLICT (event_id, location_id) DO NOTHING;

DO NOTHING (not DO UPDATE) is deliberate: a posted variance is never rewritten, preserving a tamper-evident trail for financial close.

Verification and Validation

Confirm the router behaves before you let it feed reconciliation.

• Every event lands in exactly one bucket. No null buckets, only known values:

  assert routed["variance_bucket"].notna().all()
  assert routed["variance_bucket"].isin(set(VARIANCE_MAP.values())).all()

• Missing costs are flagged, not zeroed. Assert that every cost_missing row carries routing_status == "flagged_for_review" and that no such row was posted to the ledger.

• Yield scaling is bounded. For prep_trim, assert effective_quantity <= base_quantity; for every other type, assert they are equal.

• Idempotency holds. Run the Step 6 insert twice against the same staging frame and confirm the variance_ledger row count is unchanged — the ON CONFLICT clause must swallow the replay.

• Buckets sum to the ledger. Assert routed.groupby("variance_bucket")["cost_impact"].sum() reconciles to the day’s total waste cost within a rounding cent.

A healthy run ends with zero unrouted events, every unmapped SKU visibly flagged, and an unchanged ledger count on re-run.

Gotchas and Edge Cases

IEEE-754 drift on cost-denominated arithmetic

Multiplying a float quantity by a float unit cost across thousands of daily events accumulates sub-cent error that quietly shifts variance totals over a reporting month. Keep quantities as floats but hold every currency value in decimal.Decimal or PostgreSQL NUMERIC, and quantize once at the end. The batch path in Step 5 defers the money multiply to NUMERIC in Step 6 precisely to avoid float dollars entirely.

yield_factor = 0 causing divide-by-zero or total suppression

A yield_factor of zero — from a bad import or an uninitialized default — either divides by zero if you invert it or silently zeroes the trim’s effective quantity. The Pydantic constraint gt=0.0, le=1.0 in Step 1 rejects it at the boundary. Never patch a zero to a small epsilon downstream; fix the source, because a fabricated yield distorts the yield_variance bucket for the whole location.

Regional unit aliases colliding at the base-unit boundary

lb versus lbs, oz (weight) versus fluid ounces, and locale decimal separators all defeat a naive unit map. Lowercase and strip before lookup (Step 2), keep weight and volume in separate alias tables, and raise on an unmapped alias rather than defaulting to 1.0 — a silent identity conversion mislabels a kilogram of trim as a kilogram-equivalent of the wrong scale and corrupts cross-location totals.

Unmapped SKUs silently zeroing the variance

A left join against a stale cost matrix returns NaN for any SKU not yet priced. fillna(0.0) then posts a zero-cost variance that looks like perfect efficiency — the most dangerous failure mode because it hides leakage. Route missing costs to flagged_for_review (Step 5) and exclude them from posted totals so a pricing gap surfaces as an exception, not as fake savings.

Duplicate device transmissions double-posting

Prep scales and edge terminals retransmit on flaky connectivity, so the same discard can arrive several times. Relying on a downstream dedupe is fragile; make the ledger the source of truth with a unique (event_id, location_id) constraint and ON CONFLICT DO NOTHING. Combined with UTC coercion in Step 1, this keeps a replayed event from inflating a bucket even when it lands minutes later in a different batch.

Timezone drift misaligning the reconciliation period

An event logged at 11:50 pm local can fall on either side of the close depending on how its timestamp was stored. Coerce every timestamp to UTC at ingestion (Step 1) and apply local reporting offsets only at the visualization layer. A single naive-local timestamp slipping past the boundary shifts a discard into the wrong day and manufactures a phantom variance the next morning.

Related

• Up: Waste Tracking & Routing Systems — the parent module whose ingestion and BOM-resolution architecture this router implements.
• Theoretical vs Actual Food Cost Calculation — the full variance architecture these buckets feed.
• Variance Mapping Methodologies — the layer that turns routed buckets into signed reconciliation deltas.
• Setting Dynamic Variance Thresholds — the alerting bands that decide when a routed bucket breach escalates.
• Yield Factor Calculation Frameworks — where the yield_factor applied to prep trim comes from.
• Multi-Location Cost Center Architecture — isolating each site’s variance ledger within the wider cost estate.

For deeper reference, consult the official Pydantic documentation on v2 validators and the pandas user guide on vectorized merge and column assignment.