promotional-planning

促销规划

你是 CPG 促销规划和贸易促销优化方面的专家。你的目标是帮助设计有效的促销策略，在确保供应链做好执行准备的同时，实现 ROI 最大化。

初步评估

在规划促销之前，先了解：

1. 业务背景

   • 要促销哪些产品／品类？
   • 涉及哪些零售商／渠道？（杂货店、仓储式会员店、便利店、线上）
   • 促销预算和目标是什么？
   • 历史促销效果如何（ROI、销量提升、增量）？

2. 促销策略

   • 使用哪些促销类型？（TPR、买一赠一、多件优惠、优惠券）
   • 典型的促销机制是什么？（折扣百分比、广告宣传、陈列）
   • 促销频率如何？（每个 SKU 每年的活动次数）
   • 竞争对手的促销活动如何？

3. 数据可用性

   • 是否有历史促销表现数据？
   • 是否有基线销售与促销销售数据？
   • 是否有第三方汇总数据（Nielsen、IRI）？
   • 是否能访问零售商的 POS 数据？
   • 是否有成本数据（贸易促销支出、上架费、陈列费）？

4. 供应链准备情况

   • 促销备货需要多长时间？
   • 产能是否受限？
   • 促销库存存放在哪里（提前采购）？
   • 配送中心能否应对需求激增？

---

促销规划框架

贸易促销类型

1. 临时降价（TPR）

• 从常规货架价打折
• 典型折扣：15-30%
• 最常见的促销类型
• 零售商将折扣让利给消费者

2. 广告宣传

• 产品在零售商的宣传单／广告中重点展示
• 通常与 TPR 搭配
• 曝光度高，销量提升明显
• 成本：向零售商支付广告补贴

3. 陈列

• 次要陈列位置（端架、落地架）
• 提高曝光度并促进冲动购买
• 通常与 TPR 和广告宣传搭配
• 成本：陈列补贴

4. 多件优惠

• “买 2 送 1”“3 件 5 美元”
• 鼓励消费者囤货
• 提高平均交易金额
• 可能蚕食未来销量

5. 优惠券

• 制造商或零售商提供的优惠券
• 针对特定消费者
• 可追踪核销情况
• 持续增长：数字优惠券和移动优惠券

6. 店内活动

• 试吃、产品演示
• 产品教育
• 参与度高，成本也高
• 适用于新品

7. 忠诚度计划

• 积分、奖励、个性化优惠
• 重要性日益提升（数字化）
• 因零售商而异
• 持续进行，而非仅限于特定活动

---

促销销量提升建模

基线销量与增量销量

基线销量：

• 未进行促销时预计产生的销量
• 对准确计算销量提升至关重要
• 方法：取非促销周的平均值、进行时间序列分解

增量销量：

• 促销带来的额外销量
• 增量销量 = 促销销量 - 基线销量
• 计算 ROI 的关键指标

销量提升系数：

Lift = (Promotional Sales - Baseline Sales) / Baseline Sales

促销响应模型

1. 价格弹性模型

import numpy as np
import pandas as pd
from sklearn.linear_model import LinearRegression
import statsmodels.api as sm

def estimate_price_elasticity(sales_data):
    """
    Estimate price elasticity from historical data

    Parameters:
    - sales_data: DataFrame with columns ['price', 'volume', 'feature',
                  'display', 'week', 'sku']

    Returns:
    - elasticity coefficient and model
    """

    # Log-log model for elasticity
    sales_data['log_price'] = np.log(sales_data['price'])
    sales_data['log_volume'] = np.log(sales_data['volume'])

    # Add promotional variables
    X = sales_data[['log_price', 'feature', 'display']]
    X = sm.add_constant(X)
    y = sales_data['log_volume']

    # OLS regression
    model = sm.OLS(y, X).fit()

    elasticity = {
        'price_elasticity': model.params['log_price'],
        'feature_lift': np.exp(model.params['feature']) - 1,
        'display_lift': np.exp(model.params['display']) - 1,
        'r_squared': model.rsquared,
        'model': model
    }

    return elasticity

# Example usage
sales_data = pd.DataFrame({
    'price': [3.99, 3.99, 2.99, 2.99, 3.99, 2.99],
    'volume': [1000, 1100, 1800, 2000, 950, 1850],
    'feature': [0, 0, 1, 1, 0, 1],
    'display': [0, 0, 0, 1, 0, 1]
})

