When the user wants to improve supplier relationships, implement collaboration programs, or enable information sharing with suppliers. Also use when the user mentions "supplier partnership," "collaborative planning," "VMI," "CPFR," "supplier portal," "information sharing," "supplier development," "co-innovation," or "strategic partnerships." For supplier selection, see supplier-selection. For supplier risk, see supplier-risk-management.
You are an expert in supplier collaboration and partnership management. Your goal is to help organizations build strategic supplier relationships, implement collaborative processes, and create mutual value through enhanced information sharing and joint planning.
Before implementing supplier collaboration, understand:
Collaboration Objectives
Supplier Segmentation
Current Relationship State
Organizational Readiness
Level 1: Transactional
Level 2: Cooperative
Level 3: Coordinated
Level 4: Collaborative
Level 5: Synchronized
1. Vendor-Managed Inventory (VMI)
2. Collaborative Planning, Forecasting, and Replenishment (CPFR)
3. Early Supplier Involvement (ESI)
4. Joint Business Planning (JBP)
5. Co-Innovation Partnerships
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
class VMIProgram:
"""Vendor-Managed Inventory program implementation"""
def __init__(self, buyer_id, supplier_id):
self.buyer_id = buyer_id
self.supplier_id = supplier_id
self.inventory_positions = []
self.replenishment_orders = []
self.performance_metrics = {}
def set_vmi_parameters(self, product_id, min_level, max_level,
order_multiple, lead_time_days, service_level=0.95):
"""
Set VMI parameters for a product
min_level: Reorder point
max_level: Maximum inventory level (order-up-to)
order_multiple: Minimum order quantity or multiple
"""
return {
'product_id': product_id,
'min_level': min_level,
'max_level': max_level,
'order_multiple': order_multiple,
'lead_time_days': lead_time_days,
'service_level': service_level,
'last_update': datetime.now()
}
def calculate_vmi_replenishment(self, current_inventory, pipeline_inventory,
daily_consumption, vmi_params):
"""
Calculate VMI replenishment quantity
current_inventory: On-hand inventory
pipeline_inventory: In-transit/on-order inventory
daily_consumption: Average daily consumption
vmi_params: VMI parameters dict
"""
# Inventory position = on-hand + pipeline
inventory_position = current_inventory + pipeline_inventory
# Check if replenishment needed
if inventory_position <= vmi_params['min_level']:
# Calculate order quantity to reach max level
target_inventory = vmi_params['max_level']
order_quantity = target_inventory - inventory_position
# Round to order multiple
order_multiple = vmi_params['order_multiple']
if order_multiple > 0:
order_quantity = np.ceil(order_quantity / order_multiple) * order_multiple
# Calculate days of supply
days_of_supply = order_quantity / daily_consumption if daily_consumption > 0 else 0
return {
'replenishment_needed': True,
'order_quantity': int(order_quantity),
'inventory_position': inventory_position,
'target_inventory': target_inventory,
'days_of_supply': round(days_of_supply, 1),
'reason': 'Inventory below minimum threshold'
}
else:
return {
'replenishment_needed': False,
'order_quantity': 0,
'inventory_position': inventory_position,
'days_of_supply': round(inventory_position / daily_consumption if daily_consumption > 0 else 0, 1),
'reason': 'Inventory sufficient'
}
def simulate_vmi_performance(self, demand_series, vmi_params, initial_inventory):
"""
Simulate VMI program performance
demand_series: Daily demand data
vmi_params: VMI parameters
initial_inventory: Starting inventory
"""
results = []
current_inventory = initial_inventory
pipeline = 0
orders_placed = []
for day, demand in enumerate(demand_series):
# Check for arriving orders
arriving_orders = [o for o in orders_placed if o['arrival_day'] == day]
for order in arriving_orders:
current_inventory += order['quantity']
pipeline -= order['quantity']
orders_placed.remove(order)
# Satisfy demand
demand_satisfied = min(demand, current_inventory)
stockout = max(0, demand - current_inventory)
current_inventory = max(0, current_inventory - demand)
# Check if replenishment needed
daily_avg_consumption = np.mean(demand_series[:day+1]) if day > 0 else demand
replenishment = self.calculate_vmi_replenishment(
current_inventory, pipeline, daily_avg_consumption, vmi_params
)
if replenishment['replenishment_needed']:
# Place order
order = {
'day': day,
'quantity': replenishment['order_quantity'],
'arrival_day': day + vmi_params['lead_time_days']
}
orders_placed.append(order)
pipeline += order['quantity']
results.append({
'day': day,
'demand': demand,
'demand_satisfied': demand_satisfied,
'stockout': stockout,
'ending_inventory': current_inventory,
'pipeline_inventory': pipeline,
'order_placed': replenishment['replenishment_needed'],
'order_quantity': replenishment['order_quantity']
})
df = pd.DataFrame(results)
# Calculate metrics
metrics = {
'avg_inventory': df['ending_inventory'].mean(),
'max_inventory': df['ending_inventory'].max(),
'min_inventory': df['ending_inventory'].min(),
'total_stockouts': df['stockout'].sum(),
'fill_rate': (df['demand_satisfied'].sum() / df['demand'].sum() * 100) if df['demand'].sum() > 0 else 100,
'orders_placed': df['order_placed'].sum(),
'avg_order_size': df[df['order_placed']]['order_quantity'].mean() if df['order_placed'].sum() > 0 else 0
}
return {
'simulation_results': df,
'performance_metrics': metrics
}
# Example VMI implementation
vmi = VMIProgram(buyer_id='BUYER001', supplier_id='SUP001')
# Set VMI parameters
vmi_params = vmi.set_vmi_parameters(
product_id='PROD001',
min_level=500, # Reorder point
max_level=2000, # Order-up-to level
order_multiple=100, # Order in multiples of 100
lead_time_days=7,
service_level=0.95
)
# Simulate 90 days of VMI
np.random.seed(42)
demand_series = np.random.poisson(lam=150, size=90) # Average demand of 150 units/day
simulation = vmi.simulate_vmi_performance(
demand_series, vmi_params, initial_inventory=1500
)
print("VMI Performance Metrics:")
print(f" Average Inventory: {simulation['performance_metrics']['avg_inventory']:.0f} units")
print(f" Fill Rate: {simulation['performance_metrics']['fill_rate']:.1f}%")
print(f" Total Stockouts: {simulation['performance_metrics']['total_stockouts']:.0f} units")
print(f" Orders Placed: {simulation['performance_metrics']['orders_placed']:.0f}")
print(f" Average Order Size: {simulation['performance_metrics']['avg_order_size']:.0f} units")
class CPFRProgram:
"""Collaborative Planning, Forecasting, and Replenishment program"""
def __init__(self, buyer_name, supplier_name):
self.buyer_name = buyer_name
self.supplier_name = supplier_name
self.shared_forecast = {}
self.exceptions = []
def create_joint_forecast(self, buyer_forecast, supplier_forecast, product_id):
"""
Create joint forecast from buyer and supplier inputs
Use weighted average or exception-based consensus
"""
if len(buyer_forecast) != len(supplier_forecast):
raise ValueError("Forecast lengths must match")
joint_forecast = []
exceptions = []
for i, (buyer_qty, supplier_qty) in enumerate(zip(buyer_forecast, supplier_forecast)):
# Calculate variance
variance_pct = abs(buyer_qty - supplier_qty) / ((buyer_qty + supplier_qty) / 2) * 100 if (buyer_qty + supplier_qty) > 0 else 0
# If variance > threshold, flag as exception
if variance_pct > 20: # 20% threshold
exceptions.append({
'period': i + 1,
'buyer_forecast': buyer_qty,
'supplier_forecast': supplier_qty,
'variance_pct': round(variance_pct, 1),
'status': 'Requires Resolution',
'consensus_qty': None
})
# Use weighted average (60% buyer, 40% supplier) pending resolution
consensus_qty = int(buyer_qty * 0.6 + supplier_qty * 0.4)
else:
# Use weighted average
consensus_qty = int(buyer_qty * 0.6 + supplier_qty * 0.4)
joint_forecast.append({
'period': i + 1,
'buyer_forecast': buyer_qty,
'supplier_forecast': supplier_qty,
'variance_pct': round(variance_pct, 1),
'consensus_forecast': consensus_qty,
'has_exception': variance_pct > 20
})
self.shared_forecast[product_id] = joint_forecast
self.exceptions.extend(exceptions)
return {
'joint_forecast': pd.DataFrame(joint_forecast),
'exceptions': pd.DataFrame(exceptions) if exceptions else pd.DataFrame(),
'forecast_accuracy_alignment': round(100 - variance_pct, 1) if variance_pct < 100 else 0
}
def resolve_exception(self, period, consensus_qty, resolution_notes):
"""Resolve forecast exception with agreed quantity"""
for exception in self.exceptions:
if exception['period'] == period and exception['status'] == 'Requires Resolution':
exception['consensus_qty'] = consensus_qty
exception['status'] = 'Resolved'
exception['resolution_notes'] = resolution_notes
exception['resolution_date'] = datetime.now()
# Update shared forecast
for product_id, forecast_list in self.shared_forecast.items():
for period_forecast in forecast_list:
if period_forecast['period'] == period:
period_forecast['consensus_forecast'] = consensus_qty
break
return True
return False
def calculate_collaboration_benefits(self, baseline_metrics, cpfr_metrics):
"""
Calculate benefits from CPFR collaboration
baseline_metrics: Pre-CPFR performance
cpfr_metrics: Post-CPFR performance
"""
# Forecast accuracy improvement
forecast_improvement = cpfr_metrics['forecast_accuracy'] - baseline_metrics['forecast_accuracy']
# Inventory reduction
inventory_reduction_pct = (baseline_metrics['avg_inventory'] - cpfr_metrics['avg_inventory']) / baseline_metrics['avg_inventory'] * 100
# Service level improvement
service_improvement = cpfr_metrics['fill_rate'] - baseline_metrics['fill_rate']
# Cost savings
inventory_cost_rate = 0.25 # 25% annual carrying cost
inventory_value_reduction = (baseline_metrics['avg_inventory'] - cpfr_metrics['avg_inventory']) * cpfr_metrics['unit_cost']
annual_savings = inventory_value_reduction * inventory_cost_rate
# Stockout reduction
stockout_reduction = baseline_metrics['stockout_events'] - cpfr_metrics['stockout_events']
stockout_cost_avoided = stockout_reduction * cpfr_metrics['stockout_cost_per_event']
total_annual_benefit = annual_savings + stockout_cost_avoided
return {
'forecast_accuracy_improvement_pts': round(forecast_improvement, 1),
'inventory_reduction_pct': round(inventory_reduction_pct, 1),
'inventory_value_reduction': round(inventory_value_reduction, 2),
'service_level_improvement_pts': round(service_improvement, 1),
'annual_inventory_savings': round(annual_savings, 2),
'stockout_cost_avoided': round(stockout_cost_avoided, 2),
'total_annual_benefit': round(total_annual_benefit, 2)
}
# Example CPFR implementation
cpfr = CPFRProgram(buyer_name='Acme Corp', supplier_name='Widget Manufacturing')
# Buyer and supplier forecasts for next 12 months
buyer_forecast = [1000, 1100, 1200, 1500, 1800, 1600, 1400, 1300, 1200, 1100, 1000, 950]
supplier_forecast = [1050, 1150, 1100, 1400, 2000, 1650, 1500, 1250, 1200, 1150, 1050, 1000]
# Create joint forecast
joint_forecast_result = cpfr.create_joint_forecast(
buyer_forecast, supplier_forecast, product_id='PROD001'
)
print("CPFR Joint Forecast:")
print(joint_forecast_result['joint_forecast'][['period', 'buyer_forecast', 'supplier_forecast', 'variance_pct', 'consensus_forecast', 'has_exception']])
print(f"\n\nExceptions Requiring Resolution:")
if not joint_forecast_result['exceptions'].empty:
print(joint_forecast_result['exceptions'])