Blame for src/analyzer.py

"""
数字营销专家级分析引擎。
底层使用日度数据，按 Country × Platform × Campaign Type 多维度聚合。
支持周度分析（daily→weekly 聚合）和月度分析（按自然月聚合）。
"""
from pathlib import Path
from typing import Any

import pandas as pd

from datetime import date, datetime, timedelta

from .loader import load_config, load_daily_data, get_available_weeks, get_latest_week
from .ai_analyzer import generate_ai_analysis

log = __import__("logging").getLogger("ad-agent")


def _safe_ai_analysis(
    curr_data, prev_data, wow_total, wow_by_platform, wow_by_campaign, wow_by_country,
    **kwargs,
) -> dict[str, Any]:
    """Try AI analysis first; fall back to rule-based generate_expert_analysis on failure."""
    try:
        result = generate_ai_analysis(
            curr_data, prev_data, wow_total, wow_by_platform, wow_by_campaign, wow_by_country,
            **kwargs,
        )
        if result:
            return result
    except Exception as e:
        log.warning(f"AI 分析不可用，回退到规则引擎: {e}")
    return generate_expert_analysis(
        curr_data, prev_data, wow_by_platform, wow_by_campaign, wow_by_country,
        yoy_data=kwargs.get("yoy_data"),
        yoy_total=kwargs.get("yoy_total"),
        yoy_by_platform=kwargs.get("yoy_by_platform"),
        yoy_by_country=kwargs.get("yoy_by_country"),
    )

ADDITIVE_METRICS = {
    "Cost", "Ads.Sales", "Ads.Sales(1d+7d)", "SPF.Sales",
    "GA4.Sessions", "SPF.Sessions", "LPV", "Impressions",
    "ATC", "Email", "Video Views",
}


def _safe_div(numerator: float, denominator: float) -> float:
    if denominator == 0 or pd.isna(denominator):
        return 0.0
    return numerator / denominator


def _safe_pct_change(curr: float, prev: float) -> float | None:
    if prev == 0 or pd.isna(prev):
        return None
    return (curr - prev) / prev * 100


def _agg_group(group: pd.DataFrame, metrics: list[str]) -> dict[str, float]:
    """
    正确聚合：可加指标直接 sum，比率指标通过还原分子分母重新计算。
    """
    result: dict[str, float] = {}
    g = group.copy()

    for m in ADDITIVE_METRICS:
        if m in g.columns:
            result[m] = g[m].fillna(0).sum()

    if "Impressions" in g.columns and "CTR" in g.columns:
        g["_clicks"] = g["Impressions"].fillna(0) * g["CTR"].fillna(0)
    else:
        g["_clicks"] = 0.0

    if "CVR" in g.columns and "GA4.Sessions" in g.columns:
        g["_conversions"] = g["CVR"].fillna(0) * g["GA4.Sessions"].fillna(0)
    else:
        g["_conversions"] = 0.0

    total_clicks = g["_clicks"].sum()
    total_conversions = g["_conversions"].sum()
    total_cost = result.get("Cost", 0)
    total_sales = result.get("Ads.Sales", 0)
    total_imp = result.get("Impressions", 0)
    total_sessions = result.get("GA4.Sessions", 0)
    total_lpv = result.get("LPV", 0)
    total_atc = result.get("ATC", 0)
    total_email = result.get("Email", 0)
    total_video = result.get("Video Views", 0)

    result["ROAS"] = _safe_div(total_sales, total_cost)
    result["CTR"] = _safe_div(total_clicks, total_imp)
    result["CVR"] = _safe_div(total_conversions, total_sessions)
    result["CPC"] = _safe_div(total_cost, total_clicks)
    result["CPM"] = _safe_div(total_cost, total_imp) * 1000
    result["CPSE"] = _safe_div(total_cost, total_sessions)
    result["CPLPV"] = _safe_div(total_cost, total_lpv)
    result["CP-ATC"] = _safe_div(total_cost, total_atc)
    result["CP-Email"] = _safe_div(total_cost, total_email)
    result["CP-Video Views"] = _safe_div(total_cost, total_video)

    for m, weight_col in [("Bounce Rate", "GA4.Sessions"), ("Average Duration", "GA4.Sessions")]:
        if m in g.columns and weight_col in g.columns:
            w = g[weight_col].fillna(0)
            v = g[m].fillna(0)
            total_w = w.sum()
            result[m] = _safe_div((v * w).sum(), total_w)

    for m in ["50% Play Rate", "100% Play Rate", "Hook Rate"]:
        if m in g.columns and "Video Views" in g.columns:
            w = g["Video Views"].fillna(0)
            v = g[m].fillna(0)
            total_w = w.sum()
            result[m] = _safe_div((v * w).sum(), total_w)

    return result


def aggregate_week_data(
    df: pd.DataFrame,
    year: int,
    week: int,
    metrics: list[str],
) -> dict[str, Any]:
    """
    聚合指定周的日度数据，返回多层结构：
    {
      "total": { metric: value },
      "by_country": { "CA": {...}, "US": {...} },
      "by_platform": { "Google": {...}, "Meta": {...} },
      "by_campaign_type": { "Awareness": {...}, ... },
      "by_country_platform": { "CA-Google": {...}, ... },
      "by_platform_campaign": { "Google-Awareness": {...}, ... },
    }
    """
    wk = df[(df["year"] == year) & (df["week"] == week)]
    if wk.empty:
        return {}

    out: dict[str, Any] = {}
    out["total"] = _agg_group(wk, metrics)

    for dim, col in [
        ("by_country", "country"),
        ("by_platform", "platform"),
        ("by_campaign_type", "campaign_type"),
    ]:
        out[dim] = {}
        for val, grp in wk.groupby(col):
            out[dim][val] = _agg_group(grp, metrics)

    out["by_country_platform"] = {}
    for (ctry, plat), grp in wk.groupby(["country", "platform"]):
        out["by_country_platform"][f"{ctry}-{plat}"] = _agg_group(grp, metrics)

    out["by_platform_campaign"] = {}
    for (plat, ctype), grp in wk.groupby(["platform", "campaign_type"]):
        out["by_platform_campaign"][f"{plat}-{ctype}"] = _agg_group(grp, metrics)

    return out


