B2. 预测模型与智能决策 | Prediction Models & Intelligent Decision

路径: Path B: 技术人 · 模块: B2 最后更新: 2026-07-31 难度: 中级 → 进阶 前提: B1 数据管道基础（pandas、数据清洗）、Python 基础 预计时间: 每天 1 小时，2-3 周

flowchart LR
B1["B1 数据管道"]
B1 --> B2
B2[" B2 预测模型<br/>（当前）"]:::current
B2 --> B3
B3["B3 RAG 知识库"]
B3 --> B4
B4["B4 Agent 工作流"]
B4 --> B5
B5["B5 本地模型部署"]
classDef current fill:#ff9900,stroke:#333,color:#fff,font-weight:bold

章节导航

1. 预测方法论 · 2. 工具全景 · 3. 代码实战 · 4. 模型评估 · 5. 实战项目 · 6. 常见陷阱 · 7. 学习资源

本模块你将构建

销量预测模型 + Review 主题分析系统。

完成本模块后，你将能够：

• 用 Prophet 对 SKU 做 30/60/90 天销量预测，输出预测值和置信区间
• 理解时间序列预测的核心原理（趋势 + 季节性 + 噪声分解）
• 处理电商预测的特殊挑战：大促尖峰、新品冷启动、竞品影响
• 用 AutoGluon 零配置建模，自动选择最优算法
• 用 BERTopic 从 Review 文本中自动发现主题和情感趋势
• 将预测结果转化为补货决策（连接 A5 库存模块）
• 用 MAPE/MAE/RMSE 评估模型质量，用回测验证预测可靠性

相关案例: 多语言推荐系统 另一个建模走查——跨语言、跨文化的推荐，和本章的时序预测互补。

1. 预测方法论

本节的数字是为说明而构造的，不是实测值。

相关阅读: A5 库存与供应链 销量预测在库存补货决策中的应用详见 A5。 · D3 跨平台 AI 协同策略 跨平台需求预测详见 D3

1.1 时间序列预测的第一性原理

任何时间序列数据都可以分解为三个成分：

观测值 = 趋势(Trend) + 季节性(Seasonality) + 噪声(Residual)

加法模型 vs 乘法模型：

• 加法模型：y = trend + seasonality + noise 季节性波动幅度固定（如每年 Q4 多卖 1000 件）
• 乘法模型：y = trend × seasonality × noise 季节性波动幅度随趋势变化（如每年 Q4 多卖 30%）

电商场景通常用乘法模型更准确，因为销量基数越大，季节性波动的绝对值也越大。

分解可视化：

import pandas as pd
from statsmodels.tsa.seasonal import seasonal_decompose
import matplotlib.pyplot as plt
import matplotlib

matplotlib.rcParams["font.sans-serif"] = ["PingFang SC", "Heiti TC", "Arial"]
matplotlib.rcParams["axes.unicode_minus"] = False

def decompose_sales(df: pd.DataFrame, date_col: str = "date", value_col: str = "units"):
    """
    将销量时间序列分解为趋势、季节性、噪声三个成分。

    Args:
        df: 包含日期和销量的 DataFrame
        date_col: 日期列名
        value_col: 销量列名
    """
    ts = df.set_index(date_col)[value_col]
    ts = ts.asfreq("D").fillna(method="ffill") # 补齐缺失日期

    # 乘法分解，周期=7（周季节性）
    result = seasonal_decompose(ts, model="multiplicative", period=7)

    fig, axes = plt.subplots(4, 1, figsize=(12, 8), sharex=True)
    result.observed.plot(ax=axes[0], title="原始数据 (Observed)")
    result.trend.plot(ax=axes[1], title="趋势 (Trend)")
    result.seasonal.plot(ax=axes[2], title="季节性 (Seasonality)")
    result.resid.plot(ax=axes[3], title="噪声 (Residual)")

    plt.tight_layout()
    plt.savefig("output/decomposition.png", dpi=150)
    plt.show()

    return result

# 使用示例
# df = pd.read_csv("data/daily_sales.csv")
# result = decompose_sales(df, date_col="date", value_col="units")

关键洞察：如果分解后的“噪声“成分仍然有明显的模式（比如每逢大促就有巨大残差），说明你的模型缺少对大促事件的建模。这正是 Prophet 的 holidays 参数要解决的问题。

1.2 电商预测的特殊挑战

电商销量预测比传统零售更难，因为有几个独特的干扰因素：

断货数据处理（关键！）：

def handle_stockout(df: pd.DataFrame, units_col: str = "units") -> pd.DataFrame:
    """
    处理断货期间的零销量数据。

    断货期间销量为 0 不代表需求为 0。
    策略：用断货前后的平均销量填充，避免模型学到"某些天需求为 0"。
    """
    df = df.copy()

    # 标记连续零销量（可能是断货）
    df["is_zero"] = df[units_col] == 0
    df["zero_streak"] = (
        df["is_zero"]
        .groupby((~df["is_zero"]).cumsum())
        .cumsum()
    )

    # 连续 3 天以上零销量视为断货（而非真实零需求）
    stockout_mask = df["zero_streak"] >= 3

    if stockout_mask.any():
        # 用前后各 7 天的非零均值填充
        rolling_mean = (
            df[~stockout_mask][units_col]
            .rolling(window=7, min_periods=1)
            .mean()
        )
        df.loc[stockout_mask, units_col] = rolling_mean.reindex(
            df.index
        ).ffill().bfill()

        stockout_days = stockout_mask.sum()
        print(f" 检测到 {stockout_days} 天疑似断货，已用滚动均值填充")

    df = df.drop(columns=["is_zero", "zero_streak"])
    return df

1.3 何时用简单规则 vs 何时用 ML 模型

