Utilities¶

Demand generation and the primitives for a caller-owned loop. Guides: Demand and calendars, Run your own simulation loop.

DemandGenerator ¶

DemandGenerator(
    skus: Union[List[str], ndarray],
    *,
    first_date: Timestamp,
    freq: str,
    random_seed: int | None,
    negative_demand_handling: Literal[
        "raise", "clip_zero"
    ] = "raise",
    sku_column: str = "unique_id",
    date_column: str = "date",
    period_column: str = "period",
    demand_column: str = "y",
)

Generates multi-SKU demand DataFrames for inventory simulation.

All generation methods produce DataFrames with columns

[unique_id, y, period, date]

(names set by sku_column, demand_column, period_column and date_column).

Parameters accept scalars (same for all SKUs) or dicts keyed by SKU for per-SKU configuration. sample accepts any sampler function in the same way: one for the whole panel, or a dict with one per SKU. Negative draws are rejected by default, or clipped to zero with a warning when negative_demand_handling='clip_zero'.

Parameters:

Name	Type	Description	Default
`skus`	`Union[List[str], ndarray]`	List of SKU identifiers.	required
`first_date`	`Timestamp`	Date of demand period 0, the first period generated. For `SimulationEngine` this is one period after the inventory's opening date. It must be a period date of `freq` (for example a Monday for `"W-MON"`).	required
`freq`	`str`	Length of one period, a pandas frequency such as `"D"`.	required
`random_seed`	`int \| None`	Explicit random seed, or `None` for intentionally unseeded data.	required
`negative_demand_handling`	`Literal['raise', 'clip_zero']`	Explicitly reject or clip negative draws.	`'raise'`
`sku_column`	`str`	Name of the SKU column in the output (default `"unique_id"`).	`'unique_id'`
`date_column`	`str`	Name of the date column in the output (default `"date"`).	`'date'`
`period_column`	`str`	Name of the period column in the output (default `"period"`).	`'period'`
`demand_column`	`str`	Name of the demand column in the output (default `"y"`).	`'y'`

Example

gen = DemandGenerator(
    ['A', 'B', 'C'],
    first_date='2025-01-01',
    freq='D',
    random_seed=42,
    negative_demand_handling='clip_zero',
)

# Batch: full DataFrame
df = gen.normal(n_periods=30, mean=100, std=20)

# Any distribution: a sampler(rng, periods) -> one value per period
df = gen.sample(30, lambda rng, periods: rng.poisson(100, periods.size))

constant ¶

constant(
    n_periods: int, value: Union[float, Dict[str, float]]
) -> DataFrame

Generate constant demand.

Parameters:

Name	Type	Description	Default
`n_periods`	`int`	Number of periods to generate.	required
`value`	`Union[float, Dict[str, float]]`	Demand per period. Scalar or dict keyed by SKU.	required

Returns:

Type	Description
`DataFrame`	DataFrame with columns [unique_id, y, period, date].

normal ¶

normal(
    n_periods: int,
    mean: Union[float, Dict[str, float]],
    std: Union[float, Dict[str, float]],
) -> DataFrame

Generate normally distributed demand.

Parameters:

Name	Type	Description	Default
`n_periods`	`int`	Number of periods to generate.	required
`mean`	`Union[float, Dict[str, float]]`	Mean demand. Scalar or dict keyed by SKU.	required
`std`	`Union[float, Dict[str, float]]`	Standard deviation. Scalar or dict keyed by SKU.	required

Returns:

Type	Description
`DataFrame`	DataFrame with columns [unique_id, y, period, date].

seasonal ¶

seasonal(
    n_periods: int,
    base: Union[float, Dict[str, float]],
    amplitude: Union[float, Dict[str, float]],
    season_length: int,
    std: Union[float, Dict[str, float]],
) -> DataFrame

Generate demand with sinusoidal seasonal pattern + noise.

Formula: demand = base + amplitude * sin(2π * t / season_length) + noise(0, std)

Parameters:

Name	Type	Description	Default
`n_periods`	`int`	Number of periods to generate.	required
`base`	`Union[float, Dict[str, float]]`	Base demand level. Scalar or dict keyed by SKU.	required
`amplitude`	`Union[float, Dict[str, float]]`	Seasonal swing amplitude. Scalar or dict keyed by SKU.	required
`season_length`	`int`	Explicit length of one seasonal cycle in periods.	required
`std`	`Union[float, Dict[str, float]]`	Explicit noise standard deviation.	required

Returns:

Type	Description
`DataFrame`	DataFrame with columns [unique_id, y, period, date].

trend ¶

trend(
    n_periods: int,
    initial: Union[float, Dict[str, float]],
    growth_rate: Union[float, Dict[str, float]],
    std: Union[float, Dict[str, float]],
) -> DataFrame

Generate demand with linear trend + noise.

Formula: demand = initial + growth_rate * t + noise(0, std)

Parameters:

Name	Type	Description	Default
`n_periods`	`int`	Number of periods to generate.	required
`initial`	`Union[float, Dict[str, float]]`	Starting demand level. Scalar or dict keyed by SKU.	required
`growth_rate`	`Union[float, Dict[str, float]]`	Demand increase per period. Scalar or dict keyed by SKU.	required
`std`	`Union[float, Dict[str, float]]`	Explicit noise standard deviation.	required

Returns:

Type	Description
`DataFrame`	DataFrame with columns [unique_id, y, period, date].

normal_from_history ¶

normal_from_history(
    historical_df: DataFrame,
    n_periods: int,
    *,
    demand_column: Optional[str] = None,
    sku_column: Optional[str] = None,
) -> DataFrame

Normal demand with each SKU's historical mean and standard deviation.

