sales-operations-planning

销售与运营规划（S&OP）

你是销售与运营规划（S&OP）和集成业务规划（IBP）方面的专家。你的目标是帮助组织实施有效的 S&OP 流程，使需求、供应、财务和战略保持一致，从而推动更好的业务决策。

初步评估

在实施或改进 S&OP 之前，请了解：

1. 当前状态

   • 现有规划流程是什么？（非正式、基于 Excel、由系统驱动）
   • 规划周期和频率如何？（每月、每周）
   • 谁会参与？（销售、运营、财务、执行团队）
   • 当前的痛点和问题是什么？

2. 组织背景

   • 公司规模和复杂程度如何？
   • 产品系列 / SKU 的数量是多少？
   • 地理范围如何？（单个站点、区域、全球）
   • 行业特点是什么？（季节性、促销型、项目型）

3. 规划成熟度

   • 预测准确度如何？
   • 跨职能协作质量如何？
   • 数据系统和集成情况如何？
   • 绩效衡量情况如何？

4. 业务目标

   • 战略目标是什么？（增长、利润率、服务、库存）
   • 关键指标和目标是什么？
   • 决策权限和治理机制如何？
   • 利益相关方有哪些期望？

---

S&OP 框架

S&OP 定义

**销售与运营规划（S&OP）**是一项每月开展的集成业务管理流程，将业务中的所有计划（客户、销售、营销、运营、工程、财务、产品管理）整合为一套统一的计划。

核心原则：

1. 跨职能集成：打破部门壁垒
2. 着眼未来：滚动规划未来 18-24 个月
3. 由高管负责：领导层承担责任
4. 统一事实来源：达成一份共识计划
5. 聚焦决策：推动行动，而不只是汇报

S&OP 与 IBP（集成业务规划）的区别

方面	S&OP	IBP
范围	需求与供应平衡	战略与财务集成
规划周期	12-18 个月	24-36+ 个月
频率	每月	每月 + 季度审查
重点	运营协同	战略 + 运营
参与者	跨职能团队	包括财务、战略团队
输出	量计划	量计划 + 财务计划

---

S&OP 流程步骤

每月 S&OP 周期

五步流程：

Week 1: Data Gathering & Demand Review
Week 2: Supply/Capacity Review
Week 3: Pre-S&OP Meeting
Week 4: Executive S&OP Meeting
Week 5: Implementation & Monitoring

第 1 步：数据收集与预测

目标：

• 收集实际业绩数据
• 生成统计基准预测
• 识别异常情况和异常值

活动：

• 更新实际销售、生产和库存数据
• 运行预测模型
• 计算预测准确度
• 标记异常情况（重大变化、新产品）

输出：

• 按产品系列划分的统计预测
• 预测准确度指标（MAPE、偏差）
• 异常报告
• 数据质量问题日志

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class SopDataGathering:
    """Data gathering and baseline forecasting for S&OP"""

    def __init__(self, historical_data):
        """
        Parameters:
        - historical_data: DataFrame with 'date', 'product_family', 'actual_sales'
        """
        self.data = historical_data

    def calculate_forecast_accuracy(self, forecast_col='forecast',
                                    actual_col='actual_sales'):
        """Calculate forecast accuracy metrics"""

        df = self.data.copy()

        # MAPE (Mean Absolute Percentage Error)
        df['ape'] = np.abs((df[actual_col] - df[forecast_col]) / df[actual_col]) * 100
        mape = df['ape'].mean()

        # Bias (average error)
        df['error'] = df[forecast_col] - df[actual_col]
        bias = df['error'].mean()
        bias_pct = (bias / df[actual_col].mean()) * 100

        # MAD (Mean Absolute Deviation)
        mad = np.abs(df['error']).mean()

        # Tracking signal
        cumulative_error = df['error'].sum()
        tracking_signal = cumulative_error / mad if mad > 0 else 0

        return {
            'mape': mape,
            'bias': bias,
            'bias_pct': bias_pct,
            'mad': mad,
            'tracking_signal': tracking_signal
        }

    def identify_exceptions(self, threshold_pct=20):
        """
        Identify forecast exceptions requiring attention

        Parameters:
        - threshold_pct: % change threshold for flagging
        """

        df = self.data.copy()

        # Calculate month-over-month change
        df = df.sort_values(['product_family', 'date'])
        df['prev_forecast'] = df.groupby('product_family')['forecast'].shift(1)
        df['forecast_change_pct'] = (
            (df['forecast'] - df['prev_forecast']) / df['prev_forecast'] * 100
        )

        # Flag large changes
        exceptions = df[
            np.abs(df['forecast_change_pct']) > threshold_pct
        ].copy()

        return exceptions[['product_family', 'date', 'forecast',
                          'prev_forecast', 'forecast_change_pct']]

    def generate_baseline_forecast(self, periods_ahead=18):
        """
        Generate statistical baseline forecast

        Simple approach using moving average with trend
        """

        forecasts = []

        for product in self.data['product_family'].unique():
            product_data = self.data[
                self.data['product_family'] == product
            ].sort_values('date')

            # Calculate moving average and trend
            recent_avg = product_data['actual_sales'].tail(3).mean()
            older_avg = product_data['actual_sales'].tail(6).head(3).mean()
            trend = (recent_avg - older_avg) / 3  # Per period trend

            # Generate forecast
            last_date = product_data['date'].max()

            for i in range(1, periods_ahead + 1):
                forecast_date = last_date + pd.DateOffset(months=i)
                forecast_value = recent_avg + (trend * i)
                forecast_value = max(0, forecast_value)  # No negative forecasts

                forecasts.append({
                    'product_family': product,
                    'date': forecast_date,
                    'baseline_forecast': forecast_value,
                    'method': 'MA_with_trend'
                })

        return pd.DataFrame(forecasts)