不是所有预测都需要机器学习。选择合适的方法比选择最复杂的方法更重要：

决策框架：

历史数据 > 1 年？
是 → 销量稳定？
是 → Prophet（简单高效）
否 → Prophet + 外部变量 / AutoGluon
否 → 历史数据 > 3 个月？
是 → Prophet（短周期）/ 移动平均
否 → 类比法（找相似品历史数据）

2. 工具全景

选择建议：

• 刚入门 → 从 Prophet 开始，一个下午就能出结果
• 想要自动化 → AutoGluon，零配置自动选最优模型
• 需要 Review 分析 → BERTopic，自动发现评论主题
• 需要优化补货量 → OR-Tools，数学规划求最优解

3. 代码实战

3.1 Prophet 销量预测（完整流程）

Prophet 是 Meta 开源的时间序列预测库，专为业务预测设计。它的核心优势：

• 自动处理缺失值和异常值
• 内置节假日效应
• 可解释的分解结果（趋势 + 季节性 + 节假日）
• 对非专业人员友好

完整代码：数据准备 → 训练 → 预测 → 评估 → 可视化

import pandas as pd
import numpy as np
from prophet import Prophet
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
import matplotlib

matplotlib.rcParams["font.sans-serif"] = ["PingFang SC", "Heiti TC", "Arial"]
matplotlib.rcParams["axes.unicode_minus"] = False

# ============================================================
# Step 1: 数据准备
# ============================================================

def prepare_prophet_data(
    df: pd.DataFrame,
    date_col: str = "date",
    value_col: str = "units"
) -> pd.DataFrame:
    """
    将业务数据转换为 Prophet 要求的格式。

    Prophet 要求两列：
    - ds: 日期列（datetime 类型）
    - y: 目标值列（数值类型）

    Args:
        df: 原始数据
        date_col: 日期列名
        value_col: 目标值列名

    Returns:
        Prophet 格式的 DataFrame
    """
    prophet_df = df[[date_col, value_col]].copy()
    prophet_df.columns = ["ds", "y"]

    # 确保日期格式正确
    prophet_df["ds"] = pd.to_datetime(prophet_df["ds"])
    prophet_df["y"] = pd.to_numeric(prophet_df["y"], errors="coerce")

    # 按日期排序并去重（同一天多条记录则求和）
    prophet_df = prophet_df.groupby("ds")["y"].sum().reset_index()
    prophet_df = prophet_df.sort_values("ds").reset_index(drop=True)

    # 补齐缺失日期（用 0 填充，后续可用断货处理逻辑替换）
    date_range = pd.date_range(
        start=prophet_df["ds"].min(),
        end=prophet_df["ds"].max(),
        freq="D"
    )
    prophet_df = (
        prophet_df
        .set_index("ds")
        .reindex(date_range)
        .fillna(0)
        .reset_index()
        .rename(columns={"index": "ds"})
    )

    print(f" 数据准备完成: {len(prophet_df)} 天")
    print(f" 日期范围: {prophet_df['ds'].min().date()} → {prophet_df['ds'].max().date()}")
    print(f" 日均销量: {prophet_df['y'].mean():.1f}")

    return prophet_df

# ============================================================
# Step 2: 训练模型
# ============================================================

def train_prophet(
    df: pd.DataFrame,
    yearly: bool = True,
    weekly: bool = True,
    daily: bool = False,
    changepoint_prior: float = 0.05
) -> Prophet:
    """
    训练 Prophet 模型。

    Args:
        df: Prophet 格式数据（ds, y 两列）
        yearly: 是否启用年季节性
        weekly: 是否启用周季节性
        daily: 是否启用日季节性（通常不需要）
        changepoint_prior: 趋势变化点灵敏度
            - 值越大，模型越容易捕捉趋势变化（但可能过拟合）
            - 值越小，趋势越平滑（但可能欠拟合）
            - 默认 0.05，电商场景建议 0.1-0.3（变化较快）

    Returns:
        训练好的 Prophet 模型
    """
    model = Prophet(
        yearly_seasonality=yearly,
        weekly_seasonality=weekly,
        daily_seasonality=daily,
        changepoint_prior_scale=changepoint_prior,
        interval_width=0.8, # 80% 置信区间
    )

    model.fit(df)
    print(" 模型训练完成")

    return model

# ============================================================
# Step 3: 生成预测
# ============================================================

def make_forecast(
    model: Prophet,
    periods: int = 90,
    freq: str = "D"
) -> pd.DataFrame:
    """
    生成未来 N 天的预测。

    Args:
        model: 训练好的 Prophet 模型
        periods: 预测天数
        freq: 频率（D=天, W=周, M=月）

    Returns:
        预测结果 DataFrame，包含：
        - ds: 日期
        - yhat: 预测值
        - yhat_lower: 预测下界
        - yhat_upper: 预测上界
        - trend: 趋势成分
        - weekly: 周季节性成分
        - yearly: 年季节性成分
    """
    future = model.make_future_dataframe(periods=periods, freq=freq)
    forecast = model.predict(future)

    # 预测值不能为负（销量最小为 0）
    forecast["yhat"] = forecast["yhat"].clip(lower=0)
    forecast["yhat_lower"] = forecast["yhat_lower"].clip(lower=0)

    print(f" 预测完成: 未来 {periods} 天")
    print(f" 预测均值: {forecast['yhat'].tail(periods).mean():.1f}")
    print(f" 预测区间: [{forecast['yhat_lower'].tail(periods).mean():.1f}, "
          f"{forecast['yhat_upper'].tail(periods).mean():.1f}]")

    return forecast

# ============================================================
# Step 4: 可视化
# ============================================================

