A production job: plan before sales, close after sales, triggered by events¶

The objects used for simulation also run a live replenishment job. A production day has two moments:

• Plan, in the morning before sales: receive what is due, refit the forecast on the sales up to yesterday, fit the policy, propose orders, apply the supplier rules, and send the orders.
• Close, in the evening after sales: serve the day's sales from the planned state and save the result for tomorrow.

This notebook runs that job with the setting Notebook 09 found cheapest, and builds it up the way a production system is built:

1. One day, step by step (sections 1 to 4): the saved state the job starts from, then every intermediate table of one morning and one evening.
2. The job as code (sections 5 and 6): two pure functions, plan and close, and the store around them that keeps states, sales, sent orders and decision records, with idempotent phases that are safe to retry.
3. The job in operation (sections 7 to 9): an event stream of sales that triggers the phases for two weeks, a shadow backtest with SimulationEngine that monitors the live job, and a morning when the stock count and the supplier disagree with the saved state.
4. Where each piece runs (section 10): how the same functions fit schedulers, message brokers and serverless triggers.

The M5 sales file plays the role of the live sales feed.

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. The setting from Notebook 09¶

The job uses the cheapest branch of Notebook 09, (s,S) with ETS targets:

• review every R = 2 days, lead time L = 2 days, so each target covers L + R = 4 days;
• s is Smooth's 95% upper bound for the next four days' total demand, rounded up, and S = s + half of the 4-day mean, rounded up;
• supplier rules: each order line at least 4 units and a multiple of 4.

We follow five of the 30 items. The sales up to Sunday 13 March 2016 are the history the job already knows; the sales from Monday 14 March arrive one day at a time.

In [2]:

from stockcast.core import MinimumOrderQuantity, OrderingConstraints, OrderMultiple

lead_time, review_period = 2, 2
horizon = lead_time + review_period
probability = 0.95
supplier_rules = OrderingConstraints(
    [
        MinimumOrderQuantity(4.0, mode="adjust"),
        OrderMultiple(4.0, mode="adjust"),
    ]
)

items = [
    "FOODS_1_085_CA_1",
    "FOODS_1_018_CA_1",
    "FOODS_1_046_CA_1",
    "FOODS_1_013_CA_1",
    "FOODS_1_051_CA_1",
]
data = pd.read_csv("data/m5/sample_1.csv", parse_dates=["date"])
data = data.loc[data["unique_id"].isin(items), ["unique_id", "date", "y"]]
data = data.sort_values(["unique_id", "date"]).reset_index(drop=True)

last_night = pd.Timestamp("2016-03-13")
known_sales = data[data["date"].le(last_night)]
sales_feed = data[data["date"].gt(last_night)]

from stockcast.core import MinimumOrderQuantity, OrderingConstraints, OrderMultiple lead_time, review_period = 2, 2 horizon = lead_time + review_period probability = 0.95 supplier_rules = OrderingConstraints( [ MinimumOrderQuantity(4.0, mode="adjust"), OrderMultiple(4.0, mode="adjust"), ] ) items = [ "FOODS_1_085_CA_1", "FOODS_1_018_CA_1", "FOODS_1_046_CA_1", "FOODS_1_013_CA_1", "FOODS_1_051_CA_1", ] data = pd.read_csv("data/m5/sample_1.csv", parse_dates=["date"]) data = data.loc[data["unique_id"].isin(items), ["unique_id", "date", "y"]] data = data.sort_values(["unique_id", "date"]).reset_index(drop=True) last_night = pd.Timestamp("2016-03-13") known_sales = data[data["date"].le(last_night)] sales_feed = data[data["date"].gt(last_night)]

2. Last night's saved state¶

The job starts from what it saved on Sunday evening: the stock on hand of each item, and the order lines still on their way with their due dates. These are the numbers Notebook 09's (s,S) ETS run ended with on 13 March. from_observed sets the stock on hand on the saved date, and with_open_orders adds the open orders. Due dates become due periods, counted in days from the state's date: an order due on Monday 14 March has due_period=1.

max_lead_time=4 gives the pipeline four slots, so that a delivery two days late still fits (section 9).

In [3]:

from stockcast.core import InventoryStateDataFrame

stock_on_hand = pd.DataFrame(
    {
        "unique_id": items,
        "on_hand": [0.0, 33.0, 12.0, 2.0, 10.0],
    }
)
open_orders = pd.DataFrame(
    {
        "unique_id": [
            "FOODS_1_085_CA_1",
            "FOODS_1_018_CA_1",
            "FOODS_1_046_CA_1",
            "FOODS_1_013_CA_1",
        ],
        "order_date": pd.Timestamp("2016-03-12"),
        "due_date": pd.Timestamp("2016-03-14"),
        "quantity": [88.0, 28.0, 40.0, 8.0],
    }
)


def days_after(dates, day):
    """Whole days from `day` to each date, as integer periods."""
    return (dates - day).dt.days


saved_state = InventoryStateDataFrame.from_observed(
    stock_on_hand,
    opening_date=last_night,
    max_lead_time=4,
    allow_backorders=False,
).with_open_orders(
    pd.DataFrame(
        {
            "unique_id": open_orders["unique_id"],
            "due_period": days_after(open_orders["due_date"], last_night),
            "order_period": days_after(open_orders["order_date"], last_night),
            "quantity": open_orders["quantity"],
        }
    )
)
saved_state.get_dataframe()[["unique_id", "date", "period", "on_hand", "in_transit"]]

from stockcast.core import InventoryStateDataFrame stock_on_hand = pd.DataFrame( { "unique_id": items, "on_hand": [0.0, 33.0, 12.0, 2.0, 10.0], } ) open_orders = pd.DataFrame( { "unique_id": [ "FOODS_1_085_CA_1", "FOODS_1_018_CA_1", "FOODS_1_046_CA_1", "FOODS_1_013_CA_1", ], "order_date": pd.Timestamp("2016-03-12"), "due_date": pd.Timestamp("2016-03-14"), "quantity": [88.0, 28.0, 40.0, 8.0], } ) def days_after(dates, day): """Whole days from `day` to each date, as integer periods.""" return (dates - day).dt.days saved_state = InventoryStateDataFrame.from_observed( stock_on_hand, opening_date=last_night, max_lead_time=4, allow_backorders=False, ).with_open_orders( pd.DataFrame( { "unique_id": open_orders["unique_id"], "due_period": days_after(open_orders["due_date"], last_night), "order_period": days_after(open_orders["order_date"], last_night), "quantity": open_orders["quantity"], } ) ) saved_state.get_dataframe()[["unique_id", "date", "period", "on_hand", "in_transit"]]

