Inventory Demand Sensing — 库存需求感知：实时信号融合驱动智能补货
Skill-Inventory-Demand-Sensing · 18-物流履约
causalexperimentforecastingrecommendationdata_collectionpricing广告与投放供应链与补货推荐与搜索数据采集与治理定价与利润WF-A 智能补货WF-B 广告优化WF-D 选品扫描WF-F 动态定价
年化 ROI¥25-65 万
实现难度⭐⭐⭐☆☆
业务视角
适用角色物流负责人 / 供应链负责人 · 客服负责人 · 运营负责人
适用平台FBA vs FBM vs 第三方海外仓 · 美国本土最后一公里 · 跨境退货逆向物流
什么情况下用物流时效不稳定，差评里大量「收货太慢」，影响 DSR 评分；退货率高，处理成本吃掉大量利润；旺季物流爆仓
成功是什么样的物流时效提升 20-30%，物流相关差评减少 40%，退货成本可控，旺季履约稳定不崩溃
业务痛点
1. 解决的问题
吸奶器旺季等到销量上升才补货Lead Time 45天导致旺季缺货两周——搜索趋势+广告CTR多信号感知比纯销量触发提前7-14天补货，旺季缺货减少40-60%年化GMV保护25-65万元
2. 核心算法逻辑
需求感知 = 融合多源实时信号：
3. 业务应用场景
业务问题：每年 2 月（产后恢复高峰）吸奶器销量会上升，但等到销量上升才补货，Lead Time 45 天，旺季前 2 周开始缺货。若依赖搜索趋势信号，1 月初就能看到搜索量上升，提前 3-4 周触发补货。
数据要求： - Google Trends / Helium10 关键词搜索量（周粒度） - Amazon 广告 CTR 历史（来自广告报告） - 推荐系统曝光量（Amazon Attribution 报告） - 过去 2 年销量历史
预期产出： - 需求感知综合评分（每日更新） - 旺季开始时间预测（比历史规律更精准） - 触发补货建议：提前 X 天下单，备货 Y 件
4. 输入数据要求
请查看原始代码模板获取输入规格。
5. 输出结果
请查看原始代码模板获取输出规格。
6. 业务价值 / ROI
- ROI 预估：
- 比纯销量提前 7-14 天触发补货：旺季缺货减少 40-60%，保护 ¥15-40 万 GMV
- 广告预算与库存状态联动：不在无货时投广告（节省无效广告费 ¥5-15 万/年）
- 需求下降时提前降库存：减少呆滞库存持有成本 ¥5-10 万/年
- 年化综合 ROI：¥25-65 万
- 实施难度：⭐⭐⭐☆☆（需要多数据源接入；搜索趋势 API + Amazon 报告 + Kalman 融合约 3-4 周）
7. 代码模板
代码块数量：3 · 路径：未检测到
"""
Inventory Demand Sensing
多信号融合需求感知 + 智能补货触发
"""
import numpy as np
from collections import deque
class KalmanSignalFilter:
"""Kalman 滤波器：融合多个噪声不同的信号"""
def __init__(self, process_noise: float = 1.0, measurement_noise: float = 5.0):
self.Q = process_noise # 过程噪声（真实需求的随机性）
self.R = measurement_noise # 测量噪声（信号观测误差）
self.x = 0.0 # 状态估计（真实需求）
self.P = 10.0 # 估计不确定性
def update(self, measurement: float) -> float:
"""接入新观测值，更新状态估计"""
# 预测步骤
self.P += self.Q
# 更新步骤
K = self.P / (self.P + self.R) # Kalman 增益
self.x += K * (measurement - self.x)
self.P *= (1 - K)
return self.x
class DemandSensingModel:
"""多信号融合需求感知模型"""
def __init__(self, lead_days: int = 14, window: int = 30):
self.lead_days = lead_days
self.window = window
# 各信号的权重（通过历史数据学习，这里用启发式）
self.weights = {
'search_trend': 0.35, # 搜索趋势（最强领先信号）
'ads_ctr': 0.25, # 广告 CTR（用户购买意愿）
'reco_exposure':0.20, # 推荐曝光（平台热度信号）
'historical': 0.20, # 历史销量（基础稳定性）
}
self.kalman = KalmanSignalFilter()
self.history = deque(maxlen=window)
self.baseline = None
def normalize_signal(self, values: list, name: str) -> np.ndarray:
"""信号标准化（z-score）"""
arr = np.array(values, dtype=float)
mean, std = np.mean(arr), np.std(arr)
if std < 1e-8:
return np.zeros_like(arr)
return (arr - mean) / std
def compute_sensing_score(self, signals: dict) -> np.ndarray:
"""计算每日需求感知综合评分"""
n = min(len(v) for v in signals.values())
result = np.zeros(n)
for name, weight in self.weights.items():
if name in signals and name != 'historical':8. 论文来源
- 2403.14271