Supply chains are networks of queues, buffers, and delays. Orders flow one way. Goods flow the other. Simulation reveals where they get stuck.
The Supply Chain Model
Key elements:
- Suppliers - Source of materials
- Warehouses - Storage buffers
- Manufacturing - Transformation
- Distribution - Delivery to customers
- Demand - Customer orders
Basic Inventory Model
import simpy
import random
class Warehouse:
def __init__(self, env, name, capacity, reorder_point, order_quantity):
self.env = env
self.name = name
self.inventory = simpy.Container(env, capacity=capacity, init=capacity//2)
self.reorder_point = reorder_point
self.order_quantity = order_quantity
self.orders_placed = 0
self.stockouts = 0
def fulfill_demand(self, quantity):
"""Try to fulfill demand from inventory."""
if self.inventory.level >= quantity:
yield self.inventory.get(quantity)
return True
else:
self.stockouts += 1
return False
def receive_shipment(self, quantity):
"""Receive goods into inventory."""
yield self.inventory.put(quantity)
print(f"{self.name}: Received {quantity}, level now {self.inventory.level}")
def reorder_check(self, lead_time):
"""Check and reorder when inventory is low."""
while True:
yield self.env.timeout(1) # Check daily
if self.inventory.level <= self.reorder_point:
self.orders_placed += 1
print(f"{self.name}: Ordering {self.order_quantity} at {self.env.now}")
yield self.env.timeout(lead_time)
yield from self.receive_shipment(self.order_quantity)Multi-Echelon Supply Chain
class SupplyChainNode:
def __init__(self, env, name, capacity, upstream=None):
self.env = env
self.name = name
self.inventory = simpy.Container(env, capacity=capacity, init=capacity//2)
self.upstream = upstream
self.stats = {
'fulfilled': 0,
'stockouts': 0,
'orders_placed': 0
}
def request_material(self, quantity, lead_time):
"""Request material from upstream."""
if self.upstream:
# Place order with upstream
self.stats['orders_placed'] += 1
yield self.env.timeout(lead_time)
success = yield from self.upstream.fulfill_order(quantity)
if success:
yield self.inventory.put(quantity)
else:
# We are the source - unlimited supply
yield self.env.timeout(lead_time)
yield self.inventory.put(quantity)
def fulfill_order(self, quantity):
"""Fulfill order from downstream."""
if self.inventory.level >= quantity:
yield self.inventory.get(quantity)
self.stats['fulfilled'] += 1
return True
else:
self.stats['stockouts'] += 1
return False
# Create chain: Supplier -> DC -> Store
env = simpy.Environment()
supplier = SupplyChainNode(env, "Supplier", capacity=1000, upstream=None)
dc = SupplyChainNode(env, "DC", capacity=500, upstream=supplier)
store = SupplyChainNode(env, "Store", capacity=100, upstream=dc)Order Fulfilment
class OrderManager:
def __init__(self, env, warehouse):
self.env = env
self.warehouse = warehouse
self.pending_orders = simpy.Store(env)
self.completed_orders = []
def place_order(self, order_id, quantity, customer):
order = {
'id': order_id,
'quantity': quantity,
'customer': customer,
'placed_at': self.env.now
}
yield self.pending_orders.put(order)
def process_orders(self):
while True:
order = yield self.pending_orders.get()
# Check inventory
if self.warehouse.inventory.level >= order['quantity']:
yield self.warehouse.inventory.get(order['quantity'])
order['fulfilled_at'] = self.env.now
order['status'] = 'fulfilled'
else:
order['status'] = 'backordered'
# Wait for restock then retry
yield self.env.timeout(5)
continue
self.completed_orders.append(order)
print(f"Order {order['id']} fulfilled at {self.env.now}")Transportation and Delivery
class DeliveryNetwork:
def __init__(self, env, num_trucks, warehouse):
self.env = env
self.trucks = simpy.Resource(env, capacity=num_trucks)
self.warehouse = warehouse
self.deliveries = []
def delivery(self, order_id, destination, quantity):
"""Pick, load, deliver."""
arrival = self.env.now
# Wait for truck
with self.trucks.request() as req:
yield req
# Pick from warehouse
yield self.warehouse.inventory.get(quantity)
# Load time
yield self.env.timeout(random.uniform(10, 30))
# Transit time (based on destination)
transit = destination * 0.5 # 0.5 hours per km
yield self.env.timeout(transit)
self.deliveries.append({
'order_id': order_id,
'lead_time': self.env.now - arrival
})
# Return journey (empty)
yield self.env.timeout(transit)
def stats(self):
times = [d['lead_time'] for d in self.deliveries]
return {
'deliveries': len(times),
'avg_lead_time': sum(times) / len(times) if times else 0
}Demand Forecasting Impact
def demand_generator(env, order_manager, demand_pattern):
"""Generate customer demand."""
order_id = 0
while True:
# Demand varies by day of week
day = int(env.now / 24) % 7
daily_demand = demand_pattern[day]
quantity = max(1, int(random.gauss(daily_demand, daily_demand * 0.2)))
env.process(order_manager.place_order(order_id, quantity, f"Customer"))
yield env.timeout(random.expovariate(1/4)) # ~4 hours between orders
order_id += 1
# Weekly pattern (Mon-Sun)
demand_pattern = [100, 110, 120, 115, 130, 80, 60]Complete Supply Chain Example
import simpy
import random
import numpy as np
class SupplyChainSimulation:
def __init__(self, env, config):
self.env = env
self.config = config
# Create nodes
self.supplier = simpy.Container(env, capacity=float('inf'), init=float('inf'))
self.dc = simpy.Container(env, capacity=config['dc_capacity'],
init=config['dc_initial'])
self.store = simpy.Container(env, capacity=config['store_capacity'],
init=config['store_initial'])
self.stats = {
'sales': 0,
'stockouts': 0,
'inventory_levels': [],
'dc_orders': 0,
'supplier_orders': 0
}
def customer_demand(self):
"""Generate customer demand at store."""
while True:
# Demand arrives
yield self.env.timeout(random.expovariate(self.config['demand_rate']))
demand = random.randint(1, 5)
if self.store.level >= demand:
yield self.store.get(demand)
self.stats['sales'] += demand
else:
# Partial fulfillment
if self.store.level > 0:
fulfilled = self.store.level
yield self.store.get(fulfilled)
self.stats['sales'] += fulfilled
self.stats['stockouts'] += 1
def store_reorder(self):
"""Store reorders from DC."""
while True:
yield self.env.timeout(1) # Daily check
if self.store.level <= self.config['store_reorder_point']:
order_qty = self.config['store_order_qty']
self.stats['dc_orders'] += 1
# Check DC inventory
if self.dc.level >= order_qty:
yield self.dc.get(order_qty)
yield self.env.timeout(self.config['dc_to_store_lead'])
yield self.store.put(order_qty)
def dc_reorder(self):
"""DC reorders from supplier."""
while True:
yield self.env.timeout(1)
if self.dc.level <= self.config['dc_reorder_point']:
order_qty = self.config['dc_order_qty']
self.stats['supplier_orders'] += 1
yield self.env.timeout(self.config['supplier_lead_time'])
yield self.dc.put(order_qty)
def monitor(self, interval=1):
"""Track inventory levels."""
while True:
self.stats['inventory_levels'].append({
'time': self.env.now,
'store': self.store.level,
'dc': self.dc.level
})
yield self.env.timeout(interval)
def run(self, duration):
self.env.process(self.customer_demand())
self.env.process(self.store_reorder())
self.env.process(self.dc_reorder())
self.env.process(self.monitor())
self.env.run(until=duration)
def report(self):
print("\n=== Supply Chain Report ===")
print(f"Duration: {self.env.now}")
print(f"Total sales: {self.stats['sales']}")
print(f"Stockouts: {self.stats['stockouts']}")
print(f"DC orders: {self.stats['dc_orders']}")
print(f"Supplier orders: {self.stats['supplier_orders']}")
fill_rate = self.stats['sales'] / (self.stats['sales'] + self.stats['stockouts'])
print(f"Fill rate: {fill_rate:.1%}")
levels = self.stats['inventory_levels']
store_levels = [l['store'] for l in levels]
dc_levels = [l['dc'] for l in levels]
print(f"\nAvg store inventory: {np.mean(store_levels):.0f}")
print(f"Avg DC inventory: {np.mean(dc_levels):.0f}")
# Configuration
config = {
'dc_capacity': 500,
'dc_initial': 300,
'dc_reorder_point': 150,
'dc_order_qty': 200,
'store_capacity': 100,
'store_initial': 60,
'store_reorder_point': 30,
'store_order_qty': 50,
'dc_to_store_lead': 2,
'supplier_lead_time': 7,
'demand_rate': 1/0.5 # 2 per day
}
random.seed(42)
env = simpy.Environment()
sim = SupplyChainSimulation(env, config)
sim.run(duration=90) # 90 days
sim.report()Summary
Supply chain simulation reveals:
- Inventory positioning and levels
- Lead time impacts
- Bullwhip effect dynamics
- Order policy optimisation
- Network bottlenecks
Supply chains are long queues. Simulate to shorten them.