def plot_forecast(
    model: Prophet,
    forecast: pd.DataFrame,
    actual_df: pd.DataFrame = None,
    title: str = "SKU 销量预测"
):
    """
    绘制预测结果图。

    Args:
        model: Prophet 模型
        forecast: 预测结果
        actual_df: 实际数据（用于对比）
        title: 图表标题
    """
    fig, axes = plt.subplots(2, 1, figsize=(14, 10))

    # 图 1: 预测 vs 实际
    ax1 = axes[0]
    ax1.plot(forecast["ds"], forecast["yhat"], color="#1a73e8", label="预测值")
    ax1.fill_between(
        forecast["ds"],
        forecast["yhat_lower"],
        forecast["yhat_upper"],
        alpha=0.2, color="#1a73e8", label="80% 置信区间"
    )

    if actual_df is not None:
        ax1.scatter(
            actual_df["ds"], actual_df["y"],
            color="#333", s=10, alpha=0.5, label="实际值"
        )

    ax1.set_title(title, fontsize=14, fontweight="bold")
    ax1.set_ylabel("销量 (Units)")
    ax1.legend()
    ax1.grid(True, alpha=0.3)

    # 图 2: 成分分解
    ax2 = axes[1]
    ax2.plot(forecast["ds"], forecast["trend"], label="趋势", color="#e8710a")
    if "weekly" in forecast.columns:
        ax2_twin = ax2.twinx()
        weekly_data = forecast.drop_duplicates(subset=["ds"]).tail(90)
        ax2_twin.plot(
            weekly_data["ds"], weekly_data["weekly"],
            label="周季节性", color="#0d652d", alpha=0.7
        )
        ax2_twin.set_ylabel("周季节性")

    ax2.set_title("趋势分解", fontsize=14, fontweight="bold")
    ax2.set_ylabel("趋势")
    ax2.legend(loc="upper left")
    ax2.grid(True, alpha=0.3)

    plt.tight_layout()
    plt.savefig("output/forecast.png", dpi=150, bbox_inches="tight")
    plt.show()
    print(" 图表已保存: output/forecast.png")

# ============================================================
# 完整使用示例
# ============================================================

# # 1. 加载数据
# raw_df = pd.read_csv("data/daily_sales.csv")
# prophet_df = prepare_prophet_data(raw_df, date_col="date", value_col="units")
#
# # 2. 处理断货数据
# prophet_df = handle_stockout(prophet_df, units_col="y")
#
# # 3. 训练模型
# model = train_prophet(prophet_df, changepoint_prior=0.1)
#
# # 4. 预测未来 90 天
# forecast = make_forecast(model, periods=90)
#
# # 5. 可视化
# plot_forecast(model, forecast, actual_df=prophet_df, title="ASIN-B0XXXXX 销量预测")

changepoint_prior_scale 调参指南：这是 Prophet 最重要的超参数。电商数据变化快，建议从 0.1 开始尝试。如果预测曲线太平滑（跟不上趋势变化），调大到 0.2-0.3；如果预测曲线太波动（过拟合噪声），调小到 0.01-0.05。

3.2 Prophet 进阶：节假日效应与外部变量

基础 Prophet 模型忽略了电商最重要的因素：大促事件。添加节假日效应可以显著提升预测精度。

添加电商大促事件：

def create_ecommerce_holidays(years: list[int]) -> pd.DataFrame:
    """
    创建电商大促日历。

    Prophet 的 holidays 参数接受一个 DataFrame，包含：
    - holiday: 事件名称
    - ds: 事件日期
    - lower_window: 事件前影响天数（负数）
    - upper_window: 事件后影响天数
    """
    holidays = []

    for year in years:
        # Prime Day（通常 7 月中旬，持续 2 天）
        holidays.append({
            "holiday": "prime_day",
            "ds": f"{year}-07-12",
            "lower_window": -3, # 提前 3 天开始有影响（预热期）
            "upper_window": 2, # 结束后 2 天仍有余波
        })

        # Black Friday（11 月第四个周五）
        # 简化处理：固定 11-24 附近
        holidays.append({
            "holiday": "black_friday",
            "ds": f"{year}-11-24",
            "lower_window": -7, # BFCM 周提前一周开始
            "upper_window": 3, # Cyber Monday 后几天
        })

        # Cyber Monday
        holidays.append({
            "holiday": "cyber_monday",
            "ds": f"{year}-11-27",
            "lower_window": 0,
            "upper_window": 1,
        })

        # 双十一（对中国卖家有影响）
        holidays.append({
            "holiday": "singles_day",
            "ds": f"{year}-11-11",
            "lower_window": -3,
            "upper_window": 1,
        })

        # 圣诞节前购物季
        holidays.append({
            "holiday": "christmas_shopping",
            "ds": f"{year}-12-15",
            "lower_window": -5,
            "upper_window": 10,
        })

        # 新年后低谷期
        holidays.append({
            "holiday": "post_newyear_dip",
            "ds": f"{year}-01-05",
            "lower_window": -5,
            "upper_window": 10,
        })

    return pd.DataFrame(holidays)

def train_prophet_with_holidays(
    df: pd.DataFrame,
    holidays: pd.DataFrame = None,
    changepoint_prior: float = 0.1
) -> Prophet:
    """
    训练带节假日效应的 Prophet 模型。
    """
    if holidays is None:
        years = list(range(
            df["ds"].dt.year.min(),
            df["ds"].dt.year.max() + 2 # 包含预测年份
        ))
        holidays = create_ecommerce_holidays(years)

    model = Prophet(
        yearly_seasonality=True,
        weekly_seasonality=True,
        daily_seasonality=False,
        changepoint_prior_scale=changepoint_prior,
        holidays=holidays,
        holidays_prior_scale=10.0, # 节假日效应的灵敏度
        interval_width=0.8,
    )

    model.fit(df)
    print(f" 模型训练完成（含 {len(holidays)} 个节假日事件）")

    return model

