AI驱动的需求预测数据管道：历史销量、季节性与促销因子的特征工程

AI驱动的需求预测数据管道：历史销量、季节性与促销因子的特征工程

一、"备货5000件，实际卖了150件"：预测失败的沉没成本

某生鲜电商在樱桃季前根据"去年同期销量×1.2"的简单预测，向产地采购了5000箱烟台樱桃。结果当季实际销量仅150箱。复盘发现漏掉了三个关键变量：当年是樱桃产量大年导致市场价格暴跌（供给侧冲击）、竞品平台搞了"车厘子买一送一"活动（替代品促销）、连续一周阴雨天气降低了生鲜购买意愿（天气因子）。

传统的时间序列预测（ARIMA/Holt-Winters）只能捕捉历史销量的周期性，无法融入天气、竞品促销、宏观经济等外部因子。而AI预测的关键不在于模型选型——LSTM、Prophet、XGBoost在这个问题上效果差不多——而在于特征的丰富度和新鲜度。

二、需求预测的特征工程与数据管道

三、特征管道实现

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from sklearn.preprocessing import LabelEncoder

class DemandForecastFeaturePipeline:
def __init__(self, mysql_pool, redis_client, weather_api):
self.mysql = mysql_pool
self.redis = redis_client
self.weather = weather_api

def build_features(self, sku_id: str,
forecast_date: datetime,
horizon_days: int = 30) -> pd.DataFrame:
"""构建预测所需的特征矩阵"""
features = {}

# 1. 历史销量特征（滞后特征）
features.update(self._build_lag_features(sku_id, forecast_date))

# 2. 时间特征
features.update(self._build_temporal_features(forecast_date))

# 3. 促销特征
features.update(self._build_promotion_features(
sku_id, forecast_date, horizon_days
))

# 4. 天气特征
features.update(self._build_weather_features(
forecast_date, horizon_days
))

# 5. 库存特征
features.update(self._build_inventory_features(sku_id))

# 6. 品类趋势特征
features.update(self._build_category_features(
sku_id, forecast_date
))

return pd.DataFrame([features])

def _build_lag_features(self, sku_id: str,
forecast_date: datetime) -> dict:
"""构建滞后特征：过去N天的销量"""
query = """
SELECT
date, quantity_sold, avg_price,
discount_rate, is_promotion
FROM daily_sales
WHERE sku_id = %s AND date >= %s AND date < %s
ORDER BY date
"""

start_date = forecast_date – timedelta(days=90)

try:
df = pd.read_sql(query, self.mysql.get_connection(),
params=(sku_id, start_date, forecast_date))
except Exception as e:
raise FeatureBuildException(f"销量查询失败: {sku_id}", e)

if df.empty:
return self._default_lag_features()

features = {
# 最基础：昨天、前天、上周同天的销量
'sales_lag_1': float(df['quantity_sold'].iloc[-1]) if len(df) >= 1 else 0,
'sales_lag_2': float(df['quantity_sold'].iloc[-2]) if len(df) >= 2 else 0,
'sales_lag_7': float(df['quantity_sold'].iloc[-7]) if len(df) >= 7 else 0,
'sales_lag_14': float(df['quantity_sold'].iloc[-14]) if len(df) >= 14 else 0,
'sales_lag_30': float(df['quantity_sold'].iloc[-30]) if len(df) >= 30 else 0,

# 移动平均
'sales_ma_7': float(df['quantity_sold'].tail(7).mean()) if len(df) >= 7 else 0,
'sales_ma_14': float(df['quantity_sold'].tail(14).mean()) if len(df) >= 14 else 0,
'sales_ma_30': float(df['quantity_sold'].tail(30).mean()) if len(df) >= 30 else 0,

# 趋势
'sales_trend_7': self._calc_trend(df['quantity_sold'].tail(7)),
'sales_trend_30': self._calc_trend(df['quantity_sold'].tail(30)),

# 波动性
'sales_std_7': float(df['quantity_sold'].tail(7).std()) if len(df) >= 7 else 0,
'sales_std_30': float(df['quantity_sold'].tail(30).std()) if len(df) >= 30 else 0,
}

return features

