Agent/examples/auto_put_ad/test_budget.py

"""
预算模块测试脚本 — 仅测试计算逻辑，不调用腾讯广告 API

测试内容：
1. 效率数据查询（ODPS）+ 广告状态查询（ODPS）
2. 分位数阈值计算（ROI P70/P30 + 消耗 P50）
3. ROI × 跑量 二维分类 + 5 种动作决策
4. 缩量/扩量/持平三种场景

运行方式：
    cd /path/to/Agent
    python examples/auto_put_ad/test_budget.py --budget 100000   # 缩量
    python examples/auto_put_ad/test_budget.py --budget 250000   # 扩量
    python examples/auto_put_ad/test_budget.py --budget 175000   # 持平
"""

import argparse
import sys
from collections import Counter
from datetime import datetime, timedelta
from pathlib import Path

import pandas as pd

# 项目根目录加入 sys.path
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))

from examples.auto_put_ad.odps_module import ODPSClient
from examples.auto_put_ad.tools.budget_calc import (
    MIN_BID,
    MAX_BID,
    _build_efficiency_sql,
    _classify_ad,
    _compute_thresholds,
    _decide_action,
    _determine_strategy,
    _parse_bizdate,
)


def run_budget_test(bizdate: str = "yesterday", total_budget_yuan: float = 100_000):
    print("=" * 70)
    print("  预算计算测试（ROI × 跑量 二维矩阵，不执行 API）")
    print("=" * 70)

    biz, biz_dash = _parse_bizdate(bizdate)
    print(f"\n数据日期  : {biz}（{biz_dash}）")
    print(f"今日预算  : {total_budget_yuan:,.0f} 元")

    # ── 1. ODPS 查询 ────────────────────────────────────────
    print("\n[Step 1] 连接 ODPS + 查询效率数据...")
    client = ODPSClient(project="loghubods")
    sql_efficiency = _build_efficiency_sql(biz, biz_dash)
    df_eff = client.execute_sql(sql_efficiency)
    if df_eff.empty:
        print(f"❌ 昨日({biz})效率数据为空"); return
    print(f"  → 效率数据: {len(df_eff)} 个广告")

    # ── 2. 广告状态 ─────────────────────────────────────────
    print("[Step 2] 查询广告出价/状态...")
    ad_ids = [int(x) for x in df_eff["ad_id"].dropna().unique() if str(x) != "nan"]
    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 ({",".join(map(str, ad_ids))})
"""
    df_status = client.execute_sql(sql_status)
    print(f"  → 广告状态: {len(df_status)} 个")

    # ── 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"]], on="ad_id", how="left")
    df["efficiency"] = df.apply(
        lambda r: r["fission0_count"] / r["cost"] if r["cost"] and r["cost"] > 0 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()
    print(f"  → 有效广告: {len(df_valid)}，样本不足: {len(df_nosample)}")

    # ── 4. 分位数阈值 ──────────────────────────────────────
    thresholds = _compute_thresholds(df_valid)
    print(f"\n[Step 3] 分位数阈值:")
    print(f"  ROI  P70 = {thresholds['roi_p70']:.4f}")
    print(f"  ROI  P30 = {thresholds['roi_p30']:.4f}")
    print(f"  消耗 P50 = {thresholds['cost_p50']:.0f} 元")

    # ── 5. 策略判断 ─────────────────────────────────────────
    yesterday_total = float(df_valid["cost"].sum())
    scale_ratio = total_budget_yuan / yesterday_total if yesterday_total > 0 else 1.0
    strategy = _determine_strategy(scale_ratio)
    direction = "缩量" if scale_ratio < 1 else "扩量" if scale_ratio > 1 else "持平"
    print(f"\n[Step 4] 策略判断:")
    print(f"  昨日消耗 : {yesterday_total:,.0f} 元")
    print(f"  scale_ratio : {scale_ratio:.2f}（{direction} {abs(1-scale_ratio)*100:.0f}%）")
    print(f"  策略 : {strategy}")

    # ── 6. 二维矩阵决策 ────────────────────────────────────
    print(f"\n[Step 5] 二维矩阵决策...")
    results = []
    for _, row in df_valid.iterrows():
        eff = float(row["efficiency"]) if pd.notna(row["efficiency"]) else 0.0
        cost = float(row["cost"])
        roi_level, volume_level = _classify_ad(eff, cost, thresholds)
        action, adj_ratio = _decide_action(roi_level, volume_level, strategy)

        bid = row["bid_amount"] if pd.notna(row["bid_amount"]) else None
        new_bid = None
        if bid and action in ("increase", "decrease"):
            new_bid = max(MIN_BID, min(MAX_BID, int(float(bid) * (1 + adj_ratio))))
        elif bid:
            new_bid = int(float(bid))

        results.append({
            "ad_id": int(row["ad_id"]),
            "roi_level": roi_level, "volume_level": volume_level,
            "efficiency": round(eff, 4), "cost": round(cost, 2),
            "current_bid": int(float(bid)) if bid else None,
            "new_bid": new_bid,
            "adj_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 "",
        })

    # ── 7. 结果展示 ────────────────────────────────────────
    print("\n" + "=" * 70)
    print(f"  出价调整方案（{direction} {abs(1-scale_ratio)*100:.0f}%）")
    print(f"  昨日消耗: {yesterday_total:,.0f} 元 → 今日预算: {total_budget_yuan:,.0f} 元")
    print(f"  策略: {strategy}")
    print(f"  阈值: ROI P70={thresholds['roi_p70']:.4f}, P30={thresholds['roi_p30']:.4f}, 消耗 P50={thresholds['cost_p50']:.0f}元")
    print("=" * 70)

    action_labels = [
        ("keep", "保持不动"), ("increase", "提价放量"), ("decrease", "降价控量"),
        ("close", "建议关停"), ("observe", "观察不动"),
    ]
    for act, label in action_labels:
        sub = [r for r in results if r["action"] == act]
        if not sub:
            continue
        print(f"\n【{label}（{act}）- {len(sub)}个】")
        header = f"  {'ad_id':<15} {'ROI':>4} {'量':>4} {'效率分':>8} {'消耗(元)':>10} {'当前出价':>8} {'新出价':>8} {'幅度':>6}"
        print(header)
        print("  " + "-" * (len(header) - 2))
        for item in sub[:8]:
            bid_str = str(item["current_bid"]) if item["current_bid"] else "—"
            new_str = str(item["new_bid"]) if item["new_bid"] else "—"
            print(f"  {item['ad_id']:<15} {item['roi_level']:>4} {item['volume_level']:>4} "
                  f"{item['efficiency']:>8} {item['cost']:>10,.0f} {bid_str:>8} {new_str:>8} {item['adj_ratio']:>6}")
        if len(sub) > 8:
            print(f"  ... 还有 {len(sub)-8} 个")

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

    counts = Counter(r["action"] for r in results)
    summary = " / ".join(f"{label}:{counts.get(act, 0)}" for act, label in action_labels)
    print(f"\n合计：{summary} / 样本不足:{len(df_nosample)}")
    print("\n✅ 计算完成（未调用腾讯广告 API）")
    print("=" * 70)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="预算模块计算测试（二维矩阵）")
    parser.add_argument("--date", default="yesterday", help="数据日期，YYYYMMDD 或 yesterday")
    parser.add_argument("--budget", type=float, default=100_000, help="今日总预算（元）")
    args = parser.parse_args()

    run_budget_test(bizdate=args.date, total_budget_yuan=args.budget)