def compute_wow(
    curr: dict[str, float],
    prev: dict[str, float],
    metrics: list[str],
) -> dict[str, float | None]:
    """计算两期之间的环比变化率(%)。"""
    changes: dict[str, float | None] = {}
    for m in metrics:
        c = curr.get(m, 0)
        p = prev.get(m, 0)
        changes[m] = _safe_pct_change(c, p)
    return changes


def _chg(curr_dict: dict, prev_dict: dict, key: str) -> float | None:
    return _safe_pct_change(curr_dict.get(key, 0), prev_dict.get(key, 0))


def _fmt_chg(val: float | None) -> str:
    if val is None:
        return "—"
    return f"{val:+.1f}%"


def generate_expert_analysis(
    curr: dict[str, Any],
    prev: dict[str, Any] | None,
    wow_by_platform: dict[str, dict] | None,
    wow_by_campaign: dict[str, dict] | None,
    wow_by_country: dict[str, dict] | None,
    *,
    yoy_data: dict[str, Any] | None = None,
    yoy_total: dict[str, float | None] | None = None,
    yoy_by_platform: dict[str, dict] | None = None,
    yoy_by_country: dict[str, dict] | None = None,
) -> dict[str, Any]:
    """
    始终输出完整专家分析，包含：
      summary       - 1~2 句总结评价（含 YoY）
      efficiency    - 投放效率分析
      platform      - 平台策略分析
      funnel        - 漏斗健康度诊断
      market        - 市场表现分析
      alerts        - 异动预警（仅变动较大时出现）
      recommendations - 下一步建议
    """
    tc = curr.get("total", {})
    tp = (prev or {}).get("total", {})
    has_prev = bool(tp)
    ty = (yoy_data or {}).get("total", {})
    has_yoy = bool(ty)
    plats = curr.get("by_platform", {})
    ct = curr.get("by_campaign_type", {})
    countries = curr.get("by_country", {})

    result: dict[str, Any] = {}
    alerts: list[dict[str, str]] = []
    recommendations: list[str] = []

    # 一、总结评价
    cost = tc.get("Cost", 0)
    sales = tc.get("Ads.Sales", 0)
    roas = tc.get("ROAS", 0)

    if has_prev:
        cost_chg = _chg(tc, tp, "Cost")
        sales_chg = _chg(tc, tp, "Ads.Sales")
        roas_chg = _chg(tc, tp, "ROAS")

        yoy_suffix = ""
        if has_yoy:
            yoy_roas_chg = _chg(tc, ty, "ROAS")
            yoy_sales_chg = _chg(tc, ty, "Ads.Sales")
            if yoy_roas_chg is not None:
                yoy_suffix = f" 同比去年同期: Sales {_fmt_chg(yoy_sales_chg)}，ROAS {_fmt_chg(yoy_roas_chg)}。"

        if roas_chg is not None and roas_chg > 5:
            summary = f"本期整体表现向好，花费 ${cost:,.0f}（{_fmt_chg(cost_chg)}），Sales ${sales:,.0f}（{_fmt_chg(sales_chg)}），ROAS 提升至 {roas:.2f}（{_fmt_chg(roas_chg)}），投放效率改善。{yoy_suffix}"
        elif roas_chg is not None and roas_chg < -5:
            summary = f"本期投放效率承压，花费 ${cost:,.0f}（{_fmt_chg(cost_chg)}），Sales ${sales:,.0f}（{_fmt_chg(sales_chg)}），ROAS 降至 {roas:.2f}（{_fmt_chg(roas_chg)}），需关注。{yoy_suffix}"
        else:
            summary = f"本期整体表现平稳，花费 ${cost:,.0f}（{_fmt_chg(cost_chg)}），Sales ${sales:,.0f}（{_fmt_chg(sales_chg)}），ROAS {roas:.2f}（{_fmt_chg(roas_chg)}）。{yoy_suffix}"
    else:
        summary = f"花费 ${cost:,.0f}，Sales ${sales:,.0f}，ROAS {roas:.2f}。"
    result["summary"] = summary

    # 二、投放效率分析
    eff_lines: list[str] = []
    if has_prev:
        if cost_chg is not None and sales_chg is not None:
            if cost_chg < 0 and sales_chg > 0:
                eff_lines.append(f"花费下降 {abs(cost_chg):.1f}% 的同时 Sales 增长 {sales_chg:.1f}%，边际效率提升，预算使用更精准。")
            elif cost_chg > 0 and sales_chg > cost_chg:
                eff_lines.append(f"花费增长 {cost_chg:.1f}%，但 Sales 增幅更大（{sales_chg:.1f}%），规模扩张有效。")
            elif cost_chg > 0 and sales_chg < cost_chg:
                eff_lines.append(f"花费增长 {cost_chg:.1f}% 但 Sales 增幅仅 {sales_chg:.1f}%，边际回报递减，需评估扩量策略。")
                recommendations.append("审视近期增投的广告组/受众，暂停 ROAS 低于均值的投放")
            elif cost_chg < 0 and sales_chg < cost_chg:
                eff_lines.append(f"花费缩减 {abs(cost_chg):.1f}%，Sales 降幅更大（{sales_chg:.1f}%），缩量可能砍到了高效投放。")
                recommendations.append("排查被缩减的广告是否包含高效计划，避免误伤")
        cpc_chg = _chg(tc, tp, "CPC")
        cpm_chg = _chg(tc, tp, "CPM")
        if cpc_chg is not None:
            if cpc_chg > 10:
                eff_lines.append(f"CPC 上涨 {cpc_chg:.1f}%，流量获取成本走高，关注竞争环境和受众新鲜度。")
            elif cpc_chg < -10:
                eff_lines.append(f"CPC 下降 {abs(cpc_chg):.1f}%，流量成本改善。")
            else:
                eff_lines.append(f"CPC 变动 {_fmt_chg(cpc_chg)}，流量成本基本稳定。")
    if not eff_lines:
        eff_lines.append(f"ROAS {roas:.2f}，CPC ${tc.get('CPC', 0):.2f}，CPM ${tc.get('CPM', 0):.2f}。")
    result["efficiency"] = eff_lines

    # 三、平台策略分析
    plat_lines: list[str] = []
    if "Google" in plats and "Meta" in plats:
        g = plats["Google"]
        m = plats["Meta"]
        g_cost, m_cost = g.get("Cost", 0), m.get("Cost", 0)
        g_roas, m_roas = g.get("ROAS", 0), m.get("ROAS", 0)
        total_c = g_cost + m_cost
        g_share = g_cost / total_c * 100 if total_c else 0

        better = "Google" if g_roas > m_roas else "Meta"
        worse = "Meta" if better == "Google" else "Google"
        b_roas = g_roas if better == "Google" else m_roas
        w_roas = m_roas if better == "Google" else g_roas
        b_share = g_share if better == "Google" else 100 - g_share

        plat_lines.append(f"预算分配: Google {g_share:.0f}% / Meta {100 - g_share:.0f}%。")
        plat_lines.append(f"{better} ROAS（{b_roas:.2f}）优于 {worse}（{w_roas:.2f}）。")

        if has_prev and wow_by_platform:
            for pn in ["Google", "Meta"]:
                pw = (wow_by_platform or {}).get(pn, {})
                p_roas_chg = pw.get("ROAS")
                p_sales_chg = pw.get("Ads.Sales")
                p_ctr_chg = pw.get("CTR")
                if p_roas_chg is not None:
                    if p_sales_chg is not None and p_sales_chg > 15:
                        plat_lines.append(f"{pn} Sales 增长 {p_sales_chg:.1f}%，ROAS {_fmt_chg(p_roas_chg)}，增长动能强劲。")
                        recommendations.append(f"{pn} 表现强劲，可测试在该平台小幅扩量，观察边际 ROAS 是否稳定")
                    elif p_roas_chg < -10:
                        plat_lines.append(f"{pn} ROAS 下降 {abs(p_roas_chg):.1f}%，效率走低，需审视素材和受众策略。")
                        recommendations.append(f"排查 {pn} 近期素材表现，考虑更新创意或调整受众定向")
                    elif p_roas_chg > 10:
                        plat_lines.append(f"{pn} ROAS 上升 {p_roas_chg:.1f}%，投放效率改善。")
                if p_ctr_chg is not None and p_ctr_chg < -10:
                    plat_lines.append(f"{pn} CTR 下降 {abs(p_ctr_chg):.1f}%，注意素材新鲜度。")
                    recommendations.append(f"更新 {pn} 广告创意，测试新的素材方向")

        if b_share < 40:
            plat_lines.append(f"注意: {better} ROAS 更高但预算占比仅 {b_share:.0f}%，存在效率错配。")
            recommendations.append(f"考虑适度向 {better} 倾斜预算，测试增投后的边际 ROAS 表现")

        if yoy_by_platform:
            for pn in ["Google", "Meta"]:
                py = yoy_by_platform.get(pn, {})
                yoy_r = py.get("ROAS")
                yoy_s = py.get("Ads.Sales")
                if yoy_r is not None:
                    plat_lines.append(f"{pn} 同比去年: ROAS {_fmt_chg(yoy_r)}，Sales {_fmt_chg(yoy_s)}。")
    result["platform"] = plat_lines

    # 四、漏斗健康度
    funnel_lines: list[str] = []
    aw = ct.get("Awareness", {})
    co = ct.get("Consideration", {})
    cv = ct.get("Conversion", {})
    if cv:
        cv_roas = cv.get("ROAS", 0)
        cv_cost = cv.get("Cost", 0)
        cv_share = cv_cost / cost * 100 if cost else 0
        funnel_lines.append(f"Conversion: 贡献核心收入，ROAS {cv_roas:.2f}，占总预算 {cv_share:.0f}%。")
    if co:
        co_roas = co.get("ROAS", 0)
        co_atc = co.get("ATC", 0)
        funnel_lines.append(f"Consideration: ROAS {co_roas:.2f}，ATC {co_atc:,.0f}，负责引流和加购。")
    if aw:
        aw_cost = aw.get("Cost", 0)
        aw_imp = aw.get("Impressions", 0)
        funnel_lines.append(f"Awareness: 花费 ${aw_cost:,.0f}，曝光 {aw_imp:,.0f}，品牌曝光职能。")

    if has_prev:
        ctr_chg = _chg(tc, tp, "CTR")
        cvr_chg = _chg(tc, tp, "CVR")
        if ctr_chg is not None and cvr_chg is not None:
            if ctr_chg > 5 and cvr_chg < -5:
                funnel_lines.append(f"漏斗诊断: CTR 上升（{_fmt_chg(ctr_chg)}）但 CVR 下降（{_fmt_chg(cvr_chg)}），广告吸引了更多点击但落地页承接不足。")
                recommendations.append("优先优化落地页体验（加载速度、首屏内容、CTA 按钮），检查受众匹配度")
                alerts.append({"level": "critical", "title": "漏斗断裂风险", "detail": f"CTR {_fmt_chg(ctr_chg)} 但 CVR {_fmt_chg(cvr_chg)}，落地页转化未能承接广告引流"})
            elif ctr_chg < -5 and cvr_chg > 5:
                funnel_lines.append(f"漏斗诊断: CTR 下降（{_fmt_chg(ctr_chg)}）但 CVR 上升（{_fmt_chg(cvr_chg)}），流量更精准，可在此基础上扩量。")
            elif ctr_chg < -10:
                funnel_lines.append(f"漏斗诊断: CTR 下降 {abs(ctr_chg):.1f}%，注意素材是否疲劳。")
                recommendations.append("更新广告素材，测试新的创意方向")
            else:
                funnel_lines.append(f"漏斗诊断: CTR {_fmt_chg(ctr_chg)}，CVR {_fmt_chg(cvr_chg)}，漏斗整体健康。")

        atc_chg = _chg(tc, tp, "ATC")
        if atc_chg is not None:
            if atc_chg < -15:
                funnel_lines.append(f"加购量（ATC）下降 {abs(atc_chg):.1f}%，中间漏斗承压，购物意向减弱。")
                recommendations.append("检查商品竞争力、库存状态与价格策略")
                alerts.append({"level": "warning", "title": "加购量下滑", "detail": f"ATC {_fmt_chg(atc_chg)}，购物意向减弱"})
            elif atc_chg > 15:
                funnel_lines.append(f"加购量（ATC）增长 {atc_chg:.1f}%，购物意向增强。")
            else:
                funnel_lines.append(f"加购量（ATC）变动 {_fmt_chg(atc_chg)}，基本稳定。")
    result["funnel"] = funnel_lines

    # 五、市场表现
    market_lines: list[str] = []
    if len(countries) > 1:
        for cn, cm in countries.items():
            c_cost = cm.get("Cost", 0)
            c_roas = cm.get("ROAS", 0)
            c_share = c_cost / cost * 100 if cost else 0
            line = f"{cn}: ROAS {c_roas:.2f}，预算占比 {c_share:.0f}%"
            if has_prev and wow_by_country and cn in wow_by_country:
                c_roas_chg = wow_by_country[cn].get("ROAS")
                if c_roas_chg is not None:
                    line += f"，环比 {_fmt_chg(c_roas_chg)}"
                    if c_roas_chg < -10:
                        alerts.append({"level": "warning", "title": f"{cn} 市场效率下滑", "detail": f"{cn} ROAS 环比 {_fmt_chg(c_roas_chg)}，需关注该市场竞争环境与受众变化"})
                    elif c_roas_chg > 15:
                        alerts.append({"level": "positive", "title": f"{cn} 市场表现突出", "detail": f"{cn} ROAS 环比 {_fmt_chg(c_roas_chg)}，可考虑加大该市场投入"})
            if yoy_by_country and cn in yoy_by_country:
                c_yoy_roas = yoy_by_country[cn].get("ROAS")
                c_yoy_sales = yoy_by_country[cn].get("Ads.Sales")
                if c_yoy_roas is not None:
                    line += f"，同比 ROAS {_fmt_chg(c_yoy_roas)}"
                    if c_yoy_sales is not None:
                        line += f" / Sales {_fmt_chg(c_yoy_sales)}"
            market_lines.append(line + "。")
    result["market"] = market_lines

    # 六、ROAS 级别异动预警
    if has_prev:
        roas_chg = _chg(tc, tp, "ROAS")
        if roas_chg is not None:
            if roas_chg <= -20:
                alerts.append({"level": "critical", "title": "ROAS 大幅下降",
                    "detail": f"ROAS {tp['ROAS']:.2f} -> {tc['ROAS']:.2f}（{roas_chg:+.1f}%），需立即排查受众/素材/出价策略"})
            elif roas_chg <= -10:
                alerts.append({"level": "warning", "title": "ROAS 走低",
                    "detail": f"ROAS {tp['ROAS']:.2f} -> {tc['ROAS']:.2f}（{roas_chg:+.1f}%），建议审视近期调整"})

        if cost_chg is not None and cost_chg > 20 and (sales_chg is None or sales_chg < cost_chg * 0.5):
            alerts.append({"level": "critical", "title": "花费激增回报不足",
                "detail": f"Cost {_fmt_chg(cost_chg)} 但 Sales {_fmt_chg(sales_chg)}，边际效率下降"})
        elif cost_chg is not None and cost_chg < -20:
            alerts.append({"level": "warning", "title": "花费大幅缩减",
                "detail": f"Cost {_fmt_chg(cost_chg)}，确认是主动调控还是投放异常"})

        if wow_by_campaign:
            cv_wow = wow_by_campaign.get("Conversion", {})
            cv_roas_chg = cv_wow.get("ROAS")
            if cv_roas_chg is not None and cv_roas_chg < -10:
                alerts.append({"level": "critical", "title": "Conversion ROAS 下滑",
                    "detail": f"核心转化广告 ROAS {_fmt_chg(cv_roas_chg)}，需排查"})

        sess_chg = _chg(tc, tp, "GA4.Sessions")
        if sess_chg is not None and sess_chg < -10:
            alerts.append({"level": "warning", "title": "站点流量下降",
                "detail": f"Sessions {_fmt_chg(sess_chg)}，确认投放覆盖是否不足"})

    # YoY 同比预警
    if has_yoy:
        yoy_roas_chg = _chg(tc, ty, "ROAS")
        yoy_sales_chg = _chg(tc, ty, "Ads.Sales")
        if yoy_roas_chg is not None:
            if yoy_roas_chg < -20:
                alerts.append({"level": "critical", "title": "同比 ROAS 大幅落后",
                    "detail": f"较去年同期 ROAS {_fmt_chg(yoy_roas_chg)}，Sales {_fmt_chg(yoy_sales_chg)}，需审视整体策略"})
            elif yoy_roas_chg < -10:
                alerts.append({"level": "warning", "title": "同比 ROAS 下滑",
                    "detail": f"较去年同期 ROAS {_fmt_chg(yoy_roas_chg)}，注意效率是否持续走低"})
            elif yoy_roas_chg > 20:
                alerts.append({"level": "positive", "title": "同比 ROAS 大幅提升",
                    "detail": f"较去年同期 ROAS {_fmt_chg(yoy_roas_chg)}，Sales {_fmt_chg(yoy_sales_chg)}，年度增长强劲"})

    seen = set()
    unique_alerts = []
    for a in alerts:
        key = a["title"]
        if key not in seen:
            seen.add(key)
            unique_alerts.append(a)
    level_order = {"critical": 0, "warning": 1, "positive": 2}
    unique_alerts.sort(key=lambda a: level_order.get(a["level"], 9))
    result["alerts"] = unique_alerts

    # 七、建议
    if not recommendations:
        if has_prev and roas_chg is not None and roas_chg > 0:
            recommendations.append("当前投放效率向好，建议维持现有策略，可小幅测试扩量")
        else:
            recommendations.append("保持现有投放节奏，持续监控核心指标变化")
    result["recommendations"] = recommendations

    return result


