Agent/examples/auto_put_ad/tools/budget_calc.py.backup

"""
预算计算引擎 — 出价调整与账户评估

核心机制：通过调整 oCPM 出价（bid_amount）控制消耗速度，不设日预算限制。
决策矩阵：ROI × 跑量 二维分类，5 种动作（keep/increase/decrease/close/observe）。
缩量、扩量、持平各一套矩阵，调整幅度 5%-15%。
"""

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

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

from examples.auto_put_ad.tools.ad_api import ad_update
from examples.auto_put_ad.tools.data_query import _get_odps_client

logger = logging.getLogger(__name__)

# ===== 常量 =====
MIN_BID = 10      # 最低出价 0.10 元（10分）
MAX_BID = 10000   # 最高出价 100 元（10000分）
TOP_RATIO = 0.30  # 优质广告占比（保留兼容，新逻辑用分位数）

# 动作类型
ACTION_KEEP = "keep"
ACTION_INCREASE = "increase"
ACTION_DECREASE = "decrease"
ACTION_CLOSE = "close"
ACTION_OBSERVE = "observe"


def _determine_strategy(scale_ratio: float) -> str:
    """根据缩量/扩量比例判断策略"""
    if scale_ratio < 0.7:
        return "aggressive_scale_down"
    elif scale_ratio < 0.95:
        return "moderate_scale_down"
    elif scale_ratio <= 1.05:
        return "maintain"
    elif scale_ratio <= 1.3:
        return "moderate_scale_up"
    else:
        return "aggressive_scale_up"


def _compute_thresholds(df_valid) -> dict:
    """基于有效广告池计算分位数阈值

    Returns:
        dict with keys: roi_p70, roi_p30, cost_p50
    """
    return {
        "roi_p70": float(df_valid["efficiency"].quantile(0.70)),
        "roi_p30": float(df_valid["efficiency"].quantile(0.30)),
        "cost_p50": float(df_valid["cost"].quantile(0.50)),
    }


def _classify_ad(efficiency: float, cost: float, thresholds: dict) -> tuple:
    """将广告分类到 ROI × 跑量 二维象限

    Returns:
        (roi_level, volume_level): e.g. ("high", "high")
    """
    if efficiency >= thresholds["roi_p70"]:
        roi_level = "high"
    elif efficiency >= thresholds["roi_p30"]:
        roi_level = "mid"
    else:
        roi_level = "low"

    volume_level = "high" if cost >= thresholds["cost_p50"] else "low"
    return roi_level, volume_level


def _decide_action(roi_level: str, volume_level: str, strategy: str) -> tuple:
    """根据 ROI×跑量 分类 + 策略，返回 (action, adj_ratio)

    三套矩阵：缩量 / 扩量 / 持平
    """
    # 缩量矩阵（aggressive / moderate）
    if strategy in ("aggressive_scale_down", "moderate_scale_down"):
        aggressive = strategy == "aggressive_scale_down"
        matrix = {
            ("high", "high"):  (ACTION_KEEP, 0.0),
            ("high", "low"):   (ACTION_KEEP, 0.0),
            ("mid", "high"):   (ACTION_DECREASE, -0.10 if aggressive else -0.05),
            ("mid", "low"):    (ACTION_OBSERVE, 0.0),
            ("low", "high"):   (ACTION_DECREASE, -0.15 if aggressive else -0.10),
            ("low", "low"):    (ACTION_CLOSE, 0.0),
        }
        return matrix[(roi_level, volume_level)]

    # 扩量矩阵（aggressive / moderate）
    if strategy in ("aggressive_scale_up", "moderate_scale_up"):
        aggressive = strategy == "aggressive_scale_up"
        matrix = {
            ("high", "high"):  (ACTION_KEEP, 0.0),
            ("high", "low"):   (ACTION_INCREASE, 0.15 if aggressive else 0.10),
            ("mid", "high"):   (ACTION_KEEP, 0.0),
            ("mid", "low"):    (ACTION_INCREASE, 0.05),
            ("low", "high"):   (ACTION_DECREASE, -0.10),
            ("low", "low"):    (ACTION_CLOSE, 0.0),
        }
        return matrix[(roi_level, volume_level)]

    # 持平矩阵
    if roi_level == "low" and volume_level == "low":
        return (ACTION_CLOSE, 0.0)
    return (ACTION_KEEP, 0.0)


