Agent/examples/auto_put_ad_mini/tools/roi_calculator.py

"""
ROI 计算工具 — auto_put_ad_mini V3

职责：
  1. 创意级数据 → 广告级聚合（GROUP BY ad_id + date, SUM）
  2. 计算 动态ROI（7 天滚动窗口）
  3. 计算昨日 ROI（简单 ROI）
  4. 输出广告级指标表（供决策引擎使用）

核心公式：
  T0裂变系数    = SUM(fission0_count) / SUM(open_count)
  arpu          = SUM(total_revenue) / SUM(total_return_count)
  当日裂变收益率 = SUM(fission0_count) * arpu / SUM(cost)
  当日回流倍数   = SUM(total_return_count) / SUM(open_count)
  回流倍数_7日均值    = mean(当日回流倍数) over 7天
  T0裂变系数_7日均值  = mean(T0裂变系数) over 7天
  裂变效率稳定因子    = 回流倍数_7日均值 / T0裂变系数_7日均值
  动态ROI       = 当日裂变收益率(当天) * 裂变效率稳定因子
  动态ROI_7日均值 = mean(动态ROI) over 7天  ← 决策参考值

前置条件：
  - 日消耗 < 100 元的天数不参与 ROI 计算
  - 至少需要 3 天有效数据才能计算 ROI 均值
"""

import logging
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional

import pandas as pd
import numpy as np

from agent.tools import tool
from agent.tools.models import ToolContext, ToolResult

logger = logging.getLogger(__name__)

# 添加当前目录到路径以导入其他工具模块
_TOOLS_DIR = Path(__file__).resolve().parent
if str(_TOOLS_DIR) not in sys.path:
    sys.path.insert(0, str(_TOOLS_DIR))

_MINI_DIR = Path(__file__).resolve().parent.parent
_RAW_DIR = _MINI_DIR / "outputs" / "raw"
_AD_STATUS_DIR = _MINI_DIR / "outputs" / "ad_status"
_MERGED_DIR = _MINI_DIR / "outputs" / "merged"

# 延迟导入 _extract_audience_tier（避免循环导入）
def _get_extract_audience_tier():
    """延迟导入人群包提取函数"""
    try:
        from ad_decision import _extract_audience_tier
        return _extract_audience_tier
    except ImportError:
        logger.warning("无法导入 _extract_audience_tier，使用默认实现")
        # 提供一个简单的默认实现
        def default_extract(ad_name: str) -> str:
            if not ad_name:
                return "default"
            # 简化版：直接匹配常见模式
            ad_name_lower = str(ad_name).lower()
            if "r500" in ad_name_lower or "r_500" in ad_name_lower:
                return "R500"
            elif "r330+" in ad_name_lower or "回流330+" in ad_name_lower:
                return "R330+"
            elif "r330" in ad_name_lower or "r_330" in ad_name_lower or "回流330" in ad_name_lower:
                return "R330"
            elif "r180" in ad_name_lower or "r_180" in ad_name_lower or "回流180" in ad_name_lower:
                return "R180"
            elif "r100" in ad_name_lower or "r_100" in ad_name_lower or "回流100" in ad_name_lower:
                return "R100"
            elif "r50" in ad_name_lower or "r_50" in ad_name_lower or "回流50" in ad_name_lower:
                return "R50"
            elif "r10" in ad_name_lower or "r_10" in ad_name_lower:
                return "R10"
            elif "r2" in ad_name_lower or "r_2" in ad_name_lower:
                return "R2"
            return "default"
        return default_extract


# ===== 创意 → 广告聚合 =====

