GEANN + Bass 新品冷启动需求预测 - 母婴跨境新品备货

1. 解决的问题

母婴跨境新品冷启动需求预测痛点:每年 20-30 款新品上市,前 8 周零销售记录,人工拍脑袋备货首批,积压或断货损失年化 300+ 万元. 本 Skill 组合两个方法:① Bass 扩散模型生成新品扩散曲线形状(创新+模仿系数);② GEANN 图迁移从相似品历史借用销售信号;③ Bass 参数从相似品加权迁移初始化,实现"形状从理论 + 规模从迁移"的双驱动.

2. 核心算法逻辑

母婴跨境新品冷启动需求预测痛点:每年 2030 款新品上市,前 8 周零销售记录,人工拍脑袋备货首批,积压或断货损失年化 300+ 万元. 本 Skill 组合两个方法:① Bass 扩散模型生成新品扩散曲线形状(创新+模仿系数);② GEANN 图迁移从相似品历史借用销售信号;③ Bass 参数从相似品加权迁移初始化,实现"形状从理论 + 规模从迁移"的双驱动.

3. 业务应用场景

- 业务问题:Momcozy 上市新品"羊奶粉 A 段 800g 德国品牌 X",历史零销售,采购无依据备货. 首批多压货占用现金流 50-100 万,少备货错失上升期 BSR 损失更大. Anker 案例:新品冷启动占总库存损失 35% - 数据要求:新品特征向量(价格/品类/产地/渠道) + 同品类相似品 52 周历史销售 - 配置流程: 1. 相似品检索:用 GEANN kNN 找 Top-3 相似品(德国奶粉 800g / 澳洲奶粉 900g / 荷兰奶粉 800g) 2. Bass 参数拟合:对每个相似品拟合 (m, p, q) 3. Softmax 加权迁移:相似度高的相似品权重大

- 业务问题:Momcozy 推出 Prime Day 季节性新品(夏季婴儿游泳圈),大促备货截止日提前 60 天(亚马逊 Prime Day 报名),但新品当时只有 3-7 天早期销售数据;过量备货占用 FBA 长期仓储费,不足备货错失大促爆发 - 数据要求:Prime Day 新品早期 3-7 天销售 + 非大促期相似品(湿巾/纸尿裤)峰值数据 - F-FOMAML 配置: - 非大促期相似品销售曲线作"代理数据" - GNN 按品类 + 价格带 + 历史峰值倍率构建商品相似图 - F-FOMAML 用 3-7 天早期数据快速 fine-tune - 输出 Prime Day 24-48

4. 输入数据要求

请查看原始代码模板获取输入规格。

5. 输出结果

请查看原始代码模板获取输出规格。

6. 业务价值 / ROI

• 易处:Bass 模型有解析解,scipy.curve_fit 可直接拟合
• 易处:PyMC-Marketing Bayesian Bass 开源完整
• 难处:Amazon GEANN/F-FOMAML 未开源,GNN 部分需自行实现
• 难处:相似品特征工程需业务专家(价格/品类/产地编码)
• 难处:市场潜力 m 估算依赖竞品数据(可借助 Helium 10 / Jungle Scout)

7. 代码模板

代码块数量：1 · 路径：未检测到

"""
新品冷启动需求预测 - Bass 扩散 + 相似品迁移最小骨架
论文 arXiv:2307.03595 + arXiv:2406.16221
依赖: pip install numpy scipy scikit-learn
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import List, Optional, Tuple

import numpy as np
from scipy.optimize import curve_fit


def bass_cumulative(t: np.ndarray, m: float, p: float, q: float) -> np.ndarray:
    """Bass 累积采用 N(t) = m*(1-e^-(p+q)t)/(1 + q/p * e^-(p+q)t)"""
    return m * (1 - np.exp(-(p + q) * t)) / (1 + (q / p) * np.exp(-(p + q) * t))


def bass_incremental(t: np.ndarray, m: float, p: float, q: float) -> np.ndarray:
    """Bass 增量 dN/dt(每周新增需求)"""
    N = bass_cumulative(t, m, p, q)
    return (p + q * N / m) * (m - N)


def fit_bass_from_history(sales_history: np.ndarray) -> Tuple[float, float, float]:
    """从相似品历史销售拟合 Bass 参数(m, p, q)"""
    t = np.arange(1, len(sales_history) + 1, dtype=float)
    cumulative = np.cumsum(sales_history)
    try:
        p0 = [cumulative[-1] * 2, 0.02, 0.4]
        bounds = ([0, 0.001, 0.01], [1e8, 0.5, 2.0])
        popt, _ = curve_fit(bass_cumulative, t, cumulative, p0=p0, bounds=bounds, maxfev=5000)
        return float(popt[0]), float(popt[1]), float(popt[2])
    except (RuntimeError, ValueError):
        return float(cumulative[-1] * 3), 0.02, 0.38


@dataclass
class Product:
    sku_id: str
    features: np.ndarray
    sales_history: Optional[np.ndarray] = None


def find_similar_products(new_product: Product, catalog: List[Product], top_k: int = 5) -> List[Tuple[Product, float]]:
    """基于特征向量余弦相似度找 Top-K 相似品"""
    new_feat = new_product.features
    similarities = []
    for prod in catalog:
        if prod.sales_history is not None:
            num = float(np.dot(new_feat, prod.features))
            denom = float(np.linalg.norm(new_feat) * np.linalg.norm(prod.features))
            sim = num / denom if denom > 0 else 0.0
            similarities.append((prod, sim))
    similarities.sort(key=lambda x: -x[1])
    return similarities[:top_k]


def transfer_bass_params(
    similar_products: List[Tuple[Product, float]],

8. 论文来源

• 2307.03595
• 2406.16221