用AI做供应链优化：从需求预测到库存分配的端到端工程方案实战

用AI做供应链优化：从需求预测到库存分配的端到端工程方案实战并不只看表面做法，关键还要理解相关条件、限制和后续影响。

用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# 端到端供应链优化的生产级Pipelineimport numpy as npfrom dataclasses import dataclass, fieldfrom datetime import datetime, timedeltafrom typing import Optionalfrom collections import defaultdict@dataclassclass SKUSalesRecord:sku_id: strwarehouse_id: strdate: datetimesales_qty: intpromo_flag: bool = Falseholiday_flag: bool = Falseprice: float = 0.0competitor_price: float = 0.0temperature: float = 0.0@dataclassclass DemandForecast:sku_id: strwarehouse_id: strforecast_date: datetimep10: float # 10%分位数: 低需求场景p50: float # 50%分位数: 中位预测p90: float # 90%分位数: 高需求场景model_type: str# lightgbm | lstmwmape: float# 回测误差@dataclassclass InventoryRecommendation:sku_id: strwarehouse_id: strtarget_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_priceif 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 featuresclass 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 / 7p50 = base * promo_boost + trend_adjp50 = 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.2p10 = 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 forecastsclass 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_rateself.shortage_rate = shortage_cost_rateself.transfer_cost = transfer_cost_per_unitdef 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.56z_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_transitreorder_qty = max(0, target_stock - effective_stock)# 成本计算avg_holding = (target_stock - lead_time_demand / 2) * unit_cost * self.holding_rate * 30 / 365avg_shortage = max(0, forecast.p90 - target_stock) * unit_cost * self.shortage_rate * 0.05return 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:continueactual = 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"] -= qtyde["qty"] -= qtyreturn transfersclass 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件"的单一数值。补货触发频率与预测更新频率分开设计，前者控制物流成本，后者保证预测时效性。