Blame for src/monitor.py

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