def _apply_platform_filter(df: pd.DataFrame, platform: str | None) -> pd.DataFrame:
    if platform and "platform" in df.columns:
        return df[df["platform"] == platform].copy()
    return df


def run_analysis(
    config_path: Path | None = None,
    target_year: int | None = None,
    target_week: int | None = None,
    trend_table: list[dict] | None = None,
    memory_context: str | None = None,
    business_context: str | None = None,
    platform: str | None = None,
) -> dict[str, Any]:
    """
    执行完整周度分析，返回结构化报告数据。
    底层从日度数据聚合到周度。
    若不指定 target_year/target_week 则分析最新一周。
    """
    config = load_config(config_path)
    metrics = config.get("metrics", [])
    core_metrics = config.get("core_metrics", metrics[:9])
    df = _apply_platform_filter(load_daily_data(config_path), platform)

    if target_year and target_week:
        curr_year, curr_week = target_year, target_week
    else:
        curr_year, curr_week = get_latest_week(df)

    weeks = get_available_weeks(df)
    curr_idx = None
    for i, (y, w) in enumerate(weeks):
        if y == curr_year and w == curr_week:
            curr_idx = i
            break

    prev_year, prev_week = (None, None)
    if curr_idx is not None and curr_idx > 0:
        prev_year, prev_week = weeks[curr_idx - 1]

    curr_data = aggregate_week_data(df, curr_year, curr_week, metrics)
    prev_data = aggregate_week_data(df, prev_year, prev_week, metrics) if prev_year else None

    wow_total = None
    wow_by_platform = None
    wow_by_campaign = None
    wow_by_country = None
    if prev_data:
        wow_total = compute_wow(curr_data.get("total", {}), prev_data.get("total", {}), metrics)
        wow_by_platform = _compute_dim_wow(curr_data, prev_data, "by_platform", metrics)
        wow_by_campaign = _compute_dim_wow(curr_data, prev_data, "by_campaign_type", metrics)
        wow_by_country = _compute_dim_wow(curr_data, prev_data, "by_country", metrics)

    # YoY — 去年同周
    yoy_year = curr_year - 1
    yoy_week = curr_week
    yoy_period = None
    yoy_data = None
    yoy_total = None
    yoy_by_platform = None
    yoy_by_campaign = None
    yoy_by_country = None
    if (yoy_year, yoy_week) in set(weeks):
        yoy_period = {"year": yoy_year, "week": yoy_week}
        yoy_data = aggregate_week_data(df, yoy_year, yoy_week, metrics)
        if yoy_data:
            yoy_total = compute_wow(curr_data.get("total", {}), yoy_data.get("total", {}), metrics)
            yoy_by_platform = _compute_dim_wow(curr_data, yoy_data, "by_platform", metrics)
            yoy_by_campaign = _compute_dim_wow(curr_data, yoy_data, "by_campaign_type", metrics)
            yoy_by_country = _compute_dim_wow(curr_data, yoy_data, "by_country", metrics)

    expert = _safe_ai_analysis(
        curr_data, prev_data, wow_total, wow_by_platform, wow_by_campaign, wow_by_country,
        yoy_data=yoy_data, yoy_total=yoy_total,
        yoy_by_platform=yoy_by_platform, yoy_by_country=yoy_by_country,
        period_label=f"{curr_year} W{curr_week}",
        trend_table=trend_table, memory_context=memory_context,
        business_context=business_context,
    )

    return {
        "current_period": {"year": curr_year, "week": curr_week},
        "previous_period": {"year": prev_year, "week": prev_week} if prev_year else None,
        "yoy_period": yoy_period,
        "current": curr_data,
        "previous": prev_data,
        "yoy": yoy_data,
        "wow_total": wow_total,
        "wow_by_platform": wow_by_platform,
        "wow_by_campaign": wow_by_campaign,
        "wow_by_country": wow_by_country,
        "yoy_total": yoy_total,
        "yoy_by_platform": yoy_by_platform,
        "yoy_by_campaign": yoy_by_campaign,
        "yoy_by_country": yoy_by_country,
        "expert": expert,
        "core_metrics": core_metrics,
        "all_metrics": metrics,
    }


