Blame for src/p3_benchmark.py

from __future__ import annotations

"""
P3 Benchmark 标注 + 诊断规则引擎。

Benchmark 全部从 Google Sheets「KPI与Benchmark」+「历史实绩」动态读取，
不做硬编码 fallback。读取不到时明确标记为 "no_benchmark"。

诊断规则来源: P3-投放数据分析框架.md
"""
import logging
from typing import Any

log = logging.getLogger("ad-agent")

L1_THRESHOLDS = {
    "budget_pacing": (90.0, 110.0),
    "revenue_vs_target": (80.0, 100.0),
}


def _rate_value(
    value: float | None,
    red_threshold: float,
    green_threshold: float,
    higher_is_better: bool,
) -> str:
    if value is None or value != value:
        return "grey"
    if higher_is_better:
        if value >= green_threshold:
            return "green"
        if value <= red_threshold:
            return "red"
        return "yellow"
    else:
        if value <= green_threshold:
            return "green"
        if value >= red_threshold:
            return "red"
        return "yellow"


# ────────────────────────────────────────────────
# Benchmark 加载（纯 Sheets 驱动）
# ────────────────────────────────────────────────

def load_benchmarks(p3_config: dict) -> dict[str, dict]:
    """
    从 p3_config（已从 Google Sheets 加载）中构建 benchmark 表。

    返回格式:
      {objective_group: {metric: (red_threshold, green_threshold, higher_is_better)}}

    如果 Sheets 中没有对应数据，该 objective_group 的 benchmarks 就是空 dict，
    后续 apply_benchmarks 会标记 "no_benchmark" 而不是虚假的红黄绿。
    """
    benchmarks: dict[str, dict] = {}
    kpi_cfg = p3_config.get("kpi_config", {})
    chain = kpi_cfg.get("benchmark_chain", {})
    reference = kpi_cfg.get("reference_2025", {})
    historical = p3_config.get("historical", {})

    # 从 benchmark_chain 提取 ROAS / CPA 等阈值
    us_roas = chain.get("ROAS", {}).get("us_example")
    us_cpa = chain.get("CPA", {}).get("us_example")
    us_cpc = chain.get("CPC", {}).get("us_example")
    us_cpm = chain.get("CPM", {}).get("us_example")
    us_cpl = chain.get("CPL", {}).get("us_example")

    # Purchase 类 — 根据 ROAS / CPA 倒推
    if us_roas and us_roas > 0:
        benchmarks["Purchase_Cold"] = {
            "ROAS": (us_roas * 0.6, us_roas, True),
        }
        benchmarks["Purchase_ASC"] = {
            "ROAS": (us_roas * 0.8, us_roas * 1.5, True),
        }
        benchmarks["Purchase_RMKT"] = {
            "ROAS": (us_roas * 1.2, us_roas * 2.5, True),
        }
    if us_cpa and us_cpa > 0:
        benchmarks.setdefault("Purchase_Cold", {})["CPA"] = (us_cpa * 2, us_cpa, False)

    # Lead — CPL
    if us_cpl and us_cpl > 0:
        benchmarks["Lead"] = {
            "CPL": (us_cpl * 2, us_cpl, False),
            "LeadRate": (0.01, 0.03, True),
        }

    # 通用 CPC / CPM — 从 chain 中取
    if us_cpc and us_cpc > 0:
        benchmarks["Linkclick"] = {
            "CPC": (us_cpc * 2, us_cpc, False),
            "CTR": (0.008, 0.015, True),
        }
    if us_cpm and us_cpm > 0:
        benchmarks["Pageview"] = {
            "CPM": (us_cpm * 2, us_cpm, False),
        }

    # 从历史实绩中补充平台级 benchmark
    us_roas_by_type = historical.get("us_campaign_roas", {})
    for campaign_type, info in us_roas_by_type.items():
        avg = info.get("avg_roas")
        if avg and avg > 0:
            key = f"_hist_{campaign_type}"
            benchmarks[key] = {"ROAS": (avg * 0.6, avg, True)}

    loaded_groups = [k for k in benchmarks if not k.startswith("_hist_")]
    if loaded_groups:
        log.info(f"从 Sheets 加载了 {len(loaded_groups)} 组 Benchmark: {loaded_groups}")
    else:
        log.warning("Sheets 中未读取到任何 Benchmark 数据，指标将标记为 no_benchmark")

    return benchmarks


