用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件"的单一数值。补货触发频率与预测更新频率分开设计，前者控制物流成本，后者保证预测时效性。