def _compute_dim_wow(
    curr_data: dict[str, Any],
    prev_data: dict[str, Any],
    dim_key: str,
    metrics: list[str],
) -> dict[str, dict]:
    result = {}
    curr_dim = curr_data.get(dim_key, {})
    prev_dim = prev_data.get(dim_key, {})
    for key in curr_dim:
        if key in prev_dim:
            result[key] = compute_wow(curr_dim[key], prev_dim[key], metrics)
    return result


# ─────────────────────────────────────────────
# 月度分析 — 基于日度数据按自然月聚合
# ─────────────────────────────────────────────

def aggregate_month_data(
    df: pd.DataFrame,
    year: int,
    month: int,
    metrics: list[str],
) -> dict[str, Any]:
    """对指定年月的日度数据做聚合（与 aggregate_week_data 结构一致）。"""
    cal_year = df["date"].dt.year
    subset = df[(cal_year == year) & (df["month"] == month)]
    if subset.empty:
        return {}
    out: dict[str, Any] = {}
    out["total"] = _agg_group(subset, metrics)
    for dim, col in [
        ("by_country", "country"),
        ("by_platform", "platform"),
        ("by_campaign_type", "campaign_type"),
    ]:
        out[dim] = {}
        for val, grp in subset.groupby(col):
            out[dim][val] = _agg_group(grp, metrics)
    return out