elasticity = estimate_price_elasticity(sales_data)
print(f"Price Elasticity: {elasticity['price_elasticity']:.2f}")
print(f"Feature Lift: {elasticity['feature_lift']:.1%}")
print(f"Display Lift: {elasticity['display_lift']:.1%}")

2. 乘法促销模型

def promotional_uplift_model(baseline, price_discount, feature, display):
    """
    Calculate promotional uplift using multiplicative model

    Parameters:
    - baseline: baseline sales volume
    - price_discount: % discount (e.g., 0.20 for 20% off)
    - feature: 1 if featured, 0 otherwise
    - display: 1 if on display, 0 otherwise

    Returns:
    - predicted promotional sales
    """

    # Typical lift factors (calibrate from historical data)
    PRICE_ELASTICITY = -2.5  # 10% price cut → 25% volume increase
    FEATURE_LIFT = 1.5       # Feature alone increases sales 50%
    DISPLAY_LIFT = 2.0       # Display alone doubles sales

    # Price effect
    price_effect = (1 + price_discount) ** PRICE_ELASTICITY

    # Feature effect
    feature_effect = FEATURE_LIFT if feature else 1.0

    # Display effect
    display_effect = DISPLAY_LIFT if display else 1.0

    # Multiplicative model
    promotional_sales = baseline * price_effect * feature_effect * display_effect

    return promotional_sales

# Example: 20% discount with feature and display
baseline = 1000
promo_sales = promotional_uplift_model(
    baseline=baseline,
    price_discount=0.20,
    feature=1,
    display=1
)

lift = (promo_sales - baseline) / baseline
print(f"Baseline: {baseline}")
print(f"Promotional Sales: {promo_sales:.0f}")
print(f"Lift: {lift:.1%}")

3. 机器学习促销模型

from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_absolute_error, r2_score

def train_promotional_model(historical_data):
    """
    Train ML model to predict promotional sales

    Parameters:
    - historical_data: DataFrame with promotional events and outcomes

    Returns:
    - trained model and performance metrics
    """

    # Features
    feature_cols = [
        'regular_price',
        'promotional_price',
        'discount_pct',
        'feature',
        'display',
        'brand',
        'category',
        'retailer',
        'week_of_year',
        'holiday_week',
        'competitive_promos',
        'inventory_level',
        'days_since_last_promo'
    ]

    X = historical_data[feature_cols]
    y = historical_data['promotional_sales']

    # Train/test split
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42
    )

    # Gradient Boosting model
    model = GradientBoostingRegressor(
        n_estimators=200,
        learning_rate=0.1,
        max_depth=5,
        random_state=42
    )

    model.fit(X_train, y_train)

    # Evaluate
    y_pred = model.predict(X_test)
    mae = mean_absolute_error(y_test, y_pred)
    r2 = r2_score(y_test, y_pred)

    # Feature importance
    feature_importance = pd.DataFrame({
        'feature': feature_cols,
        'importance': model.feature_importances_
    }).sort_values('importance', ascending=False)

    return {
        'model': model,
        'mae': mae,
        'r2': r2,
        'feature_importance': feature_importance
    }

---

促销 ROI 分析

ROI 计算框架

def calculate_promotional_roi(promo_data):
    """
    Calculate comprehensive promotional ROI

    Parameters:
    - promo_data: dict with promotional details

    Returns:
    - ROI metrics and profitability analysis
    """

    # Sales metrics
    baseline_units = promo_data['baseline_units']
    promo_units = promo_data['promo_units']
    incremental_units = promo_units - baseline_units

    regular_price = promo_data['regular_price']
    promo_price = promo_data['promo_price']

    # Revenue
    baseline_revenue = baseline_units * regular_price
    promo_revenue = promo_units * promo_price
    incremental_revenue = promo_revenue - baseline_revenue

    # Costs
    cogs = promo_data['cogs_per_unit']
    baseline_cogs = baseline_units * cogs
    promo_cogs = promo_units * cogs
    incremental_cogs = promo_cogs - baseline_cogs

    # Trade spending
    trade_spend = {
        'price_discount': baseline_units * (regular_price - promo_price),
        'off_invoice': promo_data.get('off_invoice', 0),
        'feature_allowance': promo_data.get('feature_cost', 0),
        'display_allowance': promo_data.get('display_cost', 0),
        'slotting_fees': promo_data.get('slotting', 0),
        'coop_advertising': promo_data.get('coop_ad', 0)
    }

    total_trade_spend = sum(trade_spend.values())

    # Incremental profit
    baseline_profit = baseline_revenue - baseline_cogs
    promo_gross_profit = promo_revenue - promo_cogs
    incremental_gross_profit = promo_gross_profit - baseline_profit

    # Net incremental profit (after trade spend)
    net_incremental_profit = incremental_gross_profit - total_trade_spend

    # ROI
    roi = net_incremental_profit / total_trade_spend if total_trade_spend > 0 else 0

    # Payback
    payback = total_trade_spend / incremental_gross_profit if incremental_gross_profit > 0 else float('inf')

    return {
        'baseline_units': baseline_units,
        'promo_units': promo_units,
        'incremental_units': incremental_units,
        'lift_pct': (incremental_units / baseline_units * 100),
        'baseline_revenue': baseline_revenue,
        'promo_revenue': promo_revenue,
        'incremental_revenue': incremental_revenue,
        'baseline_profit': baseline_profit,
        'promo_profit': promo_gross_profit,
        'incremental_profit': incremental_gross_profit,
        'trade_spend': total_trade_spend,
        'trade_spend_breakdown': trade_spend,
        'net_profit': net_incremental_profit,
        'roi': roi,
        'roi_pct': roi * 100,
        'payback': payback
    }

