Inventory processes: expiry, discards and returns as stock flows¶

Sales and deliveries are not the only flows that change stock. Stock can expire, be written off, or come back. In Stockcast such a flow is an inventory process: a small object the engine calls at fixed points of every period. The process only reports a quantity. The engine applies it to the stock on hand, records it in the event table and checks that the stock balance still holds.

This notebook shows:

• the built-in ShelfLife process, which tracks stock by delivery date and removes expired units at the start of a period;
• a custom outflow, written with the after_demand hook;
• a custom inflow, written with the before_demand hook, and why an inflow needs a date when shelf life is active;
• the stock balance that ties every flow together.

The data are synthetic: two perishable items, yogurt and salad (in kg), with Poisson demand.

In [1]:

import numpy as np
import pandas as pd

# Show every number with two decimals.
pd.options.display.float_format = "{:.2f}".format

import numpy as np import pandas as pd # Show every number with two decimals. pd.options.display.float_format = "{:.2f}".format

1. Setup: demand, policy and opening stock¶

Two items with 28 days of Poisson demand, drawn once with a fixed seed. Every run below replays exactly this table, so any difference between runs comes from the processes alone. Demand that cannot be served is lost (allow_backorders=False). The stock is counted on 1 March 2026, so the first demand period is 2 March.

In [2]:

origin = pd.Timestamp("2026-03-01")  # a Sunday; demand starts on Monday
skus = ["yogurt", "salad"]
n_days = 28
rates = {"yogurt": 6.0, "salad": 9.0}  # mean kg sold per day

# One fixed demand path, shared by every run in this notebook.
rng = np.random.default_rng(7)
demand = pd.DataFrame(
    [
        {
            "unique_id": sku,
            "period": day,
            "date": origin + pd.Timedelta(days=day + 1),
            "y": float(rng.poisson(rates[sku])),
        }
        for day in range(n_days)
        for sku in skus
    ]
)

origin = pd.Timestamp("2026-03-01") # a Sunday; demand starts on Monday skus = ["yogurt", "salad"] n_days = 28 rates = {"yogurt": 6.0, "salad": 9.0} # mean kg sold per day # One fixed demand path, shared by every run in this notebook. rng = np.random.default_rng(7) demand = pd.DataFrame( [ { "unique_id": sku, "period": day, "date": origin + pd.Timedelta(days=day + 1), "y": float(rng.poisson(rates[sku])), } for day in range(n_days) for sku in skus ] )

In [3]:

import matplotlib.dates as mdates
import matplotlib.pyplot as plt

YOGURT_COLOR = "#167d9a"
SALAD_COLOR = "#d58228"
PLAIN_COLOR = "#9CA3AF"
EXPIRED_COLOR = "#B279A2"
DISCARDED_COLOR = "#F58518"


def style_axis(ax):
    ax.set_ylim(0, ax.get_ylim()[1] * 1.05)
    ax.grid(axis="y", alpha=0.2)
    ax.spines[["top", "right"]].set_visible(False)


fig, ax = plt.subplots(figsize=(9.5, 3.3), layout="constrained")
for sku, color in [("yogurt", YOGURT_COLOR), ("salad", SALAD_COLOR)]:
    rows = demand.loc[demand["unique_id"].eq(sku)]
    ax.plot(
        rows["date"],
        rows["y"],
        marker="o",
        markersize=2.5,
        color=color,
        label=sku,
    )
ax.set(
    title="One shared 28-day demand path",
    xlabel="Demand date",
    ylabel="Demand (kg per day)",
)
ax.xaxis.set_major_formatter(mdates.DateFormatter("%d %b"))
style_axis(ax)
ax.legend(frameon=False, loc="lower right", ncol=2)
plt.show()

