AI 预测配送时间：特征工程与模型选型的工程决策
一、深度引言与场景痛点："预计 30 分钟送达"的真实含义
在配送场景中，"预计送达时间（ETA）"是用户体验的核心指标。但 ETA 预测的难度在于：影响配送时间的因素太多了。天气、交通、同时配送的订单数、取餐等待时间、目的地的小区门禁、是否有电梯——这些因素叠加在一起，让简单的公式（距离 ÷ 速度）完全失效。
更关键的是，预测不准的代价是不对称的。预测过于乐观（"25 分钟送到"结果花了 40 分钟）会被差评，预测过于保守（"60 分钟"实际只要 35 分钟）则会让用户选择竞品。好的 ETA 预测需要精准且有一定的"安全裕量"。
二、底层机制与原理深度剖析
三、生产级代码实现与最佳实践
# 配送时间预测 —— XGBoost 方案
import xgboost as xgb
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
class DeliveryTimePredictor:
"""配送送达时间预测器
为什么选择 XGBoost 而不是深度学习？
1. 特征以表格数据为主，GBDT 系列模型天然适合
2. 训练速度快，方便快速迭代特征
3. 可解释性强（Feature Importance），便于运营分析
4. 对缺失值鲁棒，不需要复杂的填充策略
"""
# 特征定义列表 —— 每一行是一个特征的名称和说明
FEATURES = {
# 订单维度
"order_distance_km": "配送距离（公里）",
"order_weight_kg": "订单重量（公斤）",
"order_category": "品类（餐饮/生鲜/蛋糕/药品）",
"order_item_count": "商品件数",
"is_rush_hour": "是否高峰期（0/1）",
"hour_of_day": "小时（0-23）",
"day_of_week": "星期几（0-6）",
"is_holiday": "是否节假日（0/1）",
# 骑手维度
"rider_experience_days": "骑手注册天数",
"rider_active_orders": "当前同时配送的订单数",
"rider_avg_rating": "骑手平均评分",
"rider_on_time_rate": "骑手准时率",
"rider_avg_delivery_time": "骑手历史平均配送时间",
# 环境维度
"weather_condition": "天气状况编码",
"temperature_celsius": "温度（摄氏度）",
"precipitation_mm": "降水量（毫米）",
"traffic_index": "实时交通指数",
# 空间维度
"pickup_area_id": "取货区域 ID",
"delivery_area_id": "送货区域 ID",
"is_downtown": "是否市中心（0/1）",
"building_type": "建筑类型（住宅/写字楼/商场）",
# 历史统计
"hist_area_avg_time": "该区域历史平均配送时间",
"hist_route_avg_time": "该路线历史平均配送时间",
"hist_hour_avg_time": "该时段历史平均配送时间",
}
def __init__(self):
self.model = None
self.feature_columns = list(self.FEATURES.keys())
def train(self, df: pd.DataFrame, target_col: str = "delivery_time_min"):
"""训练模型
Args:
df: 训练数据，包含所有特征列和目标列
target_col: 目标变量列名（配送时间，分钟）
"""
X = df[self.feature_columns]
y = df[target_col]
# 对目标变量进行 log 变换 —— 减小极端值对损失函数的影响
# 配送时间的分布通常是有偏的（右偏），log 变换后更接近正态分布
y_log = np.log1p(y)
X_train, X_val, y_train, y_val = train_test_split(
X, y_log, test_size=0.2, random_state=42
)
# XGBoost 参数配置
# 这些参数是在大量实验中调优的结果，可以根据具体数据调整
params = {
"objective": "reg:squarederror", # 回归任务
"eval_metric": "rmse",
"max_depth": 7, # 树深：太大容易过拟合，太小欠拟合
"learning_rate": 0.05,
"n_estimators": 500,
"subsample": 0.8, # 行采样：防止过拟合
"colsample_bytree": 0.8, # 列采样
"reg_alpha": 1.0, # L1 正则化
"reg_lambda": 2.0, # L2 正则化
"min_child_weight": 5,
"random_state": 42,
"n_jobs": -1, # 使用所有 CPU 核心
}
self.model = xgb.XGBRegressor(**params)
self.model.fit(
X_train, y_train,
eval_set=[(X_val, y_val)],
verbose=100 # 每 100 轮输出一次评估信息
)
# 输出特征重要性
importance = self.model.feature_importances_
feature_importance = sorted(
zip(self.feature_columns, importance),
key=lambda x: x[1], reverse=True
)
print("Top 10 特征重要性:")
for name, score in feature_importance[:10]:
print(f" {name}: {score:.4f}")
return self
def predict(
self, X: pd.DataFrame, return_interval: bool = True
) -> dict:
"""预测配送时间
Args:
X: 预测数据的特征矩阵
return_interval: 是否返回置信区间
Returns:
包含预测值、置信区间等信息的字典
"""
if self.model is None:
raise ValueError("模型尚未训练，请先调用 train 方法")
y_pred_log = self.model.predict(X[self.feature_columns])
# 反 log 变换 —— 将预测结果转换回原始分钟数
y_pred = np.expm1(y_pred_log)
result = {
"predicted_minutes": round(float(y_pred[0]), 1),
"p50": round(float(y_pred[0]), 1),
}
if return_interval:
# 简单的置信区间估计：基于预测值的百分比
# 更精确的方案需要使用分位数回归或 quantile XGBoost
result["p10"] = round(float(y_pred[0] * 0.7), 1) # 乐观估计
result["p90"] = round(float(y_pred[0] * 1.4), 1) # 悲观估计
# 对用户展示用 P90 值，给予充分预期
result["display_eta"] = result["p90"]
return result
# ========== 特征工程函数 ==========
def engineer_features(raw_order: dict, raw_rider: dict,
env: dict, historical: dict) -> pd.DataFrame:
"""从原始数据构建特征
特征工程是模型效果的核心。同样的模型，
不同的特征决定了大半的准确率差异。
关键原则：
1. 每个特征必须有业务含义
2. 避免目标泄漏（使用未来信息预测过去）
3. 特征编码要与模型匹配
"""
features = {}
# ===== 订单特征 =====
features["order_distance_km"] = _haversine(
raw_order["pickup_lat"], raw_order["pickup_lng"],
raw_order["delivery_lat"], raw_order["delivery_lng"]
) / 1000
features["order_weight_kg"] = raw_order.get("weight_kg", 0)
features["order_category"] = _encode_category(raw_order["category"])
features["order_item_count"] = raw_order.get("item_count", 1)
# 时间特征
order_time = pd.Timestamp(raw_order["created_at"])
features["hour_of_day"] = order_time.hour
features["day_of_week"] = order_time.dayofweek
features["is_rush_hour"] = int(
order_time.hour in [7, 8, 9, 11, 12, 13, 17, 18, 19]
)
features["is_holiday"] = int(order_time.date() in HOLIDAYS)
# ===== 骑手特征 =====
features["rider_experience_days"] = raw_rider.get("experience_days", 0)
features["rider_active_orders"] = raw_rider.get("active_orders", 0)
features["rider_avg_rating"] = raw_rider.get("avg_rating", 4.5)
features["rider_on_time_rate"] = raw_rider.get("on_time_rate", 0.9)
features["rider_avg_delivery_time"] = raw_rider.get("avg_delivery_time", 25)
# ===== 环境特征 =====
features["weather_condition"] = env.get("weather_code", 0)
features["temperature_celsius"] = env.get("temperature", 20)
features["precipitation_mm"] = env.get("precipitation", 0)
features["traffic_index"] = env.get("traffic_index", 1.0)
# ===== 空间特征 =====
features["pickup_area_id"] = _geo_to_area(
raw_order["pickup_lat"], raw_order["pickup_lng"]
)
features["delivery_area_id"] = _geo_to_area(
raw_order["delivery_lat"], raw_order["delivery_lng"]
)
# ===== 历史统计特征（从离线计算的特征表中获取）=====
features["hist_area_avg_time"] = historical.get("area_avg_time", 25)
features["hist_route_avg_time"] = historical.get("route_avg_time", 30)
features["hist_hour_avg_time"] = historical.get("hour_avg_time", 28)
return pd.DataFrame([features])
四、边界分析与架构权衡
模型交付的工程化考虑
模型训练完成后，最大的挑战不是"模型准不准"，而是"如何在生产环境中低延迟、高可靠地提供预测服务"：
- 在线预测 vs 离线批量：配送场景需要在线预测（<50ms 延迟），需要将模型部署为 RPC 服务
- 模型更新：每周重新训练一次，用新数据替代旧数据，防止模型"老化"
- 特征一致性：训练阶段和预测阶段必须使用完全相同的特征计算逻辑
实时特征 vs 离线特征
特征分为两类：
- 实时特征：订单距离、骑手当前负载、天气——在预测时实时计算
- 离线特征：历史平均配送时间、骑手准时率——每日预计算，存储在特征表中
一个好的特征工程系统需要同时管理这两类特征，并确保它们的更新时效性。
五、总结
配送时间预测是一个典型的"特征驱动"问题。模型本身不是关键（XGBoost/LightGBM 都够用），关键的是你能不能把影响配送时间的信息都合理地编码为特征。
三个最重要的经验：
对用户来说，他们看到的只是一个数字——"预计 30 分钟"。但这个数字背后，是一个融合了几十个特征的机器学习模型在实时运算。而这种"让复杂的事情变简单"，正是 AI 在工程中的核心价值。