# Example
promo_result = calculate_promotional_roi({
    'baseline_units': 1000,
    'promo_units': 2500,
    'regular_price': 4.99,
    'promo_price': 3.99,
    'cogs_per_unit': 2.50,
    'off_invoice': 500,
    'feature_cost': 1000,
    'display_cost': 800,
    'slotting': 0,
    'coop_ad': 500
})

print(f"Lift: {promo_result['lift_pct']:.0f}%")
print(f"Incremental Units: {promo_result['incremental_units']}")
print(f"Trade Spend: ${promo_result['trade_spend']:,.0f}")
print(f"Net Profit: ${promo_result['net_profit']:,.0f}")
print(f"ROI: {promo_result['roi_pct']:.0f}%")

促销后分析

def post_promotion_analysis(sales_data, promo_start_week, promo_duration):
    """
    Analyze post-promotion effects including cannibalization

    Parameters:
    - sales_data: weekly sales data
    - promo_start_week: week number promotion started
    - promo_duration: number of weeks promotion ran

    Returns:
    - analysis of pre, during, and post promotion periods
    """

    promo_end_week = promo_start_week + promo_duration

    # Define periods
    pre_promo = sales_data[
        (sales_data['week'] >= promo_start_week - 4) &
        (sales_data['week'] < promo_start_week)
    ]

    during_promo = sales_data[
        (sales_data['week'] >= promo_start_week) &
        (sales_data['week'] < promo_end_week)
    ]

    post_promo = sales_data[
        (sales_data['week'] >= promo_end_week) &
        (sales_data['week'] < promo_end_week + 4)
    ]

    # Calculate averages
    avg_pre = pre_promo['sales'].mean()
    avg_during = during_promo['sales'].mean()
    avg_post = post_promo['sales'].mean()

    # Pantry loading effect (post-promo dip)
    pantry_loading = avg_pre - avg_post if avg_post < avg_pre else 0

    # True incrementality (accounting for pantry loading)
    total_promo_sales = during_promo['sales'].sum()
    expected_baseline = avg_pre * promo_duration
    post_promo_impact = pantry_loading * len(post_promo)

    true_incremental = total_promo_sales - expected_baseline - post_promo_impact

    return {
        'avg_pre_promo': avg_pre,
        'avg_during_promo': avg_during,
        'avg_post_promo': avg_post,
        'lift_during': (avg_during - avg_pre) / avg_pre * 100,
        'post_promo_dip': (avg_pre - avg_post) / avg_pre * 100,
        'pantry_loading': pantry_loading,
        'true_incremental_sales': true_incremental,
        'true_incrementality_pct': true_incremental / expected_baseline * 100
    }

---

促销优化

最优促销日历

from pulp import *
import pandas as pd
import numpy as np

