Spaces:
Runtime error
Runtime error
| """ | |
| 预算追踪：从 Google Sheets 读取 Budget 工作表， | |
| 结合实际花费数据计算 pacing 和 KPI 对比。 | |
| Budget 工作表表头（宽格式，每行 = Month × Country）: | |
| Month, Country, SPF_Revenue_Target, DM_Revenue_Target, | |
| Conversion_Budget, Consideration_Budget, Awareness_Budget, | |
| DM_Total_Budget, Target_ROAS, | |
| Conversion_%, Consideration_%, Awareness_% | |
| """ | |
| import calendar | |
| import logging | |
| from datetime import date, datetime | |
| from pathlib import Path | |
| from typing import Any | |
| import pandas as pd | |
| from src.loader import load_config, load_daily_data, _get_spreadsheet | |
| log = logging.getLogger("ad-agent") | |
| FUNNEL_COLS = { | |
| "Conversion": "Conversion_Budget", | |
| "Consideration": "Consideration_Budget", | |
| "Awareness": "Awareness_Budget", | |
| } | |
| FUNNEL_PCT_COLS = { | |
| "Conversion": "Conversion_%", | |
| "Consideration": "Consideration_%", | |
| "Awareness": "Awareness_%", | |
| } | |
| NUMERIC_COLS = [ | |
| "SPF_Revenue_Target", "DM_Revenue_Target", | |
| "Conversion_Budget", "Consideration_Budget", "Awareness_Budget", | |
| "DM_Total_Budget", "Target_ROAS", | |
| ] | |
| PCT_COLS = ["Conversion_%", "Consideration_%", "Awareness_%"] | |
| def _read_sheet(config: dict, name: str) -> pd.DataFrame | None: | |
| try: | |
| sh = _get_spreadsheet(config) | |
| ws = sh.worksheet(name) | |
| rows = ws.get_all_values() | |
| if len(rows) < 2: | |
| return None | |
| return pd.DataFrame(rows[1:], columns=[c.strip() for c in rows[0]]) | |
| except Exception as e: | |
| log.warning(f"Sheet [{name}] 读取失败: {e}") | |
| return None | |
| def load_budget_config(config: dict) -> pd.DataFrame | None: | |
| df = _read_sheet(config, "Budget") | |
| if df is None: | |
| return None | |
| for col in NUMERIC_COLS: | |
| if col in df.columns: | |
| cleaned = df[col].astype(str).str.replace(r"[$,\s]", "", regex=True) | |
| df[col] = pd.to_numeric(cleaned, errors="coerce").fillna(0) | |
| for col in PCT_COLS: | |
| if col in df.columns: | |
| cleaned = df[col].astype(str).str.replace(r"[%\s]", "", regex=True) | |
| df[col] = pd.to_numeric(cleaned, errors="coerce").fillna(0) | |
| return df | |
| def _pacing_status(pacing: float | None) -> str: | |
| if pacing is None: | |
| return "no_data" | |
| if pacing > 1.15: | |
| return "overspending" | |
| if pacing < 0.85: | |
| return "underspending" | |
| return "on_track" | |
| def _safe_cost(df: pd.DataFrame) -> float: | |
| if df.empty or "Cost" not in df.columns: | |
| return 0.0 | |
| return float(df["Cost"].sum()) | |
| def _safe_revenue(df: pd.DataFrame) -> float: | |
| if df.empty or "Ads.Sales" not in df.columns: | |
| return 0.0 | |
| return float(df["Ads.Sales"].sum()) | |
| def _node( | |
| budget: float, | |
| spent: float, | |
| pct_elapsed: float, | |
| days_elapsed: int, | |
| days_in_month: int, | |
| ) -> dict: | |
| pct = spent / budget if budget > 0 else 0 | |
| pacing = pct / pct_elapsed if pct_elapsed > 0 else 0 | |
| projected = spent / days_elapsed * days_in_month if days_elapsed > 0 else 0 | |
| return { | |
| "budget": round(budget, 2), | |
| "spent": round(spent, 2), | |
| "pct_used": round(pct * 100, 1), | |
| "pacing": round(pacing, 2), | |
| "pacing_status": _pacing_status(pacing if budget > 0 else None), | |
| "projected": round(projected, 2), | |
| } | |
| def _compute_roas(actual_df: pd.DataFrame) -> float | None: | |
| if actual_df.empty: | |
| return None | |
| cost = actual_df["Cost"].sum() if "Cost" in actual_df.columns else 0 | |
| sales = actual_df["Ads.Sales"].sum() if "Ads.Sales" in actual_df.columns else 0 | |
| if cost > 0: | |
| return round(sales / cost, 2) | |
| return None | |
| def compute_budget_tracking( | |
| config_path: Path | None = None, | |
| target_month: str | None = None, | |
| ) -> dict[str, Any]: | |
| config = load_config(config_path) | |
| bcfg = config.get("budget", {}) | |
| if not bcfg.get("enabled"): | |
| return {"error": "budget_disabled"} | |
| budget_df = load_budget_config(config) | |
| if budget_df is None or budget_df.empty: | |
| return {"error": "no_budget_data"} | |
| today = date.today() | |
| if target_month: | |
| try: | |
| ym = datetime.strptime(target_month, "%Y-%m") | |
| ty, tm = ym.year, ym.month | |
| except ValueError: | |
| return {"error": "bad_month"} | |
| else: | |
| ty, tm = today.year, today.month | |
| ms = f"{ty}-{tm:02d}" | |
| _, dim = calendar.monthrange(ty, tm) | |
| if ty == today.year and tm == today.month: | |
| de = max(today.day - 1, 1) | |
| elif date(ty, tm, 1) > today: | |
| de = 1 | |
| else: | |
| de = dim | |
| pe = de / dim | |
| mb = budget_df[budget_df["Month"].astype(str).str.strip() == ms] | |
| avail = sorted(budget_df["Month"].astype(str).str.strip().unique().tolist()) | |
| if mb.empty: | |
| return {"error": "no_month_data", "available_months": avail, "month": ms} | |
| try: | |
| adf = load_daily_data(config_path) | |
| except Exception as e: | |
| return {"error": f"data_load_fail: {e}"} | |
| m_start = pd.Timestamp(ty, tm, 1) | |
| m_end = pd.Timestamp(ty, tm, dim) | |
| am = adf[(adf["date"] >= m_start) & (adf["date"] <= m_end)].copy() | |
| # ── Total level ── | |
| tb = float(mb["DM_Total_Budget"].sum()) | |
| ts = _safe_cost(am) | |
| total_rev = _safe_revenue(am) | |
| am_conv = am[am["campaign_type"].astype(str).str.strip() == "Conversion"] if "campaign_type" in am.columns else am | |
| total_roas_actual = _compute_roas(am_conv) | |
| total_spf = float(mb["SPF_Revenue_Target"].sum()) | |
| total_dm_rev = float(mb["DM_Revenue_Target"].sum()) | |
| weights = mb[["Target_ROAS", "DM_Total_Budget"]].copy() | |
| weights = weights[weights["Target_ROAS"] > 0] | |
| if not weights.empty and weights["DM_Total_Budget"].sum() > 0: | |
| total_target_roas = round( | |
| (weights["Target_ROAS"] * weights["DM_Total_Budget"]).sum() | |
| / weights["DM_Total_Budget"].sum(), 2 | |
| ) | |
| else: | |
| total_target_roas = None | |
| nd = _node(tb, ts, pe, de, dim) | |
| total_conv_rev = _safe_revenue(am_conv) | |
| nd.update({ | |
| "month": ms, | |
| "remaining": round(tb - ts, 2), | |
| "days_elapsed": de, | |
| "days_total": dim, | |
| "days_remaining": dim - de, | |
| "pct_elapsed": round(pe * 100, 1), | |
| "available_months": avail, | |
| "revenue_target": { | |
| "spf": round(total_spf, 2), | |
| "dm": round(total_dm_rev, 2), | |
| }, | |
| "actual_revenue": round(total_conv_rev, 2), | |
| "target_roas": total_target_roas, | |
| "kpi": {}, | |
| }) | |
| if total_roas_actual is not None: | |
| kpi_roas: dict[str, Any] = {"actual": total_roas_actual} | |
| if total_target_roas and total_target_roas > 0: | |
| kpi_roas["target"] = total_target_roas | |
| kpi_roas["diff_pct"] = round( | |
| (total_roas_actual - total_target_roas) / total_target_roas * 100, 1 | |
| ) | |
| nd["kpi"]["ROAS"] = kpi_roas | |
| # ── Country → Funnel → Platform ── | |
| hier: dict[str, Any] = {} | |
| for country in sorted(mb["Country"].astype(str).str.strip().unique()): | |
| crow = mb[mb["Country"].astype(str).str.strip() == country] | |
| ca = ( | |
| am[am["country"].astype(str).str.strip() == country] | |
| if "country" in am.columns | |
| else pd.DataFrame() | |
| ) | |
| c_budget = float(crow["DM_Total_Budget"].sum()) | |
| c_spent = _safe_cost(ca) | |
| c_rev = _safe_revenue(ca) | |
| ca_conv = ca[ca["campaign_type"].astype(str).str.strip() == "Conversion"] if not ca.empty and "campaign_type" in ca.columns else ca | |
| c_roas_actual = _compute_roas(ca_conv) | |
| c_target_roas = float(crow["Target_ROAS"].iloc[0]) if len(crow) > 0 else None | |
| cn = _node(c_budget, c_spent, pe, de, dim) | |
| cn["revenue_target"] = { | |
| "spf": round(float(crow["SPF_Revenue_Target"].sum()), 2), | |
| "dm": round(float(crow["DM_Revenue_Target"].sum()), 2), | |
| } | |
| c_conv_rev = _safe_revenue(ca_conv) | |
| cn["actual_revenue"] = round(c_conv_rev, 2) | |
| cn["target_roas"] = round(c_target_roas, 2) if c_target_roas else None | |
| kpi_c: dict[str, Any] = {} | |
| if c_roas_actual is not None: | |
| kpi_entry: dict[str, Any] = {"actual": c_roas_actual} | |
| if c_target_roas and c_target_roas > 0: | |
| kpi_entry["target"] = round(c_target_roas, 2) | |
| kpi_entry["diff_pct"] = round( | |
| (c_roas_actual - c_target_roas) / c_target_roas * 100, 1 | |
| ) | |
| kpi_c["ROAS"] = kpi_entry | |
| cn["kpi"] = kpi_c | |
| # ── Funnel level ── | |
| cn["by_funnel"] = {} | |
| for funnel_name, budget_col in FUNNEL_COLS.items(): | |
| if budget_col not in crow.columns: | |
| continue | |
| f_budget = float(crow[budget_col].sum()) | |
| pct_col = FUNNEL_PCT_COLS.get(funnel_name, "") | |
| f_budget_pct = float(crow[pct_col].iloc[0]) if pct_col in crow.columns and len(crow) > 0 else 0 | |
| fa = ( | |
| ca[ca["campaign_type"].astype(str).str.strip() == funnel_name] | |
| if not ca.empty and "campaign_type" in ca.columns | |
| else pd.DataFrame() | |
| ) | |
| f_spent = _safe_cost(fa) | |
| fn = _node(f_budget, f_spent, pe, de, dim) | |
| fn["budget_pct"] = round(f_budget_pct, 1) | |
| # ── Platform level (from actual data only) ── | |
| fn["by_platform"] = {} | |
| if not fa.empty and "platform" in fa.columns: | |
| for plat in sorted(fa["platform"].astype(str).str.strip().unique()): | |
| pa = fa[fa["platform"].astype(str).str.strip() == plat] | |
| p_spent = _safe_cost(pa) | |
| p_rev = _safe_revenue(pa) | |
| p_roas = _compute_roas(pa) | |
| fn["by_platform"][plat] = { | |
| "spent": round(p_spent, 2), | |
| "revenue": round(p_rev, 2), | |
| "roas": p_roas, | |
| } | |
| cn["by_funnel"][funnel_name] = fn | |
| hier[country] = cn | |
| nd["by_country"] = hier | |
| # ── 日度 pacing 曲线数据 ── | |
| if not am.empty and "date" in am.columns: | |
| daily_spend = am.groupby(am["date"].dt.day)["Cost"].sum().sort_index() | |
| cum_actual = [] | |
| running = 0.0 | |
| pacing_dates = [] | |
| daily_ideal = tb / dim if dim > 0 else 0 | |
| cum_ideal = [] | |
| for day_num in range(1, dim + 1): | |
| pacing_dates.append(f"{ms}-{day_num:02d}") | |
| running += float(daily_spend.get(day_num, 0)) | |
| cum_actual.append(round(running, 2)) | |
| cum_ideal.append(round(daily_ideal * day_num, 2)) | |
| nd["pacing_chart"] = { | |
| "dates": pacing_dates, | |
| "cumulative_actual": cum_actual, | |
| "cumulative_ideal": cum_ideal, | |
| "budget_total": round(tb, 2), | |
| } | |
| return nd | |