def _parse_bizdate(bizdate: str) -> tuple:
    """解析业务日期，返回 (YYYYMMDD, YYYY-MM-DD)"""
    if bizdate in ("yesterday", ""):
        biz = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
    else:
        biz = bizdate.replace("-", "")
    biz_dash = f"{biz[:4]}-{biz[4:6]}-{biz[6:]}"
    return biz, biz_dash


def _build_efficiency_sql(biz: str, biz_dash: str) -> str:
    """构建昨日效率数据 SQL（广告维度聚合）

    包含冷启动保护所需字段：
    - create_time: 广告创建时间（判定冷启动期 48h）
    - conversions_count: 转化量（判定赔付门槛 6 次）
    """
    return f"""
SELECT
    a.account_id,
    a.ad_id,
    c.ad_name,
    c.create_time,
    SUM(b.cost/100)          AS cost,
    SUM(b.valid_click_count) AS valid_click_count,
    SUM(b.conversions_count) AS conversions_count,
    SUM(t.首层小程序打开数)   AS open_count,
    SUM(t.裂变0层回流数)     AS fission0_count,
    SUM(t.总回流人数)        AS total_return_count
FROM (
    SELECT
        IF(c.creative_name IS NOT NULL, c.creative_name, t.rootsourceid) AS creative_name,
        t.*
    FROM loghubods.touliu_data t
    LEFT JOIN (
        SELECT DISTINCT creative_name,
            SPLIT(GET_JSON_OBJECT(page_spec,'$.wechat_mini_program_spec.mini_program_path'),'rootSourceId%3D')[1] AS rootsourceid
        FROM loghubods.ad_put_tencent_creative_components a
        LEFT JOIN loghubods.ad_put_tencent_creative_day b ON a.creative_id = b.creative_id
        WHERE page_type = 'PAGE_TYPE_WECHAT_MINI_PROGRAM'
    ) c ON c.rootsourceid = t.rootsourceid
    WHERE t.dt = '{biz}'
) t
LEFT JOIN loghubods.ad_put_tencent_creative_day a ON t.creative_name = a.creative_name
LEFT JOIN loghubods.ad_put_tencent_ad c ON a.ad_id = c.ad_id
LEFT JOIN (
    SELECT creative_id, valid_click_count, cost, conversions_count
    FROM (
        SELECT creative_id, valid_click_count, cost, conversions_count,
            ROW_NUMBER() OVER (PARTITION BY creative_id ORDER BY update_time DESC) AS rank
        FROM loghubods.ad_put_tencent_creative_data_day
        WHERE dt = '{biz_dash}'
    ) t WHERE rank = 1
) b ON a.creative_id = b.creative_id
WHERE t.dt = '{biz}'
  AND a.ad_id IS NOT NULL
  AND b.cost IS NOT NULL
  AND b.cost > 0
GROUP BY a.account_id, a.ad_id, c.ad_name, c.create_time
"""


# ===== 账户级评估 =====