def optimize_promotional_calendar(products, budget, constraints):
    """
    Optimize promotional calendar to maximize ROI within budget

    Parameters:
    - products: DataFrame with product promotional opportunities
    - budget: total promotional budget
    - constraints: dict with business constraints

    Returns:
    - optimal promotional calendar
    """

    # Create problem
    prob = LpProblem("Promotional_Optimization", LpMaximize)

    # Decision variables: promote product i in week w
    promo_vars = {}
    for idx, row in products.iterrows():
        for week in range(1, 53):
            var_name = f"promo_{row['sku']}_{week}"
            promo_vars[row['sku'], week] = LpVariable(
                var_name,
                cat='Binary'
            )

    # Objective: Maximize total incremental profit
    prob += lpSum([
        promo_vars[row['sku'], week] * row['incremental_profit']
        for idx, row in products.iterrows()
        for week in range(1, 53)
    ])

    # Constraint 1: Total budget
    prob += lpSum([
        promo_vars[row['sku'], week] * row['trade_spend']
        for idx, row in products.iterrows()
        for week in range(1, 53)
    ]) <= budget

    # Constraint 2: Minimum weeks between promotions per SKU
    min_gap = constraints.get('min_weeks_between_promos', 8)
    for idx, row in products.iterrows():
        sku = row['sku']
        for week in range(1, 53):
            # If promoting in week w, can't promote in next min_gap weeks
            prob += lpSum([
                promo_vars[sku, w]
                for w in range(week, min(week + min_gap, 53))
            ]) <= 1

    # Constraint 3: Maximum promotions per week (capacity)
    max_promos_per_week = constraints.get('max_promos_per_week', 10)
    for week in range(1, 53):
        prob += lpSum([
            promo_vars[row['sku'], week]
            for idx, row in products.iterrows()
        ]) <= max_promos_per_week

    # Constraint 4: Minimum promotions per SKU per year
    min_promos_per_sku = constraints.get('min_promos_per_sku', 2)
    for idx, row in products.iterrows():
        sku = row['sku']
        prob += lpSum([
            promo_vars[sku, week]
            for week in range(1, 53)
        ]) >= min_promos_per_sku

    # Solve
    prob.solve(PULP_CBC_CMD(msg=0))

    # Extract solution
    calendar = []
    for idx, row in products.iterrows():
        for week in range(1, 53):
            if promo_vars[row['sku'], week].varValue > 0.5:
                calendar.append({
                    'sku': row['sku'],
                    'week': week,
                    'trade_spend': row['trade_spend'],
                    'incremental_profit': row['incremental_profit'],
                    'roi': row['incremental_profit'] / row['trade_spend']
                })

    calendar_df = pd.DataFrame(calendar)

    results = {
        'status': LpStatus[prob.status],
        'total_profit': value(prob.objective),
        'total_spend': calendar_df['trade_spend'].sum(),
        'avg_roi': calendar_df['roi'].mean(),
        'num_promotions': len(calendar_df),
        'calendar': calendar_df.sort_values(['week', 'sku'])
    }

    return results

促销价格优化

def optimize_promotional_price(baseline_volume, regular_price, cogs,
                                elasticity, fixed_costs):
    """
    Find optimal promotional price to maximize profit

    Parameters:
    - baseline_volume: baseline unit sales at regular price
    - regular_price: regular shelf price
    - cogs: cost of goods sold per unit
    - elasticity: price elasticity coefficient
    - fixed_costs: fixed promotional costs (feature, display, etc.)

    Returns:
    - optimal promotional price and expected profit
    """

    def profit_function(promo_price):
        """Calculate profit at given promotional price"""

        # Price change
        price_change = (promo_price - regular_price) / regular_price

        # Volume with price elasticity
        volume = baseline_volume * (1 + elasticity * price_change)

        # Revenue and costs
        revenue = volume * promo_price
        variable_costs = volume * cogs
        total_costs = variable_costs + fixed_costs

        # Profit
        profit = revenue - total_costs

        return -profit  # Negative for minimization

    from scipy.optimize import minimize_scalar

    # Optimize price (bounded between 50% and 90% of regular price)
    result = minimize_scalar(
        profit_function,
        bounds=(regular_price * 0.5, regular_price * 0.9),
        method='bounded'
    )

    optimal_price = result.x
    optimal_profit = -result.fun

    # Calculate metrics at optimal price
    price_change = (optimal_price - regular_price) / regular_price
    optimal_volume = baseline_volume * (1 + elasticity * price_change)
    discount_pct = (1 - optimal_price / regular_price) * 100

    return {
        'optimal_price': optimal_price,
        'discount_pct': discount_pct,
        'expected_volume': optimal_volume,
        'expected_profit': optimal_profit,
        'lift': (optimal_volume - baseline_volume) / baseline_volume * 100
    }

# Example
result = optimize_promotional_price(
    baseline_volume=1000,
    regular_price=4.99,
    cogs=2.50,
    elasticity=-2.5,
    fixed_costs=2000
)

