大促实时售罄模拟与流量协同 — 电商大促中实时预测与紧急资源调配

"""
大促实时售罄模拟与流量协同系统
功能：Bayesian实时销售速率估算 + 售罄模拟 + 流量协同决策
"""
import numpy as np
import pandas as pd
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
import warnings
warnings.filterwarnings('ignore')


@dataclass
class PromoSKU:
    """大促SKU状态"""
    sku_id: str
    initial_stock: int          # 大促前库存
    current_stock: int          # 当前库存
    target_sellthrough: float   # 目标售罄率（如0.85）
    prior_hourly_sales: float   # 先验每小时销量（基于历史大促）
    prior_confidence: float = 0.3  # 先验置信度（0-1）
    hourly_sales_history: List[float] = field(default_factory=list)


class BayesianSalesRateEstimator:
    """Bayesian实时销售速率估算"""
    
    def __init__(self, prior_mean: float, prior_std: float):
        self.mu = prior_mean      # 均值
        self.sigma2 = prior_std ** 2  # 方差
        self.observations = []
    
    def update(self, observed_sales: float, obs_noise_std: float = None) -> Tuple[float, float]:
        """Bayesian更新"""
        if obs_noise_std is None:
            obs_noise_std = max(self.mu * 0.3, 1.0)
        
        obs_sigma2 = obs_noise_std ** 2
        
        # Bayesian更新公式
        prior_precision = 1 / self.sigma2
        obs_precision = 1 / obs_sigma2
        
        new_precision = prior_precision + obs_precision
        new_mu = (prior_precision * self.mu + obs_precision * observed_sales) / new_precision
        new_sigma2 = 1 / new_precision
        
        self.mu = new_mu
        self.sigma2 = new_sigma2
        self.observations.append(observed_sales)
        
        return self.mu, np.sqrt(self.sigma2)
    
    @property
    def current_rate(self) -> float:
        return max(self.mu, 0.1)
    
    @property
    def uncertainty(self) -> float:
        return np.sqrt(self.sigma2)