""" 日常数据监控引擎：检测指标异常、数据更新状态。 不做业务归因，只客观报告异常波动。 """ 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)