Out[3]:

	unique_id	date	period	on_hand	in_transit
0	FOODS_1_085_CA_1	2016-03-13	0.00	0.00	[88.0, 0.0, 0.0, 0.0]
1	FOODS_1_018_CA_1	2016-03-13	0.00	33.00	[28.0, 0.0, 0.0, 0.0]
2	FOODS_1_046_CA_1	2016-03-13	0.00	12.00	[40.0, 0.0, 0.0, 0.0]
3	FOODS_1_013_CA_1	2016-03-13	0.00	2.00	[8.0, 0.0, 0.0, 0.0]
4	FOODS_1_051_CA_1	2016-03-13	0.00	10.00	[0.0, 0.0, 0.0, 0.0]

in_transit holds the pipeline: slot 0 arrives tomorrow morning, slot 1 the day after, and so on. FOODS_1_085_CA_1 has nothing on the shelf, but 88 units arrive on Monday morning.

3. Monday morning: receive, forecast, order¶

Receive. advance_period moves the state to Monday and receives slot 0 of the pipeline. Monday is a review day (is_review_period=True), so the job may order today.

In [4]:

monday = saved_state.advance_period(freq="D", is_review_period=True)
monday.get_dataframe()[
    ["unique_id", "date", "latest_received", "on_hand", "in_transit"]
]

monday = saved_state.advance_period(freq="D", is_review_period=True) monday.get_dataframe()[ ["unique_id", "date", "latest_received", "on_hand", "in_transit"] ]

Out[4]:

	unique_id	date	latest_received	on_hand	in_transit
0	FOODS_1_085_CA_1	2016-03-14	88.00	88.00	[0.0, 0.0, 0.0, 0.0]
1	FOODS_1_018_CA_1	2016-03-14	28.00	61.00	[0.0, 0.0, 0.0, 0.0]
2	FOODS_1_046_CA_1	2016-03-14	40.00	52.00	[0.0, 0.0, 0.0, 0.0]
3	FOODS_1_013_CA_1	2016-03-14	8.00	10.00	[0.0, 0.0, 0.0, 0.0]
4	FOODS_1_051_CA_1	2016-03-14	0.00	10.00	[0.0, 0.0, 0.0, 0.0]

Forecast. The forecast is refitted on every item's sales up to Sunday, the last day the job knows. forecast_targets returns one row per item with s, S and end, the last day the targets cover. This function is the job's own code; any forecaster can take its place.

In [5]:

import math

from smooth import ES


def forecast_targets(known_sales, origin):
    """Refit ETS per item on the sales up to `origin` and return s and S."""
    rows = []
    for item, item_sales in known_sales.groupby("unique_id"):
        model = ES("ANN")
        model.fit(item_sales["y"].astype(float).reset_index(drop=True))
        forecast = model.predict(
            h=horizon,
            interval="prediction",
            level=probability,
            side="upper",
            cumulative=True,
        )
        s = float(math.ceil(forecast.upper.iloc[0, 0]))
        expected = float(forecast.mean.iloc[0])
        rows.append(
            {
                "unique_id": item,
                "s": s,
                "S": s + math.ceil(expected * review_period / horizon),
                "end": origin + pd.Timedelta(days=horizon),
            }
        )
    return pd.DataFrame(rows)


monday_targets = forecast_targets(known_sales, origin=last_night)
monday_targets

import math from smooth import ES def forecast_targets(known_sales, origin): """Refit ETS per item on the sales up to `origin` and return s and S.""" rows = [] for item, item_sales in known_sales.groupby("unique_id"): model = ES("ANN") model.fit(item_sales["y"].astype(float).reset_index(drop=True)) forecast = model.predict( h=horizon, interval="prediction", level=probability, side="upper", cumulative=True, ) s = float(math.ceil(forecast.upper.iloc[0, 0])) expected = float(forecast.mean.iloc[0]) rows.append( { "unique_id": item, "s": s, "S": s + math.ceil(expected * review_period / horizon), "end": origin + pd.Timedelta(days=horizon), } ) return pd.DataFrame(rows) monday_targets = forecast_targets(known_sales, origin=last_night) monday_targets

Out[5]:

	unique_id	s	S	end
0	FOODS_1_013_CA_1	17.00	22.00	2016-03-17
1	FOODS_1_018_CA_1	68.00	90.00	2016-03-17
2	FOODS_1_046_CA_1	52.00	67.00	2016-03-17
3	FOODS_1_051_CA_1	8.00	11.00	2016-03-17
4	FOODS_1_085_CA_1	85.00	114.00	2016-03-17

Order. The targets are fitted into a ReorderPointPolicy, which proposes an order for every item whose inventory position (stock on hand plus stock on order) is at or below s: enough to bring it up to S. The supplier rules then adjust the proposal. apply returns the order to send (accepted.order) and a table of what the rules changed (accepted.audit).

In [6]:

from stockcast.core import ConstraintContext
from stockcast.policies import ReorderPointPolicy


def fit_policy(targets):
    """Fit the (s,S) rule on one day's targets."""
    policy = ReorderPointPolicy(
        lead_time=lead_time,
        review_period=review_period,
        freq="D",
        service_level=probability,
        allow_backorders=False,
        date_column="end",
    )
    return policy.fit(targets, reorder_point_column="s", order_up_to_column="S")


monday_policy = fit_policy(monday_targets)
period = int(monday.get_dataframe()["period"].iloc[0])
proposal = monday_policy.predict(monday, current_period=period)
accepted = supplier_rules.apply(
    proposal,
    ConstraintContext(inventory=monday, policy=monday_policy, decision_period=period),
)
monday_view = monday.inventory_position()[["unique_id", "inventory_position"]].merge(
    monday_targets[["unique_id", "s", "S"]]
)
monday_view.merge(
    accepted.audit[
        [
            "unique_id",
            "requested_order_quantity",
            "constrained_order_quantity",
            "binding_constraints",
        ]
    ]
)

