Compute Inventory Turnover from Warehouse Shipments in Python and Publish as a Cloud API for Non‑Engineers

This article shows how to turn a single-file Python script that computes inventory turnover from warehouse shipment exports into a cloud-hosted API that non-engineers can call from a UI or other systems. We'll cover the exact input CSV schemas, the transformation and calculation logic, and a practical Functory deployment pattern that removes servers and cron jobs.

Long-tail phrases used naturally here include: "python inventory turnover API", "compute inventory turnover from warehouse CSV", and "warehouse shipments inventory turnover script".

What this function does (precise explanation)

This function expects:

• A shipments CSV (UTF-8, comma-separated) with columns: shipment_id (string), shipped_at (ISO 8601 datetime, e.g. 2024-07-01T14:12:00Z), sku (string), quantity (integer), unit_cost_usd (decimal), and warehouse_id (string).
• An inventory snapshot CSV with columns: snapshot_date (ISO date), sku, inventory_qty (integer), and unit_cost_usd (decimal). The pipeline uses snapshots closest to the requested start and end dates to compute average inventory value.

Transformations performed:

1. Filter shipments to the requested date range: start_date ≤ shipped_at <= end_date.
2. Compute COGS (cost of goods sold) for each shipment row: line_cogs = quantity * unit_cost_usd, then aggregate to per-SKU and per-warehouse totals.
3. Find beginning and ending inventory values per SKU: inventory_value = inventory_qty * unit_cost_usd for snapshots nearest the start and end dates, then compute avg_inventory = (begin_value + end_value) / 2.
4. Compute inventory turnover: turnover = total_cogs / avg_inventory. If avg_inventory==0, return null and a flag to investigate zero inventory snapshots.

Output produced:

• CSV or JSON table with columns: sku, warehouse_id (optional), period_start, period_end, total_cogs_usd, avg_inventory_usd, inventory_turnover (float), rows_considered.
• Example JSON row: {"sku": "SKU-123", "period_start": "2024-01-01", "period_end": "2024-03-31", "total_cogs_usd": 120000.0, "avg_inventory_usd": 30000.0, "inventory_turnover": 4.0}.

Real‑world scenario: editorial merch startup

Context: A content-driven media startup sells branded merchandise and wants weekly inventory health metrics in their ops dashboard. Warehouse exports arrive daily as CSVs from a 3PL with filenames like shipments_2024-07-01.csv. Inventory snapshots are exported nightly as inventory_snapshot_2024-07-01.csv.

Concrete inputs:

• shipments.csv — 1,000 rows for Q2 with columns: shipment_id, shipped_at, sku, quantity, unit_cost_usd, warehouse_id.
• inventory_snapshots.csv — 200 rows containing snapshot records for 100 SKUs at start and end of quarter.

Concrete output for SKU-123 (example):

• Total COGS for period: 120,000.00 USD
• Beginning inventory value: 25,000.00 USD
• Ending inventory value: 35,000.00 USD
• Average inventory: 30,000.00 USD
• Inventory turnover: 120,000 / 30,000 = 4.0

Example dataset and the specific problem solved

Fabricated dataset: 1,000 shipments (rows) covering 100 SKUs during 2024 Q2, shipments include timestamps and per-unit cost; inventory snapshots provide beginning and end quarter quantities and unit cost per SKU. Problem: operations wants a reliable weekly refresh of inventory turnover per SKU and per warehouse to identify slow-moving SKUs and avoid stockouts.

Why this function matters: it standardizes the COGS and inventory calculation so non-engineers can request a turnover report via a simple API or UI, avoiding spreadsheets and manual joins that are error-prone.

Step‑by‑step developer workflow

1. Place raw CSVs into a shared location (S3, Google Drive, or upload via UI).
2. Call the function with parameters: shipments_csv, inventory_snapshot_csv, start_date, end_date, optional warehouse_id or sku.
3. The function filters shipments, computes per-line COGS, aggregates to per‑SKU totals, locates nearest inventory snapshots, and computes inventory_turnover.
4. The function returns a CSV or JSON that the dashboard ingests; errors (zero inventory) are returned with diagnostic fields.
5. Schedule or trigger the function from the dashboard, an LLM agent, or another backend with the Functory API (no cron servers required).