添加外部回归变量（广告花费、竞品价格）：

def train_prophet_with_regressors(
    df: pd.DataFrame,
    regressor_cols: list[str] = None
) -> Prophet:
    """
    训练带外部回归变量的 Prophet 模型。

    外部变量可以是：
    - ad_spend: 广告花费（广告投入越多，销量越高）
    - competitor_price: 竞品价格（竞品涨价，自己销量可能上升）
    - bsr_rank: BSR 排名（排名越高，曝光越多）
    - coupon_active: 是否有优惠券（0/1）

    注意：预测时也需要提供未来的外部变量值！
    """
    years = list(range(
        df["ds"].dt.year.min(),
        df["ds"].dt.year.max() + 2
    ))
    holidays = create_ecommerce_holidays(years)

    model = Prophet(
        yearly_seasonality=True,
        weekly_seasonality=True,
        changepoint_prior_scale=0.1,
        holidays=holidays,
        interval_width=0.8,
    )

    # 添加外部回归变量
    regressor_cols = regressor_cols or []
    for col in regressor_cols:
        if col in df.columns:
            model.add_regressor(col, standardize=True)
            print(f" 添加回归变量: {col}")

    model.fit(df)
    print(" 模型训练完成（含外部变量）")

    return model

def forecast_with_regressors(
    model: Prophet,
    periods: int = 90,
    future_regressors: pd.DataFrame = None
) -> pd.DataFrame:
    """
    带外部变量的预测。

    Args:
        model: 训练好的模型
        periods: 预测天数
        future_regressors: 未来的外部变量值
            如果不提供，用历史均值填充（不推荐，会降低精度）
    """
    future = model.make_future_dataframe(periods=periods)

    # 合并未来的外部变量
    if future_regressors is not None:
        future = future.merge(future_regressors, on="ds", how="left")

    # 对缺失的外部变量用历史均值填充
    for col in future.columns:
        if col not in ["ds"] and future[col].isna().any():
            fill_value = future[col].dropna().mean()
            future[col] = future[col].fillna(fill_value)
            print(f" {col} 有缺失值，用均值 {fill_value:.2f} 填充")

    forecast = model.predict(future)
    forecast["yhat"] = forecast["yhat"].clip(lower=0)
    forecast["yhat_lower"] = forecast["yhat_lower"].clip(lower=0)

    return forecast

# 使用示例
# df = prepare_prophet_data(raw_df)
# df["ad_spend"] = ad_data["spend"] # 合并广告花费数据
# df["competitor_price"] = competitor_data["price"] # 合并竞品价格
#
# model = train_prophet_with_regressors(df, regressor_cols=["ad_spend", "competitor_price"])
#
# # 预测时需要提供未来的广告预算和竞品价格预估
# future_regs = pd.DataFrame({
# "ds": pd.date_range("2025-04-01", periods=90, freq="D"),
# "ad_spend": [500] * 90, # 假设未来每天广告预算 $500
# "competitor_price": [29.99] * 90, # 假设竞品价格不变
# })
# forecast = forecast_with_regressors(model, periods=90, future_regressors=future_regs)

外部变量的陷阱：预测时你需要提供未来的外部变量值。如果你不知道未来的广告预算，那添加 ad_spend 作为回归变量反而会降低预测精度。只添加你能合理预估未来值的变量。

3.3 AutoGluon 自动化预测（零配置建模）

AutoGluon 是 Amazon 开源的自动化机器学习框架。它的时间序列模块可以自动尝试多种模型（Prophet、ETS、DeepAR、Theta 等），选择最优的一个。适合不想手动调参的场景。

from autogluon.timeseries import TimeSeriesDataFrame, TimeSeriesPredictor

def autogluon_forecast(
    df: pd.DataFrame,
    date_col: str = "date",
    value_col: str = "units",
    item_col: str = "asin",
    prediction_length: int = 30,
    time_limit: int = 300
) -> pd.DataFrame:
    """
    用 AutoGluon 自动化预测多个 SKU 的销量。

    AutoGluon 的优势：
    - 零配置：不需要选模型、调参数
    - 多 SKU：一次训练，同时预测所有 SKU
    - 自动集成：自动尝试多种模型并集成最优结果

    Args:
        df: 包含日期、销量、SKU 标识的 DataFrame
        date_col: 日期列名
        value_col: 目标值列名
        item_col: SKU 标识列名
        prediction_length: 预测天数
        time_limit: 训练时间限制（秒）

    Returns:
        预测结果 DataFrame
    """
    # 1. 转换为 AutoGluon 格式
    ag_df = df.rename(columns={
        date_col: "timestamp",
        value_col: "target",
        item_col: "item_id"
    })
    ag_df["timestamp"] = pd.to_datetime(ag_df["timestamp"])

    ts_df = TimeSeriesDataFrame.from_data_frame(
        ag_df,
        id_column="item_id",
        timestamp_column="timestamp"
    )

    print(f" 数据: {ts_df.num_items} 个 SKU, "
          f"{len(ts_df)} 条记录")

    # 2. 训练（AutoGluon 自动选择最优模型）
    predictor = TimeSeriesPredictor(
        prediction_length=prediction_length,
        target="target",
        eval_metric="MAPE", # 用 MAPE 作为评估指标
    )

    predictor.fit(
        train_data=ts_df,
        time_limit=time_limit, # 限制训练时间
        presets="medium_quality", # fast / medium / high / best
    )

    # 3. 查看模型排行榜
    leaderboard = predictor.leaderboard(ts_df)
    print("\n 模型排行榜:")
    print(leaderboard[["model", "score_val"]].to_string(index=False))

    # 4. 生成预测
    predictions = predictor.predict(ts_df)

    print(f"\n 预测完成: {ts_df.num_items} 个 SKU × {prediction_length} 天")

    return predictions