@tool(description="评估各账户昨日投放表现，输出账户健康度和稳定性标签")
async def account_evaluate(
    bizdate: str = "yesterday",
    context: Optional[ToolContext] = None,
) -> ToolResult:
    """
    账户级评估：查询各账户昨日汇总数据，按消耗量判断稳定性。

    Args:
        bizdate: 数据日期，"yesterday" 或 YYYYMMDD 格式
    """
    import pandas as pd

    try:
        client = _get_odps_client()
        if client is None:
            return ToolResult(title="account_evaluate 失败", output="ODPS 客户端未初始化")

        biz, biz_dash = _parse_bizdate(bizdate)

        # 查询各账户昨日汇总（复用效率 SQL，按 account_id 聚合）
        inner_sql = _build_efficiency_sql(biz, biz_dash)
        sql = f"""
SELECT
    account_id,
    COUNT(DISTINCT ad_id) AS ad_count,
    SUM(cost)             AS total_cost,
    SUM(open_count)       AS total_open,
    SUM(fission0_count)   AS total_fission0
FROM ({inner_sql}) t
GROUP BY account_id
"""
        df = client.execute_sql(sql)
        if df.empty:
            return ToolResult(title="account_evaluate", output=f"昨日({biz})无账户数据")

        # 计算效率分均值
        df["avg_efficiency"] = df.apply(
            lambda r: r["total_fission0"] / r["total_cost"] if r["total_cost"] and r["total_cost"] > 0 else 0,
            axis=1,
        )

        # 按消耗量判断稳定性（中位数为阈值）
        median_cost = df["total_cost"].median()
        p30_cost = df["total_cost"].quantile(0.30)

        def label_stability(cost):
            if cost >= median_cost:
                return "稳定"
            elif cost >= p30_cost:
                return "一般"
            else:
                return "低量"

        df["stability"] = df["total_cost"].apply(label_stability)
        df = df.sort_values("total_cost", ascending=False).reset_index(drop=True)

        # 格式化输出
        lines = [
            f"账户评估（{biz}，共 {len(df)} 个账户）",
            f"消耗中位数: {median_cost:,.0f}元（≥中位数=稳定，≥P30=一般，<P30=低量）",
            "",
            f"{'账户ID':<15} {'广告数':>6} {'昨日消耗(元)':>12} {'效率分均值':>10} {'稳定性':>6}",
            "-" * 55,
        ]
        for _, row in df.iterrows():
            lines.append(
                f"{int(row['account_id']):<15} {int(row['ad_count']):>6} "
                f"{row['total_cost']:>12,.0f} {row['avg_efficiency']:>10.4f} "
                f"{row['stability']:>6}"
            )

        # 标记建议扩量的账户
        stable_high_eff = df[(df["stability"] == "稳定") & (df["avg_efficiency"] > df["avg_efficiency"].median())]
        if not stable_high_eff.empty:
            lines += ["", "扩量建议账户（稳定 + 效率分高于中位数）："]
            for _, row in stable_high_eff.iterrows():
                lines.append(f"  账户 {int(row['account_id'])}（消耗 {row['total_cost']:,.0f}元，效率分 {row['avg_efficiency']:.4f}）")

        return ToolResult(
            title=f"账户评估（{len(df)}个账户）",
            output="\n".join(lines),
            metadata={
                "accounts": df.to_dict("records"),
                "median_cost": median_cost,
                "bizdate": biz,
            },
        )

    except Exception as e:
        logger.error("account_evaluate 失败: %s", e, exc_info=True)
        return ToolResult(title="account_evaluate 失败", output=str(e))


# ===== 出价调整 =====