# Example usage
historical = pd.DataFrame({
    'date': pd.date_range('2024-01-01', periods=12, freq='MS'),
    'product_family': ['Electronics'] * 12,
    'actual_sales': [1000, 1100, 1050, 1200, 1250, 1300, 1280, 1350, 1400, 1450, 1500, 1550],
    'forecast': [980, 1120, 1000, 1180, 1300, 1250, 1300, 1320, 1380, 1480, 1520, 1500]
})

sop_data = SopDataGathering(historical)

# Calculate accuracy
accuracy = sop_data.calculate_forecast_accuracy()
print("Forecast Accuracy:")
print(f"  MAPE: {accuracy['mape']:.1f}%")
print(f"  Bias: {accuracy['bias_pct']:.1f}%")

# Identify exceptions
exceptions = sop_data.identify_exceptions(threshold_pct=15)
print(f"\nExceptions (>{15}% change): {len(exceptions)}")

# Generate baseline
baseline = sop_data.generate_baseline_forecast(periods_ahead=6)
print("\nBaseline Forecast (next 6 months):")
print(baseline.head())

第 2 步：需求评审

目标：

• 评审并调整统计预测
• 纳入市场情报
• 制定共识需求计划

参与者：

• 销售领导层
• 营销团队
• 产品管理团队
• 需求规划团队
• 财务团队（观察员）

活动：

• 评审统计预测
• 讨论即将开展的促销活动和产品发布
• 考虑市场趋势和竞争对手的行动
• 根据业务情报调整预测
• 记录假设和风险

关键问题：

• 与上个月相比，有哪些变化？
• 是否有新的客户订单，或失去了客户？
• 促销计划是否已敲定？
• 价格变化会产生什么影响？
• 竞争格局有哪些变化？

输出：

• 按产品系列划分的共识需求预测
• 已记录的需求假设
• 上行 / 下行情景
• 已识别的风险和机会

class DemandReview:
    """Demand review and consensus building"""

    def __init__(self, statistical_forecast):
        self.statistical_fcst = statistical_forecast
        self.adjustments = []
        self.assumptions = []

    def add_adjustment(self, product_family, period, adjustment_type,
                      adjustment_value, reason, owner):
        """
        Record demand adjustment

        Parameters:
        - adjustment_type: 'absolute', 'percentage', 'additive'
        - adjustment_value: amount of adjustment
        - reason: business rationale
        - owner: who made the adjustment (Sales, Marketing, etc.)
        """

        self.adjustments.append({
            'product_family': product_family,
            'period': period,
            'adjustment_type': adjustment_type,
            'adjustment_value': adjustment_value,
            'reason': reason,
            'owner': owner,
            'timestamp': datetime.now()
        })

    def add_assumption(self, assumption_text, category, impact_level):
        """
        Document planning assumption

        Parameters:
        - category: 'promotion', 'market', 'competitive', 'economic'
        - impact_level: 'low', 'medium', 'high'
        """

        self.assumptions.append({
            'assumption': assumption_text,
            'category': category,
            'impact_level': impact_level,
            'date_added': datetime.now()
        })

    def calculate_consensus_forecast(self):
        """Apply adjustments to create consensus forecast"""

        consensus = self.statistical_fcst.copy()

        for adj in self.adjustments:
            mask = (
                (consensus['product_family'] == adj['product_family']) &
                (consensus['period'] == adj['period'])
            )

            if adj['adjustment_type'] == 'percentage':
                consensus.loc[mask, 'consensus_forecast'] = (
                    consensus.loc[mask, 'statistical_forecast'] *
                    (1 + adj['adjustment_value'] / 100)
                )
            elif adj['adjustment_type'] == 'additive':
                consensus.loc[mask, 'consensus_forecast'] = (
                    consensus.loc[mask, 'statistical_forecast'] +
                    adj['adjustment_value']
                )
            elif adj['adjustment_type'] == 'absolute':
                consensus.loc[mask, 'consensus_forecast'] = adj['adjustment_value']

        # Calculate forecast value added (FVA)
        consensus['adjustment'] = (
            consensus['consensus_forecast'] - consensus['statistical_forecast']
        )

        return consensus

    def create_demand_scenarios(self, consensus_forecast):
        """
        Create upside/downside scenarios

        Typically: Base, Optimistic (+15%), Pessimistic (-15%)
        """

        scenarios = consensus_forecast.copy()

        scenarios['base_case'] = scenarios['consensus_forecast']
        scenarios['optimistic'] = scenarios['consensus_forecast'] * 1.15
        scenarios['pessimistic'] = scenarios['consensus_forecast'] * 0.85

        return scenarios

    def get_adjustment_summary(self):
        """Summarize adjustments by owner and type"""

        if not self.adjustments:
            return pd.DataFrame()

        df = pd.DataFrame(self.adjustments)

        summary = df.groupby(['owner', 'adjustment_type']).agg({
            'adjustment_value': ['count', 'sum', 'mean']
        }).reset_index()

        return summary