def run_monthly_analysis(
    config_path: Path | None = None,
    target_year: int = 2025,
    target_month: int = 12,
    trend_table: list[dict] | None = None,
    memory_context: str | None = None,
    business_context: str | None = None,
    platform: str | None = None,
) -> dict[str, Any]:
    """
    月度汇总分析。基于日度数据按自然月聚合，与上月和去年同月对比。
    """
    config = load_config(config_path)
    metrics = config.get("metrics", [])
    core_metrics = config.get("core_metrics", metrics[:9])
    df = _apply_platform_filter(load_daily_data(config_path), platform)

    curr_data = aggregate_month_data(df, target_year, target_month, metrics)
    if not curr_data:
        return {
            "period_label": f"{target_year}年{target_month}月",
            "current": {},
            "previous": None,
            "wow_total": None,
            "expert": {},
            "core_metrics": core_metrics,
            "all_metrics": metrics,
            "error": f"{target_year}年{target_month}月 无可用数据",
        }

    # 上月
    prev_month = target_month - 1
    prev_year = target_year
    if prev_month < 1:
        prev_month = 12
        prev_year -= 1

    prev_data = aggregate_month_data(df, prev_year, prev_month, metrics)
    prev_data = prev_data if prev_data else None

    # MoM (环比)
    wow_total = None
    if prev_data and curr_data.get("total") and prev_data.get("total"):
        wow_total = compute_wow(curr_data["total"], prev_data["total"], metrics)
    wow_by_platform = _compute_dim_wow(curr_data, prev_data, "by_platform", metrics) if prev_data else None
    wow_by_campaign = _compute_dim_wow(curr_data, prev_data, "by_campaign_type", metrics) if prev_data else None
    wow_by_country = _compute_dim_wow(curr_data, prev_data, "by_country", metrics) if prev_data else None

    # YoY — 去年同月
    yoy_year = target_year - 1
    yoy_data = aggregate_month_data(df, yoy_year, target_month, metrics)
    yoy_data = yoy_data if yoy_data else None
    yoy_total = None
    yoy_by_platform = None
    yoy_by_campaign = None
    yoy_by_country = None
    yoy_label = None
    if yoy_data and curr_data.get("total") and yoy_data.get("total"):
        yoy_label = f"{yoy_year}年{target_month}月"
        yoy_total = compute_wow(curr_data["total"], yoy_data["total"], metrics)
        yoy_by_platform = _compute_dim_wow(curr_data, yoy_data, "by_platform", metrics)
        yoy_by_campaign = _compute_dim_wow(curr_data, yoy_data, "by_campaign_type", metrics)
        yoy_by_country = _compute_dim_wow(curr_data, yoy_data, "by_country", metrics)

    expert = _safe_ai_analysis(
        curr_data, prev_data, wow_total, wow_by_platform, wow_by_campaign, wow_by_country,
        yoy_data=yoy_data, yoy_total=yoy_total,
        yoy_by_platform=yoy_by_platform, yoy_by_country=yoy_by_country,
        period_label=f"{target_year}年{target_month}月",
        trend_table=trend_table, memory_context=memory_context,
        business_context=business_context,
    ) if prev_data else {}

    prev_label = f"{prev_year}年{prev_month}月" if prev_data else None

    return {
        "period_label": f"{target_year}年{target_month}月",
        "prev_label": prev_label,
        "yoy_label": yoy_label,
        "current": curr_data,
        "previous": prev_data,
        "yoy": yoy_data,
        "wow_total": wow_total,
        "wow_by_platform": wow_by_platform,
        "wow_by_campaign": wow_by_campaign,
        "wow_by_country": wow_by_country,
        "yoy_total": yoy_total,
        "yoy_by_platform": yoy_by_platform,
        "yoy_by_campaign": yoy_by_campaign,
        "yoy_by_country": yoy_by_country,
        "expert": expert,
        "core_metrics": core_metrics,
        "all_metrics": metrics,
    }