# 使用示例
# df = pd.read_csv("data/daily_sales_all_skus.csv")
# predictions = autogluon_forecast(
# df,
# date_col="date",
# value_col="units",
# item_col="asin",
# prediction_length=30,
# time_limit=600 # 10 分钟
# )
#
# # 查看某个 SKU 的预测
# sku_pred = predictions.loc["B0XXXXX"]
# print(sku_pred)

AutoGluon vs Prophet 怎么选？

• 单个 SKU 深度分析 → Prophet（可解释性强，可以添加节假日和外部变量）
• 批量预测 100+ SKU → AutoGluon（自动化程度高，一次搞定）
• 不确定用什么 → 先用 AutoGluon 跑一遍基线，再用 Prophet 对重点 SKU 精调

参考：AutoGluon 时间序列文档

3.4 BERTopic Review 主题分析

BERTopic 可以从大量 Review 文本中自动发现主题，帮助你理解客户在说什么。这比一条条手动读 Review 快得多。

from bertopic import BERTopic
from sentence_transformers import SentenceTransformer
import pandas as pd

def analyze_review_topics(
    reviews: list[str],
    language: str = "english",
    nr_topics: int = "auto",
    min_topic_size: int = 10
) -> tuple:
    """
    用 BERTopic 从 Review 文本中自动发现主题。

    工作原理：
    1. 用 Sentence-BERT 将每条 Review 转换为向量
    2. 用 UMAP 降维
    3. 用 HDBSCAN 聚类
    4. 用 c-TF-IDF 提取每个主题的关键词

    Args:
        reviews: Review 文本列表
        language: 语言 ("english" 或 "chinese")
        nr_topics: 主题数量（"auto" 自动确定）
        min_topic_size: 最小主题大小（Review 数少于此值的主题会被合并）

    Returns:
        (topic_model, topics, probs)
        - topic_model: 训练好的 BERTopic 模型
        - topics: 每条 Review 的主题编号
        - probs: 每条 Review 属于各主题的概率
    """
    # 选择嵌入模型
    if language == "chinese":
        embedding_model = SentenceTransformer(
            "paraphrase-multilingual-MiniLM-L12-v2"
        )
    else:
        embedding_model = SentenceTransformer(
            "all-MiniLM-L6-v2"
        )

    # 创建 BERTopic 模型
    topic_model = BERTopic(
        embedding_model=embedding_model,
        nr_topics=nr_topics,
        min_topic_size=min_topic_size,
        language=language,
        verbose=True
    )

    # 训练
    topics, probs = topic_model.fit_transform(reviews)

    # 输出主题概览
    topic_info = topic_model.get_topic_info()
    print("\n 发现的主题:")
    for _, row in topic_info.head(10).iterrows():
        if row["Topic"] != -1: # -1 是离群点
            print(f" 主题 {row['Topic']}: {row['Name']} "
                  f"({row['Count']} 条 Review)")

    return topic_model, topics, probs

def get_topic_summary(
    topic_model: BERTopic,
    reviews: list[str],
    topics: list[int],
    ratings: list[int] = None
) -> pd.DataFrame:
    """
    生成主题摘要报告。

    Args:
        topic_model: 训练好的模型
        reviews: Review 文本
        topics: 主题编号
        ratings: 评分（1-5），用于分析每个主题的情感倾向

    Returns:
        主题摘要 DataFrame
    """
    summary_data = []
    topic_info = topic_model.get_topic_info()

    for _, row in topic_info.iterrows():
        topic_id = row["Topic"]
        if topic_id == -1:
            continue

        # 获取该主题的关键词
        keywords = topic_model.get_topic(topic_id)
        keyword_str = ", ".join([w for w, _ in keywords[:5]])

        # 获取该主题的 Review 索引
        topic_mask = [t == topic_id for t in topics]
        topic_reviews = [r for r, m in zip(reviews, topic_mask) if m]

        entry = {
            "topic_id": topic_id,
            "keywords": keyword_str,
            "review_count": len(topic_reviews),
            "sample_review": topic_reviews[0][:200] if topic_reviews else "",
        }

        # 如果有评分数据，计算该主题的平均评分
        if ratings:
            topic_ratings = [r for r, m in zip(ratings, topic_mask) if m]
            entry["avg_rating"] = round(sum(topic_ratings) / len(topic_ratings), 2) if topic_ratings else None
            entry["negative_pct"] = round(
                sum(1 for r in topic_ratings if r <= 2) / len(topic_ratings) * 100, 1
            ) if topic_ratings else None

        summary_data.append(entry)

    summary = pd.DataFrame(summary_data)

    if "avg_rating" in summary.columns:
        summary = summary.sort_values("avg_rating", ascending=True)

    return summary

# 使用示例
# reviews_df = pd.read_csv("data/reviews.csv")
# reviews = reviews_df["review_text"].tolist()
# ratings = reviews_df["rating"].tolist()
#
# topic_model, topics, probs = analyze_review_topics(reviews, language="english")
#
# # 生成主题摘要
# summary = get_topic_summary(topic_model, reviews, topics, ratings)
# print(summary.to_string(index=False))
#
# # 可视化主题分布
# fig = topic_model.visualize_topics()
# fig.write_html("output/review_topics.html")
#
# # 查看主题随时间的变化（需要日期数据）
# timestamps = reviews_df["date"].tolist()
# topics_over_time = topic_model.topics_over_time(reviews, topics, timestamps)
# fig = topic_model.visualize_topics_over_time(topics_over_time)
# fig.write_html("output/topics_over_time.html")