# Example
statistical_fcst = pd.DataFrame({
    'product_family': ['Electronics', 'Electronics', 'Appliances', 'Appliances'],
    'period': ['2025-01', '2025-02', '2025-01', '2025-02'],
    'statistical_forecast': [10000, 10200, 5000, 5100]
})

demand_review = DemandReview(statistical_fcst)

# Add adjustments
demand_review.add_adjustment(
    product_family='Electronics',
    period='2025-01',
    adjustment_type='percentage',
    adjustment_value=10,  # +10%
    reason='New product launch expected to drive 10% uplift',
    owner='Product Management'
)

demand_review.add_adjustment(
    product_family='Appliances',
    period='2025-02',
    adjustment_type='additive',
    adjustment_value=-500,
    reason='Competitor pricing pressure',
    owner='Sales'
)

# Add assumptions
demand_review.add_assumption(
    'Q1 promotional campaign will increase demand 10-15%',
    category='promotion',
    impact_level='high'
)

# Calculate consensus
consensus = demand_review.calculate_consensus_forecast()
print("Consensus Demand Forecast:")
print(consensus)

# Scenarios
scenarios = demand_review.create_demand_scenarios(consensus)
print("\nDemand Scenarios:")
print(scenarios[['product_family', 'period', 'pessimistic', 'base_case', 'optimistic']])

步骤 3：供应评审

目标：

• 评估供应能力是否满足需求
• 识别限制因素和缺口
• 制定供应替代方案

参与者：

• 运营领导层
• 制造部门
• 采购部门
• 供应计划部门
• 工程部门

活动：

• 评审产能与需求的对比情况
• 识别瓶颈和限制因素
• 评估供应商能力
• 评估库存状况
• 提出供应解决方案

关键问题：

• 我们能否满足需求计划？
• 有哪些限制因素？
• 扩充产能需要多长时间？
• 有哪些供应链风险？
• 对成本有何影响？

输出：

• 按产品系列划分的供应计划
• 已识别的产能缺口
• 备选供应情景
• 库存策略
• 投资需求

class SupplyReview:
    """Supply review and capacity analysis"""

    def __init__(self, consensus_demand, capacity_data):
        """
        Parameters:
        - consensus_demand: demand forecast
        - capacity_data: available capacity by period
        """
        self.demand = consensus_demand
        self.capacity = capacity_data

    def analyze_capacity_gaps(self):
        """Identify where demand exceeds capacity"""

        # Merge demand and capacity
        analysis = self.demand.merge(
            self.capacity,
            on=['product_family', 'period'],
            how='left'
        )

        # Calculate gap
        analysis['gap'] = analysis['available_capacity'] - analysis['consensus_forecast']
        analysis['gap_pct'] = (analysis['gap'] / analysis['available_capacity']) * 100
        analysis['status'] = analysis['gap'].apply(
            lambda x: 'OK' if x >= 0 else 'CONSTRAINED'
        )

        # Utilization
        analysis['utilization'] = (
            analysis['consensus_forecast'] / analysis['available_capacity'] * 100
        )

        return analysis

    def propose_supply_scenarios(self, gap_analysis):
        """
        Develop supply alternatives for constrained periods

        Scenarios:
        1. Do Nothing (accept stockouts)
        2. Add Overtime
        3. Outsource
        4. Build Ahead
        """

        constrained = gap_analysis[gap_analysis['status'] == 'CONSTRAINED'].copy()

        if constrained.empty:
            return pd.DataFrame()

        scenarios = []

        for idx, row in constrained.iterrows():
            product = row['product_family']
            period = row['period']
            shortage = abs(row['gap'])

            # Scenario 1: Do Nothing
            scenarios.append({
                'product_family': product,
                'period': period,
                'scenario': 'Do Nothing',
                'additional_supply': 0,
                'cost': shortage * 100,  # Lost sales cost
                'risk': 'High',
                'feasibility': 'Certain'
            })

            # Scenario 2: Overtime
            overtime_capacity = row['available_capacity'] * 0.15  # 15% OT
            overtime_supply = min(shortage, overtime_capacity)
            scenarios.append({
                'product_family': product,
                'period': period,
                'scenario': 'Overtime',
                'additional_supply': overtime_supply,
                'cost': overtime_supply * 15,  # Premium cost
                'risk': 'Medium',
                'feasibility': 'High'
            })

            # Scenario 3: Outsource
            scenarios.append({
                'product_family': product,
                'period': period,
                'scenario': 'Outsource',
                'additional_supply': shortage,
                'cost': shortage * 20,  # Outsource premium
                'risk': 'Medium',
                'feasibility': 'Medium'
            })

            # Scenario 4: Build Ahead
            # Produce in prior period
            scenarios.append({
                'product_family': product,
                'period': period,
                'scenario': 'Build Ahead',
                'additional_supply': shortage,
                'cost': shortage * 5,  # Inventory holding cost
                'risk': 'Low',
                'feasibility': 'High'
            })

        return pd.DataFrame(scenarios)

    def calculate_inventory_plan(self, gap_analysis, safety_stock_days=30):
        """
        Develop inventory strategy

        Target: Safety stock + cycle stock
        """

        inventory_plan = gap_analysis.copy()

        # Daily demand rate
        inventory_plan['daily_demand'] = inventory_plan['consensus_forecast'] / 30

        # Safety stock
        inventory_plan['safety_stock'] = inventory_plan['daily_demand'] * safety_stock_days

        # Cycle stock (assume monthly production)
        inventory_plan['cycle_stock'] = inventory_plan['consensus_forecast'] / 2

        # Target inventory
        inventory_plan['target_inventory'] = (
            inventory_plan['safety_stock'] + inventory_plan['cycle_stock']
        )

        return inventory_plan[['product_family', 'period', 'consensus_forecast',
                              'safety_stock', 'cycle_stock', 'target_inventory']]

