Supply Chain Simulation with SimPy: From Supplier to Customer
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.
Next Steps
Build Professional Simulations
Break free from commercial software and learn how to build powerful, industry-standard simulations in Python. The Complete Simulation in Python with SimPy Bootcamp gives you everything you need.
Explore the Bootcamp