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(days90) 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(dayshorizon) 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 fweather: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需求预测中多维特征工程的数据管道设计。