BERTopic 的实际价值：假设你有 5000 条竞品 Review，手动读完需要几天。BERTopic 5 分钟就能告诉你：主题 1 是“电池续航差“（平均评分 2.1），主题 2 是“画质好“（平均评分 4.5），主题 3 是“App 难用“（平均评分 1.8）。这直接告诉你产品改进方向。

参考：BERTopic 官方文档 | BERTopic 最佳实践

3.5 预测结果转化为补货决策

预测本身不是目的，补货决策才是。这一节把预测结果连接到 A5 库存模块 的补货逻辑。

def forecast_to_reorder(
    forecast: pd.DataFrame,
    current_stock: int,
    lead_time_days: int = 30,
    safety_stock_days: int = 14,
    moq: int = 100
) -> dict:
    """
    将预测结果转化为补货建议。

    Args:
        forecast: Prophet 预测结果
        current_stock: 当前库存数量
        lead_time_days: 供应商交货周期（天）
        safety_stock_days: 安全库存天数
        moq: 最小起订量

    Returns:
        补货建议字典
    """
    # 取未来预测数据
    future_data = forecast[forecast["ds"] > pd.Timestamp.now()]

    if future_data.empty:
        return {"error": "无未来预测数据"}

    # 计算日均预测销量（用上界做保守估计）
    daily_forecast = future_data["yhat"].mean()
    daily_upper = future_data["yhat_upper"].mean()

    # 安全库存 = 安全天数 × 日均销量上界
    safety_stock = int(safety_stock_days * daily_upper)

    # Lead Time 期间的预期消耗
    lt_consumption = int(lead_time_days * daily_forecast)

    # 再订货点 = Lead Time 消耗 + 安全库存
    reorder_point = lt_consumption + safety_stock

    # 当前库存可支撑天数
    days_of_stock = int(current_stock / daily_forecast) if daily_forecast > 0 else 999

    # 建议订货量 = 90 天预测需求 - 当前库存 + 安全库存
    forecast_90d = int(future_data["yhat"].head(90).sum())
    suggested_qty = max(forecast_90d - current_stock + safety_stock, 0)

    # 向上取整到 MOQ 的倍数
    if suggested_qty > 0:
        suggested_qty = max(
            ((suggested_qty + moq - 1) // moq) * moq,
            moq
        )

    # 紧急程度判断
    if current_stock <= reorder_point * 0.5:
        urgency = " 紧急补货"
    elif current_stock <= reorder_point:
        urgency = " 建议补货"
    else:
        urgency = " 库存充足"

    result = {
        "urgency": urgency,
        "current_stock": current_stock,
        "days_of_stock": days_of_stock,
        "daily_forecast": round(daily_forecast, 1),
        "safety_stock": safety_stock,
        "reorder_point": reorder_point,
        "suggested_qty": suggested_qty,
        "forecast_90d": forecast_90d,
        "lead_time_days": lead_time_days,
    }

    print(f"\n 补货建议:")
    print(f" 状态: {urgency}")
    print(f" 当前库存: {current_stock} 件（可支撑 {days_of_stock} 天）")
    print(f" 日均预测: {daily_forecast:.1f} 件/天")
    print(f" 安全库存: {safety_stock} 件")
    print(f" 再订货点: {reorder_point} 件")
    print(f" 建议订货: {suggested_qty} 件（MOQ={moq}）")

    return result

# 使用示例
# reorder = forecast_to_reorder(
# forecast=forecast,
# current_stock=500,
# lead_time_days=30,
# safety_stock_days=14,
# moq=200
# )

4. 模型评估

4.1 评估指标

电商场景推荐用 WAPE：MAPE 在实际销量接近 0 时会趋向无穷大（除以接近 0 的数），而 WAPE 用总销量做分母，更稳定。

def evaluate_forecast(
    actual: pd.Series,
    predicted: pd.Series
) -> dict:
    """
    计算预测评估指标。

    Args:
        actual: 实际值
        predicted: 预测值

    Returns:
        评估指标字典
    """
    actual = actual.values
    predicted = predicted.values

    mae = np.mean(np.abs(actual - predicted))
    rmse = np.sqrt(np.mean((actual - predicted) ** 2))

    # MAPE（过滤掉实际值为 0 的天）
    nonzero_mask = actual > 0
    if nonzero_mask.any():
        mape = np.mean(
            np.abs((actual[nonzero_mask] - predicted[nonzero_mask])
                   / actual[nonzero_mask])
        ) * 100
    else:
        mape = float("inf")

    # WAPE（更稳健）
    wape = np.sum(np.abs(actual - predicted)) / np.sum(actual) * 100 if np.sum(actual) > 0 else float("inf")

    metrics = {
        "MAE": round(mae, 2),
        "RMSE": round(rmse, 2),
        "MAPE": round(mape, 2),
        "WAPE": round(wape, 2),
    }

    print(" 评估结果:")
    for k, v in metrics.items():
        unit = "%" if k in ("MAPE", "WAPE") else "units"
        print(f" {k}: {v} {unit}")

    return metrics

MAPE 参考基准（电商场景）：

4.2 回测（Backtesting）

回测是验证预测模型最可靠的方法：用历史数据模拟“如果当时用这个模型预测，结果会怎样“。

def backtest_prophet(
    df: pd.DataFrame,
    initial_days: int = 180,
    horizon_days: int = 30,
    period_days: int = 30
) -> pd.DataFrame:
    """
    Prophet 回测：滚动窗口验证。

    原理：
    1. 用前 initial_days 天数据训练
    2. 预测未来 horizon_days 天
    3. 与实际值对比
    4. 窗口向前滑动 period_days 天，重复

    Args:
        df: Prophet 格式数据
        initial_days: 初始训练数据天数
        horizon_days: 每次预测的天数
        period_days: 窗口滑动步长

    Returns:
        回测结果 DataFrame
    """
    from prophet.diagnostics import cross_validation, performance_metrics

    model = Prophet(
        yearly_seasonality=True,
        weekly_seasonality=True,
        changepoint_prior_scale=0.1,
        interval_width=0.8,
    )
    model.fit(df)

    # 交叉验证
    cv_results = cross_validation(
        model,
        initial=f"{initial_days} days",
        period=f"{period_days} days",
        horizon=f"{horizon_days} days"
    )

    # 计算性能指标
    perf = performance_metrics(cv_results)

    print(" 回测结果:")
    print(f" MAE: {perf['mae'].mean():.2f}")
    print(f" RMSE: {perf['rmse'].mean():.2f}")
    print(f" MAPE: {perf['mape'].mean() * 100:.2f}%")

    return cv_results, perf

# 使用示例
# cv_results, perf = backtest_prophet(prophet_df, initial_days=180, horizon_days=30)
#
# # 可视化回测结果
# from prophet.plot import plot_cross_validation_metric
# fig = plot_cross_validation_metric(cv_results, metric="mape")
# plt.savefig("output/backtest_mape.png", dpi=150)

4.3 预测区间的使用

预测值是一个点估计，但业务决策需要考虑不确定性。Prophet 的预测区间告诉你“真实值大概率落在这个范围内“。

5. 实战项目：构建 SKU 销量预测系统

5.1 项目架构

sales-forecaster/
config.py # 配置（数据路径、模型参数）
requirements.txt # 依赖

data/ # 数据目录
raw/ # 原始销量数据
processed/ # 清洗后的数据

models/ # 模型存储
prophet/ # Prophet 模型文件

src/
data_prep.py # 数据准备（清洗、断货处理）
prophet_model.py # Prophet 训练和预测
autogluon_model.py # AutoGluon 批量预测
review_analysis.py # BERTopic Review 分析
evaluator.py # 模型评估和回测
reorder.py # 补货决策

output/ # 输出
forecasts/ # 预测结果 CSV
reports/ # HTML 报告
plots/ # 图表

run_forecast.py # 主入口：单 SKU 预测
run_batch_forecast.py # 批量预测入口
README.md

5.2 主入口脚本

# run_forecast.py 单 SKU 销量预测
import argparse
import pandas as pd
from pathlib import Path

from src.data_prep import prepare_prophet_data, handle_stockout
from src.prophet_model import (
    train_prophet_with_holidays,
    make_forecast,
    plot_forecast
)
from src.evaluator import evaluate_forecast, backtest_prophet
from src.reorder import forecast_to_reorder

def run(
    data_path: str,
    asin: str = None,
    forecast_days: int = 90,
    current_stock: int = None,
    lead_time: int = 30
):
    """
    完整预测流程：数据准备 → 训练 → 预测 → 评估 → 补货建议
    """
    print(f" 开始预测流程")

    # 1. 加载数据
    df = pd.read_csv(data_path)
    if asin and "asin" in df.columns:
        df = df[df["asin"] == asin]
        print(f" SKU: {asin}")

    # 2. 数据准备
    prophet_df = prepare_prophet_data(df, date_col="date", value_col="units")
    prophet_df = handle_stockout(prophet_df, units_col="y")

    # 3. 训练（带节假日效应）
    model = train_prophet_with_holidays(prophet_df, changepoint_prior=0.1)

    # 4. 预测
    forecast = make_forecast(model, periods=forecast_days)

    # 5. 评估（用最后 30 天做验证）
    if len(prophet_df) > 30:
        train_df = prophet_df.iloc[:-30]
        test_df = prophet_df.iloc[-30:]

        eval_model = train_prophet_with_holidays(train_df)
        eval_forecast = make_forecast(eval_model, periods=30)

        eval_pred = eval_forecast.tail(30)["yhat"].values
        eval_actual = test_df["y"].values
        metrics = evaluate_forecast(
            pd.Series(eval_actual), pd.Series(eval_pred)
        )

    # 6. 可视化
    output_dir = Path("output")
    output_dir.mkdir(exist_ok=True)

    plot_forecast(
        model, forecast, actual_df=prophet_df,
        title=f"{'ASIN ' + asin if asin else 'SKU'} 销量预测 ({forecast_days}天)"
    )

    # 7. 保存预测结果
    forecast_output = forecast[["ds", "yhat", "yhat_lower", "yhat_upper"]].tail(forecast_days)
    forecast_output.to_csv(
        output_dir / f"forecast_{asin or 'sku'}_{forecast_days}d.csv",
        index=False
    )

    # 8. 补货建议
    if current_stock is not None:
        reorder = forecast_to_reorder(
            forecast,
            current_stock=current_stock,
            lead_time_days=lead_time
        )

    print(f"\n 预测完成！结果已保存到 output/")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="SKU 销量预测")
    parser.add_argument("--data", required=True, help="销量数据 CSV 路径")
    parser.add_argument("--asin", help="ASIN 编号")
    parser.add_argument("--days", type=int, default=90, help="预测天数")
    parser.add_argument("--stock", type=int, help="当前库存")
    parser.add_argument("--lead-time", type=int, default=30, help="交货周期(天)")
    args = parser.parse_args()

    run(
        data_path=args.data,
        asin=args.asin,
        forecast_days=args.days,
        current_stock=args.stock,
        lead_time=args.lead_time
    )

