Write your own ordering rule: code it, tune it on costs, extend it¶

This tutorial implements a small reorder rule as a BasePolicy subclass, tests its decisions, tunes one parameter against declared costs, and makes one controlled change to the rule. The policy requests orders; Stockcast owns demand processing, lead-time transitions, inventory changes, and the event table.

The starting rule is the familiar (s,Q) reorder point policy: when inventory position is at or below threshold s, request a fixed quantity Q. Inventory position is on-hand + pipeline − backorders, so stock already on its way matters.

Stockcast's built-in ReorderPointPolicy(policy_type='sQ') can express this rule. We write it ourselves here to show the extension contract and make its choices inspectable. The threshold and quantity are declared inputs; Stockcast does not infer them from demand or manufacture a forecast target. Later, we will tune Q against explicit costs and extend the request formula. For ordinary (s,Q) use, the built-in policy is the simpler choice.

In [1]:

import pandas as pd

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

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

1. Define the policy contract¶

A custom policy inherits operating settings from BasePolicy: lead time, decision schedule, shortage mode, and optional service probability. Here __init__ validates the declared parameters, and fit() marks the policy ready; there is no forecast table to estimate. The engine calls predict() only at scheduled opportunities.

predict() reads inventory state and returns an OrderDecision. That object carries the requested quantity, the position used to decide, and order/delivery periods. It does not place the order or edit on-hand stock; the engine handles acceptance, timing, and physical state.

In [2]:

import math

from stockcast.core import BasePolicy, OrderDecision


class ReorderWhenBelowPolicy(BasePolicy):
    """Request a fixed quantity when inventory position reaches a threshold."""

    def __init__(
        self,
        reorder_threshold,
        order_quantity,
        parameter_source,
        **kwargs,
    ):
        # BasePolicy validates lead time, review schedule, and shortage mode.
        super().__init__(**kwargs)
        # Fail early before an invalid operating input reaches the engine.
        self.reorder_threshold = float(reorder_threshold)
        self.order_quantity = float(order_quantity)
        if not math.isfinite(self.reorder_threshold) or self.reorder_threshold < 0:
            raise ValueError("reorder_threshold must be finite and non-negative")
        if not math.isfinite(self.order_quantity) or self.order_quantity <= 0:
            raise ValueError("order_quantity must be finite and positive")
        if not isinstance(parameter_source, str) or not parameter_source.strip():
            raise ValueError("parameter_source must be a non-empty string")
        self.parameter_source = parameter_source.strip()

    def fit(self, forecast_df=None, **kwargs):
        # The parameters were checked in __init__; no forecast is fitted here.
        # The engine rejects a policy until this flag is set.
        self.fitted_ = True
        return self

    def predict(self, inventory_state_df, current_period=0, **kwargs):
        if not self.fitted_:
            raise ValueError("policy must be fitted before prediction")

        # Use on-hand + in-transit - backorders, so outstanding stock counts.
        # Work on a new table; predict() must leave the supplied state untouched.
        positions = inventory_state_df.inventory_position()
        sku_column = inventory_state_df.sku_column
        decision = positions[[sku_column, "inventory_position"]].copy()
        # Each SKU requests Q at or below s and zero above s.
        decision["order_quantity"] = (
            decision["inventory_position"].le(self.reorder_threshold).astype(float)
            * self.order_quantity
        )
        # This is a fixed-quantity rule, so an order-up-to target is not applicable.
        decision["target_level"] = math.nan
        decision["reorder_point"] = self.reorder_threshold
        # The engine interprets these zero-based demand-period coordinates.
        decision["order_period"] = current_period
        decision["expected_delivery_period"] = current_period + self.lead_time
        # Return a request, not a mutation of stock or pipeline.
        return OrderDecision(
            decision[
                [
                    sku_column,
                    "order_quantity",
                    "target_level",
                    "inventory_position",
                    "reorder_point",
                    "order_period",
                    "expected_delivery_period",
                ]
            ],
            sku_column=sku_column,
            lead_time=self.lead_time,
        )

import math from stockcast.core import BasePolicy, OrderDecision class ReorderWhenBelowPolicy(BasePolicy): """Request a fixed quantity when inventory position reaches a threshold.""" def __init__( self, reorder_threshold, order_quantity, parameter_source, **kwargs, ): # BasePolicy validates lead time, review schedule, and shortage mode. super().__init__(**kwargs) # Fail early before an invalid operating input reaches the engine. self.reorder_threshold = float(reorder_threshold) self.order_quantity = float(order_quantity) if not math.isfinite(self.reorder_threshold) or self.reorder_threshold < 0: raise ValueError("reorder_threshold must be finite and non-negative") if not math.isfinite(self.order_quantity) or self.order_quantity <= 0: raise ValueError("order_quantity must be finite and positive") if not isinstance(parameter_source, str) or not parameter_source.strip(): raise ValueError("parameter_source must be a non-empty string") self.parameter_source = parameter_source.strip() def fit(self, forecast_df=None, **kwargs): # The parameters were checked in __init__; no forecast is fitted here. # The engine rejects a policy until this flag is set. self.fitted_ = True return self def predict(self, inventory_state_df, current_period=0, **kwargs): if not self.fitted_: raise ValueError("policy must be fitted before prediction") # Use on-hand + in-transit - backorders, so outstanding stock counts. # Work on a new table; predict() must leave the supplied state untouched. positions = inventory_state_df.inventory_position() sku_column = inventory_state_df.sku_column decision = positions[[sku_column, "inventory_position"]].copy() # Each SKU requests Q at or below s and zero above s. decision["order_quantity"] = ( decision["inventory_position"].le(self.reorder_threshold).astype(float) * self.order_quantity ) # This is a fixed-quantity rule, so an order-up-to target is not applicable. decision["target_level"] = math.nan decision["reorder_point"] = self.reorder_threshold # The engine interprets these zero-based demand-period coordinates. decision["order_period"] = current_period decision["expected_delivery_period"] = current_period + self.lead_time # Return a request, not a mutation of stock or pipeline. return OrderDecision( decision[ [ sku_column, "order_quantity", "target_level", "inventory_position", "reorder_point", "order_period", "expected_delivery_period", ] ], sku_column=sku_column, lead_time=self.lead_time, )