print(f"Optimal Price: ${result['optimal_price']:.2f}")
print(f"Discount: {result['discount_pct']:.0f}%")
print(f"Expected Volume: {result['expected_volume']:.0f}")
print(f"Expected Profit: ${result['expected_profit']:.0f}")

---

促销供应链

促销需求预测

def forecast_promotional_demand(sku, retailer, promo_mechanics,
                                 historical_promos):
    """
    Forecast demand for upcoming promotion

    Parameters:
    - sku: product SKU
    - retailer: retailer identifier
    - promo_mechanics: dict with discount, feature, display
    - historical_promos: DataFrame with past promotional performance

    Returns:
    - promotional forecast
    """

    # Filter to similar promotions
    similar_promos = historical_promos[
        (historical_promos['sku'] == sku) &
        (historical_promos['retailer'] == retailer)
    ]

    if len(similar_promos) == 0:
        # No history, use category average
        similar_promos = historical_promos[
            historical_promos['category'] == get_category(sku)
        ]

    # Calculate average lift for similar mechanics
    discount_range = promo_mechanics['discount_pct']
    has_feature = promo_mechanics['feature']
    has_display = promo_mechanics['display']

    similar_mechanics = similar_promos[
        (similar_promos['discount_pct'].between(discount_range - 5, discount_range + 5)) &
        (similar_promos['feature'] == has_feature) &
        (similar_promos['display'] == has_display)
    ]

    if len(similar_mechanics) > 0:
        avg_lift = similar_mechanics['lift'].mean()
        std_lift = similar_mechanics['lift'].std()
    else:
        # Use regression model
        avg_lift = predict_lift_from_model(promo_mechanics)
        std_lift = 0.3  # Assume 30% std dev

    # Get baseline forecast
    baseline = get_baseline_forecast(sku, retailer)

    # Promotional forecast
    forecast = {
        'baseline': baseline,
        'expected_lift': avg_lift,
        'promo_forecast': baseline * (1 + avg_lift),
        'low_forecast': baseline * (1 + avg_lift - std_lift),  # P10
        'high_forecast': baseline * (1 + avg_lift + std_lift),  # P90
        'confidence': 'high' if len(similar_mechanics) >= 3 else 'low'
    }

    return forecast

促销库存规划

def plan_promotional_inventory(promo_forecast, lead_time_days,
                                safety_factor=1.2):
    """
    Calculate promotional inventory requirements

    Parameters:
    - promo_forecast: dict from forecast_promotional_demand
    - lead_time_days: production/procurement lead time
    - safety_factor: buffer multiplier (1.2 = 20% safety stock)

    Returns:
    - inventory plan with build schedule
    """

    # Expected promotional volume
    expected_volume = promo_forecast['promo_forecast']

    # Safety stock for uncertainty
    forecast_uncertainty = (
        promo_forecast['high_forecast'] - promo_forecast['low_forecast']
    ) / 2

    safety_stock = forecast_uncertainty * safety_factor

    # Total inventory needed
    total_needed = expected_volume + safety_stock

    # Build schedule (when to produce)
    promo_start_date = promo_forecast.get('start_date')
    build_start_date = promo_start_date - pd.Timedelta(days=lead_time_days + 7)

    # Pre-build strategy
    inventory_plan = {
        'promotional_forecast': expected_volume,
        'safety_stock': safety_stock,
        'total_inventory_needed': total_needed,
        'build_start_date': build_start_date,
        'promo_start_date': promo_start_date,
        'lead_time_days': lead_time_days,
        'recommended_locations': determine_forward_stock_locations(
            promo_forecast
        )
    }

    return inventory_plan

def determine_forward_stock_locations(promo_forecast):
    """
    Determine where to pre-position promotional inventory
    """

    retailer = promo_forecast.get('retailer')
    volume = promo_forecast['promo_forecast']

    # Strategy based on volume
    if volume > 10000:  # Large promotion
        return {
            'strategy': 'forward_stock',
            'locations': ['retailer_dc', 'manufacturer_dc'],
            'split': {'retailer_dc': 0.7, 'manufacturer_dc': 0.3}
        }
    elif volume > 5000:  # Medium promotion
        return {
            'strategy': 'partial_forward',
            'locations': ['retailer_dc'],
            'split': {'retailer_dc': 1.0}
        }
    else:  # Small promotion
        return {
            'strategy': 'pull_from_stock',
            'locations': ['manufacturer_dc'],
            'split': {'manufacturer_dc': 1.0}
        }

促销产能规划