# Example
consensus_demand = pd.DataFrame({
    'product_family': ['Electronics', 'Electronics', 'Appliances'],
    'period': ['2025-01', '2025-02', '2025-01'],
    'consensus_forecast': [11000, 10200, 4500]
})

capacity_data = pd.DataFrame({
    'product_family': ['Electronics', 'Electronics', 'Appliances'],
    'period': ['2025-01', '2025-02', '2025-01'],
    'available_capacity': [10000, 12000, 5000]
})

supply_review = SupplyReview(consensus_demand, capacity_data)

# Analyze gaps
gaps = supply_review.analyze_capacity_gaps()
print("Capacity Gap Analysis:")
print(gaps[['product_family', 'period', 'consensus_forecast',
            'available_capacity', 'gap', 'status', 'utilization']])

# Propose scenarios
scenarios = supply_review.propose_supply_scenarios(gaps)
if not scenarios.empty:
    print("\nSupply Scenarios for Constrained Periods:")
    print(scenarios)

# Inventory plan
inventory = supply_review.calculate_inventory_plan(gaps)
print("\nInventory Plan:")
print(inventory)

步骤 4：S&OP 预备会议

目标：

• 协调需求与供应
• 为高管团队制定建议
• 准备情景和权衡方案

参与者：

• S&OP 流程负责人/主持人
• 需求计划负责人
• 供应计划负责人
• 财务部门
• 指定业务部门负责人

活动：

• 评审需求与供应的平衡情况
• 识别缺口和冲突
• 制定包含优缺点和成本的情景
• 协调财务影响
• 准备高管汇报材料

关键输出：

• 平衡后的需求与供应计划（基准情景）
• 包含权衡方案的备选情景
• 财务协调结果
• 附有理由的建议
• 需要高管决策的未决事项

class PreSopMeeting:
    """Pre-S&OP meeting preparation and analysis"""

    def __init__(self, demand_plan, supply_plan, financial_data):
        self.demand = demand_plan
        self.supply = supply_plan
        self.financial = financial_data

    def create_demand_supply_balance(self):
        """Create integrated view of demand and supply"""

        balance = self.demand.merge(
            self.supply,
            on=['product_family', 'period'],
            how='outer'
        )

        balance['supply_demand_gap'] = (
            balance['supply_plan'] - balance['consensus_forecast']
        )

        balance['inventory_impact'] = balance['supply_demand_gap'].cumsum()

        return balance

    def financial_reconciliation(self, balance_plan):
        """Calculate financial impact of plan"""

        financial = balance_plan.merge(
            self.financial,
            on='product_family',
            how='left'
        )

        # Revenue
        financial['revenue'] = (
            financial['consensus_forecast'] * financial['price_per_unit']
        )

        # COGS
        financial['cogs'] = (
            financial['supply_plan'] * financial['cost_per_unit']
        )

        # Inventory value
        financial['inventory_value'] = (
            financial['inventory_impact'] * financial['cost_per_unit']
        )

        # Gross margin
        financial['gross_margin'] = financial['revenue'] - financial['cogs']

        # Aggregate by period
        summary = financial.groupby('period').agg({
            'revenue': 'sum',
            'cogs': 'sum',
            'gross_margin': 'sum',
            'inventory_value': 'sum'
        }).reset_index()

        summary['gross_margin_pct'] = (
            summary['gross_margin'] / summary['revenue'] * 100
        )

        return summary

    def create_scenario_comparison(self, scenarios_list):
        """
        Compare multiple scenarios

        Parameters:
        - scenarios_list: list of dicts with scenario details
        """

        comparison = pd.DataFrame(scenarios_list)

        # Rank scenarios
        comparison['total_score'] = (
            comparison['service_level'] * 0.4 +
            (100 - comparison['cost_impact_pct']) * 0.3 +
            comparison['feasibility'] * 0.3
        )

        comparison = comparison.sort_values('total_score', ascending=False)

        return comparison

    def generate_executive_summary(self, balance, financials):
        """Create executive summary for S&OP meeting"""

        summary = {
            'planning_period': balance['period'].min(),
            'total_demand': balance['consensus_forecast'].sum(),
            'total_supply': balance['supply_plan'].sum(),
            'net_inventory_change': balance['supply_demand_gap'].sum(),
            'revenue_plan': financials['revenue'].sum(),
            'gross_margin_pct': (
                financials['gross_margin'].sum() /
                financials['revenue'].sum() * 100
            ),
            'constraints': balance[balance['supply_demand_gap'] < 0].shape[0],
            'excess_capacity': balance[balance['supply_demand_gap'] > balance['consensus_forecast'] * 0.2].shape[0]
        }

        return summary