2. Declare the scenario¶

review_period=1 creates a daily PeriodicSchedule, including an opportunity before the first demand row. lead_time=2 makes accepted orders usable before demand two periods later. With allow_backorders=False, unmet demand becomes lost sales.

Opening stock equals the threshold, making the first positive decision easy to inspect. The short repeating demand path then lets the threshold be crossed again after receipts and consumption. These values make the mechanics visible; they are not calibrated recommendations.

In [3]:

sku = "widget_A"
opening_date = pd.Timestamp("2026-03-01")
period_frequency = "D"
lead_time = 2
review_period = 1
reorder_threshold = 6.0
fixed_order_quantity = 10.0

sku = "widget_A" opening_date = pd.Timestamp("2026-03-01") period_frequency = "D" lead_time = 2 review_period = 1 reorder_threshold = 6.0 fixed_order_quantity = 10.0

In [4]:

# No forecast quantile is used, so there is no target probability.
policy = ReorderWhenBelowPolicy(
    reorder_threshold=reorder_threshold,
    order_quantity=fixed_order_quantity,
    parameter_source="declared_scenario_input",
    lead_time=lead_time,
    review_period=review_period,
    service_level=None,
    allow_backorders=False,
)
# fit() marks this declared-parameter policy ready for the engine.
policy.fit()

pd.DataFrame(
    [
        {
            "policy": policy.policy_name,
            "fitted": policy.fitted_,
            "reorder_threshold": reorder_threshold,
            "order_quantity": fixed_order_quantity,
            "lead_time": lead_time,
            "review_period": review_period,
        }
    ]
)

# No forecast quantile is used, so there is no target probability. policy = ReorderWhenBelowPolicy( reorder_threshold=reorder_threshold, order_quantity=fixed_order_quantity, parameter_source="declared_scenario_input", lead_time=lead_time, review_period=review_period, service_level=None, allow_backorders=False, ) # fit() marks this declared-parameter policy ready for the engine. policy.fit() pd.DataFrame( [ { "policy": policy.policy_name, "fitted": policy.fitted_, "reorder_threshold": reorder_threshold, "order_quantity": fixed_order_quantity, "lead_time": lead_time, "review_period": review_period, } ] )

Out[4]:

	policy	fitted	reorder_threshold	order_quantity	lead_time	review_period
0	ReorderWhenBelowPolicy	True	6.00	10.00	2	1

In [5]:

# Opening stock equals the threshold, making the first decision easy to inspect.
opening_stock = pd.DataFrame({"unique_id": [sku], "observed_on_hand": [6.0]})
opening_stock

# Opening stock equals the threshold, making the first decision easy to inspect. opening_stock = pd.DataFrame({"unique_id": [sku], "observed_on_hand": [6.0]}) opening_stock

Out[5]:

	unique_id	observed_on_hand
0	widget_A	6.00

In [6]:

from stockcast.core import InventoryStateDataFrame


def opening_state(date, stock):
    """Physical stock at `date`; no stock or pipeline is inferred."""
    # max_lead_time allocates the two delivery slots the engine may need.
    empty_state = InventoryStateDataFrame(
        [sku],
        max_lead_time=lead_time,
        allow_backorders=False,
    )
    return empty_state.initialize_from_observed(
        stock,
        on_hand_column="observed_on_hand",
        opening_date=date,
    )


# The engine copies this caller-owned starting state for a run.
opening_inventory = opening_state(opening_date, opening_stock)
opening_inventory

from stockcast.core import InventoryStateDataFrame def opening_state(date, stock): """Physical stock at `date`; no stock or pipeline is inferred.""" # max_lead_time allocates the two delivery slots the engine may need. empty_state = InventoryStateDataFrame( [sku], max_lead_time=lead_time, allow_backorders=False, ) return empty_state.initialize_from_observed( stock, on_hand_column="observed_on_hand", opening_date=date, ) # The engine copies this caller-owned starting state for a run. opening_inventory = opening_state(opening_date, opening_stock) opening_inventory

Out[6]:

InventoryStateDataFrame(n_skus=1, total_on_hand=6, total_backorders=0, has_stockout=False, has_backorder=False)

In [7]:

demand_values = [2.0, 3.0, 4.0, 2.0, 3.0, 4.0, 2.0, 3.0, 4.0, 2.0]


def dated_demand(values, first_date):
    """The engine needs one dated row per SKU and demand period."""
    return pd.DataFrame(
        {
            "unique_id": sku,
            "period": range(len(values)),
            "date": pd.date_range(
                first_date, periods=len(values), freq=period_frequency
            ),
            "y": values,
        }
    )


# Demand period 0 is the day after opening; dates advance at frequency D.
demand = dated_demand(demand_values, opening_date + pd.Timedelta(days=1))
assert demand[["unique_id", "period"]].duplicated().sum() == 0

demand_values = [2.0, 3.0, 4.0, 2.0, 3.0, 4.0, 2.0, 3.0, 4.0, 2.0] def dated_demand(values, first_date): """The engine needs one dated row per SKU and demand period.""" return pd.DataFrame( { "unique_id": sku, "period": range(len(values)), "date": pd.date_range( first_date, periods=len(values), freq=period_frequency ), "y": values, } ) # Demand period 0 is the day after opening; dates advance at frequency D. demand = dated_demand(demand_values, opening_date + pd.Timedelta(days=1)) assert demand[["unique_id", "period"]].duplicated().sum() == 0

In [8]:

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

DEMAND_COLOR = "#4C78A8"
THRESHOLD_COLOR = "#E45756"
ORDER_COLOR = "#F58518"
RECEIPT_COLOR = "#54A24B"
STOCK_COLOR = "#2C7FB8"
PIPELINE_COLOR = "#7F7F7F"


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)


def format_dates(ax, dates):
    ax.set_xticks(dates)
    ax.xaxis.set_major_formatter(mdates.DateFormatter("%b %d"))
    ax.tick_params(axis="x", rotation=30)