def assess_promotional_capacity(promotional_calendar, production_capacity,
                                 dc_capacity):
    """
    Assess if supply chain can support promotional calendar

    Parameters:
    - promotional_calendar: DataFrame with planned promotions by week
    - production_capacity: available production capacity by week
    - dc_capacity: DC storage capacity

    Returns:
    - capacity analysis and bottlenecks
    """

    # Aggregate demand by week
    weekly_demand = promotional_calendar.groupby('week').agg({
        'forecasted_volume': 'sum',
        'num_promos': 'count'
    })

    # Add production lead time offset
    LEAD_TIME_WEEKS = 4
    weekly_demand['production_week'] = weekly_demand.index - LEAD_TIME_WEEKS

    # Check production capacity
    capacity_issues = []

    for week, row in weekly_demand.iterrows():
        prod_week = row['production_week']

        if prod_week < 1:
            continue

        required = row['forecasted_volume']
        available = production_capacity.get(prod_week, 0)

        if required > available:
            capacity_issues.append({
                'week': prod_week,
                'type': 'production',
                'required': required,
                'available': available,
                'gap': required - available,
                'utilization': required / available * 100
            })

    # Check DC capacity
    # Calculate cumulative inventory build-up
    inventory_profile = calculate_inventory_profile(
        promotional_calendar,
        production_capacity
    )

    for week, inventory in inventory_profile.items():
        if inventory > dc_capacity:
            capacity_issues.append({
                'week': week,
                'type': 'storage',
                'required': inventory,
                'available': dc_capacity,
                'gap': inventory - dc_capacity,
                'utilization': inventory / dc_capacity * 100
            })

    return {
        'capacity_ok': len(capacity_issues) == 0,
        'issues': capacity_issues,
        'max_production_utilization': max(
            i['utilization'] for i in capacity_issues
            if i['type'] == 'production'
        ) if capacity_issues else 0,
        'recommendations': generate_capacity_recommendations(capacity_issues)
    }

---

促销分析

促销表现仪表板

def generate_promotional_scorecard(promotional_data):
    """
    Generate comprehensive promotional performance scorecard

    Parameters:
    - promotional_data: DataFrame with promotional events and results

    Returns:
    - scorecard metrics
    """

    scorecard = {}

    # Overall metrics
    scorecard['total_promotions'] = len(promotional_data)
    scorecard['total_trade_spend'] = promotional_data['trade_spend'].sum()
    scorecard['total_incremental_profit'] = promotional_data['incremental_profit'].sum()

    # ROI metrics
    scorecard['avg_roi'] = (
        promotional_data['incremental_profit'].sum() /
        promotional_data['trade_spend'].sum()
    )
    scorecard['roi_pct'] = scorecard['avg_roi'] * 100

    positive_roi = promotional_data[promotional_data['incremental_profit'] > 0]
    scorecard['pct_positive_roi'] = len(positive_roi) / len(promotional_data) * 100

    # Lift metrics
    scorecard['avg_lift'] = promotional_data['lift_pct'].mean()
    scorecard['median_lift'] = promotional_data['lift_pct'].median()

    # Effectiveness by mechanic
    scorecard['mechanics'] = promotional_data.groupby('mechanic').agg({
        'roi': 'mean',
        'lift_pct': 'mean',
        'trade_spend': 'sum',
        'promo_id': 'count'
    }).rename(columns={'promo_id': 'count'})

    # Top performers
    scorecard['top_10_promos'] = promotional_data.nlargest(10, 'incremental_profit')[
        ['sku', 'retailer', 'mechanic', 'incremental_profit', 'roi']
    ]

    # Bottom performers
    scorecard['bottom_10_promos'] = promotional_data.nsmallest(10, 'incremental_profit')[
        ['sku', 'retailer', 'mechanic', 'incremental_profit', 'roi']
    ]

    # Efficiency metrics
    scorecard['spend_per_incremental_unit'] = (
        promotional_data['trade_spend'].sum() /
        promotional_data['incremental_units'].sum()
    )

    return scorecard

---

工具与库

商业软件

贸易促销管理（TPM）：

• SAP TPM：与 SAP ERP 集成
• Oracle TPM：贸易促销管理与优化
• AFS TPM：Analytical Frameworks Solutions
• Wipro Promax：基于云的 TPM
• Blacksmith Applications：TPM/TPO 软件
• IRI PromoPlanner：数据分析驱动的规划

贸易促销优化（TPO）：

• Anaplan：基于云的规划与促销优化
• o9 Solutions：AI 驱动的促销规划
• Blue Yonder：价格与促销优化
• Revionics：价格与促销优化（Oracle）
• TABS Analytics：促销预测与规划