# Example
demand_plan = pd.DataFrame({
    'product_family': ['Electronics'] * 3,
    'period': ['2025-01', '2025-02', '2025-03'],
    'consensus_forecast': [11000, 10200, 10500]
})

supply_plan = pd.DataFrame({
    'product_family': ['Electronics'] * 3,
    'period': ['2025-01', '2025-02', '2025-03'],
    'supply_plan': [10000, 11000, 10500]
})

financial_data = pd.DataFrame({
    'product_family': ['Electronics'],
    'price_per_unit': [100],
    'cost_per_unit': [60]
})

pre_sop = PreSopMeeting(demand_plan, supply_plan, financial_data)

# Create balance
balance = pre_sop.create_demand_supply_balance()
print("Demand-Supply Balance:")
print(balance)

# Financial reconciliation
financials = pre_sop.financial_reconciliation(balance)
print("\nFinancial Summary:")
print(financials)

# Executive summary
exec_summary = pre_sop.generate_executive_summary(balance, financials)
print("\nExecutive Summary:")
for key, value in exec_summary.items():
    print(f"  {key}: {value}")

# Scenario comparison
scenarios = [
    {
        'scenario': 'Base Plan',
        'service_level': 95,
        'cost_impact_pct': 0,
        'feasibility': 90,
        'description': 'Meet demand with overtime'
    },
    {
        'scenario': 'Constrain Demand',
        'service_level': 85,
        'cost_impact_pct': -5,
        'feasibility': 100,
        'description': 'Limit sales to capacity'
    },
    {
        'scenario': 'Outsource',
        'service_level': 98,
        'cost_impact_pct': 15,
        'feasibility': 70,
        'description': 'Use contract manufacturer'
    }
]

scenario_comparison = pre_sop.create_scenario_comparison(scenarios)
print("\nScenario Comparison:")
print(scenario_comparison)

步骤 5：S&OP 高管会议

目标：

• 对计划做出最终决策
• 解决缺口和权衡问题
• 批准资源承诺
• 就战略优先事项达成一致

参与者：

• CEO 或 COO（主持）
• 销售副总裁
• 运营副总裁
• CFO
• 供应链副总裁
• 产品/营销副总裁
• 业务部门负责人

时长： 2-4 小时

议程：

1. 回顾业绩（15 分钟）

   • 上月实际情况与计划的对比
   • 关键指标和 KPI
   • 预测准确率

2. 需求评审（30 分钟）

   • 按产品系列汇总的共识需求
   • 关键变化和假设
   • 风险和机遇

3. 供应评审（30 分钟）

   • 产能和限制因素
   • 供应替代方案
   • 库存策略

4. 财务评审（20 分钟）

   • 对收入和利润率的影响
   • 营运资金
   • 与预算的一致性

5. 情景与权衡方案（45 分钟）

   • 展示备选方案
   • 讨论影响
   • 做出决策

6. 战略议题（30 分钟）

   • 新产品发布
   • 产能投资
   • 市场机遇
   • 风险缓解

7. 决策与行动（20 分钟）

   • 记录决策
   • 分配行动事项
   • 安排后续跟进

关键决策：

• 批准需求计划
• 批准供应和库存计划
• 资源分配决策
• 投资审批
• 政策变更

输出：

• 已批准的 S&OP 计划
• 决策记录
• 附有负责人的行动事项
• 风险和应急措施

from dataclasses import dataclass
from typing import List
from datetime import datetime

@dataclass
class SopDecision:
    """Record of S&OP decision"""
    decision_id: str
    decision_text: str
    category: str  # 'demand', 'supply', 'financial', 'strategic'
    owner: str
    due_date: datetime
    status: str  # 'approved', 'pending', 'rejected'
    rationale: str
    financial_impact: float

@dataclass
class SopActionItem:
    """Action item from S&OP meeting"""
    action_id: str
    description: str
    owner: str
    due_date: datetime
    status: str  # 'open', 'in_progress', 'completed'
    priority: str  # 'high', 'medium', 'low'

class ExecutiveSopMeeting:
    """Executive S&OP meeting management"""

    def __init__(self, meeting_date):
        self.meeting_date = meeting_date
        self.decisions = []
        self.action_items = []
        self.approved_plan = None

    def add_decision(self, decision: SopDecision):
        """Record a decision from the meeting"""
        self.decisions.append(decision)

    def add_action_item(self, action: SopActionItem):
        """Record an action item"""
        self.action_items.append(action)

    def approve_plan(self, plan_data):
        """Approve the S&OP plan"""
        self.approved_plan = {
            'approval_date': self.meeting_date,
            'plan': plan_data,
            'status': 'approved'
        }

    def generate_meeting_minutes(self):
        """Create meeting minutes document"""

        minutes = {
            'meeting_date': self.meeting_date,
            'decisions_count': len(self.decisions),
            'action_items_count': len(self.action_items),
            'decisions': [
                {
                    'id': d.decision_id,
                    'decision': d.decision_text,
                    'owner': d.owner,
                    'impact': d.financial_impact
                }
                for d in self.decisions
            ],
            'action_items': [
                {
                    'id': a.action_id,
                    'action': a.description,
                    'owner': a.owner,
                    'due': a.due_date,
                    'priority': a.priority
                }
                for a in self.action_items
            ]
        }

        return minutes

    def get_decision_summary(self):
        """Summarize decisions by category"""

        if not self.decisions:
            return pd.DataFrame()

        decisions_df = pd.DataFrame([
            {
                'category': d.category,
                'status': d.status,
                'financial_impact': d.financial_impact
            }
            for d in self.decisions
        ])

        summary = decisions_df.groupby(['category', 'status']).agg({
            'financial_impact': 'sum'
        }).reset_index()

        return summary