def _aggregate_creative_to_ad(df: pd.DataFrame) -> pd.DataFrame:
    """
    创意级数据聚合到广告级（按 ad_id + date）。

    输入：创意级 DataFrame（多日数据，包含 creative_id）
    输出：广告级 DataFrame（每个 ad_id 每天一行）

    聚合规则：
      - 数值字段：SUM（cost, revenue, open_count, fission0_count, total_return_count, view_count, etc.）
      - 广告属性：取 FIRST（ad_name, account_id, create_time, bid_amount, configured_status）
      - 创意计数：COUNT(DISTINCT creative_id)
    """
    if df.empty:
        return pd.DataFrame()

    # 添加 date 列（从文件名或 create_time 推断，这里假设已有 bizdate 列）
    # 如果没有，需要从外部传入或从文件名解析
    # 将 bizdate 映射为 date 列（bizdate 格式: 20260412）
    if "bizdate" in df.columns:
        df = df.copy()
        df["date"] = df["bizdate"].astype(str)
    elif "date" not in df.columns:
        logger.warning("DataFrame 缺少 bizdate/date 列，无法按日聚合")
        return pd.DataFrame()

    # 列名映射：CSV 实际列名 → 内部标准名
    COLUMN_RENAME = {
        "首层小程序打开数": "open_count",
        "裂变0层回流数": "fission0_count",
        "裂变层回流数": "fission_count",
        "裂变1层回流数": "fission1_count",
        "总回流人数": "total_return_count",
        "总收入": "total_revenue",
        "ad_status": "configured_status",
    }
    # 只重命名存在的列
    rename_map = {k: v for k, v in COLUMN_RENAME.items() if k in df.columns}
    df = df.rename(columns=rename_map)

    agg_dict = {
        # 广告属性（取第一个值）
        "account_id": "first",
        "ad_name": "first",
        "create_time": "first",
        "configured_status": "first",
        "bid_amount": "first",
        "广告优化目标": "first",
        "package_name": "first",   # 人群包名称（如 R50*泛知识*生活科普）
        "人群包人数": "first",
        # 数值指标（SUM — 聚合后再计算派生比值，不能直接平均）
        "cost": "sum",
        "view_count": "sum",
        "key_page_view_count": "sum",
        "key_page_uv": "sum",
        "valid_click_count": "sum",
        "conversions_count": "sum",
        "open_count": "sum",       # 首层小程序打开数
        "fission0_count": "sum",   # 裂变0层回流数
        "fission_count": "sum",    # 裂变层回流数（总）
        "fission1_count": "sum",   # 裂变1层回流数
        "total_return_count": "sum",  # 总回流人数
        "total_revenue": "sum",       # 总收入
        # 创意计数
        "creative_id": "nunique",
    }

    # 过滤掉不存在的列
    agg_dict = {k: v for k, v in agg_dict.items() if k in df.columns}

    grouped = df.groupby(["ad_id", "date"], as_index=False).agg(agg_dict)
    grouped.rename(columns={"creative_id": "creative_count"}, inplace=True)

    return grouped


# ===== 动态ROI 计算 =====