# ─────────────────────────────────────────────
# 日度分析 — 单日数据，环比前一日，同比去年同日
# ─────────────────────────────────────────────

def aggregate_day_data(
    df: pd.DataFrame,
    target_date: date,
    metrics: list[str],
) -> dict[str, Any]:
    """对指定日期的数据做聚合（结构同 aggregate_week_data）。"""
    subset = df[df["date"].dt.date == target_date]
    if subset.empty:
        return {}
    out: dict[str, Any] = {}
    out["total"] = _agg_group(subset, metrics)
    for dim, col in [
        ("by_country", "country"),
        ("by_platform", "platform"),
        ("by_campaign_type", "campaign_type"),
    ]:
        out[dim] = {}
        for val, grp in subset.groupby(col):
            out[dim][val] = _agg_group(grp, metrics)
    return out


def run_daily_analysis(
    config_path: Path | None = None,
    target_date: date | None = None,
    trend_table: list[dict] | None = None,
    memory_context: str | None = None,
    business_context: str | None = None,
    platform: str | None = None,
) -> dict[str, Any]:
    """
    日度分析入口。
    环比: 前一日 (DoD)
    同比: 去年同日 (YoY)
    """
    config = load_config(config_path)
    metrics = config.get("metrics", [])
    core_metrics = config.get("core_metrics", metrics[:9])
    df = _apply_platform_filter(load_daily_data(config_path), platform)

    if target_date is None:
        target_date = df["date"].dt.date.max()

    date_str = target_date.strftime("%Y-%m-%d")
    curr_data = aggregate_day_data(df, target_date, metrics)
    if not curr_data:
        return {
            "period_label": date_str,
            "current": {},
            "previous": None,
            "wow_total": None,
            "expert": {},
            "core_metrics": core_metrics,
            "all_metrics": metrics,
            "error": f"{date_str} 无可用数据",
        }

    prev_date = target_date - timedelta(days=1)
    prev_data = aggregate_day_data(df, prev_date, metrics)
    prev_data = prev_data if prev_data else None

    dod_total = None
    dod_by_platform = None
    dod_by_campaign = None
    dod_by_country = None
    if prev_data and curr_data.get("total") and prev_data.get("total"):
        dod_total = compute_wow(curr_data["total"], prev_data["total"], metrics)
        dod_by_platform = _compute_dim_wow(curr_data, prev_data, "by_platform", metrics)
        dod_by_campaign = _compute_dim_wow(curr_data, prev_data, "by_campaign_type", metrics)
        dod_by_country = _compute_dim_wow(curr_data, prev_data, "by_country", metrics)

    yoy_date = target_date.replace(year=target_date.year - 1)
    yoy_data = aggregate_day_data(df, yoy_date, metrics)
    yoy_data = yoy_data if yoy_data else None
    yoy_total = None
    yoy_by_platform = None
    yoy_by_country = None
    yoy_label = None
    if yoy_data and curr_data.get("total") and yoy_data.get("total"):
        yoy_label = yoy_date.strftime("%Y-%m-%d")
        yoy_total = compute_wow(curr_data["total"], yoy_data["total"], metrics)
        yoy_by_platform = _compute_dim_wow(curr_data, yoy_data, "by_platform", metrics)
        yoy_by_country = _compute_dim_wow(curr_data, yoy_data, "by_country", metrics)

    expert = _safe_ai_analysis(
        curr_data, prev_data, dod_total, dod_by_platform, dod_by_campaign, dod_by_country,
        yoy_data=yoy_data, yoy_total=yoy_total,
        yoy_by_platform=yoy_by_platform, yoy_by_country=yoy_by_country,
        period_label=date_str,
        trend_table=trend_table, memory_context=memory_context,
        business_context=business_context,
    ) if prev_data else {}

    prev_label = prev_date.strftime("%Y-%m-%d") if prev_data else None

    return {
        "period_label": date_str,
        "prev_label": prev_label,
        "yoy_label": yoy_label,
        "current": curr_data,
        "previous": prev_data,
        "yoy": yoy_data,
        "wow_total": dod_total,
        "wow_by_platform": dod_by_platform,
        "wow_by_campaign": dod_by_campaign,
        "wow_by_country": dod_by_country,
        "yoy_total": yoy_total,
        "yoy_by_platform": yoy_by_platform,
        "yoy_by_country": yoy_by_country,
        "expert": expert,
        "core_metrics": core_metrics,
        "all_metrics": metrics,
    }


