Spaces:
Runtime error
Runtime error
| from __future__ import annotations | |
| """ | |
| P3 ๆๆพๅๆๅผๆ โ L1-L4 ๅๅฑ่ๅใ | |
| L1 ๅธๅบ: Budget Pacing + Revenue vs Target | |
| L2 funnelรไผๅ็ฎๆ : Spend Share + ๅ็ฎๆ ๆ็ๆๆ | |
| L3 ๅนณๅฐ: Spend/Revenue Share + ้็จๆ็ๆๆ | |
| L4 Campaign: ๅไธช campaign ๆ็ป + ็ฏๆฏ | |
| """ | |
| import logging | |
| from datetime import date, timedelta | |
| from pathlib import Path | |
| from typing import Any | |
| import pandas as pd | |
| from .loader import load_config, load_daily_data | |
| from .campaign_parser import enrich_dataframe | |
| from .p3_metrics import compute_group_metrics, get_core_metrics_for_objective | |
| from .p3_config import load_all_p3_config | |
| log = logging.getLogger("ad-agent") | |
| def _safe_div(n: float, d: float) -> float: | |
| if d == 0 or pd.isna(d): | |
| return 0.0 | |
| return n / d | |
| def _filter_period( | |
| df: pd.DataFrame, | |
| period_type: str, | |
| target_year: int | None, | |
| target_period: int | None, | |
| start_date: date | None, | |
| end_date: date | None, | |
| ) -> tuple[pd.DataFrame, str]: | |
| """ๆ นๆฎๅจๆ็ฑปๅ็ญ้ๆฐๆฎ๏ผ่ฟๅ (filtered_df, period_label)ใ""" | |
| if period_type == "range" and start_date and end_date: | |
| mask = (df["date"].dt.date >= start_date) & (df["date"].dt.date <= end_date) | |
| label = f"{start_date} ~ {end_date}" | |
| return df[mask].copy(), label | |
| if period_type == "day" and start_date: | |
| mask = df["date"].dt.date == start_date | |
| label = str(start_date) | |
| return df[mask].copy(), label | |
| if period_type == "month" and target_year and target_period: | |
| cal_year = df["date"].dt.year | |
| mask = (cal_year == target_year) & (df["month"] == target_period) | |
| label = f"{target_year}ๅนด{target_period}ๆ" | |
| return df[mask].copy(), label | |
| # default: week | |
| if target_year and target_period: | |
| mask = (df["year"] == target_year) & (df["week"] == target_period) | |
| label = f"{target_year} W{target_period}" | |
| return df[mask].copy(), label | |
| # latest week | |
| latest_year = df["year"].max() | |
| latest_week = df[df["year"] == latest_year]["week"].max() | |
| mask = (df["year"] == latest_year) & (df["week"] == latest_week) | |
| label = f"{latest_year} W{latest_week}" | |
| return df[mask].copy(), label | |
| def _get_prev_period( | |
| df: pd.DataFrame, | |
| period_type: str, | |
| target_year: int | None, | |
| target_period: int | None, | |
| start_date: date | None, | |
| end_date: date | None, | |
| ) -> tuple[pd.DataFrame, str | None]: | |
| """่ทๅไธไธๆๆฐๆฎ็จไบ็ฏๆฏใ""" | |
| if period_type == "range" and start_date and end_date: | |
| days = (end_date - start_date).days + 1 | |
| prev_end = start_date - timedelta(days=1) | |
| prev_start = prev_end - timedelta(days=days - 1) | |
| return _filter_period(df, "range", None, None, prev_start, prev_end) | |
| if period_type == "day" and start_date: | |
| prev = start_date - timedelta(days=1) | |
| return _filter_period(df, "day", None, None, prev, None) | |
| if period_type == "month" and target_year and target_period: | |
| pm = target_period - 1 | |
| py = target_year | |
| if pm < 1: | |
| pm = 12 | |
| py -= 1 | |
| return _filter_period(df, "month", py, pm, None, None) | |
| # week | |
| weeks = ( | |
| df[["year", "week"]] | |
| .drop_duplicates() | |
| .sort_values(["year", "week"]) | |
| .values.tolist() | |
| ) | |
| curr = [target_year, target_period] if target_year and target_period else None | |
| if not curr and weeks: | |
| curr = weeks[-1] | |
| if curr and curr in weeks: | |
| idx = weeks.index(curr) | |
| if idx > 0: | |
| py, pw = weeks[idx - 1] | |
| return _filter_period(df, "week", py, pw, None, None) | |
| return pd.DataFrame(), None | |
| def _infer_month(period_df: pd.DataFrame) -> int | None: | |
| """ไปๅฝๆๆฐๆฎๆจๆญๆๅฑๆไปฝ๏ผๅไธญไฝๆฅๆ็ๆไปฝ๏ผใ""" | |
| if period_df.empty or "date" not in period_df.columns: | |
| return None | |
| dates = period_df["date"].dropna() | |
| if dates.empty: | |
| return None | |
| mid = dates.sort_values().iloc[len(dates) // 2] | |
| return mid.month | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| # L1 ๅธๅบๅฑ | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| def _build_l1( | |
| curr_df: pd.DataFrame, | |
| prev_df: pd.DataFrame | None, | |
| p3_config: dict, | |
| target_month: int | None, | |
| ) -> list[dict[str, Any]]: | |
| if curr_df.empty or "market" not in curr_df.columns: | |
| return [] | |
| targets = p3_config.get("monthly_targets", {}) | |
| rows = [] | |
| for mkt, grp in curr_df.groupby("market"): | |
| m = compute_group_metrics(grp) | |
| cost = m["Cost"] | |
| revenue = m["Revenue"] | |
| roas = m["ROAS"] | |
| row: dict[str, Any] = { | |
| "market": mkt, | |
| "spend": cost, | |
| "revenue": revenue, | |
| "roas": roas, | |
| "cpa": m["CPA"], | |
| "impressions": m.get("Impressions", 0), | |
| } | |
| mkt_targets = targets.get(mkt, {}).get(target_month, {}) if target_month else {} | |
| budget = mkt_targets.get("dm_budget") | |
| dm_target = mkt_targets.get("dm_target") | |
| target_roas = mkt_targets.get("target_roas") | |
| row["budget"] = budget | |
| row["budget_pacing"] = _safe_div(cost, budget) * 100 if budget else None | |
| row["dm_target"] = dm_target | |
| row["revenue_vs_target"] = _safe_div(revenue, dm_target) * 100 if dm_target else None | |
| row["target_roas"] = target_roas | |
| if prev_df is not None and not prev_df.empty: | |
| prev_grp = prev_df[prev_df["market"] == mkt] if "market" in prev_df.columns else pd.DataFrame() | |
| if not prev_grp.empty: | |
| pm = compute_group_metrics(prev_grp) | |
| row["prev_spend"] = pm["Cost"] | |
| row["prev_revenue"] = pm["Revenue"] | |
| row["prev_roas"] = pm["ROAS"] | |
| row["spend_chg"] = _safe_div(cost - pm["Cost"], pm["Cost"]) * 100 if pm["Cost"] else None | |
| row["revenue_chg"] = _safe_div(revenue - pm["Revenue"], pm["Revenue"]) * 100 if pm["Revenue"] else None | |
| row["roas_chg"] = _safe_div(roas - pm["ROAS"], pm["ROAS"]) * 100 if pm["ROAS"] else None | |
| rows.append(row) | |
| return rows | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| # L2 funnelรไผๅ็ฎๆ ๅฑ | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| def _build_l2( | |
| curr_df: pd.DataFrame, | |
| prev_df: pd.DataFrame | None, | |
| p3_config: dict, | |
| target_month: int | None, | |
| ) -> list[dict[str, Any]]: | |
| needed_cols = {"ad_objective", "audience_type", "objective_group", "funnel"} | |
| if not needed_cols.issubset(curr_df.columns): | |
| return [] | |
| total_cost = curr_df["Cost"].fillna(0).sum() if "Cost" in curr_df.columns else 0 | |
| rows = [] | |
| for obj_group, grp in curr_df.groupby("objective_group"): | |
| objective = grp["ad_objective"].iloc[0] if not grp.empty else None | |
| audience = grp["audience_type"].iloc[0] if not grp.empty else "cold" | |
| funnel = grp["funnel"].iloc[0] if not grp.empty else "" | |
| m = compute_group_metrics(grp, objective=objective) | |
| cost = m["Cost"] | |
| row: dict[str, Any] = { | |
| "objective_group": obj_group, | |
| "ad_objective": objective, | |
| "audience_type": audience, | |
| "funnel": funnel, | |
| "spend": cost, | |
| "spend_share": _safe_div(cost, total_cost) * 100, | |
| **{k: m.get(k) for k in get_core_metrics_for_objective(objective or "")}, | |
| "impressions": m.get("Impressions", 0), | |
| "clicks": m.get("Clicks", 0), | |
| } | |
| if prev_df is not None and not prev_df.empty and "objective_group" in prev_df.columns: | |
| prev_grp = prev_df[prev_df["objective_group"] == obj_group] | |
| if not prev_grp.empty: | |
| pm = compute_group_metrics(prev_grp, objective=objective) | |
| for key in get_core_metrics_for_objective(objective or ""): | |
| curr_val = m.get(key, 0) | |
| prev_val = pm.get(key, 0) | |
| row[f"{key}_chg"] = ( | |
| _safe_div(curr_val - prev_val, prev_val) * 100 | |
| if prev_val else None | |
| ) | |
| rows.append(row) | |
| rows.sort(key=lambda r: r.get("spend", 0), reverse=True) | |
| return rows | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| # L3 ๅนณๅฐๅฑ | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| def _build_l3( | |
| curr_df: pd.DataFrame, | |
| prev_df: pd.DataFrame | None, | |
| p3_config: dict, | |
| target_month: int | None, | |
| ) -> list[dict[str, Any]]: | |
| plat_col = "parsed_platform" if "parsed_platform" in curr_df.columns else "platform" | |
| if plat_col not in curr_df.columns: | |
| return [] | |
| total_cost = curr_df["Cost"].fillna(0).sum() if "Cost" in curr_df.columns else 0 | |
| total_revenue = curr_df["Ads.Sales"].fillna(0).sum() if "Ads.Sales" in curr_df.columns else 0 | |
| channel_budgets = p3_config.get("channel_budgets", {}) | |
| rows = [] | |
| for plat, grp in curr_df.groupby(plat_col): | |
| m = compute_group_metrics(grp) | |
| cost = m["Cost"] | |
| revenue = m["Revenue"] | |
| row: dict[str, Any] = { | |
| "platform": plat, | |
| "spend": cost, | |
| "spend_share": _safe_div(cost, total_cost) * 100, | |
| "revenue": revenue, | |
| "revenue_share": _safe_div(revenue, total_revenue) * 100, | |
| "roas": m["ROAS"], | |
| "impressions": m.get("Impressions", 0), | |
| "cpm": m["CPM"], | |
| "clicks": m["Clicks"], | |
| "ctr": m["CTR"], | |
| "cpc": m["CPC"], | |
| "cpa": m["CPA"], | |
| "conversions": m["Conversions"], | |
| } | |
| # ไธๆธ ้้ข็ฎๅฏน็ ง | |
| us_budgets = channel_budgets.get("US", {}).get("channel_budgets", {}) | |
| if target_month and target_month in us_budgets: | |
| planned = us_budgets[target_month].get(plat) | |
| row["planned_budget"] = planned | |
| row["budget_pacing"] = _safe_div(cost, planned) * 100 if planned else None | |
| if prev_df is not None and not prev_df.empty and plat_col in prev_df.columns: | |
| prev_grp = prev_df[prev_df[plat_col] == plat] | |
| if not prev_grp.empty: | |
| pm = compute_group_metrics(prev_grp) | |
| row["prev_spend"] = pm["Cost"] | |
| row["prev_roas"] = pm["ROAS"] | |
| row["spend_chg"] = _safe_div(cost - pm["Cost"], pm["Cost"]) * 100 if pm["Cost"] else None | |
| row["roas_chg"] = _safe_div(m["ROAS"] - pm["ROAS"], pm["ROAS"]) * 100 if pm["ROAS"] else None | |
| row["cpc_chg"] = _safe_div(m["CPC"] - pm["CPC"], pm["CPC"]) * 100 if pm["CPC"] else None | |
| rows.append(row) | |
| rows.sort(key=lambda r: r.get("spend", 0), reverse=True) | |
| return rows | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| # L4 Campaign ๅฑ | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| def _build_l4( | |
| curr_df: pd.DataFrame, | |
| prev_df: pd.DataFrame | None, | |
| ) -> list[dict[str, Any]]: | |
| if "campaign_name" not in curr_df.columns: | |
| return [] | |
| rows = [] | |
| dim_cols = [ | |
| "campaign_name", "market", "funnel", "parsed_platform", | |
| "ad_objective", "product_line", "audience_type", "objective_group", | |
| ] | |
| existing_dims = [c for c in dim_cols if c in curr_df.columns] | |
| for name, grp in curr_df.groupby("campaign_name"): | |
| objective = grp["ad_objective"].iloc[0] if "ad_objective" in grp.columns else None | |
| m = compute_group_metrics(grp, objective=objective) | |
| row: dict[str, Any] = {"campaign_name": name} | |
| for c in existing_dims: | |
| if c != "campaign_name": | |
| row[c] = grp[c].iloc[0] if c in grp.columns else None | |
| row.update({ | |
| "spend": m["Cost"], | |
| "impressions": m.get("Impressions", 0), | |
| "cpm": m["CPM"], | |
| "clicks": m["Clicks"], | |
| "ctr": m["CTR"], | |
| "cpc": m["CPC"], | |
| "conversions": m["Conversions"], | |
| "revenue": m["Revenue"], | |
| "cpa": m["CPA"], | |
| "roas": m["ROAS"], | |
| }) | |
| if objective: | |
| for key in get_core_metrics_for_objective(objective): | |
| if key not in row: | |
| row[key] = m.get(key) | |
| if prev_df is not None and not prev_df.empty and "campaign_name" in prev_df.columns: | |
| prev_grp = prev_df[prev_df["campaign_name"] == name] | |
| if not prev_grp.empty: | |
| pm = compute_group_metrics(prev_grp, objective=objective) | |
| row["prev_roas"] = pm["ROAS"] | |
| row["roas_chg"] = ( | |
| _safe_div(m["ROAS"] - pm["ROAS"], pm["ROAS"]) * 100 | |
| if pm["ROAS"] else None | |
| ) | |
| row["prev_spend"] = pm["Cost"] | |
| row["spend_chg"] = ( | |
| _safe_div(m["Cost"] - pm["Cost"], pm["Cost"]) * 100 | |
| if pm["Cost"] else None | |
| ) | |
| rows.append(row) | |
| rows.sort(key=lambda r: r.get("spend", 0), reverse=True) | |
| return rows | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| # ไธปๅ ฅๅฃ | |
| # โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ | |
| def run_p3_analysis( | |
| config_path: Path | None = None, | |
| period_type: str = "week", | |
| target_year: int | None = None, | |
| target_period: int | None = None, | |
| start_date: date | None = None, | |
| end_date: date | None = None, | |
| market: str | None = None, | |
| ) -> dict[str, Any]: | |
| """ | |
| P3 ๆๆพๅๆไธปๅ ฅๅฃใ | |
| ๅ ่ฝฝๆฐๆฎ -> ่งฃๆ campaign name -> ๅๅฑ่ๅ -> ่ฟๅ L1-L4 ็ปๆใ | |
| """ | |
| config_path = config_path or (Path(__file__).resolve().parent.parent / "config.yaml") | |
| df = load_daily_data(config_path) | |
| df = enrich_dataframe(df) | |
| if market and "market" in df.columns: | |
| df = df[df["market"] == market].copy() | |
| curr_df, period_label = _filter_period( | |
| df, period_type, target_year, target_period, start_date, end_date | |
| ) | |
| if curr_df.empty: | |
| return { | |
| "period_label": period_label, | |
| "error": f"{period_label} ๆ ๅฏ็จๆฐๆฎ", | |
| "L1": [], "L2": [], "L3": [], "L4": [], | |
| } | |
| prev_df, prev_label = _get_prev_period( | |
| df, period_type, target_year, target_period, start_date, end_date | |
| ) | |
| prev_enriched = enrich_dataframe(prev_df) if prev_df is not None and not prev_df.empty else None | |
| p3_config = load_all_p3_config(config_path) | |
| target_month = _infer_month(curr_df) | |
| l1 = _build_l1(curr_df, prev_enriched, p3_config, target_month) | |
| l2 = _build_l2(curr_df, prev_enriched, p3_config, target_month) | |
| l3 = _build_l3(curr_df, prev_enriched, p3_config, target_month) | |
| l4 = _build_l4(curr_df, prev_enriched) | |
| parse_ok = curr_df["parse_ok"].sum() if "parse_ok" in curr_df.columns else 0 | |
| parse_total = len(curr_df) | |
| return { | |
| "period_label": period_label, | |
| "prev_label": prev_label, | |
| "period_type": period_type, | |
| "target_month": target_month, | |
| "market_filter": market, | |
| "parse_rate": _safe_div(parse_ok, parse_total) * 100 if parse_total else 0, | |
| "row_count": parse_total, | |
| "L1": l1, | |
| "L2": l2, | |
| "L3": l3, | |
| "L4": l4, | |
| "p3_config": p3_config, | |
| } | |