物流行业说到底就两个核心问题:货什么时候来,车怎么跑。前者是需求预测,属于机器学习;后者是路径优化,属于运筹学。这两个领域各自都有成熟的技术栈,但真正难的是把它们接成一个闭环系统——预测给调度提供输入,调度的执行数据又回流修正预测。这篇结合一个城配项目的经历,聊聊两条主线怎么分别落地、又怎么打通。
一、问题拆解:预测与调度是两条主线
先把问题边界划清楚。需求预测回答"明天每个片区有多少单、多少体积、多少重量",输出决定仓储排班、运力储备、车辆预调。路径优化回答"已知今天的订单和车队,每辆车按什么顺序跑哪些点",输出是具体的行驶计划。
一个常见的认知误区是拼命优化求解器,却用着拍脑袋出来的预测量。输入错 10%,再优的解也是精确的错误。所以我的建议是:预测和调度的研发投入七三开。
二、需求预测:LightGBM 依然是主力
试过 Prophet、ARIMA、LSTM、TFT 之后,城配单量预测的主力还是 LightGBM 加特征工程,原因很现实:精度够用、训练快、特征重要性可解释,业务方能看懂。特征体系四大块:
- 滞后特征:近 7/14/28 天同片区单量,以及去年同期;
- 滚动统计:7 天均值、标准差、环比变化;
- 日历特征:节假日、调休、促销日、发薪日(对快消品类影响明显)、学校开学;
- 外生变量:天气(降雨对即时配送单量拉动显著)、竞对活动。
两个工程细节值得注意:一是单量零值多,objective 用 tweedie 比 mse 稳;二是层级一致性——大区预测之和要等于全国预测,用层级协调(reconciliation)做后处理,别让仓网计划互相打架。大促(618/双 11)历史样本太少,单独建模,用相似品类的大促曲线做迁移。
import lightgbm as lgb features = ["lag_7", "lag_14", "roll_mean_7", "roll_std_7", "is_holiday", "is_promo", "weekday", "temp", "rain"] model = lgb.LGBMRegressor( objective="tweedie", # 单量零膨胀, 比 mse 更稳 n_estimators=800, learning_rate=0.05, ) model.fit( train[features], train["orders"], eval_set=[(valid[features], valid["orders"])], callbacks=[lgb.early_stopping(50)], ) # 输出特征重要性, 每周和业务方过一遍, 建立信任 print(sorted(zip(features, model.feature_importances_), key=lambda x: -x[1]))三、路径优化:VRP 问题的工程化解法
路径优化的问题家族:TSP(单旅行者)→ VRP(多车辆)→ CVRP(带容量)→ CVRPTW(再加时间窗)。城配真实场景基本是 CVRPTW 加一堆脏约束:车型限制、装卸后进先出、司机工时、区域限行、客户自提时间窗。这类问题是 NP-hard,万级节点别想精确解,工业界标配是 OR-Tools 的引导局部搜索(Guided Local Search),几十秒内能拿到可接受解。
落地时的节奏设计比算法本身更重要:夜间批量全量重排(时间预算可以给到分钟级),白天插单走局部优化(只重排受影响车辆的后续节点,秒级返回),别每次插单都全量重算。插单局部优化的实现上,我们把每条线路的后续节点抽象成一个小 VRP 子问题,节点数控制在五十以内,求解预算两秒,超时或失败就退化为按距离贪心插入,保证任何情况下都有兜底方案。
还有几个容易踩的坑:一是评估指标别用 MAPE,单量小的片区会把 MAPE 拉爆,用按单量加权的 WMAPE 更客观;二是距离矩阵别用直线距离,接真实路网服务(高德或自建 OSRM)预计算片区级矩阵缓存下来,能省掉大量在线计算;三是约束要在建模阶段就和一线调度员对齐,他们脑子里"这条路货车走不了""这个客户只收上午"这类知识不进模型,系统排的线路没人执行;四是训练数据要清洗,大促和极端天气的单量是离群点,直接喂进日常模型会带歪基线,要么剔除要么打标单独处理。最后一定要保留人工调整入口:调度员手动拖单改线的比例本身就是核心指标,拖单率高说明模型和现实脱节,比任何离线指标都诚实。
from ortools.constraint_solver import routing_enums_pb2, pywrapcp def solve_cvrp(dist_matrix, demands, vehicle_caps, depot=0): manager = pywrapcp.RoutingIndexManager( len(dist_matrix), len(vehicle_caps), depot) routing = pywrapcp.RoutingModel(manager) dist_cb = routing.RegisterTransitCallback( lambda i, j: dist_matrix[manager.IndexToNode(i)] [manager.IndexToNode(j)]) routing.SetArcCostEvaluatorOfAllVehicles(dist_cb) demand_cb = routing.RegisterUnaryTransitCallback( lambda i: demands[manager.IndexToNode(i)]) routing.AddDimensionWithVehicleCapacity( demand_cb, 0, vehicle_caps, True, "Capacity&qu