from stockcast.core import ConstraintContext from stockcast.policies import ReorderPointPolicy def fit_policy(targets): """Fit the (s,S) rule on one day's targets.""" policy = ReorderPointPolicy( lead_time=lead_time, review_period=review_period, freq="D", service_level=probability, allow_backorders=False, date_column="end", ) return policy.fit(targets, reorder_point_column="s", order_up_to_column="S") monday_policy = fit_policy(monday_targets) period = int(monday.get_dataframe()["period"].iloc[0]) proposal = monday_policy.predict(monday, current_period=period) accepted = supplier_rules.apply( proposal, ConstraintContext(inventory=monday, policy=monday_policy, decision_period=period), ) monday_view = monday.inventory_position()[["unique_id", "inventory_position"]].merge( monday_targets[["unique_id", "s", "S"]] ) monday_view.merge( accepted.audit[ [ "unique_id", "requested_order_quantity", "constrained_order_quantity", "binding_constraints", ] ] )

Out[6]:

	unique_id	inventory_position	s	S	requested_order_quantity	constrained_order_quantity	binding_constraints
0	FOODS_1_085_CA_1	88.00	85.00	114.00	0.00	0.00
1	FOODS_1_018_CA_1	61.00	68.00	90.00	29.00	32.00	order_multiple
2	FOODS_1_046_CA_1	52.00	52.00	67.00	15.00	16.00	order_multiple
3	FOODS_1_013_CA_1	10.00	17.00	22.00	12.00	12.00
4	FOODS_1_051_CA_1	10.00	8.00	11.00	0.00	0.00

What to notice: the policy orders only where the inventory position is at or below s:

• FOODS_1_085_CA_1: the 88 units received this morning lift its position to 88, above s = 85, so it does not order.
• FOODS_1_018_CA_1: position 61 is below s = 68, so it requests 90 − 61 = 29 units, and the case rule rounds them up to 32.
• FOODS_1_046_CA_1: position 52 equals s, so it requests 67 − 52 = 15 units, rounded up to 16.
• FOODS_1_013_CA_1: position 10 is below s = 17, so it requests 22 − 10 = 12 units, already a multiple of 4.
• FOODS_1_051_CA_1: position 10 is above s = 8, so it does not order.

Send and record. update_inventory_with_orders puts the accepted order into the pipeline, two slots ahead. This planned state is saved before the shop opens, together with the targets, the proposal and accepted.audit, which together explain the order.

In [7]:

from stockcast.utils import update_inventory_with_orders

monday_planned = update_inventory_with_orders(
    monday,
    accepted.order,
    policy=monday_policy,
)

from stockcast.utils import update_inventory_with_orders monday_planned = update_inventory_with_orders( monday, accepted.order, policy=monday_policy, )

4. Monday evening: close with the day's sales¶

When Monday's sales arrive, fulfill_demand serves them from the planned state. Here the M5 sales stand in for the day's demand, so they can be more than the simulated stock; the units beyond it are lost (latest_shortage), because the policies use allow_backorders=False. In a shop, sales never exceed the shelf: the till records what was sold, and the stock count of section 9 keeps the saved state true. The closed state is what the job saves tonight.

In [8]:

monday_sales = sales_feed[sales_feed["date"].eq(pd.Timestamp("2016-03-14"))]
monday_closed = monday_planned.fulfill_demand(monday_sales)
monday_closed.get_dataframe()[
    [
        "unique_id",
        "date",
        "latest_incoming_demand",
        "latest_fulfilled",
        "latest_shortage",
        "on_hand",
        "in_transit",
    ]
]

monday_sales = sales_feed[sales_feed["date"].eq(pd.Timestamp("2016-03-14"))] monday_closed = monday_planned.fulfill_demand(monday_sales) monday_closed.get_dataframe()[ [ "unique_id", "date", "latest_incoming_demand", "latest_fulfilled", "latest_shortage", "on_hand", "in_transit", ] ]

Out[8]:

	unique_id	date	latest_incoming_demand	latest_fulfilled	latest_shortage	on_hand	in_transit
0	FOODS_1_085_CA_1	2016-03-14	11.00	11.00	0.00	77.00	[0.0, 0.0, 0.0, 0.0]
1	FOODS_1_018_CA_1	2016-03-14	5.00	5.00	0.00	56.00	[0.0, 32.0, 0.0, 0.0]
2	FOODS_1_046_CA_1	2016-03-14	13.00	13.00	0.00	39.00	[0.0, 16.0, 0.0, 0.0]
3	FOODS_1_013_CA_1	2016-03-14	9.00	9.00	0.00	1.00	[0.0, 12.0, 0.0, 0.0]
4	FOODS_1_051_CA_1	2016-03-14	0.00	0.00	0.00	10.00	[0.0, 0.0, 0.0, 0.0]

What to notice: all of Monday's sales are served, and FOODS_1_013_CA_1 ends the day with 1 unit. The three orders sit in slot 1 of in_transit (32, 16 and 12 units): they arrive on Wednesday morning, two days after they were placed.

5. The job as two pure functions¶

The steps above become two functions. They take objects and return objects: no files, no clock, no network. That keeps the decision logic easy to test with fixed inputs, and lets any trigger call it (section 7).

• plan(state, known_sales, is_review) receives what is due and, on a review day, refits the forecast, fits the policy, proposes and applies the supplier rules. It returns the planned state, the order to send, and a decision record: one row per item with the forecast origin, s, S, the inventory position, the proposed and the sent quantity, and the rule that changed it.
• close(planned, sales) serves the day's sales.

predict already returns each item's inventory position, s (reorder_point) and S (target_level) next to the proposed quantity, so the record is a merge of the proposal with the rules' table.

In [9]:

JOB_VERSION = "sS-ETS-95 v1"  # names this setting in every decision record


def plan(state, known_sales, is_review):
    """Morning: receive, and on a review day forecast, propose and apply the rules."""
    today = state.advance_period(freq="D", is_review_period=is_review)
    if not is_review:
        return today, None, None
    yesterday = pd.Timestamp(state.get_dataframe()["date"].iloc[0])
    policy = fit_policy(forecast_targets(known_sales, origin=yesterday))
    now = int(today.get_dataframe()["period"].iloc[0])
    proposal = policy.predict(today, current_period=now)
    accepted = supplier_rules.apply(
        proposal,
        ConstraintContext(inventory=today, policy=policy, decision_period=now),
    )
    record = proposal.get_dataframe()[
        [
            "unique_id",
            "inventory_position",
            "reorder_point",
            "target_level",
            "order_quantity",
        ]
    ].rename(columns={"order_quantity": "proposed"})
    record = record.merge(
        accepted.audit[
            ["unique_id", "constrained_order_quantity", "binding_constraints"]
        ]
    ).rename(columns={"constrained_order_quantity": "sent"})
    record.insert(0, "date", today.get_dataframe()["date"].iloc[0])
    record.insert(1, "forecast_origin", yesterday)
    record["job_version"] = JOB_VERSION
    planned = update_inventory_with_orders(today, accepted.order, policy=policy)
    return planned, accepted.order, record