def _calculate_f7_dynamic_roi(
    ad_df: pd.DataFrame,
    min_daily_cost: float = 100.0
) -> pd.DataFrame:
    """
    计算 动态ROI（每个广告每天一个值）。

    输入：广告级 DataFrame（ad_id + date + 聚合指标）
    输出：添加以下列的 DataFrame
      - T0裂变系数, arpu（每天基础指标）
      - 当日裂变收益率, 当日回流倍数（每天派生指标）
      - T0裂变系数_7日均值, 回流倍数_7日均值（7 天滚动均值）
      - 裂变效率稳定因子（= 回流倍数_7日均值 / T0裂变系数_7日均值）
      - 动态ROI（= 当日裂变收益率 × 裂变效率稳定因子，单日值）
      - 动态ROI_7日均值（决策参考值）

    前置条件：
      - 日消耗 < min_daily_cost 的天数：T0裂变系数/arpu/a/b 设为 NaN，不参与 7 日均值计算
    """
    if ad_df.empty:
        return ad_df

    # 确保按 ad_id + date 排序
    ad_df = ad_df.sort_values(["ad_id", "date"]).reset_index(drop=True)

    # 计算每天的基础指标
    ad_df["T0裂变系数"] = np.where(
        (ad_df["open_count"] > 0) & (ad_df["cost"] >= min_daily_cost),
        ad_df["fission0_count"] / ad_df["open_count"],
        np.nan
    )

    ad_df["arpu"] = np.where(
        (ad_df["total_return_count"] > 0) & (ad_df["cost"] >= min_daily_cost),
        ad_df["total_revenue"] / ad_df["total_return_count"],
        np.nan
    )

    # 当日裂变收益率 = T0裂变数 × arpu / cost
    ad_df["当日裂变收益率"] = np.where(
        (ad_df["cost"] > 0) & (ad_df["cost"] >= min_daily_cost),
        ad_df["fission0_count"] * ad_df["arpu"] / ad_df["cost"],
        np.nan
    )

    # 当日回流倍数 = 总回流人数 / 首层打开数
    ad_df["当日回流倍数"] = np.where(
        (ad_df["open_count"] > 0) & (ad_df["cost"] >= min_daily_cost),
        ad_df["total_return_count"] / ad_df["open_count"],
        np.nan
    )

    # 7 天滚动均值（按 ad_id 分组，至少 3 天数据即可计算）
    ad_df["T0裂变系数_7日均值"] = (
        ad_df.groupby("ad_id")["T0裂变系数"]
        .transform(lambda x: x.rolling(window=7, min_periods=3).mean())
    )

    ad_df["回流倍数_7日均值"] = (
        ad_df.groupby("ad_id")["当日回流倍数"]
        .transform(lambda x: x.rolling(window=7, min_periods=3).mean())
    )

    # 裂变效率稳定因子 = 回流倍数_7日均值 / T0裂变系数_7日均值
    ad_df["裂变效率稳定因子"] = np.where(
        ad_df["T0裂变系数_7日均值"] > 0,
        ad_df["回流倍数_7日均值"] / ad_df["T0裂变系数_7日均值"],
        np.nan
    )

    # 动态ROI = 当日裂变收益率 × 裂变效率稳定因子
    ad_df["动态ROI"] = ad_df["当日裂变收益率"] * ad_df["裂变效率稳定因子"]

    # 动态ROI_7日均值（决策参考值）
    ad_df["动态ROI_7日均值"] = (
        ad_df.groupby("ad_id")["动态ROI"]
        .transform(lambda x: x.rolling(window=7, min_periods=3).mean())
    )

    # ===== 新增：计算有效ROI数据天数（非NaN的天数）=====
    ad_df["roi_valid_days"] = (
        ad_df.groupby("ad_id")["动态ROI"]
        .transform(lambda x: x.notna().sum())
    )

    return ad_df


# ===== 昨日 ROI 计算 =====

def _calculate_yesterday_roi(ad_df: pd.DataFrame, yesterday: str) -> pd.DataFrame:
    """
    计算昨日 ROI（简单 ROI = revenue / cost）。

    输入：广告级 DataFrame
    输出：添加 yesterday_roi 列
    """
    if ad_df.empty:
        return ad_df

    # 筛选昨日数据
    yesterday_df = ad_df[ad_df["date"] == yesterday].copy()

    yesterday_df["yesterday_roi"] = np.where(
        yesterday_df["cost"] > 0,
        yesterday_df["total_revenue"] / yesterday_df["cost"],
        np.nan
    )

    # 同时保留昨日消耗（用于关停消耗门槛判断，投手经验2.4）
    yesterday_df["yesterday_cost"] = yesterday_df["cost"]

    # 合并回原 DataFrame（保留昨日的 ROI + 昨日消耗）
    ad_df = ad_df.merge(
        yesterday_df[["ad_id", "yesterday_roi", "yesterday_cost"]],
        on="ad_id",
        how="left"
    )

    return ad_df


# ===== 7 日汇总指标 =====