import matplotlib.dates as mdates import matplotlib.pyplot as plt YOGURT_COLOR = "#167d9a" SALAD_COLOR = "#d58228" PLAIN_COLOR = "#9CA3AF" EXPIRED_COLOR = "#B279A2" DISCARDED_COLOR = "#F58518" def style_axis(ax): ax.set_ylim(0, ax.get_ylim()[1] * 1.05) ax.grid(axis="y", alpha=0.2) ax.spines[["top", "right"]].set_visible(False) fig, ax = plt.subplots(figsize=(9.5, 3.3), layout="constrained") for sku, color in [("yogurt", YOGURT_COLOR), ("salad", SALAD_COLOR)]: rows = demand.loc[demand["unique_id"].eq(sku)] ax.plot( rows["date"], rows["y"], marker="o", markersize=2.5, color=color, label=sku, ) ax.set( title="One shared 28-day demand path", xlabel="Demand date", ylabel="Demand (kg per day)", ) ax.xaxis.set_major_formatter(mdates.DateFormatter("%d %b")) style_axis(ax) ax.legend(frameon=False, loc="lower right", ncol=2) plt.show()

The policy is an order-up-to rule with daily review (R = 1) and a one-day lead time (L = 1): each period it orders the gap between a fixed level and the inventory position (stock on hand plus stock on order). The levels, 18 kg and 26 kg, are set by hand and claim no service level (service_level=None), because this notebook is about processes, not about choosing levels. The opening state holds 12 kg and 15 kg on hand and nothing on order.

In [4]:

from stockcast.core import InventoryStateDataFrame
from stockcast.policies import OrderUpToPolicy

lead_time, review_period = 1, 1
targets = pd.DataFrame(
    {
        "unique_id": skus,
        "order_up_to": [18.0, 26.0],
        "target_end_date": origin + pd.Timedelta(days=lead_time + review_period),
    }
)
policy = OrderUpToPolicy(
    lead_time=lead_time,
    review_period=review_period,
    freq="D",
    service_level=None,  # levels set by hand: no service level is claimed
    allow_backorders=False,
    date_column="target_end_date",
)
policy.fit(
    targets,
    target_column="order_up_to",
    forecast_origin=origin,
)

empty_state = InventoryStateDataFrame(
    skus,
    max_lead_time=lead_time,
    allow_backorders=False,
)
inventory = empty_state.initialize_from_observed(
    pd.DataFrame({"unique_id": skus, "on_hand": [12.0, 15.0]}),
    on_hand_column="on_hand",
    opening_date=origin,
)

