Spaces:
Runtime error
Runtime error
| 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), "排查原因，评估调整策略") | |