def load_platform_benchmarks(p3_config: dict) -> dict[str, dict]:
    """
    从历史实绩中构建平台级 benchmark。
    读取不到则返回空 dict，后续标记 "no_benchmark"。
    """
    platform_bm: dict[str, dict] = {}
    historical = p3_config.get("historical", {})
    ca_cf = historical.get("ca_channel_funnel", {})

    for key, info in ca_cf.items():
        parts = key.split("_", 1)
        if len(parts) < 2:
            continue
        platform = parts[0]
        if platform not in platform_bm:
            platform_bm[platform] = {}

        cpc = info.get("cpc")
        ctr = info.get("ctr")
        cpm = info.get("cpm")
        if cpc and cpc > 0:
            platform_bm[platform]["CPC"] = (cpc * 2, cpc, False)
        if ctr and ctr > 0:
            platform_bm[platform]["CTR"] = (ctr * 0.5, ctr, True)
        if cpm and cpm > 0:
            platform_bm[platform]["CPM"] = (cpm * 2, cpm, False)

    if platform_bm:
        log.info(f"从 Sheets 加载了 {len(platform_bm)} 个平台 Benchmark")
    else:
        log.warning("Sheets 中未读取到平台级 Benchmark，平台指标将标记为 no_benchmark")

    return platform_bm


def apply_benchmarks(
    data: list[dict], benchmarks: dict, level: str,
    platform_benchmarks: dict | None = None,
) -> list[dict]:
    """
    给 L1-L4 数据标注 red/yellow/green。
    没有 benchmark 数据的指标标记为 "no_benchmark"。
    """
    for row in data:
        ratings: dict[str, str] = {}

        if level == "L1":
            bp = row.get("budget_pacing")
            rvt = row.get("revenue_vs_target")

            if bp is None and rvt is None:
                ratings["_status"] = "no_benchmark"
            else:
                if bp is not None:
                    lo, hi = L1_THRESHOLDS["budget_pacing"]
                    if bp < lo:
                        ratings["budget_pacing"] = "yellow"
                    elif bp > hi:
                        ratings["budget_pacing"] = "red"
                    else:
                        ratings["budget_pacing"] = "green"

                if rvt is not None:
                    lo, hi = L1_THRESHOLDS["revenue_vs_target"]
                    if rvt < lo:
                        ratings["revenue_vs_target"] = "red"
                    elif rvt >= hi:
                        ratings["revenue_vs_target"] = "green"
                    else:
                        ratings["revenue_vs_target"] = "yellow"

        elif level == "L2":
            obj_group = row.get("objective_group", "")
            bm = benchmarks.get(obj_group, {})
            if not bm:
                ratings["_status"] = "no_benchmark"
            else:
                for metric, (red_t, green_t, hib) in bm.items():
                    val = row.get(metric)
                    ratings[metric] = _rate_value(val, red_t, green_t, hib)

        elif level == "L3":
            plat = row.get("platform", "")
            bm = (platform_benchmarks or {}).get(plat, {})
            if not bm:
                ratings["_status"] = "no_benchmark"
            else:
                for metric, (red_t, green_t, hib) in bm.items():
                    val = row.get(metric.lower(), row.get(metric))
                    ratings[metric] = _rate_value(val, red_t, green_t, hib)

        elif level == "L4":
            obj_group = row.get("objective_group", "")
            bm = benchmarks.get(obj_group, {})
            if not bm:
                ratings["_status"] = "no_benchmark"
            else:
                for metric, (red_t, green_t, hib) in bm.items():
                    val = row.get(metric, row.get(metric.lower()))
                    ratings[metric] = _rate_value(val, red_t, green_t, hib)

        row["ratings"] = ratings

    return data