def close(planned, sales):
    """Evening: serve the day's sales."""
    return planned.fulfill_demand(sales)

JOB_VERSION = "sS-ETS-95 v1" # names this setting in every decision record def plan(state, known_sales, is_review): """Morning: receive, and on a review day forecast, propose and apply the rules.""" today = state.advance_period(freq="D", is_review_period=is_review) if not is_review: return today, None, None yesterday = pd.Timestamp(state.get_dataframe()["date"].iloc[0]) policy = fit_policy(forecast_targets(known_sales, origin=yesterday)) now = int(today.get_dataframe()["period"].iloc[0]) proposal = policy.predict(today, current_period=now) accepted = supplier_rules.apply( proposal, ConstraintContext(inventory=today, policy=policy, decision_period=now), ) record = proposal.get_dataframe()[ [ "unique_id", "inventory_position", "reorder_point", "target_level", "order_quantity", ] ].rename(columns={"order_quantity": "proposed"}) record = record.merge( accepted.audit[ ["unique_id", "constrained_order_quantity", "binding_constraints"] ] ).rename(columns={"constrained_order_quantity": "sent"}) record.insert(0, "date", today.get_dataframe()["date"].iloc[0]) record.insert(1, "forecast_origin", yesterday) record["job_version"] = JOB_VERSION planned = update_inventory_with_orders(today, accepted.order, policy=policy) return planned, accepted.order, record def close(planned, sales): """Evening: serve the day's sales.""" return planned.fulfill_demand(sales)

6. What the job keeps between runs¶

A live job runs in short-lived processes: each morning and each evening a new process starts, reads what it needs, does one phase and stops. Everything that must survive lives in a store. Here the store is a folder of plain files; in production it is a database table or an object store, with the same four parts:

• states: the state after each phase, one file per date and phase (2016-03-14-plan.json). The file name is the phase's idempotency key: if the file exists, that phase has already run for that date.
• sales: every closed day's sales, the history the forecast is refitted on, one file per date (2016-03-14.csv). A retried close writes the same file again, so a day's sales are never counted twice.
• outbox: the orders handed to the supplier, one file per date with one line per order and an order id made of the date and the item. A date whose file exists is never sent again.
• decision records: the rows plan returns, one file per review date, so every order can be explained later.

Each file is written to a temporary file first and then renamed, so the destination is replaced only after the write completes. This example runs phases one at a time.

In [10]:

import io
import json
import tempfile
from pathlib import Path


def write_file(path, text):
    """Write to a temporary file, then rename: the write is all-or-nothing."""
    temporary = path.with_suffix(".tmp")
    temporary.write_text(text)
    temporary.replace(path)


class JobStore:
    """Everything the job keeps between runs, in one folder."""

    def __init__(self, folder):
        self.folder = Path(folder)
        for part in ["states", "sales", "outbox", "records"]:
            (self.folder / part).mkdir(parents=True, exist_ok=True)

    # States, keyed by date and phase.
    def _state_path(self, date, phase):
        return self.folder / "states" / f"{date:%Y-%m-%d}-{phase}.json"

    def has_state(self, date, phase):
        return self._state_path(date, phase).exists()

    def save_state(self, date, phase, state):
        write_file(
            self._state_path(date, phase),
            state.get_dataframe().to_json(orient="records", date_format="iso"),
        )

    def load_state(self, date, phase):
        frame = pd.read_json(
            self._state_path(date, phase),
            orient="records",
            dtype=False,
        )
        frame["date"] = pd.to_datetime(frame["date"])
        frame["in_transit"] = frame["in_transit"].map(
            lambda slots: np.array(slots, dtype=float)
        )
        return InventoryStateDataFrame(frame, max_lead_time=4, allow_backorders=False)

    # Sales history, keyed by date: saving a date again replaces its file.
    def save_sales(self, date, sales):
        write_file(
            self.folder / "sales" / f"{date:%Y-%m-%d}.csv",
            sales.to_csv(index=False),
        )

    def load_sales(self):
        files = sorted((self.folder / "sales").glob("*.csv"))
        return pd.concat(
            [pd.read_csv(file, parse_dates=["date"]) for file in files],
            ignore_index=True,
        )

    # Orders, keyed by date: a date whose file exists has been sent.
    def send_orders(self, date, order):
        path = self.folder / "outbox" / f"{date:%Y-%m-%d}.jsonl"
        if path.exists():
            return  # sent by an earlier attempt
        rows = order.get_dataframe()
        lines = [
            json.dumps(
                {
                    "order_id": f"{date:%Y%m%d}-{row.unique_id}",
                    "unique_id": row.unique_id,
                    "quantity": row.order_quantity,
                }
            )
            for row in rows[rows["order_quantity"] > 0].itertuples()
        ]
        write_file(path, "".join(line + "\n" for line in lines))

    def load_outbox(self):
        files = sorted((self.folder / "outbox").glob("*.jsonl"))
        lines = "".join(file.read_text() for file in files)
        return pd.read_json(io.StringIO(lines), lines=True)

    # Decision records, keyed by date.

    def save_record(self, date, record):
        write_file(
            self.folder / "records" / f"{date:%Y-%m-%d}.jsonl",
            record.to_json(orient="records", lines=True, date_format="iso"),
        )

    def load_records(self):
        files = sorted((self.folder / "records").glob("*.jsonl"))
        records = pd.concat(
            [pd.read_json(file, lines=True, dtype=False) for file in files],
            ignore_index=True,
        )
        return records.astype(
            {"date": "datetime64[ns]", "forecast_origin": "datetime64[ns]"}
        )


store = JobStore(tempfile.mkdtemp(prefix="stockcast-job-"))
store.save_state(last_night, "close", saved_state)
store.save_sales(last_night, known_sales)