# ─────────────────────────────────────────────
# 自定义时间段分析
# ─────────────────────────────────────────────

def aggregate_range_data(
    df: pd.DataFrame,
    start_date: date,
    end_date: date,
    metrics: list[str],
) -> dict[str, Any]:
    """对指定日期范围 [start, end] 的数据做聚合。"""
    mask = (df["date"].dt.date >= start_date) & (df["date"].dt.date <= end_date)
    subset = df[mask]
    if subset.empty:
        return {}
    out: dict[str, Any] = {}
    out["total"] = _agg_group(subset, metrics)
    for dim, col in [
        ("by_country", "country"),
        ("by_platform", "platform"),
        ("by_campaign_type", "campaign_type"),
    ]:
        out[dim] = {}
        for val, grp in subset.groupby(col):
            out[dim][val] = _agg_group(grp, metrics)
    return out


def run_range_analysis(
    config_path: Path | None = None,
    start_date: date | None = None,
    end_date: date | None = None,
    trend_table: list[dict] | None = None,
    memory_context: str | None = None,
    business_context: str | None = None,
    platform: str | None = None,
) -> dict[str, Any]:
    """
    自定义时间段分析。
    环比: 等长的前一段 (PoP)
    同比: 去年同日期段 (YoY)
    """
    config = load_config(config_path)
    metrics = config.get("metrics", [])
    core_metrics = config.get("core_metrics", metrics[:9])
    df = _apply_platform_filter(load_daily_data(config_path), platform)

    if start_date is None or end_date is None:
        return {"error": "需要提供起止日期"}

    if start_date > end_date:
        start_date, end_date = end_date, start_date

    days_count = (end_date - start_date).days + 1
    period_label = f"{start_date.strftime('%Y-%m-%d')} ~ {end_date.strftime('%Y-%m-%d')}"

    curr_data = aggregate_range_data(df, start_date, end_date, metrics)
    if not curr_data:
        return {
            "period_label": period_label,
            "current": {},
            "previous": None,
            "wow_total": None,
            "expert": {},
            "core_metrics": core_metrics,
            "all_metrics": metrics,
            "error": f"{period_label} 无可用数据",
        }

    prev_end = start_date - timedelta(days=1)
    prev_start = prev_end - timedelta(days=days_count - 1)
    prev_data = aggregate_range_data(df, prev_start, prev_end, metrics)
    prev_data = prev_data if prev_data else None

    pop_total = None
    pop_by_platform = None
    pop_by_campaign = None
    pop_by_country = None
    if prev_data and curr_data.get("total") and prev_data.get("total"):
        pop_total = compute_wow(curr_data["total"], prev_data["total"], metrics)
        pop_by_platform = _compute_dim_wow(curr_data, prev_data, "by_platform", metrics)
        pop_by_campaign = _compute_dim_wow(curr_data, prev_data, "by_campaign_type", metrics)
        pop_by_country = _compute_dim_wow(curr_data, prev_data, "by_country", metrics)

    try:
        yoy_start = start_date.replace(year=start_date.year - 1)
        yoy_end = end_date.replace(year=end_date.year - 1)
    except ValueError:
        yoy_start = yoy_end = None

    yoy_data = None
    yoy_total = None
    yoy_by_platform = None
    yoy_by_campaign = None
    yoy_by_country = None
    yoy_label = None
    if yoy_start and yoy_end:
        yoy_data = aggregate_range_data(df, yoy_start, yoy_end, metrics)
        yoy_data = yoy_data if yoy_data else None
        if yoy_data and curr_data.get("total") and yoy_data.get("total"):
            yoy_label = f"{yoy_start.strftime('%Y-%m-%d')} ~ {yoy_end.strftime('%Y-%m-%d')}"
            yoy_total = compute_wow(curr_data["total"], yoy_data["total"], metrics)
            yoy_by_platform = _compute_dim_wow(curr_data, yoy_data, "by_platform", metrics)
            yoy_by_campaign = _compute_dim_wow(curr_data, yoy_data, "by_campaign_type", metrics)
            yoy_by_country = _compute_dim_wow(curr_data, yoy_data, "by_country", metrics)

    prev_label = f"{prev_start.strftime('%Y-%m-%d')} ~ {prev_end.strftime('%Y-%m-%d')}" if prev_data else None

    expert = _safe_ai_analysis(
        curr_data, prev_data, pop_total, pop_by_platform, pop_by_campaign, pop_by_country,
        yoy_data=yoy_data, yoy_total=yoy_total,
        yoy_by_platform=yoy_by_platform, yoy_by_country=yoy_by_country,
        period_label=period_label,
        trend_table=trend_table, memory_context=memory_context,
        business_context=business_context,
    ) if prev_data else {}

    return {
        "period_label": period_label,
        "prev_label": prev_label,
        "yoy_label": yoy_label,
        "days_count": days_count,
        "current": curr_data,
        "previous": prev_data,
        "yoy": yoy_data,
        "wow_total": pop_total,
        "wow_by_platform": pop_by_platform,
        "wow_by_campaign": pop_by_campaign,
        "wow_by_country": pop_by_country,
        "yoy_total": yoy_total,
        "yoy_by_platform": yoy_by_platform,
        "yoy_by_campaign": yoy_by_campaign,
        "yoy_by_country": yoy_by_country,
        "expert": expert,
        "core_metrics": core_metrics,
        "all_metrics": metrics,
    }