# The short repeating path lets the threshold trigger several decisions.
fig, ax = plt.subplots(figsize=(9, 3.2), layout="constrained")
ax.plot(
    demand["date"],
    demand["y"],
    marker="o",
    linewidth=1.8,
    color=DEMAND_COLOR,
    label="Declared demand",
)
ax.set(title="Deterministic demand scenario", xlabel="Date", ylabel="Demand (units)")
format_dates(ax, list(demand["date"][::2]))
ax.legend(frameon=False)
style_axis(ax)
plt.show()

import matplotlib.dates as mdates import matplotlib.pyplot as plt DEMAND_COLOR = "#4C78A8" THRESHOLD_COLOR = "#E45756" ORDER_COLOR = "#F58518" RECEIPT_COLOR = "#54A24B" STOCK_COLOR = "#2C7FB8" PIPELINE_COLOR = "#7F7F7F" 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) def format_dates(ax, dates): ax.set_xticks(dates) ax.xaxis.set_major_formatter(mdates.DateFormatter("%b %d")) ax.tick_params(axis="x", rotation=30) # The short repeating path lets the threshold trigger several decisions. fig, ax = plt.subplots(figsize=(9, 3.2), layout="constrained") ax.plot( demand["date"], demand["y"], marker="o", linewidth=1.8, color=DEMAND_COLOR, label="Declared demand", ) ax.set(title="Deterministic demand scenario", xlabel="Date", ylabel="Demand (units)") format_dates(ax, list(demand["date"][::2])) ax.legend(frameon=False) style_axis(ax) plt.show()

3. Inspect one decision¶

Opening inventory position equals the threshold, so the policy requests 10 units. The decision table keeps the state used by the rule (inventory_position), its trigger (reorder_point), the requested quantity, and delivery timing together. predict() only reads the state, and we check that the inventory table is identical before and after the call.

The expected delivery period is the order period plus the two-day lead time. target_level is intentionally empty because this policy requests a fixed quantity rather than ordering up to a target.

In [9]:

state_before_prediction = opening_inventory.get_dataframe()
opening_decision = policy.predict(opening_inventory, current_period=0)
assert state_before_prediction.equals(opening_inventory.get_dataframe())

opening_decision.get_dataframe()

state_before_prediction = opening_inventory.get_dataframe() opening_decision = policy.predict(opening_inventory, current_period=0) assert state_before_prediction.equals(opening_inventory.get_dataframe()) opening_decision.get_dataframe()

Out[9]:

	unique_id	order_quantity	target_level	inventory_position	reorder_point	order_period	expected_delivery_period
0	widget_A	10.00	NaN	6.00	6.00	0	2

4. Let the engine run the policy¶

SimulationEngine.run copies the caller's inventory and policy. At each daily opportunity it uses the fitted rule before that day's demand; accepted orders enter the pipeline and arrive according to lead_time. The result carries both the final state and an event table with one row per day. We pass the complete dated demand path, so no demand is generated and the replay is reproducible. All ten periods are scored.

In [10]:

from stockcast.core import SimulationEngine

result = SimulationEngine().run(
    policy=policy,
    demand_source=demand,
    inventory=opening_inventory,
    freq=period_frequency,  # a custom policy records no frequency
)