def _build_temporal_features(self, date: datetime) -> dict:
"""时间特征"""
return {
'day_of_week': date.weekday(),
'is_weekend': 1 if date.weekday() >= 5 else 0,
'day_of_month': date.day,
'month': date.month,
'quarter': (date.month – 1) // 3 + 1,
'is_month_start': 1 if date.day <= 3 else 0,
'is_month_end': 1 if date.day >= 28 else 0,
# 是否为节假日
'is_holiday': self._is_holiday(date),
'days_to_holiday': self._days_to_next_holiday(date),
}

def _build_promotion_features(self, sku_id: str,
forecast_date: datetime,
horizon: int) -> dict:
"""促销特征：预测期内是否有计划促销"""
end_date = forecast_date + timedelta(days=horizon)

try:
promotions = pd.read_sql("""
SELECT
COUNT(*) AS promo_count,
AVG(discount_rate) AS avg_discount,
MAX(discount_rate) AS max_discount
FROM planned_promotions
WHERE sku_id = %s
AND start_date <= %s
AND end_date >= %s
""", self.mysql.get_connection(),
params=(sku_id, end_date, forecast_date))

if not promotions.empty:
row = promotions.iloc[0]
return {
'has_promotion': 1 if row['promo_count'] > 0 else 0,
'promo_avg_discount': row['avg_discount'] or 0,
'promo_max_discount': row['max_discount'] or 0,
}
except Exception:
pass

return {'has_promotion': 0, 'promo_avg_discount': 0,
'promo_max_discount': 0}

def _build_weather_features(self, date: datetime,
horizon: int) -> dict:
"""天气特征（从天气API获取）"""
try:
cache_key = f"weather:forecast:{date:%Y%m%d}:{horizon}"
cached = self.redis.get(cache_key)

if cached:
return json.loads(cached)

weather = self.weather.get_forecast(date, horizon)

features = {
'avg_temp': weather.get('avg_temp', 20),
'min_temp': weather.get('min_temp', 15),
'max_temp': weather.get('max_temp', 25),
'precipitation_prob': weather.get('precip_prob', 0),
'precipitation_mm': weather.get('precip_mm', 0),
'is_rain': 1 if weather.get('precip_mm', 0) > 1 else 0,
'is_extreme': 1 if (
weather.get('max_temp', 0) > 38 or
weather.get('min_temp', 0) < -5
) else 0,
}

# 缓存3小时
self.redis.setex(cache_key, 10800, json.dumps(features))
return features

except Exception:
return {'avg_temp': 20, 'min_temp': 15, 'max_temp': 25,
'precipitation_prob': 0, 'precipitation_mm': 0,
'is_rain': 0, 'is_extreme': 0}

四、需求预测的四个特征陷阱

陷阱一：促销的"挤占效应"。大促前7天的销量通常会下降50%（消费者在等降价），大促后7天也会下降30%（需求被提前透支）。简单的促销特征只关注了促销期间，漏掉了促销前后的异常波动。需要将促销窗口扩展为[-7, +7]天的特征。

陷阱二：新品冷启动的特征缺失。新SKU没有历史销量数据，滞后特征全为0。此时需要依赖"同品类相似SKU的历史表现"作为代理特征——用协同过滤从历史SKU中找最相似的品类。

陷阱三：特征的时变衰减。6个月前的销量对今天的预测权重应该比昨天的销量低得多。在模型层面可以通过时间衰减加权采样（样本权重 = exp(-delta_days/90)）来处理。

陷阱四：外部特征的采集延迟。天气API的预报数据可能在凌晨2点才更新，而凌晨3点就需要跑当日预测。采集管道的定时任务和重试机制需要精确到分钟级。

五、总结

AI需求预测的精度90%取决于特征工程，10%取决于模型选型。历史销量提供基数（"通常卖多少"），时间特征提供模式（"周末通常比工作日多"），促销特征提供扰动（"打了5折会多卖多少"），天气特征提供外部冲击（"高温天雪糕销量翻倍"）。

特征管道是"数据基础设施+业务理解的结合体"——管道的代码任何人都能写，但哪些特征对销量有预测力，只有深入业务的人才知道。

本文属于「行业场景与项目复盘」系列，解析AI需求预测中多维特征工程的数据管道设计。