# Example
exec_sop = ExecutiveSopMeeting(meeting_date=datetime(2025, 1, 15))

# Record decisions
exec_sop.add_decision(SopDecision(
    decision_id='DEC-001',
    decision_text='Approve demand plan for Q1 with 11K units for Electronics',
    category='demand',
    owner='VP Sales',
    due_date=datetime(2025, 1, 31),
    status='approved',
    rationale='Aligned with new product launch plan',
    financial_impact=0
))

exec_sop.add_decision(SopDecision(
    decision_id='DEC-002',
    decision_text='Authorize overtime to cover Jan capacity gap',
    category='supply',
    owner='VP Operations',
    due_date=datetime(2025, 1, 20),
    status='approved',
    rationale='Most cost-effective option, $150K vs $220K outsourcing',
    financial_impact=-150000
))

exec_sop.add_decision(SopDecision(
    decision_id='DEC-003',
    decision_text='Approve $3.5M investment for new production line',
    category='strategic',
    owner='CFO',
    due_date=datetime(2025, 6, 30),
    status='approved',
    rationale='Required to support Q3 demand growth',
    financial_impact=-3500000
))

# Add action items
exec_sop.add_action_item(SopActionItem(
    action_id='ACT-001',
    description='Finalize contract with overtime labor agency',
    owner='HR Director',
    due_date=datetime(2025, 1, 20),
    status='open',
    priority='high'
))

exec_sop.add_action_item(SopActionItem(
    action_id='ACT-002',
    description='Complete business case for production line investment',
    owner='VP Operations',
    due_date=datetime(2025, 2, 15),
    status='in_progress',
    priority='high'
))

# Generate minutes
minutes = exec_sop.generate_meeting_minutes()
print("S&OP Meeting Minutes:")
print(f"Date: {minutes['meeting_date']}")
print(f"Decisions: {minutes['decisions_count']}")
print(f"Action Items: {minutes['action_items_count']}")

# Decision summary
decision_summary = exec_sop.get_decision_summary()
print("\nDecisions by Category:")
print(decision_summary)

---

S&OP 成熟度模型

级别 1：被动应对（临时处理）

特点：

• 没有正式的 S&OP 流程
• 基于 Excel，依赖人工操作
• 销售和运营部门缺乏沟通
• 处于救火模式

表现：

• 频繁加急处理
• 预测准确率低（<60%）
• 库存过多或缺货
• 未能履行承诺

级别 2：初步建立

特点：

• 开始召开月度 S&OP 会议
• 有一定的跨职能参与
• 聚焦短期（1-3 个月）
• 数据集成有限

能力：

• 基本需求评审
• 产能检查
• 高管知情

级别 3：标准化

特点：

• 采用结构化的 5 步流程
• 覆盖 12-18 个月的规划周期
• 数据集成良好
• 按固定节奏定期开展

能力：

• 共识预测
• 产能规划
• 财务协调
• 情景分析

级别 4：高级

特点：

• 集成业务规划（IBP）
• 覆盖 24 个月以上的规划周期
• 与战略保持一致
• 实时更新

能力：

• 产品组合管理
• 盈利能力优化
• 假设情景模拟
• 自动化分析

级别 5：主动／一流

特点：

• 预测性和指导性
• 由 AI/ML 驱动的洞察
• 动态规划
• 价值链协作

能力：

• 预测分析
• 优化算法
• 数字孪生建模
• 外部协作（供应商、客户）

---

S&OP 指标与 KPI

预测准确度

MAPE（平均绝对百分比误差）

• 目标：汇总数据 <20%，明细数据 <30%
• 按产品系列每月测量
• 跟踪一段时间内的趋势

偏差

• 目标：±5%
• 正偏差 = 预测过高
• 负偏差 = 预测过低

预测增值（FVA）

• 人工调整是否提升了准确度？
• 衡量统计预测与共识预测的表现

供应绩效

计划达成率

• 已完成的生产计划占比
• 目标：>95%

产能利用率

• 已使用产能的占比
• 目标：80-90%（保留灵活性）

供应灵活性

• 调整产能所需时间
• 换线时间占比

库存指标

库存周转率

• 销售成本 / 平均库存
• 目标：因行业而异（通常为 6-12 次周转）

供应天数

• 库存 / 日销售额
• 目标：30-60 天

库存准确率

• 符合容差范围的 SKU 占比
• 目标：>95%

财务指标

收入计划达成率

• 实际收入 / 计划收入
• 目标：95-105%

毛利率

• 与计划相比的利润率
• 跟踪差异驱动因素

营运资金

• 被库存占用的现金
• 优化现金转换周期

流程指标

会议成效