import io import json import tempfile from pathlib import Path def write_file(path, text): """Write to a temporary file, then rename: the write is all-or-nothing.""" temporary = path.with_suffix(".tmp") temporary.write_text(text) temporary.replace(path) class JobStore: """Everything the job keeps between runs, in one folder.""" def __init__(self, folder): self.folder = Path(folder) for part in ["states", "sales", "outbox", "records"]: (self.folder / part).mkdir(parents=True, exist_ok=True) # States, keyed by date and phase. def _state_path(self, date, phase): return self.folder / "states" / f"{date:%Y-%m-%d}-{phase}.json" def has_state(self, date, phase): return self._state_path(date, phase).exists() def save_state(self, date, phase, state): write_file( self._state_path(date, phase), state.get_dataframe().to_json(orient="records", date_format="iso"), ) def load_state(self, date, phase): frame = pd.read_json( self._state_path(date, phase), orient="records", dtype=False, ) frame["date"] = pd.to_datetime(frame["date"]) frame["in_transit"] = frame["in_transit"].map( lambda slots: np.array(slots, dtype=float) ) return InventoryStateDataFrame(frame, max_lead_time=4, allow_backorders=False) # Sales history, keyed by date: saving a date again replaces its file. def save_sales(self, date, sales): write_file( self.folder / "sales" / f"{date:%Y-%m-%d}.csv", sales.to_csv(index=False), ) def load_sales(self): files = sorted((self.folder / "sales").glob("*.csv")) return pd.concat( [pd.read_csv(file, parse_dates=["date"]) for file in files], ignore_index=True, ) # Orders, keyed by date: a date whose file exists has been sent. def send_orders(self, date, order): path = self.folder / "outbox" / f"{date:%Y-%m-%d}.jsonl" if path.exists(): return # sent by an earlier attempt rows = order.get_dataframe() lines = [ json.dumps( { "order_id": f"{date:%Y%m%d}-{row.unique_id}", "unique_id": row.unique_id, "quantity": row.order_quantity, } ) for row in rows[rows["order_quantity"] > 0].itertuples() ] write_file(path, "".join(line + "\n" for line in lines)) def load_outbox(self): files = sorted((self.folder / "outbox").glob("*.jsonl")) lines = "".join(file.read_text() for file in files) return pd.read_json(io.StringIO(lines), lines=True) # Decision records, keyed by date. def save_record(self, date, record): write_file( self.folder / "records" / f"{date:%Y-%m-%d}.jsonl", record.to_json(orient="records", lines=True, date_format="iso"), ) def load_records(self): files = sorted((self.folder / "records").glob("*.jsonl")) records = pd.concat( [pd.read_json(file, lines=True, dtype=False) for file in files], ignore_index=True, ) return records.astype( {"date": "datetime64[ns]", "forecast_origin": "datetime64[ns]"} ) store = JobStore(tempfile.mkdtemp(prefix="stockcast-job-")) store.save_state(last_night, "close", saved_state) store.save_sales(last_night, known_sales)

The store starts with Sunday's saved state (section 2) and the sales history up to Sunday, saved under Sunday's date. Two small functions run one phase each. They read their inputs from the store, call plan or close, and write the results back. The state file is written last: if the process stops before it is saved, the phase is not marked complete and the next attempt runs it again. If it stops after saving the outbox but before writing the state, the next attempt finds the existing batch and keeps it. In production, the supplier or ERP must also accept each order id only once. The decision record and the day's sales are saved under their date, so a retry with the same inputs replaces them: if the close crashes after saving the sales but before the state, the retry saves the same day again and the history still holds it once.

In [11]:

FIRST_REVIEW = pd.Timestamp("2016-03-14")


def is_review_day(date):
    """The job reviews every second day, starting on Monday 14 March."""
    return (date - FIRST_REVIEW).days % review_period == 0


def run_plan(store, date):
    """Morning phase for `date`; does nothing if it already ran."""
    if store.has_state(date, "plan"):
        return "already planned"
    state = store.load_state(date - pd.Timedelta(days=1), "close")
    planned, order, record = plan(state, store.load_sales(), is_review_day(date))
    if order is not None:
        store.send_orders(date, order)
        store.save_record(date, record)
    store.save_state(date, "plan", planned)  # last: marks the phase as done
    return "planned"


def run_close(store, date, sales):
    """Evening phase for `date`; does nothing if it already ran."""
    if store.has_state(date, "close"):
        return "already closed"
    closed = close(store.load_state(date, "plan"), sales)
    store.save_sales(date, sales)
    store.save_state(date, "close", closed)  # last: marks the phase as done
    return "closed"

FIRST_REVIEW = pd.Timestamp("2016-03-14") def is_review_day(date): """The job reviews every second day, starting on Monday 14 March.""" return (date - FIRST_REVIEW).days % review_period == 0 def run_plan(store, date): """Morning phase for `date`; does nothing if it already ran.""" if store.has_state(date, "plan"): return "already planned" state = store.load_state(date - pd.Timedelta(days=1), "close") planned, order, record = plan(state, store.load_sales(), is_review_day(date)) if order is not None: store.send_orders(date, order) store.save_record(date, record) store.save_state(date, "plan", planned) # last: marks the phase as done return "planned" def run_close(store, date, sales): """Evening phase for `date`; does nothing if it already ran.""" if store.has_state(date, "close"): return "already closed" closed = close(store.load_state(date, "plan"), sales) store.save_sales(date, sales) store.save_state(date, "close", closed) # last: marks the phase as done return "closed"

A scheduler or a message broker may start the same phase twice, for example after a timeout. Running Monday's plan twice shows what the key does: the second call finds the state file and returns at once, and the outbox holds Monday's three orders once.

In [12]:

monday_date = pd.Timestamp("2016-03-14")
print(run_plan(store, monday_date))
print(run_plan(store, monday_date))
store.load_outbox()

monday_date = pd.Timestamp("2016-03-14") print(run_plan(store, monday_date)) print(run_plan(store, monday_date)) store.load_outbox()

planned
already planned

Out[12]:

	order_id	unique_id	quantity
0	20160314-FOODS_1_018_CA_1	FOODS_1_018_CA_1	32
1	20160314-FOODS_1_046_CA_1	FOODS_1_046_CA_1	16
2	20160314-FOODS_1_013_CA_1	FOODS_1_013_CA_1	12

7. The trigger: an event stream¶

The job does not loop over dates. It reacts to events. In a shop, the till system publishes each sale as it happens, and the job listens:

• shop opens (07:00): run the plan for today;
• sale: one unit sold, added to today's running totals;
• shop closes (22:00): turn the totals into today's sales table, one row per item including items that sold nothing, and run the close.

M5 gives only daily totals, so we split each day's total into single-unit sales at random times between 08:00 and 21:00 (seeded). A Python generator plays the role of the message broker (Kafka, Kinesis or Pub/Sub in production); the consumer handles one event at a time.

In [13]:

def event_stream(sales_feed, first_day, n_days, seed=10):
    """Yield store-open, sale and store-close events in time order."""
    rng = np.random.default_rng(seed)
    for date in pd.date_range(first_day, periods=n_days, freq="D"):
        events = [{"time": date + pd.Timedelta(hours=7), "type": "shop opens"}]
        for row in sales_feed[sales_feed["date"].eq(date)].itertuples():
            minutes = rng.integers(8 * 60, 21 * 60, size=int(row.y))
            events += [
                {
                    "time": date + pd.Timedelta(minutes=int(m)),
                    "type": "sale",
                    "unique_id": row.unique_id,
                }
                for m in minutes
            ]
        events.append({"time": date + pd.Timedelta(hours=22), "type": "shop closes"})
        yield from sorted(events, key=lambda event: event["time"])


class DailyJobConsumer:
    """Turns the event stream into plan and close phases."""

    def __init__(self, store, items):
        self.store = store
        self.items = items
        self.today_totals = {}
        self.log = []

    def handle(self, event):
        date = event["time"].normalize()
        if event["type"] == "shop opens":
            self.today_totals = dict.fromkeys(self.items, 0.0)
            self.log.append((event["time"], run_plan(self.store, date)))
        elif event["type"] == "sale":
            self.today_totals[event["unique_id"]] += 1.0
        elif event["type"] == "shop closes":
            sales = pd.DataFrame(
                {
                    "unique_id": self.items,
                    "date": date,
                    "y": [self.today_totals[i] for i in self.items],
                }
            )
            self.log.append((event["time"], run_close(self.store, date, sales)))

def event_stream(sales_feed, first_day, n_days, seed=10): """Yield store-open, sale and store-close events in time order.""" rng = np.random.default_rng(seed) for date in pd.date_range(first_day, periods=n_days, freq="D"): events = [{"time": date + pd.Timedelta(hours=7), "type": "shop opens"}] for row in sales_feed[sales_feed["date"].eq(date)].itertuples(): minutes = rng.integers(8 * 60, 21 * 60, size=int(row.y)) events += [ { "time": date + pd.Timedelta(minutes=int(m)), "type": "sale", "unique_id": row.unique_id, } for m in minutes ] events.append({"time": date + pd.Timedelta(hours=22), "type": "shop closes"}) yield from sorted(events, key=lambda event: event["time"]) class DailyJobConsumer: """Turns the event stream into plan and close phases.""" def __init__(self, store, items): self.store = store self.items = items self.today_totals = {} self.log = [] def handle(self, event): date = event["time"].normalize() if event["type"] == "shop opens": self.today_totals = dict.fromkeys(self.items, 0.0) self.log.append((event["time"], run_plan(self.store, date))) elif event["type"] == "sale": self.today_totals[event["unique_id"]] += 1.0 elif event["type"] == "shop closes": sales = pd.DataFrame( { "unique_id": self.items, "date": date, "y": [self.today_totals[i] for i in self.items], } ) self.log.append((event["time"], run_close(self.store, date, sales)))

Two weeks of the job, from Monday 14 to Sunday 27 March, now run from the stream alone. Monday's plan already ran in section 6, so the consumer finds its key and moves on.

In [14]:

n_days = 14
consumer = DailyJobConsumer(store, items)
n_events = 0
for event in event_stream(sales_feed, first_day=monday_date, n_days=n_days):
    consumer.handle(event)
    n_events += 1

print(f"{n_events} events handled")
pd.DataFrame(consumer.log, columns=["time", "phase"]).head(4)

n_days = 14 consumer = DailyJobConsumer(store, items) n_events = 0 for event in event_stream(sales_feed, first_day=monday_date, n_days=n_days): consumer.handle(event) n_events += 1 print(f"{n_events} events handled") pd.DataFrame(consumer.log, columns=["time", "phase"]).head(4)

538 events handled

Out[14]:

	time	phase
0	2016-03-14 07:00:00	already planned
1	2016-03-14 22:00:00	closed
2	2016-03-15 07:00:00	planned
3	2016-03-15 22:00:00	closed

What to notice: the stream carried 538 events over the 14 days: 14 openings, 14 closings and 510 single sales. Monday's opening found the plan from section 6 (already planned), and every other phase ran exactly once. A day without a review, such as 15 March, still has a plan phase: it receives the deliveries due that morning and moves the state to the new day, without forecasting or ordering.

Every order the job sent can be explained from its decision record: the forecast it came from, s and S, the inventory position at the review, what the policy proposed and what the supplier rules made of it.

In [15]:

records = store.load_records()
records[records["sent"] > 0].head(8)

records = store.load_records() records[records["sent"] > 0].head(8)

Out[15]:

	date	forecast_origin	unique_id	inventory_position	reorder_point	target_level	proposed	sent	binding_constraints	job_version
1	2016-03-14	2016-03-13	FOODS_1_018_CA_1	61.00	68.00	90.00	29.00	32.00	order_multiple	sS-ETS-95 v1
2	2016-03-14	2016-03-13	FOODS_1_046_CA_1	52.00	52.00	67.00	15.00	16.00	order_multiple	sS-ETS-95 v1
3	2016-03-14	2016-03-13	FOODS_1_013_CA_1	10.00	17.00	22.00	12.00	12.00		sS-ETS-95 v1
5	2016-03-16	2016-03-15	FOODS_1_085_CA_1	66.00	81.00	108.00	42.00	44.00	order_multiple	sS-ETS-95 v1
7	2016-03-16	2016-03-15	FOODS_1_046_CA_1	52.00	52.00	67.00	15.00	16.00	order_multiple	sS-ETS-95 v1
8	2016-03-16	2016-03-15	FOODS_1_013_CA_1	12.00	18.00	24.00	12.00	12.00		sS-ETS-95 v1
15	2016-03-20	2016-03-19	FOODS_1_085_CA_1	61.00	78.00	104.00	43.00	44.00	order_multiple	sS-ETS-95 v1
16	2016-03-20	2016-03-19	FOODS_1_018_CA_1	47.00	63.00	83.00	36.00	36.00		sS-ETS-95 v1

What to notice: each row explains one order. On 14 March, FOODS_1_018_CA_1 had an inventory position of 61, below s = 68, so the policy proposed 90 − 61 = 29 units and the case rule sent 32: the same order as in section 3. Every row also carries the forecast origin, the day before the decision, and the job version, so orders made under an old setting can be told apart from new ones.

8. Monitoring: a shadow backtest and an alert¶