# ─────────────────────────────────────────────
# 趋势引擎 — 拉取最近 N 个同粒度周期的核心指标
# ─────────────────────────────────────────────

TREND_METRICS = ["Cost", "Ads.Sales", "ROAS", "CPC", "CTR", "CVR", "GA4.Sessions"]


def build_trend_table(
    config_path: Path | None = None,
    mode: str = "week",
    current_year: int | None = None,
    current_period: int | None = None,
    current_date: date | None = None,
    n: int = 8,
) -> list[dict[str, Any]]:
    """
    返回最近 n 个同粒度周期的核心指标列表（含当期），按时间升序。
    每个元素: {"label": "W12", "Cost": ..., "ROAS": ..., ...}
    """
    config = load_config(config_path)
    metrics = config.get("metrics", [])
    df = load_daily_data(config_path)
    rows: list[dict[str, Any]] = []

    if mode == "week":
        weeks = get_available_weeks(df)
        if current_year and current_period:
            idx = None
            for i, (y, w) in enumerate(weeks):
                if y == current_year and w == current_period:
                    idx = i
                    break
            if idx is not None:
                weeks = weeks[max(0, idx - n + 1): idx + 1]
            else:
                weeks = weeks[-n:]
        else:
            weeks = weeks[-n:]
        for y, w in weeks:
            data = aggregate_week_data(df, y, w, metrics)
            total = data.get("total", {})
            row = {"label": f"{y} W{w}"}
            for m in TREND_METRICS:
                row[m] = total.get(m)
            rows.append(row)

    elif mode == "month":
        from .loader import get_available_months
        months = get_available_months(df)
        if current_year and current_period:
            idx = None
            for i, (y, mo) in enumerate(months):
                if y == current_year and mo == current_period:
                    idx = i
                    break
            if idx is not None:
                months = months[max(0, idx - n + 1): idx + 1]
            else:
                months = months[-n:]
        else:
            months = months[-n:]
        for y, mo in months:
            data = aggregate_month_data(df, y, mo, metrics)
            total = data.get("total", {})
            row = {"label": f"{y}年{mo}月"}
            for m in TREND_METRICS:
                row[m] = total.get(m)
            rows.append(row)

    elif mode == "day":
        all_dates = sorted(df["date"].dt.date.unique())
        if current_date:
            try:
                idx = all_dates.index(current_date)
                all_dates = all_dates[max(0, idx - n + 1): idx + 1]
            except ValueError:
                all_dates = all_dates[-n:]
        else:
            all_dates = all_dates[-n:]
        for d in all_dates:
            data = aggregate_day_data(df, d, metrics)
            total = data.get("total", {})
            row = {"label": str(d)}
            for m in TREND_METRICS:
                row[m] = total.get(m)
            rows.append(row)

    return rows