def _calculate_7d_summary(ad_df: pd.DataFrame, end_date: str) -> pd.DataFrame:
    """
    计算最近 7 天汇总指标（供决策引擎使用）。

    输出列：
      - cost_7d_total, cost_7d_avg
      - revenue_7d_total
      - 动态ROI_latest（单日值，最新一天）
      - 动态ROI_7日均值_latest（决策参考值，最新一天的7日均值）
    """
    if ad_df.empty:
        return pd.DataFrame()

    end_dt = datetime.strptime(end_date, "%Y%m%d")
    start_dt = end_dt - timedelta(days=6)
    start_date = start_dt.strftime("%Y%m%d")

    # 筛选 7 天数据
    df_7d = ad_df[
        (ad_df["date"] >= start_date) & (ad_df["date"] <= end_date)
    ].copy()

    # 按 ad_id 聚合
    agg_cols = {"cost": "sum", "total_revenue": "sum"}
    # 7日累计 click/view（供人群包基线计算 CTR）
    if "valid_click_count" in df_7d.columns:
        agg_cols["valid_click_count"] = "sum"
    if "view_count" in df_7d.columns:
        agg_cols["view_count"] = "sum"

    summary = df_7d.groupby("ad_id", as_index=False).agg(agg_cols)

    summary.rename(columns={
        "cost": "cost_7d_total",
        "total_revenue": "revenue_7d_total",
    }, inplace=True)

    summary["cost_7d_avg"] = summary["cost_7d_total"] / 7

    # 获取最新一天的 动态ROI + T0裂变系数_7日均值
    latest_cols_7d = ["ad_id", "动态ROI", "动态ROI_7日均值"]
    if "T0裂变系数_7日均值" in ad_df.columns:
        latest_cols_7d.append("T0裂变系数_7日均值")
    latest_df = ad_df[ad_df["date"] == end_date][[c for c in latest_cols_7d if c in ad_df.columns]].copy()
    latest_df.rename(columns={
        "动态ROI": "动态ROI_latest",
        "动态ROI_7日均值": "动态ROI_7日均值_latest"
    }, inplace=True)

    summary = summary.merge(latest_df, on="ad_id", how="left")

    return summary


# ===== 30 日汇总指标 =====

def _calculate_30d_summary(ad_df: pd.DataFrame) -> pd.DataFrame:
    """
    计算 30 天汇总指标。

    输出列：
      - cost_30d_total, cost_30d_avg
      - stable_spend_days_30d（消耗 > 100 元的天数）
    """
    if ad_df.empty:
        return pd.DataFrame()

    summary = ad_df.groupby("ad_id", as_index=False).agg({
        "cost": ["sum", "mean"],
    })

    summary.columns = ["ad_id", "cost_30d_total", "cost_30d_avg"]

    # 稳定消耗天数（cost > 100）
    stable_days = (
        ad_df[ad_df["cost"] > 100]
        .groupby("ad_id", as_index=False)
        .size()
        .rename(columns={"size": "stable_spend_days_30d"})
    )

    summary = summary.merge(stable_days, on="ad_id", how="left")
    summary["stable_spend_days_30d"] = summary["stable_spend_days_30d"].fillna(0).astype(int)

    return summary


# ===== 工具：计算 ROI 指标 =====