A separate monitoring job, run for example once a week, replays the past days with SimulationEngine, using only what the store holds: the first saved state, the sales, and the targets in the decision records, refitted into the same policies. The engine runs the same steps in the same order (receive, decide, serve), so its closing stock must equal the stock the live job saved every evening. A difference means the live job and the backtest no longer agree, for example after a code change, and the job should be stopped before it sends more orders.

When they agree, the backtest's event table measures the live job with the evaluator, with the same metrics used to choose the setting in Notebook 09. With sales standing in for demand, the fill rate and lost units are simulated service figures; in a shop, the units customers wanted from an empty shelf never appear in the sales. The alert flags any item whose fill rate fell below fill_rate_goal, a unit fill-rate goal the planning team declares on its own. The targets' 95% is the level of the forecast's upper bound for total demand over four days; fill_rate_goal is a separate operational choice.

In [16]:

from stockcast.core import SimulationEngine
from stockcast.evaluation import (
    InventoryEvaluator,
    avg_on_hand,
    fill_rate,
    lost_sales_units,
    order_units,
    sku_order_line_count,
)

# The policies of the live job, refitted from the decision records.
review_policies = {}
for date, rows in records.groupby("date"):
    targets = rows.rename(columns={"reorder_point": "s", "target_level": "S"}).assign(
        end=rows["forecast_origin"] + pd.Timedelta(days=horizon)
    )
    review_policies[(date - monday_date).days] = fit_policy(
        targets[["unique_id", "s", "S", "end"]]
    )

days = pd.date_range(monday_date, periods=n_days, freq="D")
shadow = SimulationEngine().run(
    policy=review_policies[0],
    policy_schedule={day: policy for day, policy in review_policies.items() if day > 0},
    demand_source=store.load_sales().query("date >= @monday_date"),
    inventory=store.load_state(last_night, "close"),
    freq="D",
    order_constraints=supplier_rules,
)

# The live job's closing stock, read back from the store.
live_stock = pd.concat(
    [
        store.load_state(day, "close").get_dataframe()[["unique_id", "date", "on_hand"]]
        for day in days
    ]
)
shadow_stock = shadow.to_event_frame()[["unique_id", "date", "ending_on_hand"]]
compared = live_stock.merge(
    shadow_stock,
    on=["unique_id", "date"],
    validate="one_to_one",
)
assert np.array_equal(compared["on_hand"], compared["ending_on_hand"])

evaluator = InventoryEvaluator()
evaluator.fit(shadow)
health = evaluator.evaluate(
    [fill_rate, lost_sales_units, avg_on_hand, order_units, sku_order_line_count],
    groupby=["unique_id"],
)
fill_rate_goal = 0.95  # the planning team's goal for the share of units served
health["alert"] = health["fill_rate"] < fill_rate_goal
health

from stockcast.core import SimulationEngine from stockcast.evaluation import ( InventoryEvaluator, avg_on_hand, fill_rate, lost_sales_units, order_units, sku_order_line_count, ) # The policies of the live job, refitted from the decision records. review_policies = {} for date, rows in records.groupby("date"): targets = rows.rename(columns={"reorder_point": "s", "target_level": "S"}).assign( end=rows["forecast_origin"] + pd.Timedelta(days=horizon) ) review_policies[(date - monday_date).days] = fit_policy( targets[["unique_id", "s", "S", "end"]] ) days = pd.date_range(monday_date, periods=n_days, freq="D") shadow = SimulationEngine().run( policy=review_policies[0], policy_schedule={day: policy for day, policy in review_policies.items() if day > 0}, demand_source=store.load_sales().query("date >= @monday_date"), inventory=store.load_state(last_night, "close"), freq="D", order_constraints=supplier_rules, ) # The live job's closing stock, read back from the store. live_stock = pd.concat( [ store.load_state(day, "close").get_dataframe()[["unique_id", "date", "on_hand"]] for day in days ] ) shadow_stock = shadow.to_event_frame()[["unique_id", "date", "ending_on_hand"]] compared = live_stock.merge( shadow_stock, on=["unique_id", "date"], validate="one_to_one", ) assert np.array_equal(compared["on_hand"], compared["ending_on_hand"]) evaluator = InventoryEvaluator() evaluator.fit(shadow) health = evaluator.evaluate( [fill_rate, lost_sales_units, avg_on_hand, order_units, sku_order_line_count], groupby=["unique_id"], ) fill_rate_goal = 0.95 # the planning team's goal for the share of units served health["alert"] = health["fill_rate"] < fill_rate_goal health

Out[16]:

	unique_id	fill_rate	lost_sales_units	avg_on_hand	order_units	sku_order_line_count	alert
0	FOODS_1_085_CA_1	1.00	0.00	59.14	140.00	4	False
1	FOODS_1_018_CA_1	0.98	3.00	44.07	148.00	4	False
2	FOODS_1_046_CA_1	1.00	0.00	35.71	132.00	5	False
3	FOODS_1_013_CA_1	0.97	1.00	12.93	32.00	3	False
4	FOODS_1_051_CA_1	1.00	0.00	8.36	8.00	2	False

What to notice:

• The assert passes: on all 14 days, the shadow backtest's closing stock equals the stock the live job saved, item by item. The event table and the metrics therefore describe what the live job did.
• The backtest was rebuilt from the store alone: the targets in the decision records are enough to refit every policy the job used.
• No item triggers the alert. Three items lost nothing; FOODS_1_018_CA_1 lost 3 units (fill rate 0.98) and FOODS_1_013_CA_1 lost 1 (0.97).

9. When the records disagree¶

So far the job trusted its own saved state. In the shop, the stock count and the supplier have the final word. On Monday 28 March:

• the stock count finds 6 fewer units of FOODS_1_085_CA_1 than the saved state;
• the supplier reports that the delivery of FOODS_1_046_CA_1 due that morning will arrive two days late, on Wednesday 30 March.

The job then rebuilds the morning state from those records, with the same two calls as in section 2: from_observed with the counted stock and with_open_orders with the open-order list as the supplier reports it. Because plan is a pure function, the table below runs it on both states and compares them without touching the store. In production the rebuilt state is saved as a new version next to the old one, never over it, so both stay available for checking.

In [17]:

sunday = monday_date + pd.Timedelta(days=n_days - 1)
saved_tonight = store.load_state(sunday, "close")
saved_frame = saved_tonight.get_dataframe()
saved_period = int(saved_frame["period"].iloc[0])

# The stock count: 6 units of FOODS_1_085_CA_1 fewer than the saved state.
counted = saved_frame[["unique_id", "on_hand"]].copy()
counted.loc[counted["unique_id"].eq("FOODS_1_085_CA_1"), "on_hand"] -= 6