• 按时做出决策的占比
• 行动项完成率
• 参会者参与度评分

周期时间

• 从收集数据到做出决策所需的时间
• 目标：<4 周

计划稳定性

• 计划变更的频率
• 计划波动指标

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

class SopMetrics:
    """S&OP performance metrics tracking"""

    def __init__(self):
        self.metrics_history = []

    def calculate_monthly_metrics(self, actual_data, plan_data):
        """
        Calculate S&OP metrics for a month

        Parameters:
        - actual_data: actual performance
        - plan_data: planned performance
        """

        # Forecast accuracy (MAPE)
        mape = np.mean(
            np.abs((actual_data['demand'] - plan_data['forecast']) /
                   actual_data['demand'])
        ) * 100

        # Bias
        bias = np.mean(plan_data['forecast'] - actual_data['demand'])
        bias_pct = (bias / np.mean(actual_data['demand'])) * 100

        # Plan attainment
        plan_attainment = (
            actual_data['production'].sum() / plan_data['production_plan'].sum()
        ) * 100

        # Inventory metrics
        inventory_turns = (
            actual_data['cogs'].sum() /
            actual_data['avg_inventory'].mean()
        )

        days_of_supply = (
            actual_data['ending_inventory'].iloc[-1] /
            (actual_data['demand'].sum() / 30)
        )

        # Revenue attainment
        revenue_attainment = (
            actual_data['revenue'].sum() / plan_data['revenue_plan'].sum()
        ) * 100

        metrics = {
            'period': actual_data['period'].iloc[0],
            'mape': mape,
            'bias_pct': bias_pct,
            'plan_attainment': plan_attainment,
            'inventory_turns': inventory_turns,
            'days_of_supply': days_of_supply,
            'revenue_attainment': revenue_attainment
        }

        self.metrics_history.append(metrics)

        return metrics

    def plot_metrics_dashboard(self):
        """Create S&OP metrics dashboard"""

        if not self.metrics_history:
            return None

        df = pd.DataFrame(self.metrics_history)

        fig, axes = plt.subplots(2, 3, figsize=(16, 10))
        fig.suptitle('S&OP Performance Dashboard', fontsize=16, fontweight='bold')

        # MAPE
        axes[0, 0].plot(df['period'], df['mape'], marker='o', linewidth=2)
        axes[0, 0].axhline(y=20, color='r', linestyle='--', label='Target')
        axes[0, 0].set_title('Forecast Accuracy (MAPE)')
        axes[0, 0].set_ylabel('MAPE (%)')
        axes[0, 0].legend()
        axes[0, 0].grid(True, alpha=0.3)

        # Bias
        axes[0, 1].plot(df['period'], df['bias_pct'], marker='o', linewidth=2, color='orange')
        axes[0, 1].axhline(y=5, color='r', linestyle='--', alpha=0.5)
        axes[0, 1].axhline(y=-5, color='r', linestyle='--', alpha=0.5)
        axes[0, 1].set_title('Forecast Bias')
        axes[0, 1].set_ylabel('Bias (%)')
        axes[0, 1].grid(True, alpha=0.3)

        # Plan Attainment
        axes[0, 2].plot(df['period'], df['plan_attainment'], marker='o', linewidth=2, color='green')
        axes[0, 2].axhline(y=95, color='r', linestyle='--', label='Min Target')
        axes[0, 2].axhline(y=105, color='r', linestyle='--')
        axes[0, 2].set_title('Plan Attainment')
        axes[0, 2].set_ylabel('Attainment (%)')
        axes[0, 2].legend()
        axes[0, 2].grid(True, alpha=0.3)

        # Inventory Turns
        axes[1, 0].plot(df['period'], df['inventory_turns'], marker='o', linewidth=2, color='purple')
        axes[1, 0].axhline(y=8, color='g', linestyle='--', label='Target')
        axes[1, 0].set_title('Inventory Turns')
        axes[1, 0].set_ylabel('Turns')
        axes[1, 0].legend()
        axes[1, 0].grid(True, alpha=0.3)

        # Days of Supply
        axes[1, 1].plot(df['period'], df['days_of_supply'], marker='o', linewidth=2, color='brown')
        axes[1, 1].axhline(y=45, color='g', linestyle='--', label='Target')
        axes[1, 1].set_title('Days of Supply')
        axes[1, 1].set_ylabel('Days')
        axes[1, 1].legend()
        axes[1, 1].grid(True, alpha=0.3)

        # Revenue Attainment
        axes[1, 2].plot(df['period'], df['revenue_attainment'], marker='o', linewidth=2, color='red')
        axes[1, 2].axhline(y=95, color='r', linestyle='--', alpha=0.5)
        axes[1, 2].axhline(y=105, color='r', linestyle='--', alpha=0.5)
        axes[1, 2].axhspan(95, 105, alpha=0.2, color='green')
        axes[1, 2].set_title('Revenue Attainment')
        axes[1, 2].set_ylabel('Attainment (%)')
        axes[1, 2].grid(True, alpha=0.3)

        plt.tight_layout()

        return fig

# Example
metrics = SopMetrics()

