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

用AI做供应链优化从需求预测到库存分配的端到端工程方案实战一、供应链优化的核心矛盾需求不确定性与库存成本的对立供应链管理的本质是平衡两个对立目标最大化服务水平不缺货、交付及时和最小化库存成本减少资金占用、降低仓储费用。传统方法依赖历史平均法和安全库存公式前者用过去6个月的平均销量作为下月预测后者用固定天数的安全库存覆盖需求波动。这两种方法在面对需求突变促销、季节性、突发事件时表现糟糕——平均法无法捕捉趋势变化固定安全库存要么过量滞销品积压要么不足爆款缺货。AI介入供应链不是替换传统方法而是在需求不确定性更高的环节提供概率预估能力。传统方法擅长稳定品类的长期规划AI擅长波动品类的短期预测和动态调整。端到端方案的工程目标是将需求预测从平均数安全天数升级为概率分布动态分配将库存策略从固定阈值升级为按品类/时段/仓库动态优化。端到端方案有三个工程环节需求预测输入历史销量外部特征输出未来N天的销量概率分布库存优化输入销量分布成本参数输出各仓库的推荐库存量动态调整输入实时库存在途订单预测更新输出补货触发决策和调拨指令。每个环节的输出是下一个环节的输入形成闭环。二、端到端供应链优化的架构设计与数据流需求预测层的核心设计是模型组合策略。稳定品类如日用品的需求波动小LightGBM的树模型已经足够它对滞后特征的利用效率高于LSTM。波动品类如时尚服饰、电子产品的需求有明显的趋势和周期性变化LSTM的时序建模能力优于树模型。按品类自动切换模型的策略每日回测各品类的预测误差WMAPELightGBM和LSTM并行预测取误差更低的模型输出作为最终预测值。概率预测使用分位数回归而非均值回归。均值回归给出下月销量预计500件的单一数值分位数回归给出10%概率低于300件90%概率低于700件的分布信息。这对库存决策至关重要——滞销品的库存上限应参考p90避免过量库存爆款的补货量应参考p50p90的安全余量避免缺货。LightGBM和LSTM都支持分位数输出前者通过objectivequantile参数后者通过修改损失函数为分位数损失。三、生产级供应链优化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(keylambda 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_idfeat[sku_id], warehouse_idfeat[warehouse_id], forecast_datefeat[date] timedelta( dayshorizon_days ), p10p10, p50p50, p90p90, model_typemodel_type, wmapewmape, )) 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单仓库库存优化 # 预测期总需求取p50p90安全余量 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_idforecast.sku_id, warehouse_idforecast.warehouse_id, target_stocktarget_stock, reorder_pointreorder_point, reorder_qtyreorder_qty, safety_stocksafety_stock, holding_costavg_holding, shortage_costavg_shortage, total_costavg_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的分位数回归参数objectivequantile, alpha0.1/0.5/0.9是工程落地的直接路径。第二个误区是忽略品类差异化策略。SKU按需求特征分为三类稳定品波动率15%、波动品波动率15%-40%、新品无历史数据。三类品类的预测模型、库存参数和安全库存策略完全不同。稳定品用LightGBM低安全库存波动品用LSTM高安全库存新品用同类品类比人工修正。不做品类区分会导致稳定品库存积压、波动品频繁缺货。第三个误区是忽略成本参数的校准。库存优化的核心公式目标库存提前期需求安全库存中的Z值取决于缺货成本和持有成本的比值。如果缺货成本是持有成本的5倍爆款Z值应取1.6595%服务水平如果缺货成本仅为持有成本的1倍滞销品Z值应取0.8480%服务水平。不校准成本参数意味着用同一服务水平管理所有SKU——这是传统方法的根本缺陷。关键决策是补货触发机制的频率。每日重新预测和调整库存建议会导致频繁补货指令增加物流成本。工程折中方案稳定品每周重新预测波动品每日重新预测但仅在实际库存跌破reorder_point时触发补货。预测更新频率和补货触发频率分开设计——前者保证预测时效性后者控制物流成本。五、总结端到端供应链AI优化分三个工程环节需求预测层使用概率分位数回归p10/p50/p90而非均值回归稳定品类用LightGBM波动品类用LSTM按回测WMAPE自动切换模型库存优化层基于概率分布计算目标库存提前期需求Z×σ×√LT安全库存Z值由缺货成本与持有成本比值决定而非固定95%动态调整层按品类差异化设计补货触发频率稳定品每周/波动品每日和跨仓调拨决策调拨成本低于缺货成本阈值时执行。SKU按需求波动率分为稳定品、波动品和新品三类各用不同模型和库存参数。概率预测替代点预测是核心范式升级——库存决策基于10%概率低于X、90%概率低于Y而非预计500件的单一数值。补货触发频率与预测更新频率分开设计前者控制物流成本后者保证预测时效性。