Spaces:
Runtime error
Runtime error
| """ | |
| 日常数据监控引擎：检测指标异常、数据更新状态。 | |
| 不做业务归因，只客观报告异常波动。 | |
| """ | |
| import logging | |
| from datetime import date, timedelta | |
| from pathlib import Path | |
| from typing import Any | |
| import pandas as pd | |
| from src.loader import load_config, load_daily_data | |
| from src.campaign_parser import enrich_dataframe | |
| log = logging.getLogger("ad-agent") | |
| FILTER_DIMS = ("country", "platform", "ad_objective", "product_line", "product_model", "project_type") | |
| def _load_enriched(config_path: Path) -> pd.DataFrame: | |
| """加载日度数据并用 campaign_parser 解析出额外维度列。""" | |
| df = load_daily_data(config_path) | |
| df = enrich_dataframe(df) | |
| return df | |
| MONITOR_LABEL_ZH = { | |
| "Cost": "花费", | |
| "Ads.Sales": "销售额", | |
| "ROAS": "ROAS", | |
| "CPC": "CPC", | |
| "CTR": "CTR", | |
| "CVR": "转化率", | |
| "GA4.Sessions": "会话", | |
| "ATC": "加购", | |
| "Impressions": "展示量", | |
| } | |
| MONITOR_METRICS = { | |
| "Cost": {"unit": "$", "threshold": 0.30}, | |
| "Ads.Sales": {"unit": "$", "threshold": 0.30}, | |
| "ROAS": {"unit": "", "threshold": 0.25}, | |
| "CPC": {"unit": "$", "threshold": 0.30}, | |
| "CTR": {"unit": "%", "threshold": 0.25}, | |
| "CVR": {"unit": "%", "threshold": 0.25}, | |
| "GA4.Sessions": {"unit": "", "threshold": 0.30}, | |
| "ATC": {"unit": "", "threshold": 0.35}, | |
| "Impressions": {"unit": "", "threshold": 0.30}, | |
| } | |
| def _fmt_value(val: float, unit: str) -> str: | |
| if val != val or val is None: | |
| return "—" | |
| if unit == "%": | |
| return f"{val * 100:.2f}%" if abs(val) < 1 else f"{val:.2f}%" | |
| if unit == "$": | |
| if abs(val) >= 1e6: | |
| return f"${val / 1e6:.2f}M" | |
| if abs(val) >= 1e3: | |
| return f"${val / 1e3:.1f}K" | |
| return f"${val:.2f}" | |
| if abs(val) >= 1e6: | |
| return f"{val / 1e6:.2f}M" | |
| if abs(val) >= 1e3: | |
| return f"{val / 1e3:.1f}K" | |
| return f"{val:.2f}" | |
| def _conv_filter(df: pd.DataFrame) -> pd.DataFrame: | |
| """返回仅 Conversion funnel 的行，用于 ROAS 计算。""" | |
| if "campaign_type" in df.columns: | |
| return df[df["campaign_type"] == "Conversion"] | |
| return df | |
| def _conv_roas(df: pd.DataFrame) -> float | None: | |
| """用 Conversion 行计算 ROAS。""" | |
| conv = _conv_filter(df) | |
| cost = conv["Cost"].sum() if "Cost" in conv.columns else 0 | |
| sales = conv["Ads.Sales"].sum() if "Ads.Sales" in conv.columns else 0 | |
| return sales / cost if cost > 0 else None | |
| def _agg_metrics(today_df: pd.DataFrame, ref_df: pd.DataFrame) -> dict[str, Any]: | |
| """对一组数据计算核心指标当日值和 7 日均值变化。ROAS 只用 Conversion 数据。""" | |
| cost = today_df["Cost"].sum() if "Cost" in today_df.columns else 0 | |
| sales = today_df["Ads.Sales"].sum() if "Ads.Sales" in today_df.columns else 0 | |
| imp = today_df["Impressions"].sum() if "Impressions" in today_df.columns else 0 | |
| roas = _conv_roas(today_df) | |
| result = { | |
| "Cost": round(cost, 2), | |
| "Ads.Sales": round(sales, 2), | |
| "ROAS": round(roas, 2) if roas else None, | |
| "Impressions": int(imp), | |
| } | |
| if not ref_df.empty: | |
| c7 = ref_df.groupby("date")["Cost"].sum().mean() if "Cost" in ref_df.columns else 0 | |
| s7 = ref_df.groupby("date")["Ads.Sales"].sum().mean() if "Ads.Sales" in ref_df.columns else 0 | |
| conv_ref = _conv_filter(ref_df) | |
| rc7 = conv_ref.groupby("date")["Cost"].sum().mean() if not conv_ref.empty and "Cost" in conv_ref.columns else 0 | |
| rs7 = conv_ref.groupby("date")["Ads.Sales"].sum().mean() if not conv_ref.empty and "Ads.Sales" in conv_ref.columns else 0 | |
| r7 = rs7 / rc7 if rc7 > 0 else None | |
| if c7 > 0: | |
| result["cost_chg"] = round((cost - c7) / c7 * 100, 1) | |
| if s7 > 0: | |
| result["sales_chg"] = round((sales - s7) / s7 * 100, 1) | |
| if r7 and r7 > 0 and roas: | |
| result["roas_chg"] = round((roas - r7) / r7 * 100, 1) | |
| return result | |
| _TOP_CAMPAIGNS_N = 10 | |
| def _agg_campaign_metrics(camp_today: pd.DataFrame, camp_ref: pd.DataFrame) -> dict[str, Any]: | |
| """Campaign 级聚合：Cost / Sales / ROAS + vs 7日均值变化。ROAS 只用 Conversion。""" | |
| cost = camp_today["Cost"].sum() if "Cost" in camp_today.columns else 0 | |
| sales = camp_today["Ads.Sales"].sum() if "Ads.Sales" in camp_today.columns else 0 | |
| roas = _conv_roas(camp_today) | |
| result: dict[str, Any] = { | |
| "Cost": round(cost, 2), | |
| "Ads.Sales": round(sales, 2), | |
| "ROAS": round(roas, 2) if roas else None, | |
| } | |
| if not camp_ref.empty: | |
| conv_ref = _conv_filter(camp_ref) | |
| rc7 = conv_ref.groupby("date")["Cost"].sum().mean() if not conv_ref.empty and "Cost" in conv_ref.columns else 0 | |
| rs7 = conv_ref.groupby("date")["Ads.Sales"].sum().mean() if not conv_ref.empty and "Ads.Sales" in conv_ref.columns else 0 | |
| r7 = rs7 / rc7 if rc7 > 0 else None | |
| c7 = camp_ref.groupby("date")["Cost"].sum().mean() if "Cost" in camp_ref.columns else 0 | |
| result["avg7_cost"] = round(c7, 2) if c7 else None | |
| result["avg7_roas"] = round(r7, 2) if r7 else None | |
| if c7 > 0: | |
| result["cost_chg"] = round((cost - c7) / c7 * 100, 1) | |
| if r7 and r7 > 0 and roas: | |
| result["roas_chg"] = round((roas - r7) / r7 * 100, 1) | |
| return result | |
| def _build_hierarchy( | |
| day_df: pd.DataFrame, ref_df: pd.DataFrame, | |
| full_df: pd.DataFrame | None = None, target_date: date | None = None, | |
| ) -> dict[str, Any]: | |
| """ | |
| 构建 国家 -> 平台 -> Campaign Type -> Campaign Name 的四层聚合。 | |
| 对当天无数据的平台，自动回退取前一天数据并标记 fallback_date。 | |
| Campaign Name 层只展示花费 Top N，避免数据量爆炸。 | |
| """ | |
| has_country = "country" in day_df.columns | |
| has_platform = "platform" in day_df.columns | |
| has_type = "campaign_type" in day_df.columns | |
| has_campaign = "campaign_name" in day_df.columns | |
| fallback_platforms: dict[str, str] = {} | |
| if full_df is not None and has_platform and target_date is not None: | |
| present = set(day_df["platform"].dropna().unique()) if not day_df.empty else set() | |
| all_plats = set(full_df["platform"].dropna().unique()) if "platform" in full_df.columns else set() | |
| for plat in all_plats - present: | |
| p_df = full_df[full_df["platform"] == plat] | |
| if p_df.empty: | |
| continue | |
| prev_date = p_df["date"].max() | |
| if (pd.Timestamp(target_date) - prev_date).days <= 3: | |
| fb_rows = p_df[p_df["date"] == prev_date] | |
| day_df = pd.concat([day_df, fb_rows], ignore_index=True) | |
| fallback_platforms[plat] = str(prev_date.date()) | |
| total = _agg_metrics(day_df, ref_df) | |
| by_country: dict[str, Any] = {} | |
| countries = sorted(day_df["country"].dropna().unique()) if has_country else [] | |
| for country in countries: | |
| c_today = day_df[day_df["country"] == country] | |
| c_ref = ref_df[ref_df["country"] == country] if has_country and not ref_df.empty else pd.DataFrame() | |
| c_data: dict[str, Any] = {"total": _agg_metrics(c_today, c_ref), "by_platform": {}} | |
| platforms = sorted(c_today["platform"].dropna().unique()) if has_platform else [] | |
| for plat in platforms: | |
| p_today = c_today[c_today["platform"] == plat] | |
| p_ref = c_ref[c_ref["platform"] == plat] if has_platform and not c_ref.empty else pd.DataFrame() | |
| p_data: dict[str, Any] = {"total": _agg_metrics(p_today, p_ref), "by_type": {}} | |
| if plat in fallback_platforms: | |
| p_data["fallback_date"] = fallback_platforms[plat] | |
| types = sorted(p_today["campaign_type"].dropna().unique()) if has_type else [] | |
| for ctype in types: | |
| t_today = p_today[p_today["campaign_type"] == ctype] | |
| t_ref = p_ref[p_ref["campaign_type"] == ctype] if has_type and not p_ref.empty else pd.DataFrame() | |
| type_node = _agg_metrics(t_today, t_ref) | |
| if has_campaign and not t_today.empty: | |
| camp_costs = t_today.groupby("campaign_name")["Cost"].sum() | |
| top_names = camp_costs.nlargest(_TOP_CAMPAIGNS_N).index.tolist() | |
| top_campaigns = [] | |
| for cname in top_names: | |
| cm_today = t_today[t_today["campaign_name"] == cname] | |
| cm_ref = ( | |
| t_ref[t_ref["campaign_name"] == cname] | |
| if has_campaign and not t_ref.empty and "campaign_name" in t_ref.columns | |
| else pd.DataFrame() | |
| ) | |
| cm = _agg_campaign_metrics(cm_today, cm_ref) | |
| cm["campaign_name"] = cname | |
| top_campaigns.append(cm) | |
| type_node["top_campaigns"] = top_campaigns | |
| p_data["by_type"][ctype] = type_node | |
| c_data["by_platform"][plat] = p_data | |
| by_country[country] = c_data | |
| return {"total": total, "by_country": by_country, "fallback_platforms": fallback_platforms} | |
| def _build_campaign_alerts( | |
| day_df: pd.DataFrame, ref_df: pd.DataFrame, | |
| ) -> list[dict[str, Any]]: | |
| """ | |
| 检测 Campaign 级异动： | |
| 1. 花费暴涨：今日花费 > 7日均值 * 2.5（且今日花费 > $200 避免噪声） | |
| 2. ROAS 骤降：ROAS 下降 > 50%（且今日花费 > $100） | |
| 3. 花费归零：7日内有花费（均值 > $100）但今日为 0 | |
| """ | |
| if "campaign_name" not in day_df.columns: | |
| return [] | |
| campaign_alerts: list[dict[str, Any]] = [] | |
| conv_day = _conv_filter(day_df) | |
| today_by_camp = day_df.groupby("campaign_name").agg( | |
| cost=("Cost", "sum"), | |
| ).reset_index() if "Cost" in day_df.columns else pd.DataFrame() | |
| if today_by_camp.empty: | |
| return campaign_alerts | |
| conv_today_by_camp = conv_day.groupby("campaign_name").agg( | |
| conv_cost=("Cost", "sum"), | |
| conv_sales=("Ads.Sales", "sum"), | |
| ).reset_index() if not conv_day.empty and "Cost" in conv_day.columns else pd.DataFrame(columns=["campaign_name", "conv_cost", "conv_sales"]) | |
| today_by_camp = today_by_camp.merge(conv_today_by_camp, on="campaign_name", how="left") | |
| today_by_camp["conv_cost"] = today_by_camp["conv_cost"].fillna(0) | |
| today_by_camp["conv_sales"] = today_by_camp["conv_sales"].fillna(0) | |
| today_by_camp["roas"] = today_by_camp.apply( | |
| lambda r: r["conv_sales"] / r["conv_cost"] if r["conv_cost"] > 0 else None, axis=1 | |
| ) | |
| ref_by_camp: dict[str, dict] = {} | |
| if not ref_df.empty and "campaign_name" in ref_df.columns and "Cost" in ref_df.columns: | |
| conv_ref = _conv_filter(ref_df) | |
| for cname, grp in ref_df.groupby("campaign_name"): | |
| daily_cost = grp.groupby("date")["Cost"].sum() | |
| avg_cost = daily_cost.mean() | |
| c_grp = conv_ref[conv_ref["campaign_name"] == cname] if not conv_ref.empty else pd.DataFrame() | |
| if not c_grp.empty and "Cost" in c_grp.columns: | |
| c_daily_cost = c_grp.groupby("date")["Cost"].sum() | |
| c_daily_sales = c_grp.groupby("date")["Ads.Sales"].sum() if "Ads.Sales" in c_grp.columns else pd.Series(dtype=float) | |
| avg_conv_cost = c_daily_cost.mean() | |
| avg_conv_sales = c_daily_sales.mean() if not c_daily_sales.empty else 0 | |
| avg_roas = avg_conv_sales / avg_conv_cost if avg_conv_cost > 0 else None | |
| else: | |
| avg_roas = None | |
| ref_by_camp[str(cname)] = { | |
| "avg_cost": avg_cost, | |
| "avg_roas": avg_roas, | |
| } | |
| for _, row in today_by_camp.iterrows(): | |
| cname = str(row["campaign_name"]) | |
| cost = float(row["cost"]) | |
| roas = row["roas"] | |
| ref = ref_by_camp.get(cname, {}) | |
| avg_cost = ref.get("avg_cost", 0) | |
| avg_roas = ref.get("avg_roas") | |
| country = "" | |
| platform = "" | |
| funnel = "" | |
| camp_rows = day_df[day_df["campaign_name"] == cname] | |
| if not camp_rows.empty: | |
| if "country" in camp_rows.columns: | |
| country = str(camp_rows["country"].iloc[0]) | |
| if "platform" in camp_rows.columns: | |
| platform = str(camp_rows["platform"].iloc[0]) | |
| if "campaign_type" in camp_rows.columns: | |
| funnel = str(camp_rows["campaign_type"].iloc[0]) | |
| if avg_cost > 0 and cost > 200 and cost > avg_cost * 2.5: | |
| chg = round((cost - avg_cost) / avg_cost * 100, 1) | |
| campaign_alerts.append({ | |
| "type": "cost_spike", | |
| "level": "red", | |
| "campaign_name": cname, | |
| "country": country, | |
| "platform": platform, | |
| "funnel": funnel, | |
| "cost": round(cost, 2), | |
| "avg7_cost": round(avg_cost, 2), | |
| "change_pct": chg, | |
| "message": f"花费暴涨 +{chg}%（${cost:,.0f} vs 7日均 ${avg_cost:,.0f}）", | |
| }) | |
| if avg_roas and avg_roas > 0 and roas is not None and cost > 100: | |
| roas_chg = (roas - avg_roas) / avg_roas | |
| if roas_chg < -0.5: | |
| campaign_alerts.append({ | |
| "type": "roas_drop", | |
| "level": "red", | |
| "campaign_name": cname, | |
| "country": country, | |
| "platform": platform, | |
| "funnel": funnel, | |
| "roas": round(roas, 2), | |
| "avg7_roas": round(avg_roas, 2), | |
| "change_pct": round(roas_chg * 100, 1), | |
| "message": f"ROAS 骤降 {roas_chg * 100:.0f}%（{roas:.2f} vs 7日均 {avg_roas:.2f}）", | |
| }) | |
| today_names = set(today_by_camp["campaign_name"].astype(str)) | |
| for cname, ref in ref_by_camp.items(): | |
| if cname not in today_names and ref["avg_cost"] > 100: | |
| ref_rows = ref_df[ref_df["campaign_name"] == cname] | |
| country = "" | |
| platform = "" | |
| funnel = "" | |
| if not ref_rows.empty: | |
| if "country" in ref_rows.columns: | |
| country = str(ref_rows["country"].iloc[0]) | |
| if "platform" in ref_rows.columns: | |
| platform = str(ref_rows["platform"].iloc[0]) | |
| if "campaign_type" in ref_rows.columns: | |
| funnel = str(ref_rows["campaign_type"].iloc[0]) | |
| campaign_alerts.append({ | |
| "type": "stopped", | |
| "level": "yellow", | |
| "campaign_name": cname, | |
| "country": country, | |
| "platform": platform, | |
| "funnel": funnel, | |
| "avg7_cost": round(ref["avg_cost"], 2), | |
| "message": f"今日无花费（7日均 ${ref['avg_cost']:,.0f}）", | |
| }) | |
| campaign_alerts.sort(key=lambda a: ( | |
| 0 if a["level"] == "red" else 1, | |
| -a.get("cost", 0), | |
| )) | |
| return campaign_alerts | |
| def run_daily_monitor(config_path: Path, target_date: date | None = None, country: str | None = None) -> dict[str, Any]: | |
| """ | |
| 对比 target_date 与前 7 天均值，检测异常波动。 | |
| country 为 None 或 "all" 时返回全部国家汇总。 | |
| """ | |
| df = _load_enriched(config_path) | |
| if country and country.lower() != "all" and "country" in df.columns: | |
| df = df[df["country"] == country] | |
| latest_in_data = df["date"].max().date() if not df.empty else None | |
| if target_date is None: | |
| yesterday = date.today() - timedelta(days=1) | |
| target_date = min(yesterday, latest_in_data) if latest_in_data else yesterday | |
| days_behind = (date.today() - latest_in_data).days if latest_in_data else None | |
| stale_days = 2 | |
| if latest_in_data is None: | |
| data_status = {"latest_date": None, "days_behind": None, "status": "missing"} | |
| elif days_behind > stale_days: | |
| data_status = {"latest_date": str(latest_in_data), "days_behind": days_behind, "status": "stale"} | |
| else: | |
| data_status = {"latest_date": str(latest_in_data), "days_behind": days_behind, "status": "ok"} | |
| platform_freshness = {} | |
| if not df.empty and "platform" in df.columns: | |
| for plat in ("Meta", "Google", "Criteo", "Bing"): | |
| p_df = df[df["platform"] == plat] | |
| if p_df.empty: | |
| platform_freshness[plat] = {"latest_date": None, "days_behind": None, "status": "missing"} | |
| else: | |
| p_latest = p_df["date"].max().date() | |
| p_behind = (date.today() - p_latest).days | |
| if p_behind > stale_days: | |
| platform_freshness[plat] = {"latest_date": str(p_latest), "days_behind": p_behind, "status": "stale"} | |
| else: | |
| platform_freshness[plat] = {"latest_date": str(p_latest), "days_behind": p_behind, "status": "ok"} | |
| data_status["by_platform"] = platform_freshness | |
| target_ts = pd.Timestamp(target_date) | |
| day_df = df[df["date"] == target_ts] | |
| if day_df.empty: | |
| return { | |
| "date": str(target_date), | |
| "data_status": data_status, | |
| "alerts": [], | |
| "summary": {}, | |
| "by_platform": {}, | |
| "error": f"没有 {target_date} 的数据", | |
| } | |
| window_start = target_ts - pd.Timedelta(days=7) | |
| window_end = target_ts - pd.Timedelta(days=1) | |
| ref_df = df[(df["date"] >= window_start) & (df["date"] <= window_end)] | |
| total_today = {} | |
| total_avg7 = {} | |
| for m in MONITOR_METRICS: | |
| if m in day_df.columns: | |
| total_today[m] = day_df[m].sum() if m not in ("ROAS", "CTR", "CVR", "CPC") else None | |
| if m not in ("ROAS", "CTR", "CVR", "CPC"): | |
| total_today[m] = day_df[m].sum() | |
| else: | |
| total_today[m] = None | |
| cost_today = day_df["Cost"].sum() if "Cost" in day_df.columns else 0 | |
| sales_today = day_df["Ads.Sales"].sum() if "Ads.Sales" in day_df.columns else 0 | |
| imp_today = day_df["Impressions"].sum() if "Impressions" in day_df.columns else 0 | |
| sessions_today = day_df["GA4.Sessions"].sum() if "GA4.Sessions" in day_df.columns else 0 | |
| atc_today = day_df["ATC"].sum() if "ATC" in day_df.columns else 0 | |
| clicks_today = 0 | |
| if "CTR" in day_df.columns and "Impressions" in day_df.columns: | |
| clicks_today = (day_df["Impressions"].fillna(0) * day_df["CTR"].fillna(0)).sum() | |
| conversions_today = 0 | |
| if "CVR" in day_df.columns and "GA4.Sessions" in day_df.columns: | |
| conversions_today = (day_df["GA4.Sessions"].fillna(0) * day_df["CVR"].fillna(0)).sum() | |
| total_today["Cost"] = cost_today | |
| total_today["Ads.Sales"] = sales_today | |
| total_today["ROAS"] = _conv_roas(day_df) | |
| total_today["Impressions"] = imp_today | |
| total_today["GA4.Sessions"] = sessions_today | |
| total_today["ATC"] = atc_today | |
| if "CPC" in day_df.columns: | |
| total_today["CPC"] = cost_today / clicks_today if clicks_today > 0 else None | |
| if "CTR" in day_df.columns: | |
| total_today["CTR"] = clicks_today / imp_today if imp_today > 0 else None | |
| if "CVR" in day_df.columns: | |
| total_today["CVR"] = conversions_today / sessions_today if sessions_today > 0 else None | |
| if not ref_df.empty: | |
| daily_sums = ref_df.groupby("date") | |
| n_days = daily_sums.ngroups or 1 | |
| for m in MONITOR_METRICS: | |
| if m in ref_df.columns: | |
| if m in ("CTR", "CVR"): | |
| total_avg7[m] = ref_df.groupby("date")[m].mean().mean() | |
| elif m == "ROAS": | |
| conv_ref = _conv_filter(ref_df) | |
| if not conv_ref.empty: | |
| rc7 = conv_ref.groupby("date")["Cost"].sum().mean() | |
| rs7 = conv_ref.groupby("date")["Ads.Sales"].sum().mean() | |
| total_avg7[m] = rs7 / rc7 if rc7 > 0 else None | |
| else: | |
| total_avg7[m] = None | |
| elif m == "CPC": | |
| c7 = ref_df.groupby("date")["Cost"].sum().mean() | |
| imp7 = ref_df.groupby("date")["Impressions"].sum().mean() if "Impressions" in ref_df.columns else 0 | |
| ctr7 = ref_df.groupby("date")["CTR"].mean().mean() if "CTR" in ref_df.columns else 0 | |
| clicks7 = imp7 * ctr7 if imp7 and ctr7 else 0 | |
| total_avg7[m] = c7 / clicks7 if clicks7 > 0 else None | |
| else: | |
| total_avg7[m] = ref_df.groupby("date")[m].sum().mean() | |
| alerts = [] | |
| for m, cfg in MONITOR_METRICS.items(): | |
| today_val = total_today.get(m) | |
| avg_val = total_avg7.get(m) | |
| if today_val is None or avg_val is None or avg_val == 0: | |
| continue | |
| if today_val != today_val or avg_val != avg_val: | |
| continue | |
| change = (today_val - avg_val) / abs(avg_val) | |
| if abs(change) < cfg["threshold"]: | |
| continue | |
| is_cost_metric = m in ("Cost", "CPC", "Impressions") | |
| if m in ("ROAS", "Ads.Sales", "GA4.Sessions", "ATC", "CTR", "CVR"): | |
| direction = "down" if change < 0 else "up" | |
| level = "red" if (change < 0 and abs(change) >= cfg["threshold"] * 1.5) else \ | |
| "green" if change > 0 else "yellow" | |
| else: | |
| direction = "up" if change > 0 else "down" | |
| level = "red" if (change > 0 and abs(change) >= cfg["threshold"] * 1.5) else \ | |
| "green" if change < 0 else "yellow" | |
| alerts.append({ | |
| "metric": m, | |
| "value": round(today_val, 4), | |
| "value_fmt": _fmt_value(today_val, cfg["unit"]), | |
| "avg7": round(avg_val, 4), | |
| "avg7_fmt": _fmt_value(avg_val, cfg["unit"]), | |
| "change_pct": round(change * 100, 1), | |
| "level": level, | |
| "direction": direction, | |
| }) | |
| alerts.sort(key=lambda a: (0 if a["level"] == "red" else 1 if a["level"] == "yellow" else 2, -abs(a["change_pct"]))) | |
| by_platform = {} | |
| for plat in ("Google", "Meta", "Criteo", "Bing"): | |
| p_today = day_df[day_df["platform"] == plat] if "platform" in day_df.columns else pd.DataFrame() | |
| p_ref = ref_df[ref_df["platform"] == plat] if "platform" in ref_df.columns else pd.DataFrame() | |
| if p_today.empty: | |
| continue | |
| p_cost = p_today["Cost"].sum() if "Cost" in p_today.columns else 0 | |
| p_sales = p_today["Ads.Sales"].sum() if "Ads.Sales" in p_today.columns else 0 | |
| p_roas = _conv_roas(p_today) | |
| p_data = {"Cost": p_cost, "Ads.Sales": p_sales, "ROAS": p_roas} | |
| if not p_ref.empty: | |
| c7 = p_ref.groupby("date")["Cost"].sum().mean() if "Cost" in p_ref.columns else 0 | |
| s7 = p_ref.groupby("date")["Ads.Sales"].sum().mean() if "Ads.Sales" in p_ref.columns else 0 | |
| conv_p_ref = _conv_filter(p_ref) | |
| rc7 = conv_p_ref.groupby("date")["Cost"].sum().mean() if not conv_p_ref.empty and "Cost" in conv_p_ref.columns else 0 | |
| rs7 = conv_p_ref.groupby("date")["Ads.Sales"].sum().mean() if not conv_p_ref.empty and "Ads.Sales" in conv_p_ref.columns else 0 | |
| r7 = rs7 / rc7 if rc7 > 0 else None | |
| p_data["avg7_Cost"] = round(c7, 2) if c7 else None | |
| p_data["avg7_Sales"] = round(s7, 2) if s7 else None | |
| p_data["avg7_ROAS"] = round(r7, 2) if r7 else None | |
| if c7 > 0: | |
| p_data["cost_chg"] = round((p_cost - c7) / c7 * 100, 1) | |
| if s7 > 0: | |
| p_data["sales_chg"] = round((p_sales - s7) / s7 * 100, 1) | |
| if r7 and r7 > 0 and p_roas: | |
| p_data["roas_chg"] = round((p_roas - r7) / r7 * 100, 1) | |
| by_platform[plat] = p_data | |
| summary_fmt = {} | |
| for m in ("Cost", "Ads.Sales", "ROAS", "CPC", "CTR", "CVR", "GA4.Sessions", "ATC"): | |
| v = total_today.get(m) | |
| if v is not None and v == v: | |
| summary_fmt[m] = _fmt_value(v, MONITOR_METRICS.get(m, {}).get("unit", "")) | |
| hierarchy = _build_hierarchy(day_df, ref_df, full_df=df, target_date=target_date) | |
| campaign_alerts = _build_campaign_alerts(day_df, ref_df) | |
| return { | |
| "date": str(target_date), | |
| "data_status": data_status, | |
| "alerts": alerts, | |
| "campaign_alerts": campaign_alerts, | |
| "summary": total_today, | |
| "summary_fmt": summary_fmt, | |
| "by_platform": by_platform, | |
| "hierarchy": hierarchy, | |
| } | |
| def _agg_matrix_metrics(subset: pd.DataFrame) -> dict[str, Any]: | |
| """对一组行计算矩阵所需的 8 个聚合指标。ROAS 只用 Conversion 数据。""" | |
| cost = subset["Cost"].sum() if "Cost" in subset.columns else 0 | |
| revenue = subset["Ads.Sales"].sum() if "Ads.Sales" in subset.columns else 0 | |
| impressions = int(subset["Impressions"].sum()) if "Impressions" in subset.columns else 0 | |
| clicks = 0.0 | |
| if "Impressions" in subset.columns and "CTR" in subset.columns: | |
| clicks = (subset["Impressions"].fillna(0) * subset["CTR"].fillna(0)).sum() | |
| conversions = 0.0 | |
| if "Impressions" in subset.columns and "CTR" in subset.columns and "CVR" in subset.columns: | |
| row_clicks = subset["Impressions"].fillna(0) * subset["CTR"].fillna(0) | |
| conversions = (row_clicks * subset["CVR"].fillna(0)).sum() | |
| roas = _conv_roas(subset) | |
| ctr = clicks / impressions if impressions > 0 else None | |
| cpc = cost / clicks if clicks > 0 else None | |
| cpm = cost / impressions * 1000 if impressions > 0 else None | |
| cvr = conversions / clicks if clicks > 0 else None | |
| return { | |
| "cost": round(cost, 2), | |
| "revenue": round(revenue, 2), | |
| "roas": round(roas, 4) if roas is not None else None, | |
| "clicks": int(round(clicks)), | |
| "ctr": round(ctr, 4) if ctr is not None else None, | |
| "cpc": round(cpc, 2) if cpc is not None else None, | |
| "cpm": round(cpm, 2) if cpm is not None else None, | |
| "cvr": round(cvr, 4) if cvr is not None else None, | |
| } | |
| def compute_trend_data(config_path: Path, start: date, end: date, country: str | None = None) -> dict[str, Any]: | |
| """ | |
| 返回 [start, end] 区间内按日聚合的趋势数据，用于图表展示。 | |
| 包含按漏斗、按渠道、按渠道×漏斗的花费分布以及 Conversion ROAS 折线。 | |
| country 为 None 或 "all" 时返回全部国家汇总。 | |
| """ | |
| df = _load_enriched(config_path) | |
| mask = (df["date"] >= pd.Timestamp(start)) & (df["date"] <= pd.Timestamp(end)) | |
| period = df[mask].copy() | |
| if country and country.lower() != "all" and "country" in period.columns: | |
| period = period[period["country"] == country] | |
| countries = sorted(period["country"].dropna().unique()) if "country" in period.columns else [] | |
| if period.empty: | |
| return {"dates": [], "by_funnel": {}, "by_platform": {}, "by_platform_funnel": {}, "roas": [], "countries": countries} | |
| period["ds"] = period["date"].dt.strftime("%Y-%m-%d") | |
| dates = sorted(period["ds"].unique()) | |
| funnels = ["Conversion", "Consideration", "Awareness"] | |
| platforms = sorted(period["platform"].dropna().unique()) if "platform" in period.columns else [] | |
| by_funnel: dict[str, list[float]] = {} | |
| piv_f = period.groupby(["ds", "campaign_type"])["Cost"].sum().unstack(fill_value=0) | |
| for f in funnels: | |
| by_funnel[f] = [round(float(piv_f.loc[d, f]), 2) if d in piv_f.index and f in piv_f.columns else 0 for d in dates] | |
| by_platform: dict[str, list[float]] = {} | |
| piv_p = period.groupby(["ds", "platform"])["Cost"].sum().unstack(fill_value=0) | |
| for p in platforms: | |
| by_platform[p] = [round(float(piv_p.loc[d, p]), 2) if d in piv_p.index and p in piv_p.columns else 0 for d in dates] | |
| by_platform_funnel: dict[str, dict[str, list[float]]] = {} | |
| piv_pf = period.groupby(["ds", "platform", "campaign_type"])["Cost"].sum() | |
| for p in platforms: | |
| by_platform_funnel[p] = {} | |
| for f in funnels: | |
| vals = [] | |
| for d in dates: | |
| try: | |
| vals.append(round(float(piv_pf.loc[(d, p, f)]), 2)) | |
| except KeyError: | |
| vals.append(0) | |
| by_platform_funnel[p][f] = vals | |
| conv = period[period["campaign_type"] == "Conversion"] | |
| conv_daily = conv.groupby("ds").agg(cost=("Cost", "sum"), sales=("Ads.Sales", "sum")) | |
| roas_list = [] | |
| for d in dates: | |
| if d in conv_daily.index and conv_daily.loc[d, "cost"] > 0: | |
| roas_list.append(round(float(conv_daily.loc[d, "sales"] / conv_daily.loc[d, "cost"]), 2)) | |
| else: | |
| roas_list.append(None) | |
| # ── 数据矩阵：Country → Channel → Funnel → Date ── | |
| matrix_total = _agg_matrix_metrics(period) | |
| matrix: dict[str, Any] = {} | |
| ctry_list = sorted(period["country"].dropna().unique()) if "country" in period.columns else [] | |
| for cty in ctry_list: | |
| c_df = period[period["country"] == cty] | |
| c_node: dict[str, Any] = {"m": _agg_matrix_metrics(c_df), "ch": {}} | |
| plats = sorted(c_df["platform"].dropna().unique()) if "platform" in c_df.columns else [] | |
| for pl in plats: | |
| p_df = c_df[c_df["platform"] == pl] | |
| if p_df.empty: | |
| continue | |
| p_node: dict[str, Any] = {"m": _agg_matrix_metrics(p_df), "ch": {}} | |
| for fn in funnels: | |
| f_df = p_df[p_df["campaign_type"] == fn] | |
| if f_df.empty: | |
| continue | |
| f_node: dict[str, Any] = {"m": _agg_matrix_metrics(f_df), "ch": {}} | |
| for dt in dates: | |
| d_df = f_df[f_df["ds"] == dt] | |
| if d_df.empty: | |
| continue | |
| f_node["ch"][dt] = _agg_matrix_metrics(d_df) | |
| p_node["ch"][fn] = f_node | |
| c_node["ch"][pl] = p_node | |
| matrix[cty] = c_node | |
| return { | |
| "dates": dates, | |
| "by_funnel": by_funnel, | |
| "by_platform": by_platform, | |
| "by_platform_funnel": by_platform_funnel, | |
| "roas": roas_list, | |
| "countries": countries, | |
| "matrix": matrix, | |
| "matrix_total": matrix_total, | |
| } | |
| def _apply_filters(df: pd.DataFrame, filters: dict[str, str | None]) -> pd.DataFrame: | |
| """按筛选条件过滤数据行。filters 中值为 None / 'all' / '' 时不过滤。""" | |
| for dim in FILTER_DIMS: | |
| val = filters.get(dim) | |
| if val and val.lower() != "all" and dim in df.columns: | |
| df = df[df[dim] == val] | |
| return df | |
| def get_filter_options(config_path: Path) -> dict[str, list[str]]: | |
| """返回四个筛选维度的可选值列表（从数据中动态提取）。""" | |
| df = _load_enriched(config_path) | |
| result: dict[str, list[str]] = {} | |
| for dim in FILTER_DIMS: | |
| if dim in df.columns: | |
| vals = sorted(df[dim].dropna().astype(str).unique()) | |
| vals = [v for v in vals if v.strip()] | |
| result[dim] = vals | |
| else: | |
| result[dim] = [] | |
| return result | |
| def _agg_summary(period: pd.DataFrame) -> dict[str, Any]: | |
| """对一段区间数据计算汇总 KPI。ROAS 只用 Conversion 数据。""" | |
| cost = period["Cost"].sum() if "Cost" in period.columns else 0 | |
| sales = period["Ads.Sales"].sum() if "Ads.Sales" in period.columns else 0 | |
| imp = period["Impressions"].sum() if "Impressions" in period.columns else 0 | |
| sessions = period["GA4.Sessions"].sum() if "GA4.Sessions" in period.columns else 0 | |
| atc = period["ATC"].sum() if "ATC" in period.columns else 0 | |
| roas = _conv_roas(period) | |
| clicks = 0.0 | |
| if "Impressions" in period.columns and "CTR" in period.columns: | |
| clicks = (period["Impressions"].fillna(0) * period["CTR"].fillna(0)).sum() | |
| conversions = 0.0 | |
| if "CVR" in period.columns and "GA4.Sessions" in period.columns: | |
| conversions = (period["GA4.Sessions"].fillna(0) * period["CVR"].fillna(0)).sum() | |
| return { | |
| "Cost": round(cost, 2), | |
| "Ads.Sales": round(sales, 2), | |
| "ROAS": round(roas, 4) if roas else None, | |
| "Impressions": int(imp), | |
| "GA4.Sessions": round(sessions, 2), | |
| "ATC": round(atc, 2), | |
| "CPC": round(cost / clicks, 2) if clicks > 0 else None, | |
| "CTR": round(clicks / imp, 4) if imp > 0 else None, | |
| "CVR": round(conversions / sessions, 4) if sessions > 0 else None, | |
| } | |
| def _build_product_breakdown(period: pd.DataFrame) -> list[dict[str, Any]]: | |
| """按 产品线 -> 型号 聚合各 funnel 的 Cost / Sales / ROAS。""" | |
| if "product_line" not in period.columns: | |
| return [] | |
| funnels = ["Conversion", "Consideration", "Awareness"] | |
| breakdown: list[dict[str, Any]] = [] | |
| for pl, pl_df in period.groupby("product_line"): | |
| pl_str = str(pl) | |
| if not pl_str or pl_str == "None": | |
| continue | |
| pl_cost = pl_df["Cost"].sum() if "Cost" in pl_df.columns else 0 | |
| pl_sales = pl_df["Ads.Sales"].sum() if "Ads.Sales" in pl_df.columns else 0 | |
| pl_roas = _conv_roas(pl_df) | |
| funnel_split: dict[str, float] = {} | |
| for fn in funnels: | |
| fn_cost = pl_df[pl_df["campaign_type"] == fn]["Cost"].sum() if "campaign_type" in pl_df.columns else 0 | |
| funnel_split[fn] = round(fn_cost, 2) | |
| models: list[dict[str, Any]] = [] | |
| if "product_model" in pl_df.columns: | |
| for pm, pm_df in pl_df.groupby("product_model"): | |
| pm_str = str(pm) | |
| if not pm_str or pm_str == "None": | |
| continue | |
| pm_cost = pm_df["Cost"].sum() if "Cost" in pm_df.columns else 0 | |
| pm_sales = pm_df["Ads.Sales"].sum() if "Ads.Sales" in pm_df.columns else 0 | |
| pm_roas = _conv_roas(pm_df) | |
| pm_funnel: dict[str, float] = {} | |
| for fn in funnels: | |
| fn_cost = pm_df[pm_df["campaign_type"] == fn]["Cost"].sum() if "campaign_type" in pm_df.columns else 0 | |
| pm_funnel[fn] = round(fn_cost, 2) | |
| models.append({ | |
| "name": pm_str, | |
| "Cost": round(pm_cost, 2), | |
| "Ads.Sales": round(pm_sales, 2), | |
| "ROAS": round(pm_roas, 4) if pm_roas else None, | |
| "by_funnel": pm_funnel, | |
| }) | |
| models.sort(key=lambda x: x["Cost"], reverse=True) | |
| breakdown.append({ | |
| "name": pl_str, | |
| "Cost": round(pl_cost, 2), | |
| "Ads.Sales": round(pl_sales, 2), | |
| "ROAS": round(pl_roas, 4) if pl_roas else None, | |
| "by_funnel": funnel_split, | |
| "models": models, | |
| }) | |
| breakdown.sort(key=lambda x: x["Cost"], reverse=True) | |
| return breakdown | |
| def compute_dashboard_data( | |
| config_path: Path, | |
| start: date, | |
| end: date, | |
| filters: dict[str, str | None] | None = None, | |
| ) -> dict[str, Any]: | |
| """ | |
| 数据看板：返回选定区间 + 筛选条件下的汇总 KPI、趋势、产品线看板、矩阵。 | |
| """ | |
| df = _load_enriched(config_path) | |
| mask = (df["date"] >= pd.Timestamp(start)) & (df["date"] <= pd.Timestamp(end)) | |
| period = df[mask].copy() | |
| if filters: | |
| period = _apply_filters(period, filters) | |
| countries = sorted(period["country"].dropna().unique()) if "country" in period.columns else [] | |
| if period.empty: | |
| return { | |
| "start": str(start), "end": str(end), | |
| "summary": {}, "dates": [], "by_funnel": {}, "by_platform": {}, | |
| "roas": [], "countries": countries, "product_breakdown": [], | |
| "matrix": {}, "matrix_total": {}, | |
| } | |
| summary = _agg_summary(period) | |
| period["ds"] = period["date"].dt.strftime("%Y-%m-%d") | |
| dates = sorted(period["ds"].unique()) | |
| funnels = ["Conversion", "Consideration", "Awareness"] | |
| platforms = sorted(period["platform"].dropna().unique()) if "platform" in period.columns else [] | |
| by_funnel: dict[str, list[float]] = {} | |
| piv_f = period.groupby(["ds", "campaign_type"])["Cost"].sum().unstack(fill_value=0) | |
| for f in funnels: | |
| by_funnel[f] = [round(float(piv_f.loc[d, f]), 2) if d in piv_f.index and f in piv_f.columns else 0 for d in dates] | |
| by_platform: dict[str, list[float]] = {} | |
| piv_p = period.groupby(["ds", "platform"])["Cost"].sum().unstack(fill_value=0) | |
| for p in platforms: | |
| by_platform[p] = [round(float(piv_p.loc[d, p]), 2) if d in piv_p.index and p in piv_p.columns else 0 for d in dates] | |
| conv = period[period["campaign_type"] == "Conversion"] | |
| conv_daily = conv.groupby("ds").agg(cost=("Cost", "sum"), sales=("Ads.Sales", "sum")) | |
| roas_list: list[float | None] = [] | |
| for d in dates: | |
| if d in conv_daily.index and conv_daily.loc[d, "cost"] > 0: | |
| roas_list.append(round(float(conv_daily.loc[d, "sales"] / conv_daily.loc[d, "cost"]), 2)) | |
| else: | |
| roas_list.append(None) | |
| product_breakdown = _build_product_breakdown(period) | |
| matrix_total = _agg_matrix_metrics(period) | |
| matrix: dict[str, Any] = {} | |
| for cty in countries: | |
| c_df = period[period["country"] == cty] | |
| c_node: dict[str, Any] = {"m": _agg_matrix_metrics(c_df), "ch": {}} | |
| plats = sorted(c_df["platform"].dropna().unique()) if "platform" in c_df.columns else [] | |
| for pl in plats: | |
| p_df = c_df[c_df["platform"] == pl] | |
| if p_df.empty: | |
| continue | |
| p_node: dict[str, Any] = {"m": _agg_matrix_metrics(p_df), "ch": {}} | |
| for fn in funnels: | |
| f_df = p_df[p_df["campaign_type"] == fn] | |
| if f_df.empty: | |
| continue | |
| f_node: dict[str, Any] = {"m": _agg_matrix_metrics(f_df), "ch": {}} | |
| for dt in dates: | |
| d_df = f_df[f_df["ds"] == dt] | |
| if d_df.empty: | |
| continue | |
| f_node["ch"][dt] = _agg_matrix_metrics(d_df) | |
| p_node["ch"][fn] = f_node | |
| c_node["ch"][pl] = p_node | |
| matrix[cty] = c_node | |
| return { | |
| "start": str(start), | |
| "end": str(end), | |
| "summary": summary, | |
| "dates": dates, | |
| "by_funnel": by_funnel, | |
| "by_platform": by_platform, | |
| "roas": roas_list, | |
| "countries": countries, | |
| "product_breakdown": product_breakdown, | |
| "matrix": matrix, | |
| "matrix_total": matrix_total, | |
| } | |
| def format_monitor_dingtalk(result: dict) -> tuple[str, str]: | |
| """将监控结果格式化为钉钉 Markdown 消息。""" | |
| dt = result["date"] | |
| ds = result.get("data_status", {}) | |
| alerts = result.get("alerts", []) | |
| summary_fmt = result.get("summary_fmt", {}) | |
| by_platform = result.get("by_platform", {}) | |
| title = f"📡 数据监控 {dt}" | |
| lines = [f"## 📡 数据监控 — {dt}\n"] | |
| if ds.get("status") == "stale": | |
| lines.append(f"> ⚠️ 数据已 {ds['days_behind']} 天未更新（最新: {ds['latest_date']}）\n") | |
| elif ds.get("status") == "missing": | |
| lines.append("> ❌ 无法读取数据\n") | |
| if alerts: | |
| lines.append("### 异常指标\n") | |
| for a in alerts: | |
| icon = "🔴" if a["level"] == "red" else "🟢" if a["level"] == "green" else "🟡" | |
| arrow = "↑" if a["direction"] == "up" else "↓" | |
| lines.append(f"- {icon} **{a['metric']}**: {a['value_fmt']} ({arrow}{abs(a['change_pct'])}% vs 7日均值 {a['avg7_fmt']})") | |
| lines.append("") | |
| else: | |
| lines.append("✅ 各指标均在正常波动范围内\n") | |
| if summary_fmt: | |
| lines.append("### 今日概览\n") | |
| for k, v in summary_fmt.items(): | |
| label = MONITOR_LABEL_ZH.get(k, k) | |
| lines.append(f"- **{label}** {v}") | |
| lines.append("") | |
| if by_platform: | |
| lines.append("### 平台概览\n") | |
| for plat, pd_ in by_platform.items(): | |
| lines.append(f"**{plat}**") | |
| if "Cost" in pd_: | |
| lines.append(f"- **花费** {_fmt_value(pd_['Cost'], '$')}") | |
| if "Ads.Sales" in pd_: | |
| lines.append(f"- **销售额** {_fmt_value(pd_['Ads.Sales'], '$')}") | |
| if "ROAS" in pd_ and pd_["ROAS"]: | |
| lines.append(f"- **ROAS** {pd_['ROAS']:.2f}") | |
| lines.append("") | |
| return title, "\n".join(lines) | |