@tool(description="基于昨日裂变效率数据计算今日出价调整方案（ROI×跑量二维矩阵，5种动作）")
async def budget_calculate_from_data(
    account_id: int,
    total_budget_yuan: float,
    bizdate: str = "yesterday",
    strategy: str = "auto",
    min_bid_cents: int = MIN_BID,
    context: Optional[ToolContext] = None,
) -> ToolResult:
    """
    智能出价调整（ROI × 跑量 二维决策矩阵）：
    1. 拉取昨日效率数据（按广告维度聚合）
    2. 拉取当前广告出价/状态
    3. 计算分位数阈值（ROI P70/P30 + 消耗 P50）
    4. 每个广告分类到二维象限，决定动作（keep/increase/decrease/close/observe）
    5. 样本不足广告跳过不操作

    Args:
        account_id: 账户ID（传 0 则不过滤账户）
        total_budget_yuan: 今日总预算（元）
        bizdate: 业务日期，格式 YYYYMMDD 或 "yesterday"
        strategy: "auto" 或手动指定策略
        min_bid_cents: 最低出价（分，默认 10）
    """
    import pandas as pd

    try:
        client = _get_odps_client()
        if client is None:
            return ToolResult(title="budget_calculate_from_data 失败", output="ODPS 客户端未初始化")

        biz, biz_dash = _parse_bizdate(bizdate)

        # Step 1: 昨日效率数据
        logger.info("拉取昨日效率数据: bizdate=%s", biz)
        sql_efficiency = _build_efficiency_sql(biz, biz_dash)
        df_eff = client.execute_sql(sql_efficiency)
        if df_eff.empty:
            return ToolResult(title="budget_calculate_from_data 失败", output=f"昨日({biz})效率数据为空")

        # Step 2: 当前广告出价/状态
        ad_ids = [int(x) for x in df_eff["ad_id"].dropna().unique() if str(x) != "nan"]
        ad_ids_str = ",".join(map(str, ad_ids))
        sql_status = f"""
SELECT ad_id, ad_name, account_id, bid_amount, day_amount, ad_status, optimization_goal
FROM loghubods.ad_put_tencent_ad
WHERE ad_id IN ({ad_ids_str})
"""
        df_status = client.execute_sql(sql_status)

        # Step 3: 合并
        df_eff["ad_id"] = df_eff["ad_id"].astype(float).astype("Int64")
        df_status["ad_id"] = df_status["ad_id"].astype(float).astype("Int64")
        df = pd.merge(
            df_eff,
            df_status[["ad_id", "bid_amount", "day_amount", "ad_status", "optimization_goal"]],
            on="ad_id", how="left",
        )

        # Step 4: 效率分 + 有效广告筛选
        df["efficiency"] = df.apply(
            lambda r: r["fission0_count"] / r["cost"]
            if r["cost"] and r["cost"] > 0 and r["fission0_count"] is not None and pd.notna(r["fission0_count"])
            else None,
            axis=1,
        )
        df_valid = df[df["open_count"] >= 100].copy().sort_values("efficiency", ascending=False).reset_index(drop=True)
        df_nosample = df[df["open_count"] < 100].copy()

        if df_valid.empty:
            return ToolResult(title="budget_calculate_from_data", output="无有效广告（open_count >= 100）")

        # Step 5: 计算分位数阈值
        thresholds = _compute_thresholds(df_valid)

        # Step 6: 判断策略
        yesterday_total = float(df_valid["cost"].sum())
        scale_ratio = total_budget_yuan / yesterday_total if yesterday_total > 0 else 1.0
        if strategy == "auto":
            strategy = _determine_strategy(scale_ratio)

        # Step 7: 二维矩阵决策（含冷启动保护）
        results = []
        cold_start_count = 0
        for _, row in df_valid.iterrows():
            eff = float(row["efficiency"]) if pd.notna(row["efficiency"]) else 0.0
            cost = float(row["cost"])
            conversions = int(row["conversions_count"]) if pd.notna(row.get("conversions_count")) else 0

            # --- 冷启动保护判定（优先于决策矩阵） ---
            is_cold_start = False
            cold_start_reason = ""

            # 判定1：广告创建时间 < 48 小时
            create_time = row.get("create_time")
            if create_time and pd.notna(create_time):
                try:
                    if isinstance(create_time, str):
                        ct = datetime.strptime(create_time[:19], "%Y-%m-%d %H:%M:%S")
                    else:
                        ct = pd.Timestamp(create_time).to_pydatetime()
                    hours_since_creation = (datetime.now() - ct).total_seconds() / 3600
                    if hours_since_creation < 48:
                        is_cold_start = True
                        cold_start_reason = f"冷启动期({hours_since_creation:.0f}h<48h)"
                except (ValueError, TypeError):
                    pass

            # 判定2：转化数不足 < 6
            if conversions < 6 and not is_cold_start:
                is_cold_start = True
                cold_start_reason = f"转化不足({conversions}<6)"

            if is_cold_start:
                cold_start_count += 1
                roi_level, volume_level = _classify_ad(eff, cost, thresholds)
                action = ACTION_OBSERVE
                adj_ratio = 0.0
            else:
                roi_level, volume_level = _classify_ad(eff, cost, thresholds)
                action, adj_ratio = _decide_action(roi_level, volume_level, strategy)

                # --- 关停保护：赔付门槛检查 ---
                if action == ACTION_CLOSE and 3 <= conversions < 6:
                    action = ACTION_OBSERVE
                    adj_ratio = 0.0
                    cold_start_reason = f"接近赔付门槛({conversions}次转化,等待积累到6)"

            bid = row["bid_amount"] if pd.notna(row["bid_amount"]) else None
            new_bid = None
            if bid and action in (ACTION_INCREASE, ACTION_DECREASE):
                new_bid = max(min_bid_cents, min(MAX_BID, int(float(bid) * (1 + adj_ratio))))
            elif bid:
                new_bid = int(float(bid))  # keep/observe/close 不改出价

            results.append({
                "date": datetime.now().strftime("%Y-%m-%d"),
                "ad_id": int(row["ad_id"]),
                "ad_name": str(row["ad_name"]) if pd.notna(row["ad_name"]) else "",
                "account_id": int(row["account_id"]) if pd.notna(row["account_id"]) else 0,
                "roi_level": roi_level,
                "volume_level": volume_level,
                "efficiency": round(eff, 4),
                "cost": round(cost, 2),
                "open_count": int(row["open_count"]),
                "conversions_count": conversions,
                "is_cold_start": is_cold_start,
                "cold_start_reason": cold_start_reason,
                "current_bid": int(float(bid)) if bid else None,
                "new_bid": new_bid,
                "adjustment_ratio": f"{adj_ratio:+.0%}" if adj_ratio != 0 else "—",
                "action": action,
                "ad_status": str(row["ad_status"]) if pd.notna(row.get("ad_status")) else "",
            })

        # Step 8: 格式化输出（按动作分组）
        direction = "缩量" if scale_ratio < 1 else "扩量" if scale_ratio > 1 else "持平"
        lines = [
            f"出价调整方案（{direction} {abs(1-scale_ratio)*100:.0f}%）",
            f"昨日消耗: {yesterday_total:,.0f} 元 → 今日预算: {total_budget_yuan:,.0f} 元",
            f"策略: {strategy}",
            f"阈值: ROI P70={thresholds['roi_p70']:.4f}, P30={thresholds['roi_p30']:.4f}, 消耗 P50={thresholds['cost_p50']:.0f}元",
            "",
        ]

        action_labels = [
            (ACTION_KEEP, "保持不动"),
            (ACTION_INCREASE, "提价放量"),
            (ACTION_DECREASE, "降价控量"),
            (ACTION_CLOSE, "建议关停"),
            (ACTION_OBSERVE, "观察不动"),
        ]
        for act, label in action_labels:
            sub = [r for r in results if r["action"] == act]
            if not sub:
                continue
            lines.append(f"【{label}（{act}）- {len(sub)}个】")
            for item in sub[:5]:
                bid_info = f"出价:{item['current_bid']}→{item['new_bid']}分 {item['adjustment_ratio']}" if item["current_bid"] else "无出价"
                lines.append(
                    f"  {item['ad_id']} | ROI:{item['roi_level']}/量:{item['volume_level']} | "
                    f"效率:{item['efficiency']} | 消耗:{item['cost']:.0f}元 | {bid_info}"
                )
            if len(sub) > 5:
                lines.append(f"  ... 还有 {len(sub)-5} 个")
            lines.append("")

        if len(df_nosample) > 0:
            lines.append(f"【样本不足 - {len(df_nosample)}个，本次不操作】")
            lines.append("")

        if cold_start_count > 0:
            cold_start_ads = [r for r in results if r.get("is_cold_start")]
            lines.append(f"【冷启动保护 - {cold_start_count}个，标记observe不调价】")
            for item in cold_start_ads[:5]:
                lines.append(f"  {item['ad_id']} | {item['cold_start_reason']} | 转化:{item['conversions_count']} | 消耗:{item['cost']:.0f}元")
            if cold_start_count > 5:
                lines.append(f"  ... 还有 {cold_start_count - 5} 个")
            lines.append("")

        # 汇总
        action_counts = {}
        for r in results:
            action_counts[r["action"]] = action_counts.get(r["action"], 0) + 1
        summary_parts = [f"{label}:{action_counts.get(act, 0)}" for act, label in action_labels]
        lines.append(f"合计：{' / '.join(summary_parts)} / 样本不足:{len(df_nosample)} / 冷启动保护:{cold_start_count}")

        # Step 9: 输出 Excel（按动作颜色标识）
        try:
            import openpyxl
            from openpyxl.styles import PatternFill, Font

            ACTION_COLORS = {
                "increase": "C6EFCE",  # 绿
                "decrease": "FFEB9C",  # 橙黄
                "close":    "FFC7CE",  # 红
                "observe":  "FFFF99",  # 黄
                "keep":     "FFFFFF",  # 白
            }

            output_dir = Path(__file__).parent.parent / "outputs"
            output_dir.mkdir(exist_ok=True)
            xlsx_path = output_dir / f"adjustment_plan_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"

            headers_cn = ["日期", "账户ID", "广告ID", "广告名称", "动作", "当前出价(分)", "新出价(分)",
                          "调整幅度", "ROI等级", "跑量等级", "效率分", "昨日消耗(元)", "打开数",
                          "转化数", "冷启动", "冷启动原因", "广告状态"]
            fields_en = ["date", "account_id", "ad_id", "ad_name", "action", "current_bid", "new_bid",
                         "adjustment_ratio", "roi_level", "volume_level", "efficiency", "cost",
                         "open_count", "conversions_count", "is_cold_start", "cold_start_reason", "ad_status"]

            wb = openpyxl.Workbook()
            ws = wb.active
            ws.title = "调整方案"

            # 表头（加粗）
            ws.append(headers_cn)
            for cell in ws[1]:
                cell.font = Font(bold=True)

            # 数据行 + 颜色
            for row in results:
                ws.append([row.get(f) for f in fields_en])
                color = ACTION_COLORS.get(row.get("action", "keep"), "FFFFFF")
                fill = PatternFill(fill_type="solid", fgColor=color)
                for cell in ws[ws.max_row]:
                    cell.fill = fill

            # 冻结首行 + 列宽
            ws.freeze_panes = "A2"
            for col in ws.columns:
                ws.column_dimensions[col[0].column_letter].width = 16

            wb.save(xlsx_path)
            lines.append(f"\n📄 完整方案已输出: {xlsx_path}")
            logger.info("方案已保存: %s", xlsx_path)
        except Exception as xlsx_err:
            logger.warning("xlsx 输出失败（不影响主流程）: %s", xlsx_err)

        return ToolResult(
            title=f"出价调整方案（{len(results)}个广告，{direction}{abs(1-scale_ratio)*100:.0f}%）",
            output="\n".join(lines),
            metadata={
                "adjustment_plan": results,
                "strategy": strategy,
                "scale_ratio": scale_ratio,
                "thresholds": thresholds,
                "yesterday_total": yesterday_total,
                "total_budget_yuan": total_budget_yuan,
                "nosample_count": len(df_nosample),
                "cold_start_count": cold_start_count,
                "action_counts": action_counts,
            },
        )

    except Exception as e:
        logger.error("budget_calculate_from_data 失败: %s", e, exc_info=True)
        return ToolResult(title="budget_calculate_from_data 失败", output=str(e))