分析平台：

• Nielsen：联合销售数据与促销分析
• IRI：市场测量与促销洞察
• Numerator：消费者样本组与促销跟踪
• 84.51° (Kroger)：零售数据与促销分析

Python 库

# Core libraries for promotional analysis

# Data manipulation and analysis
import pandas as pd
import numpy as np

# Statistical modeling
import statsmodels.api as sm
from statsmodels.tsa.seasonal import seasonal_decompose
from scipy import stats
from scipy.optimize import minimize

# Machine learning
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.linear_model import LinearRegression, Ridge, Lasso
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import cross_val_score

# Optimization
from pulp import *
import pyomo.environ as pyo

# Visualization
import matplotlib.pyplot as plt
import seaborn as sns
import plotly.express as px
import plotly.graph_objects as go

# Time series
from prophet import Prophet
import pmdarima as pm

促销分析模板

class PromotionalAnalyzer:
    """
    Comprehensive promotional analysis framework
    """

    def __init__(self, sales_data, promotional_events):
        self.sales_data = sales_data
        self.promo_events = promotional_events
        self.baseline_model = None
        self.lift_model = None

    def calculate_baseline(self, method='moving_average', window=4):
        """Calculate baseline sales (non-promotional)"""

        if method == 'moving_average':
            # Use non-promotional weeks only
            non_promo = self.sales_data[self.sales_data['promo'] == 0]
            baseline = non_promo.rolling(window=window)['sales'].mean()

        elif method == 'regression':
            # Time series regression
            X = np.arange(len(self.sales_data)).reshape(-1, 1)
            y = self.sales_data[self.sales_data['promo'] == 0]['sales']
            model = LinearRegression().fit(X, y)
            baseline = model.predict(X)

        return baseline

    def measure_lift(self):
        """Measure promotional lift for each event"""

        results = []

        for idx, promo in self.promo_events.iterrows():
            # Get sales during promotion
            promo_sales = self.sales_data[
                (self.sales_data['week'] >= promo['start_week']) &
                (self.sales_data['week'] <= promo['end_week'])
            ]['sales'].sum()

            # Get baseline
            baseline = self.sales_data[
                (self.sales_data['week'] >= promo['start_week']) &
                (self.sales_data['week'] <= promo['end_week'])
            ]['baseline'].sum()

            lift = (promo_sales - baseline) / baseline

            results.append({
                'promo_id': promo['promo_id'],
                'baseline': baseline,
                'promo_sales': promo_sales,
                'incremental': promo_sales - baseline,
                'lift_pct': lift * 100
            })

        return pd.DataFrame(results)

    def build_lift_model(self):
        """Build predictive model for promotional lift"""

        # Feature engineering
        features = self.promo_events[[
            'discount_pct', 'feature', 'display', 'price'
        ]]

        # Target
        lift_results = self.measure_lift()
        target = lift_results['lift_pct']

        # Train model
        model = RandomForestRegressor(n_estimators=100, random_state=42)
        model.fit(features, target)

        self.lift_model = model

        return model

    def forecast_promotion(self, promo_mechanics):
        """Forecast results for planned promotion"""

        if self.lift_model is None:
            self.build_lift_model()

        # Predict lift
        features = pd.DataFrame([promo_mechanics])
        predicted_lift = self.lift_model.predict(features)[0]

        # Apply to baseline
        baseline = self.calculate_baseline()
        forecast = baseline * (1 + predicted_lift / 100)

        return {
            'predicted_lift_pct': predicted_lift,
            'baseline_forecast': baseline,
            'promotional_forecast': forecast
        }

---

常见挑战与解决方案

挑战：促销预测不准确

问题：

• 促销表现差异很大
• 预测偏差达 50-100%
• 缺货或库存过剩

解决方案：

• 建立促销数据库（涵盖所有过往活动）
• 按促销机制细分（TPR、TPR+F、TPR+F+D）
• 为新品使用类似产品进行类比
• 使用近期数据校准模型
• 为不确定性高的活动增加安全库存
• 实施需求感知（根据早期结果更新预测）

