""" 数字营销专家级分析引擎。 底层使用日度数据,按 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