Promotion Logistics Surge Forecast — 大促物流爆仓预测：营销-履约联动容量规划

1. 解决的问题

基于营销投放数据提前 3-7 天预测物流需求峰值，驱动仓储/运力的前置性扩容决策。

2. 核心算法逻辑

大促期间（黑五/Prime Day/双十一），营销投放与物流履约之间存在数据孤岛：营销团队在广告平台上看到 ROAS 飙升，但仓库团队直到订单涌入才发现已超容，导致延误和差评。Promotion Logistics Surge Forecast 解决的核心问题是：基于营销投放数据提前 37 天预测物流需求峰值，驱动仓储/运力的前置性扩容决策。

3. 业务应用场景

业务背景：某母婴品牌黑五前 2 周开始提升广告预算，Sponsored Products 预算从 $5,000/日提升至 $28,000/日。历史经验表明黑五 GMV 约为平日 4.5 倍，但今年新增 TikTok Shop 渠道，实际 GMV 达到平日 7.2 倍，导致 DHL Express 运力严重不足，延误率 34%。

量化收益： - 避免延误率从 34% 降至 8%，减少差评 ≈ 1,240 条 - 按每条差评影响 BSR 降权 ≈ $180 GMV 损失，节省 $223,200 - 提前预订运力节省紧急加价 ≈ $38,000（旺季紧急运力溢价通常 30-50%） - 单次大促 ROI ≈ $261,200

业务背景：轻便折叠婴儿车有 12 个 SKU（颜色×尺寸），Prime Day 期间某些颜色（如薰衣草紫）历史转化率是标准黑色的 2.3 倍，但备货比例未反映这一差异，导致爆款 SKU 断货。

4. 输入数据要求

• 广告平台数据：Amazon AMS / TikTok Ads 每日预算、CTR、CVR（过去 90 天）
• 历史促销数据：历届大促 GMV 倍数、折扣深度、持续天数
• 仓储数据：当前库存、日均运力、ERP 出库记录
• 大促前 T-14天：第一次全量预测，识别高风险 SKU
• 大促前 T-7天：第二次预测（结合实时投放数据更新），触发补仓决策

5. 输出结果

• 广告平台数据：Amazon AMS / TikTok Ads 每日预算、CTR、CVR（过去 90 天）
• 历史促销数据：历届大促 GMV 倍数、折扣深度、持续天数
• 仓储数据：当前库存、日均运力、ERP 出库记录
• 大促前 T-14天：第一次全量预测，识别高风险 SKU
• 大促前 T-7天：第二次预测（结合实时投放数据更新），触发补仓决策

6. 业务价值 / ROI

20-40 万

7. 代码模板

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

"""
Promotion Logistics Surge Forecast
整合促销需求涌浪预测 + CVaR 容量规划 + 动态扩容决策
CampaignSurge (arXiv:2411.09283) + FulfillCap (arXiv:2502.16071) + DemandDecomp-Promo (arXiv:2409.18512)
"""

import numpy as np
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
from enum import Enum
import warnings
warnings.filterwarnings("ignore")


class RiskLevel(Enum):
    LOW = "low"          # CVaR < 80% 容量
    MEDIUM = "medium"    # 80% <= CVaR < 100%
    HIGH = "high"        # 100% <= CVaR < 150%
    CRITICAL = "critical"  # CVaR >= 150%


@dataclass
class CampaignConfig:
    """大促活动配置"""
    name: str                           # 活动名（黑五/Prime Day/双十一）
    start_date: str                     # 活动开始日
    duration_days: int                  # 活动持续天数
    budget_usd_daily: float             # 日均广告预算（美元）
    discount_rate: float                # 折扣深度（0~1，0.3表示7折）
    channel_weights: Dict[str, float] = field(default_factory=lambda: {"amazon": 1.0})


@dataclass
class WarehouseCapacity:
    """仓储容量配置"""
    warehouse_id: str
    current_stock: int                  # 当前库存件数
    daily_inbound_capacity: int         # 日均入库容量
    daily_outbound_capacity: int        # 日均发货容量（运力上限）
    safety_stock_ratio: float = 0.15    # 安全库存比例


@dataclass
class SKUDemandProfile:
    """SKU 需求特征"""
    sku_id: str
    baseline_daily: float               # 平日日均需求
    promo_elasticity: float             # 促销弹性（β1）
    discount_elasticity: float          # 折扣弹性（β2）
    decay_rate: float = 0.15            # 促销衰减率 γ
    historical_std: float = 0.2         # 历史需求标准差/均值


class DemandDecomposer:
    """
    促销需求分解器
    arXiv:2409.18512 DemandDecomp-Promo
    分离 baseline / promotional / spillover 三组分
    """

8. 论文来源

• 2409.18512
• 2411.09283
• 2502.16071