# 运行示例
python3 run_forecast.py --data data/daily_sales.csv --asin B0XXXXX --days 90
python3 run_forecast.py --data data/daily_sales.csv --asin B0XXXXX --stock 500 --lead-time 30

6. 常见陷阱

7. 学习资源

7.1 免费课程与文档

7.2 推荐 GitHub 仓库

Content rephrased for compliance with licensing restrictions. Sources cited inline.

9. 完成标志

这里的数值是判断自己数据用的参考线，不是市场实测均值。跑过一轮后用你自己的中位数替换。

• 用 Prophet 对一个真实 SKU 做 90 天销量预测，输出预测值和置信区间
• 添加电商大促节假日效应（Prime Day、BFCM），对比有无节假日的预测精度差异
• 用回测（backtesting）验证模型，MAPE < 30%
• 用 AutoGluon 对 10+ SKU 做批量预测，查看模型排行榜
• 用 BERTopic 分析一组 Review 文本，发现至少 3 个有意义的主题
• 将预测结果转化为补货建议（计算再订货点和建议订货量）

完成以上所有项目后，你已经掌握了电商预测建模的核心技能。接下来进入 B3 RAG 知识库，学习如何构建基于 RAG 的智能问答系统。

什么时候这套不管用

• 历史数据不足一个完整年周期。 Prophet 的年季节性需要至少一年数据才能学到。只有几个月时，它会把促销尖峰或者一次断货当成规律。这种情况下移动平均加人工判断比模型可靠，等数据够了再上。
• 需求由事件驱动而不是时序延续。 节日礼品、新品发布带动的配件、跟热点走的品——这些的下一期需求不是上一期的函数。时序模型在这些品类上会稳定滞后。要么把事件作为外部回归变量显式加进去，要么按事件手工排产。
• 预测精度提升换不来决策变化。 举例：MAPE 往下压几个点听起来是进步，但如果你的补货批量有最小起订量、交期又以月计，这点精度差根本不改变你下什么单。先算清楚“精度提高多少才会让我改决定“，再决定值不值得调模型。
• SKU 太少或太新。 十几个 SKU 的账户，人工看趋势加经验判断通常不比模型差，而且没有维护成本。AutoGluon 这类多 SKU 批量预测的价值在几百上千个 SKU 上才体现出来。

