from typing import Dict, List
from datetime import datetime

class AdEffectivenessMonitor:
    def __init__(self):
        # 定义数据口径统一配置
        self.attribution_window_days = 7  # 归因窗口期（天）
        self.attribution_model = "last_click"  # 归因模型（last_click、first_click、linear、time_decay）
        self.timezone = "UTC"  # 时区（统一为 UTC）
    
    def collect_impression_data(self, impression_data: List[Dict]) -> List[Dict]:
        """
        曝光数据采集与校验
        """
        validated_impressions = []
        for impression in impression_data:
            # 校验曝光时间戳、广告位 ID、创意 ID 等字段完整性
            if self._validate_impression(impression):
                # 过滤无效曝光（如曝光时长<1 秒、曝光区域<50% 屏幕）
                if impression.get("duration_seconds", 0) >= 1 and impression.get("viewability_rate", 0) >= 0.5:
                    validated_impressions.append(impression)
        return validated_impressions
    
    def collect_click_data(self, click_data: List[Dict]) -> List[Dict]:
        """
        点击数据实时采集与去重
        """
        deduplicated_clicks = []
        seen_clicks = set()
        
        for click in click_data:
            # 点击去重（同一设备 1 小时内仅记录一次点击）
            device_id = click.get("device_id")
            click_hour = click.get("timestamp").replace(minute=0, second=0, microsecond=0)
            click_key = f"{device_id}_{click_hour}"
            
            if click_key not in seen_clicks:
                seen_clicks.add(click_key)
                deduplicated_clicks.append(click)
        
        return deduplicated_clicks
    
    def match_activation_data(self, click_data: List[Dict], activation_data: List[Dict]) -> List[Dict]:
        """
        激活数据 S2S 回传与归因匹配
        """
        matched_activations = []
        
        for activation in activation_data:
            # 动态级联补偿算法（IP 匹配 → 设备 ID 匹配 → 概率模型打分）
            matched_click = self._find_matching_click(click_data, activation)
            if matched_click:
                matched_activations.append({
                    "activation": activation,
                    "matched_click": matched_click,
                    "attribution_model": self.attribution_model,
                })
        
        return matched_activations
    
    def _validate_impression(self, impression: Dict) -> bool:
        """
        校验曝光数据字段完整性
        """
        required_fields = ["timestamp", "ad_placement_id", "creative_id"]
        return all(field in impression for field in required_fields)
    
    def _find_matching_click(self, click_data: List[Dict], activation: Dict) -> Dict:
        """
        查找匹配的点击数据（简化示例）
        """
        # 实际实现应基于动态级联补偿算法（IP 匹配 → 设备 ID 匹配 → 概率模型打分）
        for click in click_data:
            if click.get("device_id") == activation.get("device_id"):
                return click
        return None

# 示例：广告效果监测
monitor = AdEffectivenessMonitor()

impression_data = [
    {"timestamp": "2026-09-28 10:00:00", "ad_placement_id": "placement_1", "creative_id": "creative_1", "duration_seconds": 2, "viewability_rate": 0.8},
    {"timestamp": "2026-09-28 10:01:00", "ad_placement_id": "placement_2", "creative_id": "creative_2", "duration_seconds": 0.5, "viewability_rate": 0.3},  # 无效曝光
]
click_data = [
    {"timestamp": "2026-09-28 10:05:00", "device_id": "device_1", "click_id": "click_1"},
    {"timestamp": "2026-09-28 10:05:30", "device_id": "device_1", "click_id": "click_2"},  # 重复点击
]
activation_data = [
    {"timestamp": "2026-09-28 10:10:00", "device_id": "device_1"},
]

validated_impressions = monitor.collect_impression_data(impression_data)
print("有效曝光:", validated_impressions)

deduplicated_clicks = monitor.collect_click_data(click_data)
print("去重点击:", deduplicated_clicks)

matched_activations = monitor.match_activation_data(deduplicated_clicks, activation_data)
print("匹配激活:", matched_activations)

from typing import Dict, List

class ROICalculator:
    def __init__(self):
        pass
    
    def calculate_roi(self, revenue: float, ad_spend: float) -> float:
        """
        计算 ROI：ROI = （后端变现收入 - 前端广告消耗）/ 前端广告消耗 × 100%
        """
        if ad_spend == 0:
            return 0.0
        return (revenue - ad_spend) / ad_spend * 100
    
    def calculate_cac(self, ad_spend: float, new_users: int) -> float:
        """
        计算 CAC：CAC = 前端广告消耗 / 新增用户数
        """
        if new_users == 0:
            return 0.0
        return ad_spend / new_users
    
    def calculate_channel_roi_cac(self, channel_data: List[Dict]) -> List[Dict]:
        """
        计算分渠道 ROI 与 CAC
        """
        results = []
        for channel in channel_data:
            roi = self.calculate_roi(channel["revenue"], channel["ad_spend"])
            cac = self.calculate_cac(channel["ad_spend"], channel["new_users"])
            results.append({
                "channel": channel["channel"],
                "roi": roi,
                "cac": cac,
            })
        return results

# 示例：ROI 与 CAC 核算
calculator = ROICalculator()

channel_data = [
    {"channel": "巨量引擎", "ad_spend": 100000.0, "revenue": 300000.0, "new_users": 5000},
    {"channel": "腾讯广告", "ad_spend": 80000.0, "revenue": 200000.0, "new_users": 4000},
    {"channel": "快手", "ad_spend": 50000.0, "revenue": 150000.0, "new_users": 3000},
]

results = calculator.calculate_channel_roi_cac(channel_data)
for result in results:
    print(f"渠道：{result['channel']}, ROI: {result['roi']:.2f}%, CAC: {result['cac']:.2f}")