Algorithm (high‑level)

1. Read and parse shipments CSV; convert "shipped_at" into UTC datetimes. 2. Filter shipments between start_date and end_date. 3. Compute line COGS = quantity * unit_cost_usd; aggregate to total_cogs per sku (and warehouse if requested). 4. From inventory snapshots, select record nearest to start_date (begin) and nearest to end_date (end) for each sku; compute values. 5. avg_inventory = (begin_value + end_value) / 2; if avg_inventory <= 0, mark as insufficient data. 6. inventory_turnover = total_cogs / avg_inventory; return results with diagnostics.

Python example (single-file core)

import pandas as pd
from datetime import datetime
from typing import Optional

def main(shipments_csv_path: str,
         inventory_snapshot_csv_path: str,
         start_date: str,
         end_date: str,
         sku: Optional[str] = None,
         warehouse_id: Optional[str] = None,
         output_csv: str = 'turnover_report.csv') -> str:
    # Read shipments
    ship = pd.read_csv(shipments_csv_path, parse_dates=['shipped_at'])
    inv = pd.read_csv(inventory_snapshot_csv_path, parse_dates=['snapshot_date'])

    start = pd.to_datetime(start_date)
    end = pd.to_datetime(end_date)

    # Filter shipments by date and optional sku/warehouse
    mask = (ship['shipped_at'] >= start) & (ship['shipped_at'] <= end)
    if sku:
        mask &= (ship['sku'] == sku)
    if warehouse_id:
        mask &= (ship['warehouse_id'] == warehouse_id)
    ship = ship.loc[mask].copy()

    # Compute per-line COGS
    ship['line_cogs'] = ship['quantity'] * ship['unit_cost_usd']

    # Aggregate total COGS per sku and warehouse
    agg_cols = ['sku'] + (['warehouse_id'] if warehouse_id is None else [])
    total = ship.groupby(agg_cols).agg(total_cogs_usd=('line_cogs', 'sum'), rows_considered=('shipment_id','count')).reset_index()

    # Prepare inventory values
    # find snapshot nearest to start and end for each sku
    def nearest_snapshot(df, target_date):
        df['delta'] = (df['snapshot_date'] - target_date).abs()
        return df.loc[df.groupby('sku')['delta'].idxmin()].set_index('sku')

    inv_start = nearest_snapshot(inv, start)
    inv_end = nearest_snapshot(inv, end)

    rows = []
    for _, r in total.iterrows():
        s = r['sku']
        begin = inv_start.loc[s] if s in inv_start.index else None
        endrow = inv_end.loc[s] if s in inv_end.index else None
        begin_val = float(begin['inventory_qty'] * begin['unit_cost_usd']) if begin is not None else 0.0
        end_val = float(endrow['inventory_qty'] * endrow['unit_cost_usd']) if endrow is not None else 0.0
        avg_inventory = (begin_val + end_val) / 2.0
        turnover = None
        if avg_inventory > 0:
            turnover = float(r['total_cogs_usd']) / avg_inventory
        rows.append({
            'sku': s,
            'period_start': start.strftime('%Y-%m-%d'),
            'period_end': end.strftime('%Y-%m-%d'),
            'total_cogs_usd': float(r['total_cogs_usd']),
            'avg_inventory_usd': avg_inventory,
            'inventory_turnover': turnover,
            'rows_considered': int(r['rows_considered'])
        })

    out = pd.DataFrame(rows)
    out.to_csv(output_csv, index=False)
    return output_csv

# Example call
if __name__ == '__main__':
    # This would run locally: adjust paths as needed
    print(main('shipments_q2.csv', 'inventory_snapshots_q2.csv', '2024-04-01', '2024-06-30'))

Conclusion: Converting a deterministic inventory-turnover script into a cloud API reduces manual effort, improves reproducibility, and puts the metric in the hands of non-engineers. Next steps: (1) add automated snapshot ingestion (S3->function), and (2) add anomaly detection to flag implausible averages. Try packaging your local script as a Functory function and publish a weekly endpoint for your operations dashboard.