附录

附录 A：模型选择决策树

你的预测需求是什么？

单个 SKU 深度预测
有 1 年以上历史数据？
是 → Prophet + 节假日 + 外部变量
否 → Prophet 基础版 / 移动平均
需要解释为什么这样预测？
是 → Prophet（可分解为趋势+季节性）
否 → AutoGluon（自动选最优模型）

批量预测 100+ SKU
有 GPU？
是 → AutoGluon (high_quality preset)
否 → AutoGluon (medium_quality preset)
需要快速出结果？
是 → AutoGluon (fast_training preset)

Review 文本分析
发现主题 → BERTopic
情感分类 → scikit-learn + TF-IDF / BERT
关键词提取 → BERTopic 的 c-TF-IDF

补货优化
简单规则 → 预测值 + 安全库存公式
多约束优化 → OR-Tools（考虑 MOQ、仓容、资金）

附录 B：代码速查表

# === Prophet 基础 ===
from prophet import Prophet

df = pd.DataFrame({"ds": dates, "y": values}) # 必须是 ds 和 y
model = Prophet()
model.fit(df)
future = model.make_future_dataframe(periods=90)
forecast = model.predict(future)
model.plot(forecast) # 预测图
model.plot_components(forecast) # 成分分解图

# === Prophet 节假日 ===
holidays = pd.DataFrame({
"holiday": ["prime_day"], "ds": ["2025-07-12"],
"lower_window": [-3], "upper_window": [2]
})
model = Prophet(holidays=holidays)

# === Prophet 外部变量 ===
model = Prophet()
model.add_regressor("ad_spend")
model.fit(df) # df 必须包含 ad_spend 列
future["ad_spend"] = 500 # 预测时也要提供

# === Prophet 回测 ===
from prophet.diagnostics import cross_validation, performance_metrics
cv = cross_validation(model, initial="180 days", period="30 days", horizon="30 days")
perf = performance_metrics(cv)

# === AutoGluon ===
from autogluon.timeseries import TimeSeriesDataFrame, TimeSeriesPredictor
ts = TimeSeriesDataFrame.from_data_frame(df, id_column="item_id", timestamp_column="timestamp")
predictor = TimeSeriesPredictor(prediction_length=30, target="target")
predictor.fit(ts, time_limit=300)
predictions = predictor.predict(ts)

# === BERTopic ===
from bertopic import BERTopic
model = BERTopic(language="english", min_topic_size=10)
topics, probs = model.fit_transform(documents)
model.get_topic_info() # 主题概览
model.visualize_topics() # 主题可视化

# === 评估指标 ===
mae = np.mean(np.abs(actual - predicted))
rmse = np.sqrt(np.mean((actual - predicted) ** 2))
mape = np.mean(np.abs((actual - predicted) / actual)) * 100
wape = np.sum(np.abs(actual - predicted)) / np.sum(actual) * 100

附录 C：依赖安装

# 基础预测
pip install prophet pandas numpy matplotlib

# AutoGluon（较大，建议单独安装）
pip install autogluon.timeseries

# Review 分析
pip install bertopic sentence-transformers

# 运筹优化
pip install ortools

# 全部安装
pip install prophet pandas numpy matplotlib \
autogluon.timeseries \
bertopic sentence-transformers \
ortools scikit-learn \
statsmodels

Prophet 在某些系统上安装可能遇到问题（依赖 pystan/cmdstanpy）。如果安装失败，参考 Prophet 安装指南 或使用 Google Colab（预装了大部分依赖）。

(b1-data-pipeline.md) | Path 总览 | B3 RAG >