物流成本前中后生命周期管理KPI — 生意前模拟/生意中账单/生意后分析的三段成本闭环

"""
物流成本前中后生命周期管理KPI
基于《全链路管理》陈凤霞 第七章第七节
生意前模拟 + 生意中账单管理 + 生意后分析
"""
import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Tuple
import warnings
warnings.filterwarnings('ignore')


@dataclass
class LogisticsCostRecord:
    """物流成本记录"""
    record_id: str
    sku_id: str
    channel: str            # 'FBA', 'Own_WH', 'Direct_Mail'
    destination: str        # 'US', 'UK', 'DE'
    units: int
    declared_weight_kg: float
    actual_weight_kg: float
    declared_volume_m3: float
    actual_volume_m3: float
    invoiced_amount: float
    expected_amount: float  # 按费率表计算的期望金额


class LogisticsCostLifecycleKPI:
    """物流成本三段生命周期KPI"""

    # 各渠道费率（书中行业参考数据）
    CHANNEL_RATES = {
        'FBA_US': {'per_unit': 8.50, 'storage_per_sqft_month': 0.83},
        'FBA_UK': {'per_unit': 6.80, 'storage_per_sqft_month': 0.72},
        'Own_WH_DE': {'per_unit': 5.50, 'storage_per_sqft_month': 0.60},
        'Direct_Mail': {'per_unit': 3.50, 'storage_per_sqft_month': 0.0},
    }

    def pre_business_simulation(self, scenarios: List[Dict]) -> pd.DataFrame:
        """
        生意前成本模拟
        scenarios: [{'name': ..., 'channel_mix': {...}, 'monthly_units': N, 'avg_price': P}]
        """
        records = []
        for scenario in scenarios:
            total_logistics_cost = 0
            monthly_revenue = scenario['monthly_units'] * scenario['avg_price']

            for channel, pct in scenario['channel_mix'].items():
                units = scenario['monthly_units'] * pct
                rate = self.CHANNEL_RATES.get(channel, {}).get('per_unit', 8.0)
                channel_cost = units * rate
                total_logistics_cost += channel_cost

            logistics_rate = total_logistics_cost / max(monthly_revenue, 1)
            records.append({
                'scenario': scenario['name'],
                'monthly_units': scenario['monthly_units'],