@tool(description="计算 动态ROI + 昨日 ROI + 7日/30日汇总指标")
async def calculate_roi_metrics(
    ctx: ToolContext = None,
    end_date: str = "yesterday",
    min_daily_cost: float = 100.0
) -> ToolResult:
    """
    计算广告级 ROI 指标（V3）。

    工作流：
      1. 加载 30 天创意级 CSV
      2. 聚合到广告级（ad_id + date）
      3. 计算 动态ROI
      4. 计算昨日 ROI
      5. 计算 7 日/30 日汇总
      6. 输出广告级指标表

    Args:
        end_date: 结束日期（YYYYMMDD 或 "yesterday"）
        min_daily_cost: 日消耗阈值（元），低于此值的天数不参与 ROI 计算

    Returns:
        ToolResult 包含广告级指标 DataFrame
    """
    try:
        # 解析日期
        if end_date == "yesterday":
            end_dt = datetime.now() - timedelta(days=1)
        else:
            end_dt = datetime.strptime(end_date.replace("-", ""), "%Y%m%d")

        end_date_str = end_dt.strftime("%Y%m%d")
        start_dt = end_dt - timedelta(days=29)

        # 加载 30 天 merged 数据（已包含创意+广告状态）
        merged_dfs = []
        for i in range(30):
            date = (start_dt + timedelta(days=i)).strftime("%Y%m%d")
            csv_path = _MERGED_DIR / f"merged_{date}.csv"

            if not csv_path.exists():
                logger.warning("merged 数据缺失: %s", date)
                continue

            df = pd.read_csv(csv_path, dtype={"ad_id": str, "account_id": str})
            # bizdate 列已存在，无需手动添加 date
            merged_dfs.append(df)

        if not merged_dfs:
            return ToolResult(
                title="ROI 计算失败",
                output=f"未找到任何 merged 数据 CSV（{_MERGED_DIR}）"
            )

        creative_df = pd.concat(merged_dfs, ignore_index=True)
        logger.info("加载 merged 数据: %d 行（%d 天）", len(creative_df), len(merged_dfs))

        # Step 1: 聚合到广告级
        ad_df = _aggregate_creative_to_ad(creative_df)
        logger.info("聚合到广告级: %d 行", len(ad_df))

        # Step 2: 计算 动态ROI
        ad_df = _calculate_f7_dynamic_roi(ad_df, min_daily_cost)

        # Step 3: 计算昨日 ROI
        ad_df = _calculate_yesterday_roi(ad_df, end_date_str)

        # Step 4: 计算 7 日汇总
        summary_7d = _calculate_7d_summary(ad_df, end_date_str)

        # Step 5: 计算 30 日汇总
        summary_30d = _calculate_30d_summary(ad_df)

        # Step 6: 合并所有指标（取最新一天的广告属性）
        latest_cols = [
            "ad_id", "account_id", "ad_name", "create_time",
            "configured_status", "bid_amount", "creative_count",
            "package_name",  # 人群包名称（如 R50*泛知识*生活科普）
            "yesterday_roi", "yesterday_cost",  # 昨日ROI+昨日消耗（投手经验2.4关停门槛）
        ]
        # 只取存在的列（yesterday_cost 可能在 _calculate_yesterday_roi 中添加）
        latest_cols = [c for c in latest_cols if c in ad_df.columns]
        latest_ad = ad_df[ad_df["date"] == end_date_str][latest_cols].copy()

        result_df = latest_ad.merge(summary_7d, on="ad_id", how="left")
        result_df = result_df.merge(summary_30d, on="ad_id", how="left")

        # 计算广告年龄（天）
        result_df["ad_age_days"] = (
            (end_dt - pd.to_datetime(result_df["create_time"])).dt.days
        )

        # ===== 人群包字段 =====
        # audience_tier: 使用 package_name 原始值（如 R50*泛知识*生活科普）
        # r_tier: 从 ad_name 提取的 R 层级（如 R50），作为辅助分组
        extract_tier = _get_extract_audience_tier()
        result_df["r_tier"] = result_df["ad_name"].apply(extract_tier)
        # 优先使用 package_name 作为 audience_tier，缺失时用 r_tier 兜底
        if "package_name" in result_df.columns:
            result_df["audience_tier"] = result_df["package_name"].fillna("").replace("", pd.NA)
            result_df["audience_tier"] = result_df["audience_tier"].fillna(result_df["r_tier"])
        else:
            result_df["audience_tier"] = result_df["r_tier"]

        # 获取 roi_valid_days（从 ad_df 最新一天的数据）
        latest_roi_valid = ad_df[ad_df["date"] == end_date_str][["ad_id", "roi_valid_days"]].copy()
        result_df = result_df.merge(latest_roi_valid, on="ad_id", how="left")
        result_df["roi_valid_days"] = result_df["roi_valid_days"].fillna(0).astype(int)

        # 重命名输出列，统一供决策引擎使用
        # 动态ROI_latest → 动态ROI（单日值，反映当日效率）
        # 动态ROI_7日均值_latest → 动态ROI_7日均值（决策参考值）
        if "动态ROI_latest" in result_df.columns:
            result_df.rename(columns={"动态ROI_latest": "动态ROI"}, inplace=True)
        if "动态ROI_7日均值_latest" in result_df.columns:
            result_df.rename(columns={"动态ROI_7日均值_latest": "动态ROI_7日均值"}, inplace=True)

        # 计算全体 动态ROI_7日均值 的均值（供决策引擎做相对比较）
        f7_7d_mean_all = result_df["动态ROI_7日均值"].mean() if "动态ROI_7日均值" in result_df.columns else float("nan")

        # 保存指标 CSV（供 get_ads_for_review 读取）
        metrics_dir = _MINI_DIR / "outputs"
        metrics_dir.mkdir(parents=True, exist_ok=True)
        metrics_csv = metrics_dir / f"metrics_{end_date_str}.csv"
        result_df.to_csv(metrics_csv, index=False, encoding="utf-8-sig")
        logger.info("指标 CSV 已保存: %s", metrics_csv)

        # 同时保存为 metrics_temp.csv（最新指标，供下游工具默认路径使用）
        metrics_temp = metrics_dir / "metrics_temp.csv"
        result_df.to_csv(metrics_temp, index=False, encoding="utf-8-sig")
        logger.info("指标临时文件已更新: %s", metrics_temp)

        # ===== 自动生成 portfolio_summary（人群包基线）=====
        # 这是决策引擎的硬依赖，直接在 ROI 计算完成后生成
        portfolio_tier_count = 0
        try:
            from tools.portfolio_metrics import _describe_group, _compute_daily_tier_snapshot, _compute_market_signal
            portfolio_dir = _MINI_DIR / "outputs" / "portfolio_summary"
            portfolio_dir.mkdir(parents=True, exist_ok=True)
            portfolio_file = portfolio_dir / f"portfolio_summary_{end_date_str}.json"

            by_tier = {}
            if "audience_tier" in result_df.columns:
                for tier, group in result_df.groupby("audience_tier"):
                    by_tier[str(tier)] = _describe_group(group)

            by_tier_goal = {}
            goal_col = "广告优化目标" if "广告优化目标" in result_df.columns else None
            if "audience_tier" in result_df.columns and goal_col:
                for (tier, goal), group in result_df.groupby(["audience_tier", goal_col]):
                    by_tier_goal[f"{tier}_{goal}"] = _describe_group(group)

            global_stats = _describe_group(result_df)
            by_date = _compute_daily_tier_snapshot(end_dt, days=7)
            market_signal = _compute_market_signal(by_date)

            import json as _json
            summary = {
                "end_date": end_date_str,
                "source_csv": str(metrics_csv),
                "by_audience_tier": by_tier,
                "by_tier_goal": by_tier_goal,
                "global": global_stats,
                "by_date": by_date,
                "market_signal": market_signal,
            }
            portfolio_file.write_text(
                _json.dumps(summary, ensure_ascii=False, indent=2),
                encoding="utf-8",
            )
            portfolio_tier_count = len(by_tier)
            logger.info("人群包基线已生成: %s（%d 个人群包）", portfolio_file, portfolio_tier_count)
        except Exception as e:
            logger.warning("人群包基线生成失败（不影响主流程）: %s", e)

        output_lines = [
            f"ROI 计算完成（截至 {end_date_str}）",
            f"广告总数: {len(result_df)}",
            f"动态ROI_7日均值 全体均值: {f7_7d_mean_all:.4f}（决策参考）",
            "",
            f"指标 CSV: {metrics_csv}",
            "",
            "指标列：",
            "  - 动态ROI（单日值，反映当日效率）",
            "  - 动态ROI_7日均值（决策参考值）",
            "  - yesterday_roi（昨日简单 ROI）",
            "  - cost_7d_total, cost_7d_avg, revenue_7d_total",
            "  - cost_30d_total, cost_30d_avg, stable_spend_days_30d",
            "  - ad_age_days, creative_count",
            "  - audience_tier（人群包名称，如 R50*泛知识*生活科普，用于同类对比）",
            "  - r_tier（R层级，如 R50，辅助分组）",
            "  - roi_valid_days（有效ROI数据天数，用于置信度评估）",
        ]

        return ToolResult(
            title=f"ROI 计算完成（{len(result_df)} 个广告）",
            output="\n".join(output_lines),
            metadata={
                "metrics_csv": str(metrics_csv),
                "动态ROI_7日均值_mean_all": f7_7d_mean_all,
                "end_date": end_date_str,
                "min_daily_cost": min_daily_cost,
            }
        )

    except Exception as e:
        logger.error("ROI 计算失败: %s", e, exc_info=True)
        return ToolResult(
            title="ROI 计算失败",
            output=f"错误: {e}"
        )