The same as normal with mean and std estimated per SKU from historical_df (sample standard deviation, ddof=1). For any other distribution, for example resampling the history itself, use sample.

Parameters:

Name	Type	Description	Default
`historical_df`	`DataFrame`	DataFrame with historical demand data.	required
`n_periods`	`int`	Number of periods to generate.	required
`demand_column`	`Optional[str]`	Demand column of `historical_df`; defaults to the generator's `demand_column`, so its own output can be passed.	`None`
`sku_column`	`Optional[str]`	SKU column of `historical_df`; defaults to the generator's `sku_column`.	`None`

Returns:

Type	Description
`DataFrame`	DataFrame with the generator's columns (default
`DataFrame`	`[unique_id, y, period, date]`).

sample ¶

sample(
    n_periods: int,
    sampler: Union[Callable, Dict[str, Callable]],
) -> DataFrame

Generate demand from any sampler function.

A sampler is a function sampler(rng, periods) that returns one demand value per entry of periods. rng is the generator's seeded np.random.Generator; draw from it so the panel is reproducible from random_seed. periods holds the integer period indices, so demand can change over time.

Pass one sampler to generate every SKU in the panel from it (each SKU gets its own draws), or a dict with one sampler per SKU.

Parameters:

Name	Type	Description	Default
`n_periods`	`int`	Number of periods to generate.	required
`sampler`	`Union[Callable, Dict[str, Callable]]`	A sampler for the whole panel, or a dict keyed by SKU.	required

Returns:

Type	Description
`DataFrame`	DataFrame with columns [unique_id, y, period, date].

Example

def poisson(rng, periods):
    return rng.poisson(6.0, periods.size)

def busy_weekends(rng, periods):
    return rng.poisson(6.0 + 3.0 * (periods % 7 >= 5))

panel = gen.sample(28, poisson)
mixed = gen.sample(28, {"A": poisson, "B": busy_weekends})

update_inventory_with_orders ¶

update_inventory_with_orders(
    inventory_state: InventoryStateDataFrame,
    orders: OrderDecision,
    policy: object | None = None,
) -> InventoryStateDataFrame

Update inventory state by placing orders.

This function updates multiple columns to track the order placement

'in_transit': Orders added to array at appropriate period offset
'latest_order': Records the order quantity placed
'target_level': Stores the target level (S) from the policy

Positive-lead-time orders enter the future pipeline. Zero-lead-time orders are received immediately, clear prior backlog first, and can serve demand when this operation is called before fulfillment.

Inventory Management Logic

Lead time is automatically inferred from orders.lead_time (set by policy)
The state's backorder mode applies; a supplied policy must use the same mode
Positive-lead-time orders enter pipeline index lead_time - 1; zero-lead orders are received.
'latest_order' and 'target_level' are updated for tracking
The 'inventory_position' increases immediately: IP = on_hand + sum(in_transit) - backorders
Physical on-hand increases at receipt, immediately for zero lead time

Parameters:

Name	Type	Description	Default
`inventory_state`	`InventoryStateDataFrame`	Current multi-SKU inventory state	required
`orders`	`OrderDecision`	Order decisions to execute (must include 'order_quantity', 'target_level', 'lead_time')	required
`policy`	`object \| None`	Optional policy object; its allow_backorders must match the state's	`None`

Returns:

Type	Description
`InventoryStateDataFrame`	New InventoryStateDataFrame with updated 'in_transit', 'latest_order' and 'target_level'

Raises:

Type	Description
`ValueError`	If orders.lead_time is None, max_lead_time is insufficient, or the policy's allow_backorders differs from the state's

Example

from stockcast.core import InventoryStateDataFrame
from stockcast.utils import update_inventory_with_orders

# Current inventory state
inventory = InventoryStateDataFrame(inventory_df, max_lead_time=14)

# Generate orders from policy (automatically includes lead_time)
orders = policy.predict(inventory, current_period=10)

# Place orders (lead_time inferred automatically)
new_inventory = update_inventory_with_orders(inventory, orders, policy=policy)
# → in_transit updated, latest_order and target_level recorded

place_order_lines ¶

place_order_lines(
    inventory_state: InventoryStateDataFrame,
    order_lines: OrderLines,
    policy: object | None = None,
) -> InventoryStateDataFrame

Place order lines with their own suppliers and due periods.

This is the order-level counterpart of update_inventory_with_orders. Each row of order_lines is one scheduled delivery. Deliveries due in the current period are received immediately (clearing old backlog first, as zero-lead-time orders are); later deliveries enter the in_transit slot of their due period. latest_order accumulates each SKU's total. target_level is left unchanged because order lines carry no policy target.

The state's open-order book records every line (see InventoryStateDataFrame.open_orders()). A state whose pipeline was given only as in_transit gets that pipeline attributed to opening orders first.

Mapping from the per-SKU API: update_inventory_with_orders with lead time L is the same as one order line per positive SKU order with due_period = order_period + L.

Parameters:

Name	Type	Description	Default
`inventory_state`	`InventoryStateDataFrame`	Ready multi-SKU inventory state	required
`order_lines`	`OrderLines`	Order lines; every positive row must have `order_period` equal to the state period and `due_period <= period + max_lead_time`	required
`policy`	`object \| None`	Optional policy; its `allow_backorders` must match the state's	`None`

Returns:

Type	Description
`InventoryStateDataFrame`	New InventoryStateDataFrame with updated pipeline, receipts and book

Example

lines = OrderLines(pd.DataFrame({
    'unique_id': ['beans', 'beans'],
    'supplier_id': ['local', 'import'],
    'order_quantity': [20.0, 60.0],
    'order_period': [5, 5],
    'due_period': [6, 12],
}))
state = place_order_lines(state, lines)