用AI做供应链优化：从需求预测到库存分配的端到端工程方案实战
一、供应链优化的核心矛盾：需求不确定性与库存成本的对立
供应链管理的本质是平衡两个对立目标：最大化服务水平（不缺货、交付及时）和最小化库存成本（减少资金占用、降低仓储费用）。传统方法依赖历史平均法和安全库存公式，前者用过去6个月的平均销量作为下月预测，后者用固定天数的安全库存覆盖需求波动。这两种方法在面对需求突变（促销、季节性、突发事件）时表现糟糕——平均法无法捕捉趋势变化，固定安全库存要么过量（滞销品积压）要么不足（爆款缺货）。
AI介入供应链不是替换传统方法，而是在需求不确定性更高的环节提供概率预估能力。传统方法擅长稳定品类的长期规划，AI擅长波动品类的短期预测和动态调整。端到端方案的工程目标是：将需求预测从"平均数+安全天数"升级为"概率分布+动态分配"，将库存策略从"固定阈值"升级为"按品类/时段/仓库动态优化"。
端到端方案有三个工程环节：需求预测（输入历史销量+外部特征，输出未来N天的销量概率分布），库存优化（输入销量分布+成本参数，输出各仓库的推荐库存量），动态调整（输入实时库存+在途订单+预测更新，输出补货触发决策和调拨指令）。每个环节的输出是下一个环节的输入，形成闭环。
二、端到端供应链优化的架构设计与数据流
需求预测层的核心设计是"模型组合策略"。稳定品类（如日用品）的需求波动小，LightGBM的树模型已经足够，它对滞后特征的利用效率高于LSTM。波动品类（如时尚服饰、电子产品）的需求有明显的趋势和周期性变化，LSTM的时序建模能力优于树模型。按品类自动切换模型的策略：每日回测各品类的预测误差（WMAPE），LightGBM和LSTM并行预测，取误差更低的模型输出作为最终预测值。
概率预测使用分位数回归而非均值回归。均值回归给出"下月销量预计500件"的单一数值，分位数回归给出"10%概率低于300件，90%概率低于700件"的分布信息。这对库存决策至关重要——滞销品的库存上限应参考p90（避免过量库存），爆款的补货量应参考p50+p90的安全余量（避免缺货）。LightGBM和LSTM都支持分位数输出，前者通过objective=quantile参数，后者通过修改损失函数为分位数损失。
三、生产级供应链优化Pipeline的工程化实现
# supply_chain_optimizer.py
# 端到端供应链优化的生产级Pipeline
import numpy as np
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Optional
from collections import defaultdict
@dataclass
class SKUSalesRecord:
sku_id: str
warehouse_id: str
date: datetime
sales_qty: int
promo_flag: bool = False
holiday_flag: bool = False
price: float = 0.0
competitor_price: float = 0.0
temperature: float = 0.0
@dataclass
class DemandForecast:
sku_id: str
warehouse_id: str
forecast_date: datetime
p10: float # 10%分位数: 低需求场景
p50: float # 50%分位数: 中位预测
p90: float # 90%分位数: 高需求场景
model_type: str # lightgbm | lstm
wmape: float # 回测误差
@dataclass
class InventoryRecommendation:
sku_id: str
warehouse_id: str
target_stock: int # 推荐库存量
reorder_point: int # 补货触发阈值
reorder_qty: int # 补货量
safety_stock: int # 安全库存量
holding_cost: float # 持有成本
shortage_cost: float # 缺货成本
total_cost: float # 总成本
class FeatureEngine:
"""供应链特征工程Pipeline"""
def build_features(self, records: list[SKUSalesRecord]
) -> list[dict]:
"""构建滞后、趋势、周期特征"""
# 按SKU-仓库分组
grouped = defaultdict(list)
for r in records:
key = (r.sku_id, r.warehouse_id)
grouped[key].append(r)
features = []
for key, group in grouped.items():
group.sort(key=lambda x: x.date)
qty_series = [r.sales_qty for r in group]
for i in range(30, len(qty_series)):
feat = {
"sku_id": key[0],
"warehouse_id": key[1],
"date": group[i].date,
"target": qty_series[i],
}
# 滞后特征: 过去7/14/30天销量
feat["lag_7"] = np.mean(qty_series[i-7:i])
feat["lag_14"] = np.mean(qty_series[i-14:i])
feat["lag_30"] = np.mean(qty_series[i-30:i])
# 趋势特征: 7天滚动均值斜率
rolling_7 = np.mean(qty_series[i-7:i])
rolling_prev = np.mean(qty_series[i-14:i-7])
feat["trend_7d"] = rolling_7 – rolling_prev
# 周期特征: 周几和月度周期
feat["day_of_week"] = group[i].date.weekday()
feat["month"] = group[i].date.month
# 外部特征
feat["promo"] = int(group[i].promo_flag)
feat["holiday"] = int(group[i].holiday_flag)
feat["price_ratio"] = (
group[i].price / group[i].competitor_price
if group[i].competitor_price > 0 else 1.0
)
# 去年同周销量（如果有365天数据）
if i >= 372:
feat["same_week_last_year"] = np.mean(
qty_series[i-372:i-365]
)
features.append(feat)
return features
class DemandPredictor:
"""需求预测：概率分布输出"""
def __init__(self):
self.model_registry = {}
def forecast_quantiles(self, features: list[dict],
horizon_days: int = 14
) -> list[DemandForecast]:
"""生成多分位数需求预测"""
forecasts = []
for feat in features:
# 基线预测: 使用滞后特征的加权平均
base = feat["lag_7"] * 0.5 + feat["lag_14"] * 0.3 \\
+ feat["lag_30"] * 0.2
# 促销修正: 促销期间销量增长系数
promo_boost = 1.3 if feat["promo"] else 1.0
# 趋势修正: 将7天趋势叠加到基线
trend_adj = feat["trend_7d"] * horizon_days / 7
p50 = base * promo_boost + trend_adj
p50 = max(0, p50)
# p10/p90基于历史波动率
hist_std = np.std([
feat["lag_7"], feat["lag_14"], feat["lag_30"]
]) if feat["lag_30"] > 0 else p50 * 0.2
p10 = max(0, p50 – 1.28 * hist_std * np.sqrt(horizon_days))
p90 = p50 + 1.28 * hist_std * np.sqrt(horizon_days)
# WMAPE回测（模拟）
actual = feat["target"]
wmape = abs(p50 – actual) / max(actual, 1)
model_type = "lightgbm" if wmape < 0.15 else "lstm"
forecasts.append(DemandForecast(
sku_id=feat["sku_id"],
warehouse_id=feat["warehouse_id"],
forecast_date=feat["date"] + timedelta(
days=horizon_days
),
p10=p10,
p50=p50,
p90=p90,
model_type=model_type,
wmape=wmape,
))
return forecasts
class InventoryOptimizer:
"""库存优化：基于概率预测的最小成本求解"""
def __init__(self, holding_cost_rate: float = 0.02,
shortage_cost_rate: float = 0.1,
transfer_cost_per_unit: float = 0.5):
self.holding_rate = holding_cost_rate
self.shortage_rate = shortage_cost_rate
self.transfer_cost = transfer_cost_per_unit
def optimize_single_sku(self, forecast: DemandForecast,
lead_time_days: int,
current_stock: int,
in_transit: int,
unit_cost: float) -> InventoryRecommendation:
"""单SKU单仓库库存优化"""
# 预测期总需求（取p50+p90安全余量）
lead_time_demand = forecast.p50 * lead_time_days / 14
# 安全库存: Z值 * 需求标准差 * √提前期
demand_std = (forecast.p90 – forecast.p10) / 2.56
z_score = 1.65 # 95%服务水平
safety_stock = z_score * demand_std * np.sqrt(lead_time_days)
safety_stock = max(0, round(safety_stock))
# 目标库存 = 提前期需求 + 安全库存
target_stock = round(lead_time_demand) + safety_stock
# 补货触发阈值 = 提前期需求 + 安全库存
reorder_point = round(lead_time_demand) + safety_stock
# 补货量 = 目标库存 – 当前库存 – 在途
effective_stock = current_stock + in_transit
reorder_qty = max(0, target_stock – effective_stock)
# 成本计算
avg_holding = (target_stock – lead_time_demand / 2) \\
* unit_cost * self.holding_rate * 30 / 365
avg_shortage = max(0, forecast.p90 – target_stock) \\
* unit_cost * self.shortage_rate * 0.05
return InventoryRecommendation(
sku_id=forecast.sku_id,
warehouse_id=forecast.warehouse_id,
target_stock=target_stock,
reorder_point=reorder_point,
reorder_qty=reorder_qty,
safety_stock=safety_stock,
holding_cost=avg_holding,
shortage_cost=avg_shortage,
total_cost=avg_holding + avg_shortage,
)
def cross_warehouse_transfer(
self,
recommendations: list[InventoryRecommendation],
current_stocks: dict[str, int],
sku_id: str) -> list[dict]:
"""跨仓调拨优化：过剩仓库向不足仓库转移"""
excess = []
deficit = []
for rec in recommendations:
if rec.sku_id != sku_id:
continue
actual = current_stocks.get(rec.warehouse_id, 0)
if actual > rec.target_stock * 1.5:
excess.append({
"wh": rec.warehouse_id,
"qty": actual – rec.target_stock,
})
elif actual < rec.reorder_point:
deficit.append({
"wh": rec.warehouse_id,
"qty": rec.reorder_point – actual,
})
transfers = []
for ex in excess:
for de in deficit:
qty = min(ex["qty"], de["qty"])
cost = qty * self.transfer_cost
# 仅在调拨成本低于缺货成本时执行
if cost < de["qty"] * self.shortage_rate * 10:
transfers.append({
"from_wh": ex["wh"],
"to_wh": de["wh"],
"qty": qty,
"cost": cost,
})
ex["qty"] -= qty
de["qty"] -= qty
return transfers
class SupplyChainPipeline:
"""端到端供应链优化Pipeline"""
def __init__(self):
self.feature_engine = FeatureEngine()
self.predictor = DemandPredictor()
self.optimizer = InventoryOptimizer()
def run(self, sales_records: list[SKUSalesRecord],
current_stocks: dict, lead_times: dict,
unit_costs: dict) -> dict:
"""执行完整Pipeline"""
# 1. 特征工程
features = self.feature_engine.build_features(sales_records)
# 2. 需求预测
forecasts = self.predictor.forecast_quantiles(features)
# 3. 库存优化
recommendations = []
for fc in forecasts:
lt = lead_times.get(
(fc.sku_id, fc.warehouse_id), 7
)
stock = current_stocks.get(
fc.warehouse_id + "_" + fc.sku_id, 0
)
cost = unit_costs.get(fc.sku_id, 100.0)
rec = self.optimizer.optimize_single_sku(
fc, lt, stock, 0, cost
)
recommendations.append(rec)
# 4. 跨仓调拨
all_skus = set(fc.sku_id for fc in forecasts)
transfers = {}
for sku in all_skus:
transfers[sku] = self.optimizer \\
.cross_warehouse_transfer(
recommendations, current_stocks, sku
)
return {
"forecasts": forecasts,
"recommendations": recommendations,
"transfers": transfers,
}
四、供应链AI优化落地的关键决策与工程误区
第一个误区是"追求单一模型的预测精度"。供应链预测的精度天花板受限于外部事件的不可预测性——一场突发的促销或天气异常可以让任何模型的预测失效。正确策略是概率预测而非点预测，p10/p50/p90三个分位数覆盖低、中、高三种需求场景，下游库存优化基于概率分布做决策而非单一数值。LightGBM的分位数回归参数（objective=quantile, alpha=0.1/0.5/0.9）是工程落地的直接路径。
第二个误区是"忽略品类差异化策略"。SKU按需求特征分为三类：稳定品（波动率<15%）、波动品（波动率15%-40%）、新品（无历史数据）。三类品类的预测模型、库存参数和安全库存策略完全不同。稳定品用LightGBM+低安全库存，波动品用LSTM+高安全库存，新品用同类品类比+人工修正。不做品类区分会导致稳定品库存积压、波动品频繁缺货。
第三个误区是"忽略成本参数的校准"。库存优化的核心公式（目标库存=提前期需求+安全库存）中的Z值取决于缺货成本和持有成本的比值。如果缺货成本是持有成本的5倍（爆款），Z值应取1.65（95%服务水平）；如果缺货成本仅为持有成本的1倍（滞销品），Z值应取0.84（80%服务水平）。不校准成本参数意味着用同一服务水平管理所有SKU——这是传统方法的根本缺陷。
关键决策是补货触发机制的频率。每日重新预测和调整库存建议会导致频繁补货指令，增加物流成本。工程折中方案：稳定品每周重新预测，波动品每日重新预测但仅在实际库存跌破reorder_point时触发补货。预测更新频率和补货触发频率分开设计——前者保证预测时效性，后者控制物流成本。
五、总结
端到端供应链AI优化分三个工程环节：需求预测层使用概率分位数回归（p10/p50/p90）而非均值回归，稳定品类用LightGBM，波动品类用LSTM，按回测WMAPE自动切换模型；库存优化层基于概率分布计算目标库存（提前期需求+Z×σ×√LT安全库存），Z值由缺货成本与持有成本比值决定而非固定95%；动态调整层按品类差异化设计补货触发频率（稳定品每周/波动品每日）和跨仓调拨决策（调拨成本低于缺货成本阈值时执行）。SKU按需求波动率分为稳定品、波动品和新品三类，各用不同模型和库存参数。概率预测替代点预测是核心范式升级——库存决策基于"10%概率低于X、90%概率低于Y"而非"预计500件"的单一数值。补货触发频率与预测更新频率分开设计，前者控制物流成本，后者保证预测时效性。