# The open orders as the supplier reports them, with due dates:
# the delivery of FOODS_1_046_CA_1 due on Monday comes two days late.
supplier_orders = saved_tonight.scheduled_receipts()
days_ahead = supplier_orders["due_period"] - saved_period
supplier_orders["due_date"] = sunday + pd.to_timedelta(days_ahead, unit="D")
due_monday = supplier_orders["due_date"].eq(sunday + pd.Timedelta(days=1))
late_row = supplier_orders["unique_id"].eq("FOODS_1_046_CA_1") & due_monday
supplier_orders.loc[late_row, "due_date"] += pd.Timedelta(days=2)

rebuilt = InventoryStateDataFrame.from_observed(
    counted,
    opening_date=sunday,
    max_lead_time=4,
    allow_backorders=False,
).with_open_orders(
    pd.DataFrame(
        {
            "unique_id": supplier_orders["unique_id"],
            "due_period": days_after(supplier_orders["due_date"], sunday),
            "quantity": supplier_orders["quantity"],
        }
    )
)

# Monday morning planned twice: from the saved state and from the records.
mornings = {}
for label, state in [("saved state", saved_tonight), ("rebuilt from records", rebuilt)]:
    planned, _, record = plan(state, store.load_sales(), is_review=True)
    mornings[label] = record.set_index("unique_id")[
        ["inventory_position", "sent"]
    ].assign(on_hand=planned.get_dataframe().set_index("unique_id")["on_hand"])
pd.concat(mornings, axis=1)

sunday = monday_date + pd.Timedelta(days=n_days - 1) saved_tonight = store.load_state(sunday, "close") saved_frame = saved_tonight.get_dataframe() saved_period = int(saved_frame["period"].iloc[0]) # The stock count: 6 units of FOODS_1_085_CA_1 fewer than the saved state. counted = saved_frame[["unique_id", "on_hand"]].copy() counted.loc[counted["unique_id"].eq("FOODS_1_085_CA_1"), "on_hand"] -= 6 # The open orders as the supplier reports them, with due dates: # the delivery of FOODS_1_046_CA_1 due on Monday comes two days late. supplier_orders = saved_tonight.scheduled_receipts() days_ahead = supplier_orders["due_period"] - saved_period supplier_orders["due_date"] = sunday + pd.to_timedelta(days_ahead, unit="D") due_monday = supplier_orders["due_date"].eq(sunday + pd.Timedelta(days=1)) late_row = supplier_orders["unique_id"].eq("FOODS_1_046_CA_1") & due_monday supplier_orders.loc[late_row, "due_date"] += pd.Timedelta(days=2) rebuilt = InventoryStateDataFrame.from_observed( counted, opening_date=sunday, max_lead_time=4, allow_backorders=False, ).with_open_orders( pd.DataFrame( { "unique_id": supplier_orders["unique_id"], "due_period": days_after(supplier_orders["due_date"], sunday), "quantity": supplier_orders["quantity"], } ) ) # Monday morning planned twice: from the saved state and from the records. mornings = {} for label, state in [("saved state", saved_tonight), ("rebuilt from records", rebuilt)]: planned, _, record = plan(state, store.load_sales(), is_review=True) mornings[label] = record.set_index("unique_id")[ ["inventory_position", "sent"] ].assign(on_hand=planned.get_dataframe().set_index("unique_id")["on_hand"]) pd.concat(mornings, axis=1)

Out[17]:

	saved state			rebuilt from records
	inventory_position	sent	on_hand	inventory_position	sent	on_hand
unique_id
FOODS_1_085_CA_1	73.00	28.00	73.00	67.00	32.00	67.00
FOODS_1_018_CA_1	36.00	88.00	36.00	36.00	88.00	36.00
FOODS_1_046_CA_1	42.00	68.00	42.00	42.00	68.00	18.00
FOODS_1_013_CA_1	13.00	8.00	13.00	13.00	8.00	13.00
FOODS_1_051_CA_1	11.00	0.00	11.00	11.00	0.00	11.00

What to notice:

• The stock count lowers FOODS_1_085_CA_1's stock on hand and inventory position from 73 to 67 units, so its order rises from 28 to 32: the policy fills the larger gap up to S, and the case rule rounds it to a multiple of 4.
• The late delivery leaves FOODS_1_046_CA_1 with 18 units on the shelf instead of 42, but its inventory position stays at 42, because the 24 late units are still on order. Its order is unchanged at 68: the policy does not reorder units that are late, only units that are missing, as inside the engine in Notebook 05f.
• The other items are unchanged.

Only the starting state changed; forecast, policy and supplier rules ran as on any other day.

10. Where each piece runs in production¶

The job is the two pure functions of section 5 plus the two phase functions of section 6; everything else is infrastructure the application already has. The same run_plan and run_close fit every common setup:

Piece	In this notebook	Typical production choice
Trigger	event_stream and DailyJobConsumer	a consumer on Kafka, Kinesis or Pub/Sub; a cron or Airflow/Dagster schedule at 07:00 and 22:00; or a serverless function started when the day's sales file lands in object storage
States	JobStore.save_state / load_state	a database table keyed by date and phase, written in one transaction
Sales history	JobStore.save_sales, one file per date	the sales warehouse table the forecast reads, written by date so a rerun replaces a day
Outbox	JobStore.send_orders, one file per date, with order ids	the ERP or supplier API, which accepts an order id once
Decision records	JobStore.save_record, one file per review date	a table next to the orders, keyed by date and item, for planners and later checks
Monitoring	the shadow backtest and the fill-rate alert	a weekly job with alerts to the planning team
Setting	JOB_VERSION and the constants of section 1	configuration under version control, recorded with every decision

A few practices keep such a job safe:

• Change the setting through a backtest. A new probability, forecast or supplier rule is first replayed with SimulationEngine on recent sales (as in Notebook 09), then released with a new JOB_VERSION.
• Test the pure functions. plan and close take tables and return tables, so a test can feed them a saved day and check the orders.
• Never lose sales events. A consumer that restarts during the day must not lose the running totals: brokers replay events from the last committed position, or the close reads the day's totals from the sales system. Late events for a closed day go to the next day's records and are fixed by the next stock count.
• Trust the records over the saved state when they disagree (section 9), and keep both versions.

Callbacks, inventory processes and delivery outcomes run only inside SimulationEngine. In a live job their effects reach the state through the stock count and the open-order list, as in section 9.