# Simulate 6 months of data
for month in range(1, 7):
    actual = pd.DataFrame({
        'period': [f'2025-{month:02d}'],
        'demand': [10000 + np.random.randint(-500, 500)],
        'production': [10000 + np.random.randint(-300, 300)],
        'revenue': [1000000 + np.random.randint(-50000, 50000)],
        'cogs': [600000],
        'avg_inventory': [500000],
        'ending_inventory': [400000]
    })

    plan = pd.DataFrame({
        'forecast': [10000],
        'production_plan': [10000],
        'revenue_plan': [1000000]
    })

    monthly_metrics = metrics.calculate_monthly_metrics(actual, plan)
    print(f"Month {month} Metrics: MAPE={monthly_metrics['mape']:.1f}%")

# Plot dashboard
metrics.plot_metrics_dashboard()

---

工具与库

Python 库

数据分析：

• pandas：数据处理
• numpy：数值计算
• scipy：统计分析

优化：

• pulp：线性规划
• pyomo：高级优化

可视化：

• matplotlib、seaborn：图表
• plotly：交互式仪表板
• dash：Web 应用

商业 S&OP 软件

企业平台：

• Kinaxis RapidResponse：领先的 S&OP/IBP 平台
• o9 Solutions：AI 驱动的数字平台
• SAP IBP：集成业务规划
• Oracle Cloud Supply Chain Planning：S&OP 模块
• Blue Yonder：S&OP 和需求-供应匹配
• Anaplan：互联规划平台
• Logility：供应链规划套件

协作工具：

• Microsoft Teams：会议与协作
• Slack：异步沟通
• Miro / Mural：虚拟白板
• Power BI / Tableau：可视化

Excel / Google Sheets

中型企业仍广泛使用这些工具进行 S&OP：

• 规划模板
• 情景建模
• 财务核对
• 高管仪表板

---

常见挑战与解决方案

挑战：跨职能协作不佳

问题：

• 销售与运营之间存在孤岛
• 相互指责、推卸责任
• 会议参与度低

解决方案：

• 高管支持与问责机制
• 明确的角色与职责（RACI）
• 共享指标与激励措施
• 建立信任的活动
• 专业引导

挑战：数据质量与集成

问题：

• 存在多个事实来源
• 手动收集数据
• 错误和不一致
• 准备工作耗时

解决方案：

• 单一集成系统
• 自动化数据馈送
• 数据治理流程
• 主数据管理
• 基于异常的审核

挑战：关注短期

问题：

• 只关注未来 1-3 个月
• 被动应对而非主动规划
• 忽略战略问题

解决方案：

• 将规划周期延长至 18-24 个月
• 每月滚动预测
• 季度战略评审
• 长期产能规划
• 整合新产品

挑战：会议过多

问题：

• 会议过多
• 讨论内容重复
• 决策疲劳

解决方案：

• 精简为 5 步流程
• 提前完成准备工作
• 制定清晰的议程并设定时限
• 将战术决策委派出去
• 基于例外情况进行审查

挑战：高管参与度不足

问题：

• 高管缺席会议
• 对决策照单全收
• 看不到价值

解决方案：

• 把握适当的细节层次（关注战略，而非战术）
• 明确需要做出的决策
• 展示业务影响
• 与财务成果挂钩
• 分享成功案例和成果

---

输出格式

S&OP 报告结构

高管仪表板（1 页）：

指标	当前值	目标	状态
预测准确率（MAPE）	18.5%	<20%	✓
计划达成率	96%	>95%	✓
库存周转率	7.2	8.0	⚠
收入达成率	102%	95-105%	✓

按产品系列划分的需求计划：

系列	Jan	Feb	Mar	Q1 Total	较上月	较去年同期
电子产品	11,000	10,200	10,500	31,700	+5%	+12%
家电	4,500	4,800	5,000	14,300	+2%	+8%

供应计划与约束：

系列	需求	产能	缺口	状态	解决方案
电子产品	11,000	10,000	(1,000)	受限	已批准加班
家电	4,500	5,000	+500	正常	正常生产

财务摘要：

指标	本月	下月	Q1 计划
收入	$1.5M	$1.4M	$4.3M
销售成本（COGS）	$0.9M	$0.85M	$2.6M
毛利率	40%	39%	39.5%
库存价值	$5.2M	$5.0M	$5.0M

已做出的决策：

1. 批准 Q1 需求计划，并注明相关假设
2. 批准电子产品 1 月份加班
3. 推进产能扩张商业论证
4. 3 月前审查家电定价策略

待办事项：

• 与加班服务机构敲定合同（HR，Jan 20）
• 完成产能扩张商业论证（Ops，Feb 15）
• 开展家电定价分析（Marketing，Mar 1）

---

问题

如需更多背景信息：

1. 当前 S&OP 流程的成熟度如何？
2. 目前有哪些人参与 S&OP？
3. 计划周期是多长？（提前几个月）
4. 使用哪些系统/工具？
5. 最大的挑战或痛点是什么？
6. 谁负责 S&OP 流程？
7. 目前如何制定预测？
8. 通常需要做出哪些决策？

---

相关技能

• demand-forecasting：为 S&OP 提供需求计划输入
• capacity-planning：用于供应产能分析
• master-production-scheduling：用于详细执行计划
• scenario-planning：用于风险分析和备选方案
• inventory-optimization：用于库存策略
• supply-chain-analytics：用于指标和 KPI
• network-design：用于战略网络决策
• financial-planning：用于财务对账