from stockcast.core import InventoryStateDataFrame from stockcast.policies import OrderUpToPolicy lead_time, review_period = 1, 1 targets = pd.DataFrame( { "unique_id": skus, "order_up_to": [18.0, 26.0], "target_end_date": origin + pd.Timedelta(days=lead_time + review_period), } ) policy = OrderUpToPolicy( lead_time=lead_time, review_period=review_period, freq="D", service_level=None, # levels set by hand: no service level is claimed allow_backorders=False, date_column="target_end_date", ) policy.fit( targets, target_column="order_up_to", forecast_origin=origin, ) empty_state = InventoryStateDataFrame( skus, max_lead_time=lead_time, allow_backorders=False, ) inventory = empty_state.initialize_from_observed( pd.DataFrame({"unique_id": skus, "on_hand": [12.0, 15.0]}), on_hand_column="on_hand", opening_date=origin, )

2. The built-in ShelfLife process¶

Expiry depends on age, so ShelfLife tracks stock by delivery: each delivery is a quantity together with the date it arrived. Deliveries that arrive during the run are tracked automatically. The stock on hand at the start must be given as deliveries too, with the opening_lots argument (Stockcast's API calls a dated delivery a lot). For each item they must add up to the stock on hand; ShelfLife checks this before the run starts.

In [5]:

opening_deliveries = pd.DataFrame(
    {
        "unique_id": ["yogurt", "yogurt", "salad"],
        "received_date": [origin - pd.Timedelta(days=1), origin, origin],
        "quantity": [5.0, 7.0, 15.0],
    }
)
opening_deliveries

opening_deliveries = pd.DataFrame( { "unique_id": ["yogurt", "yogurt", "salad"], "received_date": [origin - pd.Timedelta(days=1), origin, origin], "quantity": [5.0, 7.0, 15.0], } ) opening_deliveries

Out[5]:

	unique_id	received_date	quantity
0	yogurt	2026-02-28	5.00
1	yogurt	2026-03-01	7.00
2	salad	2026-03-01	15.00

The rules of ShelfLife(shelf_life_days=3, ...):

• a delivery received on day D can serve demand on D, D+1 and D+2;
• it expires at the start of D+3, before that day's receipts, order decision and demand, so expired units are never sold;
• demand is served from the oldest delivery first (first in, first out, FIFO).

Processes are passed to the engine with processes=[...]. We run the same simulation twice, without and with ShelfLife. The policy, demand and opening state are identical, and the engine works on its own copy of the opening state in each run.

In [6]:

from stockcast.core import ShelfLife, SimulationEngine

no_process = SimulationEngine().run(
    policy=policy,
    demand_source=demand,
    inventory=inventory,
)
shelf_life_only = SimulationEngine().run(
    policy=policy,
    demand_source=demand,
    inventory=inventory,
    processes=[ShelfLife(shelf_life_days=3, opening_lots=opening_deliveries)],
)
no_process_events = no_process.to_event_frame()
shelf_life_events = shelf_life_only.to_event_frame()
shelf_life_events.loc[
    shelf_life_events["unique_id"].eq("yogurt"),
    [
        "date",
        "starting_on_hand",
        "received_units",
        "expired_units",
        "demand",
        "fulfilled_units",
        "ending_on_hand",
    ],
].head(4)

from stockcast.core import ShelfLife, SimulationEngine no_process = SimulationEngine().run( policy=policy, demand_source=demand, inventory=inventory, ) shelf_life_only = SimulationEngine().run( policy=policy, demand_source=demand, inventory=inventory, processes=[ShelfLife(shelf_life_days=3, opening_lots=opening_deliveries)], ) no_process_events = no_process.to_event_frame() shelf_life_events = shelf_life_only.to_event_frame() shelf_life_events.loc[ shelf_life_events["unique_id"].eq("yogurt"), [ "date", "starting_on_hand", "received_units", "expired_units", "demand", "fulfilled_units", "ending_on_hand", ], ].head(4)

Out[6]:

	date	starting_on_hand	received_units	expired_units	demand	fulfilled_units	ending_on_hand
0	2026-03-02	12.00	0.00	0.00	6.00	6.00	6.00
2	2026-03-03	6.00	6.00	0.00	3.00	3.00	9.00
4	2026-03-04	9.00	6.00	3.00	4.00	4.00	8.00
6	2026-03-05	8.00	6.00	0.00	4.00	4.00	10.00

Reading the event table. FIFO explains the 4 March row. On 2 March the 6 kg of demand use the whole 28 February delivery (5 kg) and 1 kg of the 1 March delivery; on 3 March another 3 kg come from the 1 March delivery. On 4 March that delivery is three days old, and its remaining 3 kg are removed before demand. They appear in expired_units, a column of its own: expired units are neither fulfilled demand nor lost sales.

In [7]:

fig, axes = plt.subplots(
    2,
    1,
    figsize=(10, 5.8),
    sharex=True,
    layout="constrained",
)
for ax, sku, color in [
    (axes[0], "yogurt", YOGURT_COLOR),
    (axes[1], "salad", SALAD_COLOR),
]:
    without = no_process_events.loc[no_process_events["unique_id"].eq(sku)]
    with_shelf_life = shelf_life_events.loc[shelf_life_events["unique_id"].eq(sku)]
    ax.step(
        without["date"],
        without["ending_on_hand"],
        where="mid",
        color=PLAIN_COLOR,
        linewidth=2.5,
        label="No process",
    )
    ax.step(
        with_shelf_life["date"],
        with_shelf_life["ending_on_hand"],
        where="mid",
        color=color,
        linewidth=1.5,
        label="With shelf life",
    )
    expired = with_shelf_life.loc[with_shelf_life["expired_units"].gt(0)]
    ax.scatter(
        expired["date"],
        expired["ending_on_hand"],
        color=EXPIRED_COLOR,
        marker="D",
        s=48,
        zorder=3,
        clip_on=False,
        label="Stock expired that morning",
    )
    ax.set(title=f"{sku.capitalize()}: end-of-day stock", ylabel="On hand (kg)")
    style_axis(ax)
    ax.legend(
        frameon=False,
        loc="upper left",
        bbox_to_anchor=(1.01, 1.0),
        fontsize=8,
    )
axes[1].set_xlabel("Demand date")
axes[1].xaxis.set_major_formatter(mdates.DateFormatter("%d %b"))
plt.show()

fig, axes = plt.subplots( 2, 1, figsize=(10, 5.8), sharex=True, layout="constrained", ) for ax, sku, color in [ (axes[0], "yogurt", YOGURT_COLOR), (axes[1], "salad", SALAD_COLOR), ]: without = no_process_events.loc[no_process_events["unique_id"].eq(sku)] with_shelf_life = shelf_life_events.loc[shelf_life_events["unique_id"].eq(sku)] ax.step( without["date"], without["ending_on_hand"], where="mid", color=PLAIN_COLOR, linewidth=2.5, label="No process", ) ax.step( with_shelf_life["date"], with_shelf_life["ending_on_hand"], where="mid", color=color, linewidth=1.5, label="With shelf life", ) expired = with_shelf_life.loc[with_shelf_life["expired_units"].gt(0)] ax.scatter( expired["date"], expired["ending_on_hand"], color=EXPIRED_COLOR, marker="D", s=48, zorder=3, clip_on=False, label="Stock expired that morning", ) ax.set(title=f"{sku.capitalize()}: end-of-day stock", ylabel="On hand (kg)") style_axis(ax) ax.legend( frameon=False, loc="upper left", bbox_to_anchor=(1.01, 1.0), fontsize=8, ) axes[1].set_xlabel("Demand date") axes[1].xaxis.set_major_formatter(mdates.DateFormatter("%d %b")) plt.show()

What to notice:

• Expiry removes 3 kg of yogurt on 4 March and 2 kg on 15 March (diamonds), so the end-of-day stock is lower on those days.
• The difference lasts one day. Expiry happens before the order decision, so the policy sees the lower inventory position the same day and orders the missing quantity, which arrives the next day.
• Salad never reaches three days of age, so the two runs coincide.

3. A custom outflow with after_demand¶

In every period the engine runs the same sequence, and a process can act at two points of it:

1. before_demand: the start of the period, before receipts and the order decision;
2. receipts arrive, the policy decides, demand is served;
3. after_demand: after demand has been served.

A custom process subclasses InventoryProcess and declares:

• name, which labels its flows in the results;
• flows, a tuple of Flow(name, direction). The direction tells the engine what to do with a reported quantity: "outflow" subtracts it from the stock on hand, "inflow" adds it. The quantity itself is always nonnegative;
• the hooks it needs. A hook receives a context with the current date and the stock on_hand per item, and returns ProcessFlows({flow_name: quantity}), or None when there is no flow in this period.

The process never edits the stock itself: it reports a quantity, and the engine applies it, records it and checks the balance.

The example outflow is a periodic write-off: every Monday after demand, 25% of the remaining stock is discarded, rounded down to whole kilograms (think of a weekly quality inspection).

In [8]:

from stockcast.core import Flow, InventoryProcess, ProcessFlows


class MondayInspection(InventoryProcess):
    """After demand on Mondays, discard a share of the stock on hand."""

    name = "inspection"
    flows = (Flow("discarded", "outflow"),)

    def __init__(self, share):
        self.share = share

    def after_demand(self, context):
        if context.date.day_name() != "Monday":
            return None  # nothing today
        return ProcessFlows({"discarded": np.floor(self.share * context.on_hand)})

from stockcast.core import Flow, InventoryProcess, ProcessFlows class MondayInspection(InventoryProcess): """After demand on Mondays, discard a share of the stock on hand.""" name = "inspection" flows = (Flow("discarded", "outflow"),) def __init__(self, share): self.share = share def after_demand(self, context): if context.date.day_name() != "Monday": return None # nothing today return ProcessFlows({"discarded": np.floor(self.share * context.on_hand)})

Order of processes. Processes run in list order, and each sees the stock left by the ones before it. ShelfLife is also informed about the flows of every other process: when the inspection removes stock, ShelfLife takes those units from its oldest deliveries, so its record of deliveries keeps matching the stock on hand.

New event-table columns. A flow from a custom process is not expiry, so the engine records it in two general columns, process_outflow_units and process_inflow_units. They appear as soon as a custom process is part of the run. The first Monday shows the outflow on a single row.

In [9]:

with_inspection = SimulationEngine().run(
    policy=policy,
    demand_source=demand,
    inventory=inventory,
    processes=[
        ShelfLife(shelf_life_days=3, opening_lots=opening_deliveries),
        MondayInspection(share=0.25),
    ],
)
inspection_events = with_inspection.to_event_frame()

first_monday = pd.Timestamp("2026-03-02")
columns = [
    "starting_on_hand",
    "demand",
    "fulfilled_units",
    "process_outflow_units",
    "ending_on_hand",
]
both_runs = pd.concat(
    {
        "Shelf life only": shelf_life_events,
        "With inspection": inspection_events,
    },
    names=["run", "row"],
)
first_monday_yogurt = both_runs.loc[
    both_runs["unique_id"].eq("yogurt") & both_runs["date"].eq(first_monday),
    columns,
]
# The shelf-life-only run has no process outflow column: nothing was discarded.
first_monday_yogurt.fillna(0.0).droplevel("row")

with_inspection = SimulationEngine().run( policy=policy, demand_source=demand, inventory=inventory, processes=[ ShelfLife(shelf_life_days=3, opening_lots=opening_deliveries), MondayInspection(share=0.25), ], ) inspection_events = with_inspection.to_event_frame() first_monday = pd.Timestamp("2026-03-02") columns = [ "starting_on_hand", "demand", "fulfilled_units", "process_outflow_units", "ending_on_hand", ] both_runs = pd.concat( { "Shelf life only": shelf_life_events, "With inspection": inspection_events, }, names=["run", "row"], ) first_monday_yogurt = both_runs.loc[ both_runs["unique_id"].eq("yogurt") & both_runs["date"].eq(first_monday), columns, ] # The shelf-life-only run has no process outflow column: nothing was discarded. first_monday_yogurt.fillna(0.0).droplevel("row")

Out[9]:

	starting_on_hand	demand	fulfilled_units	process_outflow_units	ending_on_hand
run
Shelf life only	12.00	6.00	6.00	0.00	6.00
With inspection	12.00	6.00	6.00	1.00	5.00

Both runs start 2 March with 12 kg of yogurt and serve 6 kg. After demand the inspection reports floor(0.25 × 6) = 1 kg, so the period ends with 5 kg instead of 6. The figure follows the outflow over the whole run.

In [10]:

yogurt_shelf_life = shelf_life_events.loc[shelf_life_events["unique_id"].eq("yogurt")]
yogurt_inspected = inspection_events.loc[inspection_events["unique_id"].eq("yogurt")]

fig, axes = plt.subplots(
    2,
    1,
    figsize=(10, 5.5),
    sharex=True,
    layout="constrained",
)
axes[0].step(
    yogurt_shelf_life["date"],
    yogurt_shelf_life["ending_on_hand"],
    where="mid",
    color=PLAIN_COLOR,
    linewidth=2.5,
    label="Shelf life only",
)
axes[0].step(
    yogurt_inspected["date"],
    yogurt_inspected["ending_on_hand"],
    where="mid",
    color=YOGURT_COLOR,
    linewidth=1.5,
    label="With Monday inspection",
)
axes[0].set(
    title="Yogurt: end-of-day stock",
    ylabel="On hand (kg)",
)
axes[0].legend(frameon=False, loc="upper right")
axes[1].bar(
    yogurt_inspected["date"],
    yogurt_inspected["process_outflow_units"],
    width=0.7,
    color=DISCARDED_COLOR,
    label="Discarded after demand (Mondays)",
)
expired = yogurt_inspected.loc[yogurt_inspected["expired_units"].gt(0)]
axes[1].scatter(
    expired["date"],
    expired["expired_units"],
    color=EXPIRED_COLOR,
    marker="D",
    s=44,
    zorder=3,
    clip_on=False,
    label="Expired before demand",
)
axes[1].set(
    title="Yogurt that left the shelf unsold, with the inspection",
    ylabel="kg",
    xlabel="Demand date",
)
axes[1].legend(frameon=False, loc="upper right")
for ax in axes:
    style_axis(ax)
axes[1].xaxis.set_major_formatter(mdates.DateFormatter("%d %b"))
plt.show()

yogurt_shelf_life = shelf_life_events.loc[shelf_life_events["unique_id"].eq("yogurt")] yogurt_inspected = inspection_events.loc[inspection_events["unique_id"].eq("yogurt")] fig, axes = plt.subplots( 2, 1, figsize=(10, 5.5), sharex=True, layout="constrained", ) axes[0].step( yogurt_shelf_life["date"], yogurt_shelf_life["ending_on_hand"], where="mid", color=PLAIN_COLOR, linewidth=2.5, label="Shelf life only", ) axes[0].step( yogurt_inspected["date"], yogurt_inspected["ending_on_hand"], where="mid", color=YOGURT_COLOR, linewidth=1.5, label="With Monday inspection", ) axes[0].set( title="Yogurt: end-of-day stock", ylabel="On hand (kg)", ) axes[0].legend(frameon=False, loc="upper right") axes[1].bar( yogurt_inspected["date"], yogurt_inspected["process_outflow_units"], width=0.7, color=DISCARDED_COLOR, label="Discarded after demand (Mondays)", ) expired = yogurt_inspected.loc[yogurt_inspected["expired_units"].gt(0)] axes[1].scatter( expired["date"], expired["expired_units"], color=EXPIRED_COLOR, marker="D", s=44, zorder=3, clip_on=False, label="Expired before demand", ) axes[1].set( title="Yogurt that left the shelf unsold, with the inspection", ylabel="kg", xlabel="Demand date", ) axes[1].legend(frameon=False, loc="upper right") for ax in axes: style_axis(ax) axes[1].xaxis.set_major_formatter(mdates.DateFormatter("%d %b")) plt.show()

What to notice:

• The outflow happens on three Mondays (orange bars), 1 kg each time. It is applied after demand, so the policy sees it at the next decision, and the replacement order arrives one day after that: the stock stays 1 kg lower for two days.
• Expiry happens on the same dates as before but removes less, 2 kg instead of 3 kg on 4 March. The inspection had already taken units from the oldest delivery, so fewer units reached the expiry age. This is why the blue line is above the grey one on 4 March.

The totals per item over the run:

In [11]:

def outflow_totals(events):
    """Kilograms fulfilled, expired and discarded over the run, per item."""
    totals = pd.DataFrame(
        {
            "Fulfilled": events.groupby("unique_id")["fulfilled_units"].sum(),
            "Expired": events.groupby("unique_id")["expired_units"].sum(),
        }
    )
    totals["Discarded"] = 0.0
    if "process_outflow_units" in events:
        totals["Discarded"] = events.groupby("unique_id")["process_outflow_units"].sum()
    return totals


pd.concat(
    {
        "Shelf life only": outflow_totals(shelf_life_events),
        "With inspection": outflow_totals(inspection_events),
    },
    names=["run", "unique_id"],
)

def outflow_totals(events): """Kilograms fulfilled, expired and discarded over the run, per item.""" totals = pd.DataFrame( { "Fulfilled": events.groupby("unique_id")["fulfilled_units"].sum(), "Expired": events.groupby("unique_id")["expired_units"].sum(), } ) totals["Discarded"] = 0.0 if "process_outflow_units" in events: totals["Discarded"] = events.groupby("unique_id")["process_outflow_units"].sum() return totals pd.concat( { "Shelf life only": outflow_totals(shelf_life_events), "With inspection": outflow_totals(inspection_events), }, names=["run", "unique_id"], )

Out[11]:

		Fulfilled	Expired	Discarded
run	unique_id
Shelf life only	salad	273.00	0.00	0.00
	yogurt	168.00	5.00	0.00
With inspection	salad	273.00	0.00	6.00
	yogurt	168.00	4.00	3.00

With shelf life alone, 5 kg of yogurt expire. With the inspection, 3 kg are discarded and 4 kg expire. Salad never expires but loses 6 kg to the inspection. Fulfilled demand is identical in both runs, 168 kg and 273 kg: the order-up-to policy replaces every unit that leaves, whichever flow removed it.

4. A custom inflow with before_demand¶

An inflow adds stock. In the example, 3 kg of salad come back to stock (returned goods, say) at the start of 11 March. The process uses the before_demand hook, so the units are on hand before that period's receipts and order decision, and the policy already counts them.

With ShelfLife active there is one more requirement. Every unit on hand belongs to a dated delivery, so an inflow must state how old its units are: the process passes their date with received_dates={flow_name: date} in ProcessFlows. Stockcast never assumes an age, and an inflow without a date stops the run. Here the returned units are one day old.

In [12]:

class CustomerReturns(InventoryProcess):
    """Scheduled inflows, added before receipts and the order decision."""

    name = "returns"
    flows = (Flow("returned", "inflow"),)

    def __init__(self, schedule, days_old):
        self.schedule = schedule  # {date: {item: kg}}
        self.days_old = days_old  # age of the inflowing units

    def before_demand(self, context):
        returned = self.schedule.get(context.date)
        if returned is None:
            return None  # nothing today
        delivery_date = context.date - pd.Timedelta(days=self.days_old)
        return ProcessFlows(
            {"returned": returned},
            received_dates={"returned": delivery_date},
        )


return_day = pd.Timestamp("2026-03-11")
with_returns = SimulationEngine().run(
    policy=policy,
    demand_source=demand,
    inventory=inventory,
    processes=[
        ShelfLife(shelf_life_days=3, opening_lots=opening_deliveries),
        MondayInspection(share=0.25),
        CustomerReturns({return_day: {"salad": 3.0}}, days_old=1),
    ],
)
returns_events = with_returns.to_event_frame()

class CustomerReturns(InventoryProcess): """Scheduled inflows, added before receipts and the order decision.""" name = "returns" flows = (Flow("returned", "inflow"),) def __init__(self, schedule, days_old): self.schedule = schedule # {date: {item: kg}} self.days_old = days_old # age of the inflowing units def before_demand(self, context): returned = self.schedule.get(context.date) if returned is None: return None # nothing today delivery_date = context.date - pd.Timedelta(days=self.days_old) return ProcessFlows( {"returned": returned}, received_dates={"returned": delivery_date}, ) return_day = pd.Timestamp("2026-03-11") with_returns = SimulationEngine().run( policy=policy, demand_source=demand, inventory=inventory, processes=[ ShelfLife(shelf_life_days=3, opening_lots=opening_deliveries), MondayInspection(share=0.25), CustomerReturns({return_day: {"salad": 3.0}}, days_old=1), ], ) returns_events = with_returns.to_event_frame()

The table compares salad on 11 and 12 March, without and with the inflow.

In [13]:

both_runs = pd.concat(
    {
        "Without the return": inspection_events,
        "With the return": returns_events,
    },
    names=["run", "row"],
)
salad_days = both_runs.loc[
    both_runs["unique_id"].eq("salad")
    & both_runs["date"].between(return_day, return_day + pd.Timedelta(days=1)),
    [
        "date",
        "process_inflow_units",
        "received_units",
        "order_quantity",
        "ending_on_hand",
    ],
]
salad_days.fillna(0.0).droplevel("row")

both_runs = pd.concat( { "Without the return": inspection_events, "With the return": returns_events, }, names=["run", "row"], ) salad_days = both_runs.loc[ both_runs["unique_id"].eq("salad") & both_runs["date"].between(return_day, return_day + pd.Timedelta(days=1)), [ "date", "process_inflow_units", "received_units", "order_quantity", "ending_on_hand", ], ] salad_days.fillna(0.0).droplevel("row")

Out[13]:

	date	process_inflow_units	received_units	order_quantity	ending_on_hand
run
Without the return	2026-03-11	0.00	12.00	13.00	1.00
Without the return	2026-03-12	0.00	13.00	12.00	7.00
With the return	2026-03-11	3.00	12.00	10.00	4.00
With the return	2026-03-12	0.00	10.00	12.00	7.00

On 11 March process_inflow_units records the 3 kg. The policy sees an inventory position 3 kg higher and orders 10 kg instead of 13 kg. On 12 March that smaller order is received, and both runs hold the same stock again. An inflow before the decision is offset by the next order: the policy's level, not the inflow, sets the stock.

5. The stock balance¶

Every row of the event table satisfies the same balance, and the engine checks it in every period:

ending on hand = starting on hand + received + process inflow
                 − fulfilled − expired − process outflow

Runs with backorders or stock adjustments add their own terms. Summed over the run, the balance gives a waterfall per item: the opening stock, each flow, and the ending stock. The figure uses the run with all three processes.

In [14]:

def flow_totals(events):
    """Opening stock and every flow over the run, in kg (outflows negative)."""
    return {
        "Opening stock": events["starting_on_hand"].iloc[0],
        "Received": events["received_units"].sum(),
        "Returned": events["process_inflow_units"].sum(),
        "Fulfilled": -events["fulfilled_units"].sum(),
        "Expired": -events["expired_units"].sum(),
        "Discarded": -events["process_outflow_units"].sum(),
    }


FLOW_COLORS = {
    "Opening stock": "#4C78A8",
    "Received": "#54A24B",
    "Returned": "#72B7B2",
    "Fulfilled": PLAIN_COLOR,
    "Expired": EXPIRED_COLOR,
    "Discarded": DISCARDED_COLOR,
    "Ending stock": "#4C78A8",
}

fig, axes = plt.subplots(1, 2, figsize=(12, 3.8), layout="constrained")
for ax, sku in zip(axes, skus):
    events = returns_events.loc[returns_events["unique_id"].eq(sku)]
    totals = flow_totals(events)
    ending = events["ending_on_hand"].iloc[-1]
    # The flows add up to the ending stock, item by item.
    assert np.isclose(sum(totals.values()), ending)
    bars = {**totals, "Ending stock": ending}
    running = 0.0
    for position, (name, amount) in enumerate(bars.items()):
        if name == "Ending stock":
            bottom, label = 0.0, f"{ending:.0f}"
        else:
            bottom = running if amount >= 0 else running + amount
            label = "0" if amount == 0 else f"{amount:+.0f}"
            running += amount
        ax.bar(position, abs(amount), bottom=bottom, color=FLOW_COLORS[name])
        ax.text(
            position,
            bottom + abs(amount),
            label,
            ha="center",
            va="bottom",
            fontsize=8,
        )
    ax.set_xticks(range(len(bars)), list(bars), rotation=30, ha="right")
    ax.set(title=f"{sku.capitalize()} over 28 days", ylabel="kg")
    style_axis(ax)
plt.show()

def flow_totals(events): """Opening stock and every flow over the run, in kg (outflows negative).""" return { "Opening stock": events["starting_on_hand"].iloc[0], "Received": events["received_units"].sum(), "Returned": events["process_inflow_units"].sum(), "Fulfilled": -events["fulfilled_units"].sum(), "Expired": -events["expired_units"].sum(), "Discarded": -events["process_outflow_units"].sum(), } FLOW_COLORS = { "Opening stock": "#4C78A8", "Received": "#54A24B", "Returned": "#72B7B2", "Fulfilled": PLAIN_COLOR, "Expired": EXPIRED_COLOR, "Discarded": DISCARDED_COLOR, "Ending stock": "#4C78A8", } fig, axes = plt.subplots(1, 2, figsize=(12, 3.8), layout="constrained") for ax, sku in zip(axes, skus): events = returns_events.loc[returns_events["unique_id"].eq(sku)] totals = flow_totals(events) ending = events["ending_on_hand"].iloc[-1] # The flows add up to the ending stock, item by item. assert np.isclose(sum(totals.values()), ending) bars = {**totals, "Ending stock": ending} running = 0.0 for position, (name, amount) in enumerate(bars.items()): if name == "Ending stock": bottom, label = 0.0, f"{ending:.0f}" else: bottom = running if amount >= 0 else running + amount label = "0" if amount == 0 else f"{amount:+.0f}" running += amount ax.bar(position, abs(amount), bottom=bottom, color=FLOW_COLORS[name]) ax.text( position, bottom + abs(amount), label, ha="center", va="bottom", fontsize=8, ) ax.set_xticks(range(len(bars)), list(bars), rotation=30, ha="right") ax.set(title=f"{sku.capitalize()} over 28 days", ylabel="kg") style_axis(ax) plt.show()

For yogurt: 12 kg opening + 168 kg received − 168 kg fulfilled − 4 kg expired − 3 kg discarded = 5 kg. For salad: 15 kg opening + 263 kg received + 3 kg returned − 273 kg fulfilled − 6 kg discarded = 2 kg. The code asserts these totals for each item.

with_returns.to_process_flow_frame() lists the same flows by process, one row per item, period and named flow. The inventory processes guide describes every hook, including on_receipt and the hooks ShelfLife uses to keep its record of deliveries in line with other processes.