大促中实时决策KPI与流量协同阈值 — 售罄速率监控/流量协同触发/紧急干预决策
Skill-InPromo-Realtime-Decision-KPI · 04-供应链
causalexperimentforecastingoptimization广告与投放供应链与补货WF-A 智能补货WF-B 广告优化WF-I 智能体工程
实现难度⭐⭐⭐☆☆
业务优先级⭐⭐⭐⭐⭐
业务视角
适用角色供应链负责人 · 采购负责人 · CEO / 运营 VP
适用平台Amazon FBA · 海外仓 · 多国仓位（美/欧/日）
什么情况下用库存周转率低，资金压在海外仓出不来；SKU 断货紧急空运，物流成本吃掉毛利；多仓库存分布不均
成功是什么样的库存周转天数从 90 天降到 60 天，断货率 <3%，海外仓综合成本降低 15-25%
业务痛点
1. 解决的问题
大促中供应链和广告是两个独立系统没有联动导致爆款提前售罄时广告仍在浪费——售罄速率监控+流量协同阈值（减流量/加流量/迁移预算），书中供应链-流量协同大促GMV提升约12%
2. 核心算法逻辑
书籍核心洞察（陈凤霞）：书中第六章大促中管理揭示了一个关键联动：供应链信号（库存消耗速率）应该自动触发流量策略调整（广告投入/流量分配），而不是两个独立的系统各自运转。这种"供应链流量"的实时协同，是大促期间效率最高的管理模式。
3. 业务应用场景
场景A：Prime Day吸奶器实时流量协同
- 业务问题：Prime Day第1小时吸奶器销售速率极高（按此速率4小时后售罄），但广告仍在持续投入；竞品此时也在拼广告；团队没有自动化机制识别并响应 - 流量协同机制： 1. 每小时计算滚动售罄速率 2. 发现预计在大促结束前6小时售罄→触发"减广告"信号 3. 降低吸奶器SP广告出价40%（减少流量消耗库存速度） 4. 将$2000广告预算转移到配件套装（吸奶器配件包） 5. 吸奶器延后售罄时间4小时（把握更长窗口），配件销售激增$8000 - 预期产出：大促总GMV提升约12%（来自更优化的库存时间分配）
- **业务问题**：某SKU大促前2小时实际销售只有预测的55%，即将大量积压 - **加流量协同**：立即提高广告出价+申请平台限时秒杀+站内搜索置顶，让剩余时间销售速率恢复，最终售罄率从55%提升至78%
4. 输入数据要求
请查看原始代码模板获取输入规格。
5. 输出结果
请查看原始代码模板获取输出规格。
6. 业务价值 / ROI
- ROI 预估：精准流量协同使大促GMV额外提升8-15%；避免爆款提前售罄带来的"有流量无库存"浪费（每次大促节省$3000-8000广告浪费）；系统$2万，ROI>400%
- 实施难度：⭐⭐⭐☆☆（需要实时库存API+广告平台API联动；最难点是将库存信号自动触发广告出价调整）
- 优先级：⭐⭐⭐⭐⭐（书中第六章核心，供应链-流量协同是大促效率最高的管理模式）
- 适用规模：月销>$5万且参与主要大促的卖家
- 数据依赖：实时库存数据、广告数据（每小时维度）、历史大促分时销售数据
7. 代码模板
代码块数量：2 · 路径：未检测到
"""
大促中实时决策KPI与流量协同阈值
基于《全链路管理》陈凤霞 第六章第二节
售罄速率监控 + 流量协同触发 + 紧急干预决策
"""
import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')
@dataclass
class PromoRealTimeStatus:
"""大促实时状态"""
sku_id: str
initial_stock: int
hourly_sales: List[float] # 每小时销售记录
hourly_forecast: List[float] # 每小时预测
promo_total_hours: int = 48 # 大促总时长（小时）
ad_budget_usd: float = 500.0 # 广告预算
current_ad_bid: float = 1.0 # 当前出价
class InPromoRealTimeKPI:
"""大促中实时KPI监控器"""
def __init__(self, rolling_window_hours: int = 3):
self.rolling_window = rolling_window_hours
def compute_sellthrough_status(self, status: PromoRealTimeStatus) -> Dict:
"""计算实时售罄状态"""
current_hour = len(status.hourly_sales)
cumulative_sales = sum(status.hourly_sales)
remaining_stock = max(status.initial_stock - cumulative_sales, 0)
sellthrough_rate = cumulative_sales / max(status.initial_stock, 1)
# 滚动窗口销售速率
window_sales = status.hourly_sales[-self.rolling_window:]
rolling_rate = sum(window_sales) / len(window_sales) if window_sales else 0
# 预计售罄时刻
if rolling_rate > 0:
hours_to_sellout = remaining_stock / rolling_rate
predicted_sellout_hour = current_hour + hours_to_sellout
else:
hours_to_sellout = float('inf')
predicted_sellout_hour = float('inf')
remaining_promo_hours = status.promo_total_hours - current_hour
will_sellout_before_end = predicted_sellout_hour < status.promo_total_hours
return {
'current_hour': current_hour,
'cumulative_sales': int(cumulative_sales),
'remaining_stock': int(remaining_stock),
'sellthrough_rate': sellthrough_rate,
'sellthrough_pct': f"{sellthrough_rate:.0%}",8. 论文来源
未自动抽取；请查看原始 Skill 卡片。