# ────────────────────────────────────────────────
# 诊断规则引擎
# ────────────────────────────────────────────────

def run_diagnostics(p3_result: dict, p3_config: dict) -> list[dict[str, str]]:
    """
    基于 P3 框架诊断规则表，输出诊断结论列表。
    每条: {"level": "red/yellow/green", "title": str, "detail": str, "action": str}
    """
    diags: list[dict[str, str]] = []

    has_benchmarks = any(
        row.get("ratings", {}).get("_status") != "no_benchmark"
        for layer in ("L1", "L2", "L3", "L4")
        for row in p3_result.get(layer, [])
    )
    if not has_benchmarks:
        diags.append({
            "level": "yellow", "layer": "系统",
            "title": "Benchmark 数据未加载",
            "detail": "未能从 Google Sheets 读取 Benchmark 配置，无法进行红黄绿标注",
            "action": "检查 credentials.json 或 GOOGLE_CREDENTIALS 环境变量是否配置正确",
        })

    # L1 诊断
    for row in p3_result.get("L1", []):
        mkt = row.get("market", "?")
        bp = row.get("budget_pacing")
        if bp is not None:
            if bp > 110:
                diags.append({
                    "level": "red", "layer": "L1",
                    "title": f"{mkt} 预算超支",
                    "detail": f"Budget Pacing {bp:.0f}%，已超出预算 {bp - 100:.0f}%",
                    "action": "审查近期增投，需要审批或调减预算",
                })
            elif bp < 90:
                diags.append({
                    "level": "yellow", "layer": "L1",
                    "title": f"{mkt} 预算花慢",
                    "detail": f"Budget Pacing {bp:.0f}%，存在机会成本",
                    "action": "排查是否有 campaign 暂停或受众耗尽",
                })
        elif row.get("budget") is None:
            diags.append({
                "level": "yellow", "layer": "L1",
                "title": f"{mkt} 月度预算未配置",
                "detail": "未能从 Sheets 月度目标中读取到该市场的 DM 预算",
                "action": "在 Sheets「月度目标」中填写该月预算和目标",
            })

        rvt = row.get("revenue_vs_target")
        if rvt is not None and rvt < 80:
            diags.append({
                "level": "red", "layer": "L1",
                "title": f"{mkt} 营收不达标",
                "detail": f"Revenue vs Target {rvt:.0f}%，距目标差距较大",
                "action": "排查哪个平台/漏斗拖后腿，考虑调整投放策略",
            })

    # L2 诊断
    l2_map = {r["objective_group"]: r for r in p3_result.get("L2", [])}

    for row in p3_result.get("L2", []):
        obj = row.get("objective_group", "")
        ratings = row.get("ratings", {})

        for metric, color in ratings.items():
            if metric.startswith("_"):
                continue
            if color == "red":
                diags.append({
                    "level": "red", "layer": "L2",
                    "title": f"{obj} {metric} 表现差",
                    "detail": f"{metric} = {row.get(metric, '?')}，低于 Benchmark",
                    "action": _l2_action(obj, metric),
                })

    lead = l2_map.get("Lead", {})
    lead_ratings = lead.get("ratings", {})
    if lead_ratings.get("CPL") == "red" and lead_ratings.get("LeadRate") == "red":
        diags.append({
            "level": "red", "layer": "L2",
            "title": "留邮漏斗异常",
            "detail": "CPL 高 + 留邮率低，落地页或 Offer 不够吸引",
            "action": "换 Offer（折扣码？内容？）、优化表单、检查落地页体验",
        })

    purchase_cold = l2_map.get("Purchase_Cold", {})
    cpa_chg = purchase_cold.get("CPA_chg")
    consideration_objs = [r for r in p3_result.get("L2", []) if r.get("funnel") == "Consideration"]
    mid_spend_down = any(
        r.get("spend_share", 0) < 15 or (r.get("clicks", 0) == 0)
        for r in consideration_objs
    )
    if cpa_chg and cpa_chg > 10 and mid_spend_down:
        diags.append({
            "level": "red", "layer": "L2",
            "title": "漏斗断裂风险",
            "detail": "下层 CPA 上升，中层流量不足",
            "action": "加中层预算补量，确保下层有足够意向用户",
        })

    # L3 诊断
    l3_rows = p3_result.get("L3", [])
    if len(l3_rows) >= 2:
        for row in l3_rows:
            ss = row.get("spend_share", 0)
            rs = row.get("revenue_share", 0)
            plat = row.get("platform", "")
            if ss > 30 and rs < ss * 0.5:
                diags.append({
                    "level": "yellow", "layer": "L3",
                    "title": f"{plat} 预算效率错配",
                    "detail": f"Spend Share {ss:.0f}% 但 Revenue Share 仅 {rs:.0f}%",
                    "action": f"评估是否需要从 {plat} 减预算，挪到高效平台",
                })

            roas_chg = row.get("roas_chg")
            if roas_chg is not None and roas_chg < -15:
                diags.append({
                    "level": "yellow", "layer": "L3",
                    "title": f"{plat} ROAS 下降",
                    "detail": f"ROAS 环比 {roas_chg:+.1f}%",
                    "action": f"排查 {plat} 素材/受众变化，考虑更新创意",
                })

            cpc_chg = row.get("cpc_chg")
            if cpc_chg is not None and cpc_chg > 20:
                diags.append({
                    "level": "yellow", "layer": "L3",
                    "title": f"{plat} CPC 飙升",
                    "detail": f"CPC 环比 {cpc_chg:+.1f}%",
                    "action": "检查受众重叠、竞争环境和素材新鲜度",
                })

    # L4 诊断（Top 异常 campaign）
    for row in p3_result.get("L4", [])[:20]:
        ratings = row.get("ratings", {})
        spend = row.get("spend", 0)
        roas = row.get("roas", 0)
        name = row.get("campaign_name", "?")
        roas_chg = row.get("roas_chg")

        if spend > 500 and any(v == "red" for k, v in ratings.items() if not k.startswith("_")):
            diags.append({
                "level": "red", "layer": "L4",
                "title": "高花费低效 Campaign",
                "detail": f"{name[:40]}... | Spend ${spend:,.0f}, ROAS {roas:.2f}",
                "action": "立即减预算，拆解是素材问题还是 targeting 问题",
            })

        if roas_chg is not None and roas_chg < -20 and spend > 200:
            diags.append({
                "level": "yellow", "layer": "L4",
                "title": "Campaign 效率恶化",
                "detail": f"{name[:40]}... | ROAS 环比 {roas_chg:+.1f}%",
                "action": "连续下降则减预算或暂停，查素材疲劳/受众耗尽",
            })

    seen = set()
    unique = []
    for d in diags:
        key = d["title"]
        if key not in seen:
            seen.add(key)
            unique.append(d)

    level_order = {"red": 0, "yellow": 1, "green": 2}
    unique.sort(key=lambda d: level_order.get(d["level"], 9))

    return unique


def _l2_action(objective_group: str, metric: str) -> str:
    actions = {
        ("Lead", "CPL"): "换 Offer、优化落地页表单、收紧 Targeting",
        ("Lead", "LeadRate"): "检查落地页、Offer 吸引力",
        ("Purchase_Cold", "CPA"): "审视受众/素材，暂停低效计划",
        ("Purchase_Cold", "ROAS"): "查素材和 Targeting，考虑减预算",
        ("Purchase_RMKT", "ROAS"): "检查再营销频次，是否池耗尽",
        ("ATC", "CostPerATC"): "检查商品竞争力、价格策略",
        ("Pageview", "CPM"): "换素材、扩受众，降低竞争成本",
        ("Linkclick", "CPC"): "优化素材 CTR，扩展受众",
    }
    return actions.get((objective_group, metric), "排查原因，评估调整策略")