在途ETA准确率与到货履约率KPI — 过程数据vs结果数据的全链路在途管理体系

"""
在途ETA准确率与到货履约率KPI体系
基于《全链路管理》陈凤霞 第五章第五节
过程数据(异常跟进) vs 结果数据(履约达成)
"""
import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')


@dataclass
class TransitBatchRecord:
    """在途批次记录"""
    batch_id: str
    sku_id: str
    route: str                      # 如 '上海→洛杉矶'
    planned_units: int
    actual_units: int
    planned_eta: datetime           # 计划到货日
    actual_arrival: Optional[datetime]  # 实际到货日（None=未到）
    transit_days_planned: int       # 计划在途天数
    eta_updates: List[datetime] = field(default_factory=list)  # ETA历史更新记录
    delay_cause: Optional[str] = None  # 延误原因


class InboundETAKPIAnalyzer:
    """在途ETA准确率与到货KPI分析器"""

    @staticmethod
    def compute_eta_accuracy(batch: TransitBatchRecord) -> Optional[float]:
        """计算单批次ETA准确率"""
        if batch.actual_arrival is None:
            return None
        delay_days = (batch.actual_arrival - batch.planned_eta).days
        accuracy = 1 - abs(delay_days) / max(batch.transit_days_planned, 1)
        return max(accuracy, 0.0)

    @staticmethod
    def delay_days(batch: TransitBatchRecord) -> Optional[int]:
        """计算延迟天数（正=延迟，负=提前）"""
        if batch.actual_arrival is None:
            return None
        return (batch.actual_arrival - batch.planned_eta).days

    def route_analytics(self, batches: List[TransitBatchRecord]) -> pd.DataFrame:
        """按航线分析ETA准确率分布"""
        route_data = {}
        for b in batches:
            if b.actual_arrival is None:
                continue
            dd = self.delay_days(b)
            ea = self.compute_eta_accuracy(b)
            if b.route not in route_data:
                route_data[b.route] = {'delays': [], 'accuracies': [], 'units': []}
            route_data[b.route]['delays'].append(dd)
            route_data[b.route]['accuracies'].append(ea)