# ===== 执行出价调整 =====

@tool(description="执行出价调整方案")
async def bid_adjustment_execute(
    adjustment_plan: List[Dict],
    account_id: int,
    context: Optional[ToolContext] = None,
) -> ToolResult:
    """
    批量执行出价调整

    Args:
        adjustment_plan: 调整方案列表，每项包含 ad_id, new_bid, action
        account_id: 账户ID
    """
    success_count = 0
    failed_count = 0
    errors = []

    for item in adjustment_plan:
        if item["action"] not in ("increase", "decrease"):
            continue
        try:
            await ad_update(
                account_id=account_id,
                adgroup_id=item["ad_id"],
                bid_amount=item["new_bid"]
            )
            success_count += 1
            logger.info("调整出价: ad_id=%s, new_bid=%s", item["ad_id"], item["new_bid"])
        except Exception as e:
            failed_count += 1
            error_msg = f"ad_id={item['ad_id']}: {str(e)}"
            errors.append(error_msg)
            logger.error("执行失败: %s", error_msg)

    output_lines = [
        "执行完成：",
        f"- 成功调整: {success_count} 个",
        f"- 失败: {failed_count} 个",
    ]

    if errors:
        output_lines.append("\n失败详情:")
        for err in errors[:10]:
            output_lines.append(f"  {err}")
        if len(errors) > 10:
            output_lines.append(f"  ... 还有 {len(errors)-10} 个错误")

    return ToolResult(
        title="出价调整执行结果",
        output="\n".join(output_lines),
    )