关于 multi-echelon-inventory
这是用于协调供应链多个层级库存策略的优化技能，面向库存规划与供应链运营场景。它结合网络结构、需求、提前期、成本和服务水平来制定全网库存策略。
主要功能
- 01分析多层级网络与库存位置
- 02涵盖基库存策略与保障服务模型
- 03纳入需求变异和相关性
- 04分析集中化的风险池化效应
- 05考虑持有成本与服务水平权衡
适用场景
- 01协调中心仓、区域仓与零售门店补货
- 02评估集中库存的安全库存变化
- 03按地点或产品类别设定服务目标
- 04诊断库存可视性与需求数据问题
安装
复制下面的命令，在终端中运行。
$
npx skills add https:/安装后在 Claude Code 中输入/multi-echelon-inventory确认已加载。
详细介绍
多级库存优化
你是多级库存系统和全网库存优化方面的专家。你的目标是帮助协调供应链网络中多个地点的库存策略，在满足服务水平要求的同时，最大限度地降低系统总成本。
初步评估
在优化多级库存之前，请先了解：
-
网络结构
- 有多少级/阶段？（例如：中央配送中心 → 区域配送中心 → 门店）
- 网络拓扑结构是什么？（串行、分销/树状、一般网络）
- 每个层级有多少个地点？
- 当前采用什么库存部署策略？
-
产品特征
- SKU 的复杂程度如何？（产品数量）
- 产品价值和周转速度如何（ABC 分类）？
- 是否存在产品保质期或过时方面的问题？
- 是否存在蚕食或替代效应？
-
需求模式
- 外部需求发生在哪里？（零售门店的终端客户）
- 各地点的需求波动情况如何？
- 各地点的需求之间是否相关？
- 是否存在季节性或趋势？
-
供应链流动
- 各级之间的提前期有多长？
- 各地点采用什么补货策略？
- 各地点是否可以横向调拨（同级地点之间）？
- 是否可以紧急/加急运输？
-
成本结构
- 各级的持有成本是多少？
- 地点之间的运输成本是多少？
- 订购/补货成本是多少？
- 缺货或延期交货的惩罚成本是多少？
- 成本是否因地点而异？
-
服务要求
- 各地点或整体的目标服务水平是多少？
- 哪些是关键物品，哪些是非关键物品？
- 客户优先级如何？
多级系统基础
网络结构
1. 串行系统（顺序）
Supplier → Factory → Central DC → Regional DC → Retail Store → Customer
(0) (1) (2) (3) (4)
2. 分销/树状系统（树形）
Central Warehouse
/ | \
DC1 DC2 DC3
/ | \ / | \ / | \
S1 S2 S3 S4 S5 S6 S7 S8 S9
3. 一般网络
- 多个供应商
- 同级地点之间横向调拨
- 复杂的流动
核心概念
安装库存（IS）：
- 特定地点的实物库存
- 盘点实物库存时会统计的库存
级库存（ES）：
- 安装库存 + 系统中的所有下游库存
- ES(i) = IS(i) + Σ 所有下游地点 j 的 ES(j)
基准库存策略：
- 订至目标策略：当库存位置 ≤ S 时，下单补货至 S
- S = 基准库存水平
保障服务模型（GSM）：
- 每个上游地点向下游地点保证服务时间
- 下游地点可以根据保证的服务时间进行规划
- 将多级问题拆分为单级问题
Python 实现：多级模型
采用基准库存策略的串行系统
import numpy as np
import pandas as pd
from scipy import stats
from typing import List, Dict, Tuple
import matplotlib.pyplot as plt
from collections import deque
class SerialMultiEchelonSystem:
"""
Serial multi-echelon inventory system with base stock policies
Models a sequential supply chain: Supplier → Stage 1 → Stage 2 → ... → Customer
"""
def __init__(self, num_stages: int, lead_times: List[int],
holding_costs: List[float], demand_mean: float,
demand_std: float):
"""
Parameters:
-----------
num_stages : int
Number of stages in the serial system
lead_times : list
Lead time (in periods) for each stage [L1, L2, ..., Ln]
holding_costs : list
Holding cost per unit per period at each stage [h1, h2, ..., hn]
demand_mean : float
Mean demand per period at final stage
demand_std : float
Standard deviation of demand per period
"""
self.num_stages = num_stages
self.lead_times = np.array(lead_times)
self.holding_costs = np.array(holding_costs)
self.demand_mean = demand_mean
self.demand_std = demand_std
# Echelon holding costs
self.echelon_holding_costs = self._calculate_echelon_holding_costs()
def _calculate_echelon_holding_costs(self) -> np.ndarray:
"""
Calculate echelon holding costs
Echelon holding cost h'(j) = h(j) - h(j+1)
where h(j) is installation holding cost at stage j
"""
echelon_costs = np.zeros(self.num_stages)
for j in range(self.num_stages):
echelon_costs[j] = self.holding_costs[j]
if j < self.num_stages - 1:
echelon_costs[j] -= self.holding_costs[j + 1]
return echelon_costs
def calculate_base_stock_levels(self, service_level: float = 0.95) -> Dict:
"""
Calculate optimal base stock levels using Clark-Scarf decomposition
For serial system with base stock policies, can decompose into
single-stage newsvendor problems for each echelon
Parameters:
-----------
service_level : float
Target service level (e.g., 0.95 for 95%)
Returns:
--------
dict with base stock levels and costs
"""
z = stats.norm.ppf(service_level)
base_stocks = np.zeros(self.num_stages)
safety_stocks = np.zeros(self.num_stages)
# Work backwards from downstream to upstream
for j in range(self.num_stages - 1, -1, -1):
# Cumulative lead time from stage j to end customer
if j == self.num_stages - 1:
cumulative_lt = self.lead_times[j]
else:
cumulative_lt = np.sum(self.lead_times[j:])
# Expected demand during cumulative lead time
expected_demand = self.demand_mean * cumulative_lt
# Standard deviation during cumulative lead time
std_demand = self.demand_std * np.sqrt(cumulative_lt)
# Safety stock at this echelon
safety_stock = z * std_demand
# Base stock level (echelon stock)
base_stock = expected_demand + safety_stock
base_stocks[j] = base_stock
safety_stocks[j] = safety_stock
# Calculate expected costs
total_holding_cost = np.sum(self.echelon_holding_costs * safety_stocks)
return {
'base_stock_levels': base_stocks,
'safety_stocks': safety_stocks,
'echelon_holding_costs': self.echelon_holding_costs,
'total_holding_cost': total_holding_cost,
'service_level': service_level
}
def optimize_service_level(self, max_holding_cost: float) -> Dict:
"""
Find optimal service level given a holding cost budget
Parameters:
-----------
max_holding_cost : float
Maximum acceptable holding cost per period
Returns:
--------
Optimal service level and corresponding base stocks
"""
# Binary search for optimal service level
sl_low, sl_high = 0.50, 0.9999
while sl_high - sl_low > 0.0001:
sl_mid = (sl_low + sl_high) / 2
result = self.calculate_base_stock_levels(sl_mid)
if result['total_holding_cost'] <= max_holding_cost:
sl_low = sl_mid
else:
sl_high = sl_mid
optimal_result = self.calculate_base_stock_levels(sl_low)
return optimal_result
def simulate(self, num_periods: int, base_stock_levels: np.ndarray,
review_period: int = 1) -> pd.DataFrame:
"""
Simulate multi-echelon system with base stock policies
Parameters:
-----------
num_periods : int
Number of periods to simulate
base_stock_levels : array
Base stock (order-up-to) level for each stage
review_period : int
How often to review inventory (default 1 = continuous)
Returns:
--------
DataFrame with inventory positions, orders, and costs
"""
# Initialize inventory positions (start at base stock)
inventory_position = base_stock_levels.copy()
inventory_on_hand = base_stock_levels.copy()
# Order pipelines (orders in transit)
order_pipelines = [deque() for _ in range(self.num_stages)]
results = []
for t in range(num_periods):
# Generate customer demand (at final stage)
demand = max(0, np.random.normal(self.demand_mean, self.demand_std))
# Process orders arriving at each stage
for j in range(self.num_stages):
if len(order_pipelines[j]) > 0:
# Check for arriving orders
arrivals = []
for order in order_pipelines[j]:
if order['arrival_period'] == t:
arrivals.append(order)
for order in arrivals:
inventory_on_hand[j] += order['quantity']
order_pipelines[j].remove(order)
# Satisfy demand at final stage
actual_demand_filled = min(demand, inventory_on_hand[-1])
stockout = demand - actual_demand_filled
inventory_on_hand[-1] -= actual_demand_filled
# Review and order at each stage (if review period)
orders_placed = np.zeros(self.num_stages)
if t % review_period == 0:
for j in range(self.num_stages - 1, -1, -1):
# Calculate inventory position
pipeline_inventory = sum(order['quantity']
for order in order_pipelines[j])
inventory_position[j] = inventory_on_hand[j] + pipeline_inventory
# Order up to base stock
order_qty = max(0, base_stock_levels[j] - inventory_position[j])
if order_qty > 0:
# Place order (arrives after lead time)
order_pipelines[j].append({
'quantity': order_qty,
'order_period': t,
'arrival_period': t + self.lead_times[j]
})
orders_placed[j] = order_qty
# Calculate holding costs
holding_cost = np.sum(self.holding_costs * inventory_on_hand)
results.append({
'period': t,
'demand': demand,
'stockout': stockout,
'fill_rate': actual_demand_filled / demand if demand > 0 else 1.0,
**{f'inv_stage_{j}': inventory_on_hand[j] for j in range(self.num_stages)},
**{f'order_stage_{j}': orders_placed[j] for j in range(self.num_stages)},
'holding_cost': holding_cost
})
return pd.DataFrame(results)
# Example: Serial System
def example_serial_system():
"""Example: 3-stage serial supply chain"""
print("\n" + "=" * 70)
print("MULTI-ECHELON SERIAL SYSTEM OPTIMIZATION")
print("=" * 70)
# Define system: Supplier → Warehouse → Distribution Center → Retailer
system = SerialMultiEchelonSystem(
num_stages=3,
lead_times=[10, 5, 2], # Days: Supplier→WH, WH→DC, DC→Retail
holding_costs=[1.0, 2.0, 5.0], # $/unit/day: WH, DC, Retail
demand_mean=100, # Daily demand at retail
demand_std=20 # Demand variability
)
print("\nSystem Configuration:")
print(f" Number of Stages: {system.num_stages}")
print(f" Lead Times: {system.lead_times} days")
print(f" Installation Holding Costs: ${system.holding_costs} per unit per day")
print(f" Echelon Holding Costs: ${system.echelon_holding_costs} per unit per day")
print(f" Demand: Mean={system.demand_mean}, Std={system.demand_std}")
# Calculate optimal base stocks for 95% service level
result = system.calculate_base_stock_levels(service_level=0.95)
print(f"\n{'Optimal Base Stock Levels (95% Service Level):'}")
print(f"\n{'Stage':<20} {'Lead Time':<12} {'Base Stock':<15} {'Safety Stock':<15} {'Echelon Cost':<15}")
print("-" * 77)
stage_names = ['Warehouse', 'Distribution', 'Retail']
for j in range(system.num_stages):
print(f"{stage_names[j]:<20} {system.lead_times[j]:<12} "
f"{result['base_stock_levels'][j]:<15.0f} "
f"{result['safety_stocks'][j]:<15.0f} "
f"${result['echelon_holding_costs'][j]:<14.2f}")
print(f"\n{'Total Expected Holding Cost:':<40} ${result['total_holding_cost']:.2f} per period")
# Simulate system
print("\nSimulating system for 365 days...")
np.random.seed(42)
simulation = system.simulate(
num_periods=365,
base_stock_levels=result['base_stock_levels'],
review_period=1
)
# Analyze results
avg_fill_rate = simulation['fill_rate'].mean()
total_stockouts = simulation['stockout'].sum()
avg_holding_cost = simulation['holding_cost'].mean()
print(f"\n{'Simulation Results (365 days):'}")
print(f" {'Average Fill Rate:':<35} {avg_fill_rate:.2%}")
print(f" {'Total Stockout Units:':<35} {total_stockouts:.0f}")
print(f" {'Average Daily Holding Cost:':<35} ${avg_holding_cost:.2f}")
print(f" {'Average Inventory per Stage:':<35}")
for j in range(system.num_stages):
avg_inv = simulation[f'inv_stage_{j}'].mean()
print(f" {stage_names[j]:<32} {avg_inv:.0f} units")
return system, result, simulation
if __name__ == "__main__":
system, result, simulation = example_serial_system()
分销系统（树状网络）
两级分销系统
class TwoEchelonDistributionSystem:
"""
Two-echelon distribution system: 1 central warehouse → N retailers
Implements guaranteed service model approach
"""
def __init__(self, num_retailers: int, central_lead_time: int,
delivery_lead_times: List[int], central_holding_cost: float,
retailer_holding_costs: List[float], demands_mean: List[float],
demands_std: List[float]):
"""
Parameters:
-----------
num_retailers : int
Number of retail locations
central_lead_time : int
Lead time from supplier to central warehouse
delivery_lead_times : list
Lead time from central to each retailer
central_holding_cost : float
Holding cost per unit per period at central warehouse
retailer_holding_costs : list
Holding cost per unit per period at each retailer
demands_mean : list
Mean demand per period at each retailer
demands_std : list
Std dev of demand per period at each retailer
"""
self.num_retailers = num_retailers
self.L0 = central_lead_time
self.Li = np.array(delivery_lead_times)
self.h0 = central_holding_cost
self.hi = np.array(retailer_holding_costs)
self.mu = np.array(demands_mean)
self.sigma = np.array(demands_std)
def calculate_base_stocks_guaranteed_service(self,
retailer_service_levels: List[float],
central_service_level: float) -> Dict:
"""
Calculate base stocks using Guaranteed Service Model
Approach:
1. Each retailer sets safety stock based on delivery lead time
2. Central warehouse sets safety stock to guarantee service time
"""
# Retailer base stocks
retailer_base_stocks = np.zeros(self.num_retailers)
retailer_safety_stocks = np.zeros(self.num_retailers)
for i in range(self.num_retailers):
z_i = stats.norm.ppf(retailer_service_levels[i])
# Expected demand during delivery lead time
expected_demand = self.mu[i] * self.Li[i]
# Safety stock
safety_stock = z_i * self.sigma[i] * np.sqrt(self.Li[i])
retailer_base_stocks[i] = expected_demand + safety_stock
retailer_safety_stocks[i] = safety_stock
# Central warehouse base stock
z_0 = stats.norm.ppf(central_service_level)
# Aggregate demand statistics
total_mean_demand = np.sum(self.mu)
# Assuming independent retailer demands
total_std_demand = np.sqrt(np.sum(self.sigma ** 2))
# Total lead time (external + internal)
total_lt = self.L0 + np.max(self.Li)
# Central expected demand
central_expected = total_mean_demand * total_lt
# Central safety stock
central_safety_stock = z_0 * total_std_demand * np.sqrt(total_lt)
central_base_stock = central_expected + central_safety_stock
# Calculate costs
retailer_holding_cost = np.sum(self.hi * retailer_safety_stocks)
central_holding_cost = self.h0 * central_safety_stock
total_holding_cost = retailer_holding_cost + central_holding_cost
return {
'central_base_stock': central_base_stock,
'central_safety_stock': central_safety_stock,
'retailer_base_stocks': retailer_base_stocks,
'retailer_safety_stocks': retailer_safety_stocks,
'central_holding_cost': central_holding_cost,
'retailer_holding_cost': retailer_holding_cost,
'total_holding_cost': total_holding_cost
}
def optimize_service_levels(self, total_holding_cost_budget: float) -> Dict:
"""
Optimize service levels subject to holding cost budget
This is a simplified heuristic approach
"""
from scipy.optimize import minimize
def objective(service_levels):
"""Minimize negative of total service level (maximize service)"""
retailer_sls = service_levels[:self.num_retailers]
central_sl = service_levels[self.num_retailers]
# Penalize if service levels are too different
return -np.sum(retailer_sls) - central_sl
def holding_cost_constraint(service_levels):
"""Total holding cost must be <= budget"""
retailer_sls = service_levels[:self.num_retailers]
central_sl = service_levels[self.num_retailers]
result = self.calculate_base_stocks_guaranteed_service(
retailer_sls.tolist(), central_sl
)
return total_holding_cost_budget - result['total_holding_cost']
# Initial guess: 95% everywhere
x0 = np.full(self.num_retailers + 1, 0.95)
# Constraints
cons = [{'type': 'ineq', 'fun': holding_cost_constraint}]
# Bounds: service levels between 50% and 99.9%
bounds = [(0.5, 0.999) for _ in range(self.num_retailers + 1)]
# Optimize
opt_result = minimize(objective, x0, method='SLSQP',
bounds=bounds, constraints=cons)
if opt_result.success:
retailer_sls = opt_result.x[:self.num_retailers].tolist()
central_sl = opt_result.x[self.num_retailers]
result = self.calculate_base_stocks_guaranteed_service(retailer_sls, central_sl)
result['retailer_service_levels'] = retailer_sls
result['central_service_level'] = central_sl
return result
else:
raise ValueError("Optimization failed")
def plot_network(self, result: Dict = None):
"""Visualize the distribution network"""
fig, ax = plt.subplots(figsize=(14, 8))
# Central warehouse position
central_pos = (0.5, 0.8)
# Draw central warehouse
circle = plt.Circle(central_pos, 0.05, color='blue', zorder=3)
ax.add_patch(circle)
ax.text(central_pos[0], central_pos[1] + 0.08, 'Central\nWarehouse',
ha='center', fontsize=10, fontweight='bold')
if result:
ax.text(central_pos[0], central_pos[1] - 0.08,
f"Base Stock: {result['central_base_stock']:.0f}\n"
f"Safety Stock: {result['central_safety_stock']:.0f}",
ha='center', fontsize=8)
# Retailer positions (spread out below)
retailer_positions = []
for i in range(self.num_retailers):
x = 0.1 + (0.8 * i / (self.num_retailers - 1) if self.num_retailers > 1 else 0.4)
y = 0.2
retailer_positions.append((x, y))
# Draw retailer
circle = plt.Circle((x, y), 0.04, color='green', zorder=3)
ax.add_patch(circle)
ax.text(x, y + 0.08, f'Retailer {i+1}',
ha='center', fontsize=9)
if result:
ax.text(x, y - 0.1,
f"BS: {result['retailer_base_stocks'][i]:.0f}\n"
f"SS: {result['retailer_safety_stocks'][i]:.0f}",
ha='center', fontsize=7)
# Draw connection line
ax.plot([central_pos[0], x], [central_pos[1], y],
'k-', linewidth=2, alpha=0.5)
# Add lead time label
mid_x, mid_y = (central_pos[0] + x) / 2, (central_pos[1] + y) / 2
ax.text(mid_x, mid_y, f"LT: {self.Li[i]}",
fontsize=8, bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.5))
ax.set_xlim(0, 1)
ax.set_ylim(0, 1)
ax.set_aspect('equal')
ax.axis('off')
ax.set_title('Two-Echelon Distribution Network', fontsize=14, fontweight='bold')
plt.tight_layout()
return plt
# Example: Distribution System
def example_distribution_system():
"""Example: 1 central warehouse serving 5 retailers"""
print("\n" + "=" * 70)
print("TWO-ECHELON DISTRIBUTION SYSTEM")
print("=" * 70)
# 5 retailers with different characteristics
system = TwoEchelonDistributionSystem(
num_retailers=5,
central_lead_time=14, # 14 days from supplier
delivery_lead_times=[2, 2, 3, 3, 4], # Days to each retailer
central_holding_cost=0.5, # $/unit/day
retailer_holding_costs=[2, 2, 3, 3, 4], # $/unit/day
demands_mean=[50, 80, 30, 60, 40], # Daily demand
demands_std=[10, 15, 8, 12, 10] # Demand std dev
)
print("\nSystem Configuration:")
print(f" Central Warehouse Lead Time: {system.L0} days")
print(f" Central Holding Cost: ${system.h0}/unit/day")
print(f"\n {'Retailer':<12} {'Lead Time':<12} {'Demand Mean':<15} {'Demand Std':<15} {'Holding Cost'}")
print(" " + "-" * 70)
for i in range(system.num_retailers):
print(f" {i+1:<12} {system.Li[i]:<12} {system.mu[i]:<15.0f} "
f"{system.sigma[i]:<15.1f} ${system.hi[i]:.2f}/unit/day")
# Calculate base stocks with uniform 95% service level
print("\n\nCalculating base stocks with 95% service level...")
retailer_sls = [0.95] * system.num_retailers
result = system.calculate_base_stocks_guaranteed_service(retailer_sls, 0.95)
print(f"\n{'Central Warehouse:'}")
print(f" {'Base Stock Level:':<30} {result['central_base_stock']:.0f} units")
print(f" {'Safety Stock:':<30} {result['central_safety_stock']:.0f} units")
print(f" {'Holding Cost:':<30} ${result['central_holding_cost']:.2f}/day")
print(f"\n{'Retailers:'}")
print(f" {'Retailer':<12} {'Base Stock':<15} {'Safety Stock':<15} {'Holding Cost'}")
print(" " + "-" * 60)
for i in range(system.num_retailers):
print(f" {i+1:<12} {result['retailer_base_stocks'][i]:<15.0f} "
f"{result['retailer_safety_stocks'][i]:<15.0f} "
f"${result['retailer_holding_cost']:.2f}/day")
print(f"\n{'Total System Holding Cost:':<40} ${result['total_holding_cost']:.2f}/day")
print(f"{'Annual Holding Cost:':<40} ${result['total_holding_cost'] * 365:.2f}/year")
# Visualize network
system.plot_network(result)
plt.savefig('/tmp/distribution_network.png', dpi=300, bbox_inches='tight')
print(f"\nNetwork diagram saved to /tmp/distribution_network.png")
return system, result
if __name__ == "__main__":
example_distribution_system()
高级多级优化
风险池化效应
def analyze_risk_pooling(num_retailers: int, demand_mean: float,
demand_std: float, service_level: float = 0.95) -> Dict:
"""
Analyze risk pooling benefit of centralization
Compare:
- Decentralized: Each retailer holds safety stock independently
- Centralized: Single central warehouse serves all retailers
"""
z = stats.norm.ppf(service_level)
# Decentralized: each retailer independent
safety_stock_per_retailer = z * demand_std
total_safety_stock_decentral = num_retailers * safety_stock_per_retailer
# Centralized: aggregate demand
aggregate_demand_mean = num_retailers * demand_mean
# Assuming independent demands
aggregate_demand_std = demand_std * np.sqrt(num_retailers)
safety_stock_central = z * aggregate_demand_std
# Risk pooling benefit
reduction = total_safety_stock_decentral - safety_stock_central
reduction_pct = (reduction / total_safety_stock_decentral) * 100
return {
'decentralized_safety_stock': total_safety_stock_decentral,
'centralized_safety_stock': safety_stock_central,
'inventory_reduction': reduction,
'reduction_percentage': reduction_pct,
'num_retailers': num_retailers
}
def plot_risk_pooling_effect():
"""Visualize risk pooling benefit as function of number of locations"""
num_locations = range(1, 21)
results = []
for n in num_locations:
result = analyze_risk_pooling(
num_retailers=n,
demand_mean=100,
demand_std=20,
service_level=0.95
)
results.append(result)
df = pd.DataFrame(results)
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(14, 5))
# Plot 1: Total safety stock
ax1.plot(df['num_retailers'], df['decentralized_safety_stock'],
'o-', linewidth=2, label='Decentralized', color='red')
ax1.plot(df['num_retailers'], df['centralized_safety_stock'],
's-', linewidth=2, label='Centralized', color='green')
ax1.set_xlabel('Number of Retail Locations', fontsize=12)
ax1.set_ylabel('Total Safety Stock (units)', fontsize=12)
ax1.set_title('Risk Pooling: Safety Stock vs. Number of Locations',
fontsize=13, fontweight='bold')
ax1.legend()
ax1.grid(True, alpha=0.3)
# Plot 2: Reduction percentage
ax2.plot(df['num_retailers'], df['reduction_percentage'],
'o-', linewidth=2, color='blue')
ax2.set_xlabel('Number of Retail Locations', fontsize=12)
ax2.set_ylabel('Inventory Reduction (%)', fontsize=12)
ax2.set_title('Risk Pooling Benefit: Inventory Reduction',
fontsize=13, fontweight='bold')
ax2.grid(True, alpha=0.3)
plt.tight_layout()
return plt
# Example
def example_risk_pooling():
"""Example: Quantify risk pooling benefit"""
print("\n" + "=" * 70)
print("RISK POOLING ANALYSIS")
print("=" * 70)
scenarios = [2, 5, 10, 20]
print(f"\n{'Locations':<12} {'Decentralized SS':<20} {'Centralized SS':<20} {'Reduction':<15} {'Reduction %'}")
print("-" * 80)
for n in scenarios:
result = analyze_risk_pooling(
num_retailers=n,
demand_mean=100,
demand_std=20,
service_level=0.95
)
print(f"{n:<12} {result['decentralized_safety_stock']:<20.0f} "
f"{result['centralized_safety_stock']:<20.0f} "
f"{result['inventory_reduction']:<15.0f} "
f"{result['reduction_percentage']:<.1f}%")
plot_risk_pooling_effect()
plt.savefig('/tmp/risk_pooling_analysis.png', dpi=300, bbox_inches='tight')
print(f"\nRisk pooling analysis saved to /tmp/risk_pooling_analysis.png")
if __name__ == "__main__":
example_risk_pooling()
工具与库
Python 库
优化与仿真：
numpy、scipy：数值计算、优化pandas：数据处理networkx：网络拓扑表示simpy：复杂策略的离散事件仿真
供应链专用：
supplychainpy：库存优化库- 自定义实现（如上所示）
商业软件
多级库存优化：
- Blue Yonder (JDA)：MEIO 行业领先者
- Logility Voyager：多级规划
- o9 Solutions：采用 MEIO 的数字化供应链规划
- Kinaxis RapidResponse：多级库存优化
- ToolsGroup：概率型多级优化
- SAP IBP：集成业务计划，支持多级规划
分销需求计划（DRP）：
- 大多数 ERP 系统（SAP、Oracle、Microsoft Dynamics）
常见挑战与解决方案
挑战：数据质量和可视性
问题：
- 整个网络缺乏实时库存可视性
- 下游地点的需求数据不准确
- 交付周期跟踪不佳
解决方案：
- 实施库存可视性平台
- 使用 IoT/RFID 进行实时跟踪
- 建立数据治理流程
- 从关键地点的试点开始
- 改进 POS 端的需求数据采集
挑战：需求相关性
问题：
- 零售商的需求可能存在相关性（并非相互独立）
- 风险池化的收益被高估
解决方案：
- 估算需求相关系数
- 使用多变量需求分布
- 根据相关性调整安全库存计算
- 考虑基于地理位置／产品的相关性模式
挑战：复杂的网络结构
问题：
- 真实网络并非简单的树形结构（存在横向调拨、退货）
- 多种产品在网络中流动
解决方案：
- 对复杂网络使用仿真
- 实施近似分解方法
- 将优化重点放在高价值／高销量产品上
- 使用成熟的软件平台处理通用网络
挑战：服务水平权衡
问题：
- 不同利益相关方对服务水平的要求不同
- 需要在各地点之间平衡成本与服务
解决方案：
- 按产品／客户类别制定明确的服务水平政策
- 使用服务成本分析
- 实施差异化服务水平（例如，A 类商品 99%，C 类商品 90%）
- 根据绩效定期审核和调整
挑战：实施和变更管理
问题：
- 地方经理抵制集中控制
- 难以改变既有订货模式
解决方案：
- 分阶段推广，从愿意参与的地点开始
- 展示快速成效和收益
- 提供本地可视性和覆盖默认设置的能力
- 就新政策开展培训和沟通
- 使激励机制与全系统指标保持一致
输出格式
多级库存优化报告
执行摘要：
- 当前网络：1 个中央 DC、3 个区域 DC、25 家零售门店
- 当前库存总额：$8.5M
- 优化后库存：$6.2M（减少 27%）
- 服务水平：95% → 96%（库存减少的同时有所提升）
- 实施时间：分阶段进行，历时 6 个月
当前状态分析：
| 层级 | 地点数 | 平均库存 | 库存价值 | 服务水平 | 问题 |
|---|---|---|---|---|---|
| 中央 DC | 1 | 15,000 | $2.1M | 98% | 库存过多 |
| 区域 DC | 3 | 各 8,000 | $3.4M | 94% | 波动较大 |
| 零售门店 | 25 | 各 450 | $3.0M | 92% | 经常缺货 |
| 总计 | 29 | - | $8.5M | 93% | - |
优化策略：
| 层级 | 基准库存策略 | 安全库存 | 预期库存 | 目标服务水平 |
|---|---|---|---|---|
| 中央 DC | 补货至 12,000 | 2,500 | 8,000 | 98% |
| 区域 DC 1 | 补货至 6,500 | 1,200 | 4,200 | 96% |
| 区域 DC 2 | 补货至 5,800 | 1,100 | 3,900 | 96% |
| 区域 DC 3 | 补货至 7,200 | 1,400 | 4,800 | 96% |
| 零售门店 | 平均补货至 380 | 平均 80 | 平均 250 | 95% |
财务影响：
| 指标 | 当前 | 优化后 | 改善情况 |
|---|---|---|---|
| 库存投资总额 | $8.5M | $6.2M | -$2.3M (-27%) |
| 年持有成本 (25%) | $2.1M | $1.6M | -$500K (-24%) |
| 服务水平 | 93% | 96% | +3 个百分点 |
| 每月缺货事件 | 45 | 12 | -73% |
| 释放的营运资金 | - | $2.3M | 一次性收益 |
风险池化效益：
- 集中化效应：安全库存减少 22%
- 当前分散式安全库存：8,500 件
- 优化后的协同安全库存：6,650 件
- 减少：1,850 件，价值 $250K
实施计划：
- 阶段 1（第 1-2 个月）： 在中央 DC 和 1 个试点区域 DC 实施
- 阶段 2（第 3-4 个月）： 推广至其余 2 个区域 DC
- 阶段 3（第 5-6 个月）： 分批在零售门店实施
- 持续进行： 监控绩效，每月调整参数
技术要求：
- 覆盖所有层级的库存可视化平台
- 自动化补货系统
- 改进需求感知与预测
- 绩效仪表板
需要提出的问题
如果需要更多背景信息：
- 您的网络结构如何？（层级、各级地点数）
- 客户需求在哪里产生？（零售端、配送中心，还是两者都有）
- 各层级之间的提前期是多少？
- 各地点的持有成本是多少？
- 当前服务水平和目标分别是多少？
- 您能否实施基准库存策略，还是存在限制？
- 是否有横向调拨的机会？
- 有哪些可用数据？（需求历史、提前期数据、当前库存）
- 此项计划的驱动因素是什么？（降低成本、提升服务，还是两者兼有）
- 是否有网络重设计计划？（开设或关闭设施）
相关技能
- inventory-optimization：单地点库存模型
- economic-order-quantity：EOQ 批量订货模型
- stochastic-inventory-models：安全库存与服务水平计算
- network-design：战略网络结构优化
- demand-forecasting：库存规划的需求预测
- inventory-routing-problem：库存与路径联合决策
- distribution-center-network：DC 网络设计与优化
- risk-mitigation：供应链风险管理策略
评论
支持 Markdown