def adaptive_promotional_forecast(initial_forecast, actual_sales_to_date,
                                   days_into_promo, total_promo_days):
    """
    Update promotional forecast based on early performance
    """

    # Typical promotional curve (front-loaded)
    expected_pct_complete = {
        1: 0.30,  # Day 1: 30% of sales
        2: 0.50,  # Day 2: 50% cumulative
        3: 0.65,
        4: 0.75,
        5: 0.83,
        6: 0.90,
        7: 1.00
    }

    expected_complete = expected_pct_complete.get(days_into_promo, days_into_promo / total_promo_days)
    expected_sales = initial_forecast * expected_complete

    # Actual vs expected
    performance_ratio = actual_sales_to_date / expected_sales

    # Update forecast
    updated_forecast = initial_forecast * performance_ratio

    return {
        'initial_forecast': initial_forecast,
        'expected_to_date': expected_sales,
        'actual_to_date': actual_sales_to_date,
        'performance_ratio': performance_ratio,
        'updated_forecast': updated_forecast
    }

挑战：促销投资回报率低或为负

问题：

• 许多促销活动亏损
• 贸易支出高，增量销售低
• 给本来就会购买的忠诚客户打折

解决方案：

• 预先测试促销活动（小型市场、测试门店）
• 优化促销折扣力度（价格测试）
• 定向开展促销（忠诚度计划、优惠券与 TPR 对比）
• 改善促销执行（确保陈列到位）
• 淘汰长期表现不佳的活动（砍掉表现最差的 20%）
• 专注于获取新客户，而非现有客户

挑战：促销冲突与蚕食效应

问题：

• 同一周促销多个产品
• 产品组合内部出现蚕食效应
• 零售商反对促销日历

解决方案：

• 协调整个产品组合的促销日历
• 按产品线错开促销活动
• 对蚕食效应进行建模
• 在产品组合层面而非单个 SKU 层面进行优化
• 提前与零售商协商促销日历

挑战：供应链执行

问题：

• 促销期间缺货
• 促销库存到货延迟
• DC 容量超限

解决方案：

• 延长促销备货的提前期（6-8 周）
• 在活动开始前 4 周锁定促销预测
• 提前部署库存（将库存前置到零售商 DC）
• 为促销预留产能
• 在供应网络中规划促销峰值产能
• 大型活动使用备用供应商/合同包装商

挑战：促销后销量下滑（囤货效应）

问题：

• 促销后销量骤降
• 消费者囤货
• 看起来像是促销失败

解决方案：

• 衡量真实增量（将促销后时段纳入考量）
• 限制促销时长（最多 1-2 周）
• 设置购买限额
• 避免过度促销
• 专注于试用/新客户，而非囤货

---

输出格式

促销计划模板

促销摘要：

• 产品：Brand X Cereal 18oz
• 零售商：大型杂货连锁店
• 促销期间：第 15–16 周（4 月 10–23 日）
• 促销方式：TPR + 特价宣传 + 陈列
• 常规价格：$4.99
• 促销价格：$3.49（优惠 30%）

预测业绩：

指标	基准值	促销值	增量	提升 %
件数	1,000	3,200	2,200	220%
收入	$4,990	$11,168	$6,178	124%
毛利	$2,490	$3,168	$678	27%

贸易费用：

项目	成本
发票外折扣	$1,500
特价宣传补贴	$1,200
陈列补贴	$800
联合广告	$500
贸易费用总计	$4,000

ROI 分析：

• 增量毛利：$678
• 贸易费用：$4,000
• 净利润：-$3,322
• ROI：-83%

建议：不要推进

• 预计促销会造成亏损
• 可考虑：降低贸易费用、改进促销方式或取消促销

供应链计划：

• 生产备货：第 11 周（促销前 4 周）
• 数量：3,500 件（预测量 + 10% 安全库存）
• 发货至零售商 DC：第 13 周
• 门店布置：第 14 周

---

要提出的问题

如果需要更多背景信息：

1. 正在促销哪些产品/品类？
2. 涉及哪些零售商和渠道？
3. 采用哪些促销方式？（折扣 %、特价宣传、陈列）
4. 促销预算和目标是什么？
5. 是否有历史促销数据？
6. 典型的促销 ROI 和销量提升幅度是多少？
7. 生产提前期和产能限制是什么？
8. 促销日历会提前多久敲定？

---

相关技能

• demand-forecasting：用于基准需求和促销需求预测
• markdown-optimization：用于定价和清仓优化
• retail-replenishment：用于协调促销库存
• dsd-route-optimization：用于 DSD 促销执行
• capacity-planning：用于促销生产产能规划
• inventory-optimization：用于促销安全库存规划
• sales-operations-planning：用于集成 S&OP
• supply-chain-analytics：用于促销业绩跟踪
• seasonal-planning：用于节假日和季节性促销