from stockcast.core import SimulationEngine result = SimulationEngine().run( policy=policy, demand_source=demand, inventory=opening_inventory, freq=period_frequency, # a custom policy records no frequency )

In [11]:

# The engine owns this state; the caller-owned `opening_inventory` above is unchanged.
result.inventory

# The engine owns this state; the caller-owned `opening_inventory` above is unchanged. result.inventory

Out[11]:

InventoryStateDataFrame(n_skus=1, total_on_hand=0, total_backorders=0, has_stockout=True, has_backorder=False)

The period-0 row shows that the 10-unit order enters the pipeline at once, while on-hand stock falls from 6 to 4 by that day's demand. Physical stock gains the 10 units only when the order is received two periods later.

result.to_event_frame(window="all") exposes the event table. Its decision_inventory_position is the state the policy read before that day's order. With no constraints or callbacks here, the accepted order_quantity equals the request. Receipts, lost sales, and ending stock come from the engine, so we can check the rule and its physical consequences from one table.

In [12]:

events = result.to_event_frame(window="all")
period_events = events.loc[events["event_type"].eq("period")].copy()
# With no callbacks or constraints, the accepted quantity is the policy request.
period_events["pre_order_inventory_position"] = period_events[
    "decision_inventory_position"
]
period_events["rule_triggered"] = (
    period_events["pre_order_inventory_position"] <= reorder_threshold
)

events = result.to_event_frame(window="all") period_events = events.loc[events["event_type"].eq("period")].copy() # With no callbacks or constraints, the accepted quantity is the policy request. period_events["pre_order_inventory_position"] = period_events[ "decision_inventory_position" ] period_events["rule_triggered"] = ( period_events["pre_order_inventory_position"] <= reorder_threshold )

In [13]:

period_events[
    [
        "date",
        "demand",
        "pre_order_inventory_position",
        "rule_triggered",
        "order_quantity",
        "received_units",
        "ending_on_hand",
        "on_order_end",
        "shortage_units",
    ]
].reset_index(drop=True)

period_events[ [ "date", "demand", "pre_order_inventory_position", "rule_triggered", "order_quantity", "received_units", "ending_on_hand", "on_order_end", "shortage_units", ] ].reset_index(drop=True)

Out[13]:

	date	demand	pre_order_inventory_position	rule_triggered	order_quantity	received_units	ending_on_hand	on_order_end	shortage_units
0	2026-03-02	2.00	6.00	True	10.00	0.00	4.00	10.00	0.00
1	2026-03-03	3.00	14.00	False	0.00	0.00	1.00	10.00	0.00
2	2026-03-04	4.00	11.00	False	0.00	10.00	7.00	0.00	0.00
3	2026-03-05	2.00	7.00	False	0.00	0.00	5.00	0.00	0.00
4	2026-03-06	3.00	5.00	True	10.00	0.00	2.00	10.00	0.00
5	2026-03-07	4.00	12.00	False	0.00	0.00	0.00	10.00	2.00
6	2026-03-08	2.00	10.00	False	0.00	10.00	8.00	0.00	0.00
7	2026-03-09	3.00	8.00	False	0.00	0.00	5.00	0.00	0.00
8	2026-03-10	4.00	5.00	True	10.00	0.00	1.00	10.00	0.00
9	2026-03-11	2.00	11.00	False	0.00	0.00	0.00	10.00	1.00

In [14]:

from stockcast.evaluation import (
    InventoryEvaluator,
    avg_on_hand,
    demand_period_service_level,
    fill_rate,
    order_event_count,
    order_units,
)

# evaluate() computes named metrics from the event table, not policy internals.
evaluator = InventoryEvaluator()
evaluator.fit(simulation_result=result, window="scoring")
evaluator.evaluate(
    metrics=[
        fill_rate,
        demand_period_service_level,
        avg_on_hand,
        order_units,
        order_event_count,
    ],
    groupby=[],  # Pool this one-SKU scenario into one result row.
)

from stockcast.evaluation import ( InventoryEvaluator, avg_on_hand, demand_period_service_level, fill_rate, order_event_count, order_units, ) # evaluate() computes named metrics from the event table, not policy internals. evaluator = InventoryEvaluator() evaluator.fit(simulation_result=result, window="scoring") evaluator.evaluate( metrics=[ fill_rate, demand_period_service_level, avg_on_hand, order_units, order_event_count, ], groupby=[], # Pool this one-SKU scenario into one result row. )

Out[14]:

	fill_rate	demand_period_service_level	avg_on_hand	order_units	order_event_count
0	0.90	0.80	3.30	30.00	3

The rule fired on three of the ten days, so the engine accepted three 10-unit orders, 30 units in all. Fill rate is 0.90, because 26 of the 29 demanded units were served from stock in the period they occurred. Demand-period service level is 0.80, because 2 of the 10 days had a shortage. The run averaged 3.30 units on hand. The engine's run manifest records the shared policy settings, so keep the custom s and Q and their declared source with your scenario inputs.

5. See the rule and its inventory consequences¶

The first panel checks the decision rule: red markers occur only where pre-order inventory position reaches or falls below the dashed threshold. The second panel separates requests from receipts, making the two-period delay visible. The third distinguishes usable on-hand stock from units still in transit.

The first order is placed before the first demand. It enters the pipeline and arrives before demand two periods later.

In [15]:

fig, axes = plt.subplots(3, 1, figsize=(11, 7.5), sharex=True, layout="constrained")

# Step lines suit values that remain in effect over discrete inventory periods.
axes[0].step(
    period_events["date"],
    period_events["pre_order_inventory_position"],
    where="mid",
    linewidth=2,
    color=DEMAND_COLOR,
    label="Pre-order inventory position",
)
axes[0].axhline(
    reorder_threshold,
    color=THRESHOLD_COLOR,
    linestyle="--",
    label=f"Reorder threshold ({reorder_threshold:g})",
)
triggered = period_events.loc[period_events["rule_triggered"]]
axes[0].scatter(
    triggered["date"],
    triggered["pre_order_inventory_position"],
    color=THRESHOLD_COLOR,
    zorder=3,
    label="Rule triggered",
)
axes[0].set(title="Custom decision rule", ylabel="Units")

axes[1].step(
    period_events["date"],
    period_events["order_quantity"],
    where="mid",
    linewidth=2,
    color=ORDER_COLOR,
    label="Orders placed",
)
axes[1].step(
    period_events["date"],
    period_events["received_units"],
    where="mid",
    linewidth=2,
    linestyle="--",
    color=RECEIPT_COLOR,
    label="Receipts",
)
axes[1].set(title="Engine-owned replenishment timing", ylabel="Units")

axes[2].step(
    period_events["date"],
    period_events["ending_on_hand"],
    where="mid",
    linewidth=2,
    color=STOCK_COLOR,
    label="End-of-day on hand",
)
axes[2].step(
    period_events["date"],
    period_events["on_order_end"],
    where="mid",
    linewidth=2,
    color=PIPELINE_COLOR,
    label="Unreceived orders",
)
axes[2].set(title="Physical stock and pipeline", xlabel="Date", ylabel="Units")

for ax in axes:
    ax.legend(frameon=False, loc="upper left", bbox_to_anchor=(1.01, 1.0))
    style_axis(ax)
format_dates(axes[2], list(period_events["date"][::2]))
fig.suptitle("A custom rule inside the standard simulation flow", fontsize=13)
plt.show()

fig, axes = plt.subplots(3, 1, figsize=(11, 7.5), sharex=True, layout="constrained") # Step lines suit values that remain in effect over discrete inventory periods. axes[0].step( period_events["date"], period_events["pre_order_inventory_position"], where="mid", linewidth=2, color=DEMAND_COLOR, label="Pre-order inventory position", ) axes[0].axhline( reorder_threshold, color=THRESHOLD_COLOR, linestyle="--", label=f"Reorder threshold ({reorder_threshold:g})", ) triggered = period_events.loc[period_events["rule_triggered"]] axes[0].scatter( triggered["date"], triggered["pre_order_inventory_position"], color=THRESHOLD_COLOR, zorder=3, label="Rule triggered", ) axes[0].set(title="Custom decision rule", ylabel="Units") axes[1].step( period_events["date"], period_events["order_quantity"], where="mid", linewidth=2, color=ORDER_COLOR, label="Orders placed", ) axes[1].step( period_events["date"], period_events["received_units"], where="mid", linewidth=2, linestyle="--", color=RECEIPT_COLOR, label="Receipts", ) axes[1].set(title="Engine-owned replenishment timing", ylabel="Units") axes[2].step( period_events["date"], period_events["ending_on_hand"], where="mid", linewidth=2, color=STOCK_COLOR, label="End-of-day on hand", ) axes[2].step( period_events["date"], period_events["on_order_end"], where="mid", linewidth=2, color=PIPELINE_COLOR, label="Unreceived orders", ) axes[2].set(title="Physical stock and pipeline", xlabel="Date", ylabel="Units") for ax in axes: ax.legend(frameon=False, loc="upper left", bbox_to_anchor=(1.01, 1.0)) style_axis(ax) format_dates(axes[2], list(period_events["date"][::2])) fig.suptitle("A custom rule inside the standard simulation flow", fontsize=13) plt.show()

6. Tune the fixed quantity against declared costs¶

Q trades off three things: how often we order, how much stock we hold, and how much demand goes unserved. Ordering cost alone is not the objective, since placing no orders would minimize it. For this exercise total_cost adds three illustrative charges: 5.00 for every order placed (a fixed cost per positive SKU order line, not a shared warehouse fee), 0.50 for every unit held at the end of a day, and 8.00 for every unit of lost sales. They are business assumptions, not estimated facts. The evaluator applies them; the policy does not optimize them itself. We keep s, lead time, opening stock, and shortage mode fixed.

We compare a small declared grid of Q values on a 28-day calibration path, the demand we are allowed to tune on. A separate, later 14-day held-out path is not used to select Q. It checks the choice on demand the choice never saw. Each path is an independent replay from the same opening stock. We also report terminal pipeline stock: it is visible in the results but is not charged by this cost objective.

In [16]:

from stockcast.evaluation import (
    holding_cost,
    ordering_cost,
    shortage_cost,
    sku_order_line_count,
    terminal_pipeline_units,
    total_cost,
)

candidate_quantities = [6, 8, 10, 12, 16, 20]
calibration_values = [2, 3, 4, 2, 3, 4, 3] * 4
held_out_values = [5, 3, 5, 2, 4, 3, 5, 3, 4, 2, 5, 3, 4, 3]
# Costs are evaluator inputs, not parameters that the policy optimizes itself.
cost_context = {
    "cost_components": ["ordering", "holding", "shortage"],
    "order_cost_per_sku_line": 5.0,
    "order_cost_per_unit": 0.0,  # Explicit zero: no variable ordering fee.
    "holding_cost_per_unit_period": 0.50,
    "shortage_cost_per_unit": 8.0,
}
cost_metrics = [
    ordering_cost,
    holding_cost,
    shortage_cost,
    total_cost,
    sku_order_line_count,
    fill_rate,
    avg_on_hand,
    terminal_pipeline_units,
]


def fitted_rule(quantity, rule_class=ReorderWhenBelowPolicy, **extra):
    """Only Q or the named class extension changes; all other settings stay fixed."""
    rule = rule_class(
        reorder_threshold=reorder_threshold,
        order_quantity=quantity,
        parameter_source="declared_tuning_grid",
        lead_time=lead_time,
        review_period=review_period,
        service_level=None,
        allow_backorders=False,
        **extra,
    )
    rule.fit()
    return rule


calibration_demand = dated_demand(calibration_values, pd.Timestamp("2026-04-01"))
held_out_demand = dated_demand(held_out_values, pd.Timestamp("2026-04-29"))

pd.DataFrame(
    {
        "path": ["Calibration", "Held out"],
        "first_date": [calibration_demand["date"].min(), held_out_demand["date"].min()],
        "last_date": [calibration_demand["date"].max(), held_out_demand["date"].max()],
        "days": [len(calibration_demand), len(held_out_demand)],
    }
)

from stockcast.evaluation import ( holding_cost, ordering_cost, shortage_cost, sku_order_line_count, terminal_pipeline_units, total_cost, ) candidate_quantities = [6, 8, 10, 12, 16, 20] calibration_values = [2, 3, 4, 2, 3, 4, 3] * 4 held_out_values = [5, 3, 5, 2, 4, 3, 5, 3, 4, 2, 5, 3, 4, 3] # Costs are evaluator inputs, not parameters that the policy optimizes itself. cost_context = { "cost_components": ["ordering", "holding", "shortage"], "order_cost_per_sku_line": 5.0, "order_cost_per_unit": 0.0, # Explicit zero: no variable ordering fee. "holding_cost_per_unit_period": 0.50, "shortage_cost_per_unit": 8.0, } cost_metrics = [ ordering_cost, holding_cost, shortage_cost, total_cost, sku_order_line_count, fill_rate, avg_on_hand, terminal_pipeline_units, ] def fitted_rule(quantity, rule_class=ReorderWhenBelowPolicy, **extra): """Only Q or the named class extension changes; all other settings stay fixed.""" rule = rule_class( reorder_threshold=reorder_threshold, order_quantity=quantity, parameter_source="declared_tuning_grid", lead_time=lead_time, review_period=review_period, service_level=None, allow_backorders=False, **extra, ) rule.fit() return rule calibration_demand = dated_demand(calibration_values, pd.Timestamp("2026-04-01")) held_out_demand = dated_demand(held_out_values, pd.Timestamp("2026-04-29")) pd.DataFrame( { "path": ["Calibration", "Held out"], "first_date": [calibration_demand["date"].min(), held_out_demand["date"].min()], "last_date": [calibration_demand["date"].max(), held_out_demand["date"].max()], "days": [len(calibration_demand), len(held_out_demand)], } )

Out[16]:

	path	first_date	last_date	days
0	Calibration	2026-04-01	2026-04-28	28
1	Held out	2026-04-29	2026-05-12	14

In [17]:

# The declared cost rates, shown once; every branch is charged the same way.
pd.DataFrame([cost_context]).drop(columns="cost_components")

# The declared cost rates, shown once; every branch is charged the same way. pd.DataFrame([cost_context]).drop(columns="cost_components")

Out[17]:

	order_cost_per_sku_line	order_cost_per_unit	holding_cost_per_unit_period	shortage_cost_per_unit
0	5.00	0.00	0.50	8.00

SimulationEngine.run_comparison gives every candidate the same complete calibration demand and an isolated copy of the same opening inventory. Its results mapping lets us evaluate each branch separately. InventoryEvaluator.fit(..., window="scoring") selects those event rows, and evaluate(..., context=cost_context, groupby=[]) applies the same declared cost rates and returns one pooled result row per branch.

In [18]:

def score_branches(comparison):
    """Declared costs and service for each branch of a run_comparison result."""
    frames = []
    for label, run in comparison.results.items():
        evaluator = InventoryEvaluator()
        evaluator.fit(simulation_result=run, window="scoring")
        scores = evaluator.evaluate(
            metrics=cost_metrics,
            groupby=[],
            context=cost_context,
        )
        frames.append(scores.assign(rule=label))
    return pd.concat(frames, ignore_index=True).set_index("rule")


# A separately fitted instance for each candidate keeps policy state isolated.
calibration_labels = [f"Q={quantity}" for quantity in candidate_quantities]
calibration_comparison = SimulationEngine().run_comparison(
    policies=[fitted_rule(quantity) for quantity in candidate_quantities],
    labels=calibration_labels,
    demand_source=calibration_demand,
    inventory=opening_state(
        calibration_demand["date"].min() - pd.Timedelta(days=1),
        opening_stock,
    ),
    freq=period_frequency,
)
calibration_scores = score_branches(calibration_comparison)
cost_parts = ["ordering_cost", "holding_cost", "shortage_cost"]
assert (
    (calibration_scores["total_cost"] - calibration_scores[cost_parts].sum(axis=1))
    .abs()
    .lt(1e-9)
    .all()
)

calibration_scores

def score_branches(comparison): """Declared costs and service for each branch of a run_comparison result.""" frames = [] for label, run in comparison.results.items(): evaluator = InventoryEvaluator() evaluator.fit(simulation_result=run, window="scoring") scores = evaluator.evaluate( metrics=cost_metrics, groupby=[], context=cost_context, ) frames.append(scores.assign(rule=label)) return pd.concat(frames, ignore_index=True).set_index("rule") # A separately fitted instance for each candidate keeps policy state isolated. calibration_labels = [f"Q={quantity}" for quantity in candidate_quantities] calibration_comparison = SimulationEngine().run_comparison( policies=[fitted_rule(quantity) for quantity in candidate_quantities], labels=calibration_labels, demand_source=calibration_demand, inventory=opening_state( calibration_demand["date"].min() - pd.Timedelta(days=1), opening_stock, ), freq=period_frequency, ) calibration_scores = score_branches(calibration_comparison) cost_parts = ["ordering_cost", "holding_cost", "shortage_cost"] assert ( (calibration_scores["total_cost"] - calibration_scores[cost_parts].sum(axis=1)) .abs() .lt(1e-9) .all() ) calibration_scores

Out[18]:

	ordering_cost	holding_cost	shortage_cost	total_cost	sku_order_line_count	fill_rate	avg_on_hand	terminal_pipeline_units
rule
Q=6	60.00	22.00	96.00	178.00	12	0.86	1.57	6.00
Q=8	50.00	33.50	56.00	139.50	10	0.92	2.39	8.00
Q=10	40.00	45.50	88.00	173.50	8	0.87	3.25	10.00
Q=12	35.00	57.50	64.00	156.50	7	0.90	4.11	12.00
Q=16	25.00	86.50	56.00	167.50	5	0.92	6.18	0.00
Q=20	20.00	112.50	16.00	148.50	4	0.98	8.04	0.00

In [19]:

# Choose only within the displayed grid; ties go to the smaller Q.
selected_quantity = candidate_quantities[calibration_scores["total_cost"].argmin()]
print(f"Lowest declared calibration cost in this grid: Q={selected_quantity}")

# Choose only within the displayed grid; ties go to the smaller Q. selected_quantity = candidate_quantities[calibration_scores["total_cost"].argmin()] print(f"Lowest declared calibration cost in this grid: Q={selected_quantity}")

Lowest declared calibration cost in this grid: Q=8

In [20]:

COST_COLORS = {
    "ordering_cost": ("Order lines", ORDER_COLOR),
    "holding_cost": ("Stock held", DEMAND_COLOR),
    "shortage_cost": ("Lost sales", THRESHOLD_COLOR),
}
fig, axes = plt.subplots(1, 2, figsize=(11, 3.8), layout="constrained")
for column, (label, color) in COST_COLORS.items():
    axes[0].plot(
        candidate_quantities,
        calibration_scores[column],
        marker="o",
        label=label,
        color=color,
    )
axes[0].set(
    title="Why the chosen quantity changes cost",
    xlabel="Fixed quantity Q",
    ylabel="Cost",
)
axes[0].legend(frameon=False)

axes[1].plot(
    candidate_quantities,
    calibration_scores["total_cost"],
    marker="o",
    color=STOCK_COLOR,
    label="Declared total cost",
)
axes[1].scatter(
    [selected_quantity],
    [calibration_scores.loc[f"Q={selected_quantity}", "total_cost"]],
    s=105,
    marker="D",
    color=THRESHOLD_COLOR,
    zorder=3,
    label="Grid minimum",
)
axes[1].set(
    title="Selection on calibration demand only",
    xlabel="Fixed quantity Q",
    ylabel="Cost",
)
axes[1].legend(frameon=False)
for ax in axes:
    style_axis(ax)
plt.show()

COST_COLORS = { "ordering_cost": ("Order lines", ORDER_COLOR), "holding_cost": ("Stock held", DEMAND_COLOR), "shortage_cost": ("Lost sales", THRESHOLD_COLOR), } fig, axes = plt.subplots(1, 2, figsize=(11, 3.8), layout="constrained") for column, (label, color) in COST_COLORS.items(): axes[0].plot( candidate_quantities, calibration_scores[column], marker="o", label=label, color=color, ) axes[0].set( title="Why the chosen quantity changes cost", xlabel="Fixed quantity Q", ylabel="Cost", ) axes[0].legend(frameon=False) axes[1].plot( candidate_quantities, calibration_scores["total_cost"], marker="o", color=STOCK_COLOR, label="Declared total cost", ) axes[1].scatter( [selected_quantity], [calibration_scores.loc[f"Q={selected_quantity}", "total_cost"]], s=105, marker="D", color=THRESHOLD_COLOR, zorder=3, label="Grid minimum", ) axes[1].set( title="Selection on calibration demand only", xlabel="Fixed quantity Q", ylabel="Cost", ) axes[1].legend(frameon=False) for ax in axes: style_axis(ax) plt.show()

The minimum is only the best candidate in this grid for this declared calibration path and cost context. It is not a generally optimal (s,Q) policy. In particular, the threshold remains fixed, and short finite runs can be sensitive to ending stock and orders still in transit. The reported terminal_pipeline_units is a diagnostic, not a cost component here. We report those outcomes when we check the selected rule on new demand.

7. Make one small rule extension¶

Suppose a buyer wants a larger request when the inventory position has fallen far below s. We can retain the tested base rule and add one transparent response: on a triggered decision, request Q + b × (s − inventory position) when the gap is positive. Here b=1 is a declared illustrative choice. At the threshold the gap is zero, so the original Q remains unchanged. Below it, the extra units grow with the gap. This is a custom variant, not a newly optimized policy; it introduces no forecast, hidden state, or change to engine timing.

In [21]:

class ShortfallResponsivePolicy(ReorderWhenBelowPolicy):
    """Add units in proportion to the gap below the reorder threshold."""

    def __init__(self, *, extra_per_unit_shortfall, **kwargs):
        super().__init__(**kwargs)
        self.extra_per_unit_shortfall = float(extra_per_unit_shortfall)
        if (
            not math.isfinite(self.extra_per_unit_shortfall)
            or self.extra_per_unit_shortfall < 0
        ):
            raise ValueError("extra_per_unit_shortfall must be finite and non-negative")

    def predict(self, inventory_state_df, current_period=0, **kwargs):
        # Reuse the parent's trigger, timing, and OrderDecision fields.
        base = super().predict(
            inventory_state_df,
            current_period=current_period,
            **kwargs,
        )
        decision = base.get_dataframe()
        # Only positions below s add units; positions above s keep a zero order.
        gap = (self.reorder_threshold - decision["inventory_position"]).clip(lower=0)
        decision["order_quantity"] += self.extra_per_unit_shortfall * gap
        return OrderDecision(
            decision,
            sku_column=inventory_state_df.sku_column,
            lead_time=self.lead_time,
        )

class ShortfallResponsivePolicy(ReorderWhenBelowPolicy): """Add units in proportion to the gap below the reorder threshold.""" def __init__(self, *, extra_per_unit_shortfall, **kwargs): super().__init__(**kwargs) self.extra_per_unit_shortfall = float(extra_per_unit_shortfall) if ( not math.isfinite(self.extra_per_unit_shortfall) or self.extra_per_unit_shortfall < 0 ): raise ValueError("extra_per_unit_shortfall must be finite and non-negative") def predict(self, inventory_state_df, current_period=0, **kwargs): # Reuse the parent's trigger, timing, and OrderDecision fields. base = super().predict( inventory_state_df, current_period=current_period, **kwargs, ) decision = base.get_dataframe() # Only positions below s add units; positions above s keep a zero order. gap = (self.reorder_threshold - decision["inventory_position"]).clip(lower=0) decision["order_quantity"] += self.extra_per_unit_shortfall * gap return OrderDecision( decision, sku_column=inventory_state_df.sku_column, lead_time=self.lead_time, )

In [22]:

responsive = fitted_rule(
    selected_quantity,
    rule_class=ShortfallResponsivePolicy,
    extra_per_unit_shortfall=1.0,
)
low_stock = opening_state(
    opening_date,
    pd.DataFrame({"unique_id": [sku], "observed_on_hand": [3.0]}),
)

# A position of 3 is three units below s=6: Q becomes Q+3 when b=1.
low_stock_before = low_stock.get_dataframe()
base_request = fitted_rule(selected_quantity).predict(low_stock, current_period=0)
responsive_request = responsive.predict(low_stock, current_period=0)
assert low_stock.get_dataframe().equals(low_stock_before)
base_quantity = base_request.get_dataframe()["order_quantity"].iloc[0]
responsive_quantity = responsive_request.get_dataframe()["order_quantity"].iloc[0]
assert responsive_quantity == base_quantity + 3

pd.DataFrame(
    {
        "pre_order_position": [3.0],
        "threshold": [reorder_threshold],
        "fixed_Q_request": [base_quantity],
        "responsive_request": [responsive_quantity],
    }
)

responsive = fitted_rule( selected_quantity, rule_class=ShortfallResponsivePolicy, extra_per_unit_shortfall=1.0, ) low_stock = opening_state( opening_date, pd.DataFrame({"unique_id": [sku], "observed_on_hand": [3.0]}), ) # A position of 3 is three units below s=6: Q becomes Q+3 when b=1. low_stock_before = low_stock.get_dataframe() base_request = fitted_rule(selected_quantity).predict(low_stock, current_period=0) responsive_request = responsive.predict(low_stock, current_period=0) assert low_stock.get_dataframe().equals(low_stock_before) base_quantity = base_request.get_dataframe()["order_quantity"].iloc[0] responsive_quantity = responsive_request.get_dataframe()["order_quantity"].iloc[0] assert responsive_quantity == base_quantity + 3 pd.DataFrame( { "pre_order_position": [3.0], "threshold": [reorder_threshold], "fixed_Q_request": [base_quantity], "responsive_request": [responsive_quantity], } )

Out[22]:

	pre_order_position	threshold	fixed_Q_request	responsive_request
0	3.00	6.00	8.00	11.00

Finally, compare the originally declared Q=10, the calibration-selected fixed Q, and the illustrative responsive variant on the same held-out demand. All three start from the same stock, use the same schedule and lead time, and face the same costs. We fit one policy per branch and pass the same held-out table to run_comparison. The responsive rule did not take part in the calibration grid, so its result is descriptive, not a tuned winner.

In [23]:

# The first branch is the original teaching rule; the other two use Q
# selected without consulting this later demand path.
held_out_labels = [
    f"Original Q={fixed_order_quantity:g}",
    f"Selected Q={selected_quantity}",
    "Responsive variant",
]
held_out_comparison = SimulationEngine().run_comparison(
    policies=[
        fitted_rule(fixed_order_quantity),
        fitted_rule(selected_quantity),
        responsive,
    ],
    labels=held_out_labels,
    demand_source=held_out_demand,
    inventory=opening_state(
        held_out_demand["date"].min() - pd.Timedelta(days=1),
        opening_stock,
    ),
    freq=period_frequency,
)
held_out_scores = score_branches(held_out_comparison)

held_out_scores

# The first branch is the original teaching rule; the other two use Q # selected without consulting this later demand path. held_out_labels = [ f"Original Q={fixed_order_quantity:g}", f"Selected Q={selected_quantity}", "Responsive variant", ] held_out_comparison = SimulationEngine().run_comparison( policies=[ fitted_rule(fixed_order_quantity), fitted_rule(selected_quantity), responsive, ], labels=held_out_labels, demand_source=held_out_demand, inventory=opening_state( held_out_demand["date"].min() - pd.Timedelta(days=1), opening_stock, ), freq=period_frequency, ) held_out_scores = score_branches(held_out_comparison) held_out_scores

Out[23]:

	ordering_cost	holding_cost	shortage_cost	total_cost	sku_order_line_count	fill_rate	avg_on_hand	terminal_pipeline_units
rule
Original Q=10	20.00	18.00	96.00	134.00	4	0.76	2.57	0.00
Selected Q=8	25.00	11.00	104.00	140.00	5	0.75	1.57	8.00
Responsive variant	20.00	18.50	96.00	134.50	4	0.76	2.64	0.00

Tuning on one path does not guarantee a better result on the next. On the later path the calibration pick, Q=8, cost 140.00 in total, more than the original Q=10 at 134.00, and its fill rate was 0.75 against 0.76. A short finite run is sensitive to ending stock and orders still in transit: Q=8 ended the path with 8 units in the pipeline. The responsive variant cost 134.50 with a fill rate of 0.76.

In [24]:

BAR_COLORS = ["#9CA3AF", DEMAND_COLOR, RECEIPT_COLOR]

fig, axes = plt.subplots(1, 2, figsize=(11, 3.8), layout="constrained")
axes[0].bar(held_out_scores.index, held_out_scores["total_cost"], color=BAR_COLORS)
axes[0].set(title="Cost on the held-out path", ylabel="Declared total cost")
axes[1].bar(held_out_scores.index, held_out_scores["fill_rate"], color=BAR_COLORS)
axes[1].set(title="Service on the same path", ylabel="Fill rate")
for ax in axes:
    ax.tick_params(axis="x", rotation=18)
    style_axis(ax)
axes[1].set_ylim(0, 1.05)
plt.show()

BAR_COLORS = ["#9CA3AF", DEMAND_COLOR, RECEIPT_COLOR] fig, axes = plt.subplots(1, 2, figsize=(11, 3.8), layout="constrained") axes[0].bar(held_out_scores.index, held_out_scores["total_cost"], color=BAR_COLORS) axes[0].set(title="Cost on the held-out path", ylabel="Declared total cost") axes[1].bar(held_out_scores.index, held_out_scores["fill_rate"], color=BAR_COLORS) axes[1].set(title="Service on the same path", ylabel="Fill rate") for ax in axes: ax.tick_params(axis="x", rotation=18) style_axis(ax) axes[1].set_ylim(0, 1.05) plt.show()

The figure shows where the responsive variant differs from the fixed rule it extends. It orders the same quantities on most days, but on 6 days, shaded in grey, it orders more after a deeper shortfall and skips orders the fixed rule would have placed. Selection and evaluation stay separate here: the result depends on the demand paths shown, the costs, the fixed threshold, and the finite horizon. It establishes neither forecast calibration nor general policy optimality.

In [25]:

fixed_events = held_out_comparison.results[held_out_labels[1]].to_event_frame(
    window="scoring"
)
responsive_events = held_out_comparison.results[held_out_labels[2]].to_event_frame(
    window="scoring"
)
# Both branches share the SKU/date grid, so orders can be compared row by row.
assert (
    fixed_events[["unique_id", "date"]]
    .reset_index(drop=True)
    .equals(responsive_events[["unique_id", "date"]].reset_index(drop=True))
)
order_dates = fixed_events["date"].reset_index(drop=True)
fixed_orders = fixed_events["order_quantity"].reset_index(drop=True)
responsive_orders = responsive_events["order_quantity"].reset_index(drop=True)
changed = fixed_orders.ne(responsive_orders)
assert changed.any()

fig, ax = plt.subplots(figsize=(11, 3.8), layout="constrained")
offset = pd.Timedelta(hours=5)
ax.bar(
    order_dates - offset,
    fixed_orders,
    width=0.4,
    color=DEMAND_COLOR,
    label=held_out_labels[1],
)
ax.bar(
    order_dates + offset,
    responsive_orders,
    width=0.4,
    color=RECEIPT_COLOR,
    label=held_out_labels[2],
)
for index, day in enumerate(order_dates[changed]):
    ax.axvspan(
        day - pd.Timedelta(hours=12),
        day + pd.Timedelta(hours=12),
        color="0.5",
        alpha=0.14,
        label="Orders differ" if index == 0 else None,
    )
ax.set(
    title="Held-out orders placed: fixed Q against the shortfall response",
    xlabel="Date",
    ylabel="Units ordered",
)
format_dates(ax, list(order_dates[::2]))
style_axis(ax)
ax.legend(frameon=False, loc="upper left", bbox_to_anchor=(1.01, 1.0))
plt.show()

fixed_events = held_out_comparison.results[held_out_labels[1]].to_event_frame( window="scoring" ) responsive_events = held_out_comparison.results[held_out_labels[2]].to_event_frame( window="scoring" ) # Both branches share the SKU/date grid, so orders can be compared row by row. assert ( fixed_events[["unique_id", "date"]] .reset_index(drop=True) .equals(responsive_events[["unique_id", "date"]].reset_index(drop=True)) ) order_dates = fixed_events["date"].reset_index(drop=True) fixed_orders = fixed_events["order_quantity"].reset_index(drop=True) responsive_orders = responsive_events["order_quantity"].reset_index(drop=True) changed = fixed_orders.ne(responsive_orders) assert changed.any() fig, ax = plt.subplots(figsize=(11, 3.8), layout="constrained") offset = pd.Timedelta(hours=5) ax.bar( order_dates - offset, fixed_orders, width=0.4, color=DEMAND_COLOR, label=held_out_labels[1], ) ax.bar( order_dates + offset, responsive_orders, width=0.4, color=RECEIPT_COLOR, label=held_out_labels[2], ) for index, day in enumerate(order_dates[changed]): ax.axvspan( day - pd.Timedelta(hours=12), day + pd.Timedelta(hours=12), color="0.5", alpha=0.14, label="Orders differ" if index == 0 else None, ) ax.set( title="Held-out orders placed: fixed Q against the shortfall response", xlabel="Date", ylabel="Units ordered", ) format_dates(ax, list(order_dates[::2])) style_axis(ax) ax.legend(frameon=False, loc="upper left", bbox_to_anchor=(1.01, 1.0)) plt.show()