Agent/examples/auto_put_ad/BID_ADJUSTMENT_README.md

预算约束下的智能出价调整系统

概述

本系统实现了基于裂变效率的多维出价调整决策，用于腾讯广告小程序投放的预算控制。

核心机制： 通过调整 oCPM 出价（bid_amount）来控制消耗速度，而非设置日预算限制。

业务背景

场景

• 昨日消耗：173,765 元（216 个有效广告 + 79 个样本不足广告）
• 今日预算：100,000 元
• 缩量幅度：-42%

关键发现

1. 控制机制：所有广告的 day_amount=0（不限日预算），实际通过调整 bid_amount（oCPM 出价）控制消耗
2. 决策维度：多层次决策（账户层、广告层、创意层）
3. 数据来源：
   • SQL 1：creative_detail 表 → 昨日效率数据（T0裂变系数、单用户成本）
   • SQL 2：ad_put_tencent_ad 表 → 当前广告状态（bid_amount、targeting、optimization_goal）
   • Python 合并：通过 ad_id 关联两个数据集

核心指标

效率分计算

效率分 = T0裂变系数 / 单用户成本

其中：
- T0裂变系数 = 裂变0层回流数 / 首层小程序打开数
- 单用户成本 = cost / 首层小程序打开数

数据有效性

• 有效广告：首层小程序打开数 >= 100 且有消耗
• 样本不足：首层小程序打开数 < 100 或无消耗

系统架构

新增工具

1. get_ad_current_status (data_query.py)

查询广告当前状态（出价、预算、定向等）

await get_ad_current_status(
    account_id=123456,
    ad_ids=[90397405754, 90397405755]  # 可选
)

返回字段：

• ad_id, ad_name, account_id
• bid_amount：当前出价（单位：分）
• day_amount：日预算限制（0=不限）
• ad_status：广告状态
• optimization_goal：转化目标
• targeting：定向配置（JSON）
• create_time：创建时间

2. budget_calculate_from_data (budget_calc.py) - 重构

基于昨日裂变效率数据计算今日出价调整方案

result = await budget_calculate_from_data(
    account_id=123456,
    total_budget_yuan=100000,
    bizdate="20260406",  # 默认 "yesterday"
    strategy="auto",  # 自动判断缩量/扩量
    tier1_ratio=0.15,  # Tier 1 占比
    tier2_ratio=0.35,  # Tier 2 占比
    min_bid_cents=10   # 最低出价（分）
)

核心逻辑：

1. 拉取昨日效率数据（data_query）
2. 拉取当前广告状态（get_ad_current_status）
3. 合并数据，按效率分分层
4. 计算出价调整方案
5. 输出详细调整说明

3. bid_adjustment_execute (budget_calc.py)

执行出价调整方案

result = await bid_adjustment_execute(
    adjustment_plan=plan["adjustment_plan"],
    account_id=123456
)

功能：

• 批量调整广告出价
• 暂停低效广告
• 返回执行结果统计

决策算法

Step 1: 判断缩量/扩量场景

scale_ratio = total_budget_yuan / yesterday_total

if scale_ratio < 0.7:
    strategy = "aggressive_scale_down"  # 大幅缩量（>30%）
elif scale_ratio < 1.0:
    strategy = "moderate_scale_down"    # 温和缩量
elif scale_ratio > 1.3:
    strategy = "scale_up"               # 扩量
else:
    strategy = "maintain"               # 基本持平

Step 2: 按效率分分层

# 按效率分降序排列
ads_sorted = sorted(ads_with_data, key=lambda x: x["efficiency"], reverse=True)

# 动态分层
tier1_size = min(30, int(total_count * 0.15))  # Top 15%，最多30个
tier2_size = min(70, int(total_count * 0.35))  # 中部35%，最多70个

tier1 = ads_sorted[:tier1_size]                # 核心广告
tier2 = ads_sorted[tier1_size:tier1_size+tier2_size]  # 观察广告
tier3 = ads_sorted[tier1_size+tier2_size:]     # 低效广告

Step 3: 应用出价调整策略

大幅缩量场景（aggressive_scale_down）

层级	出价调整幅度	决策原则
Tier 1（核心）	-5% ~ 0%	保护高效资产，轻微降价或不降
Tier 2（中部）	-10% ~ -15%	适度降价
Tier 3（低效）	-20% ~ -30%	大幅降价，低于最低出价则暂停
样本不足	-	全部暂停

温和缩量场景（moderate_scale_down）

层级	出价调整幅度
Tier 1	-3%
Tier 2	-8%
Tier 3	-15%
样本不足	暂停

扩量场景（scale_up）

层级	出价调整幅度	决策原则
Tier 1	+10% ~ +15%	优先加码高效广告
Tier 2	+5% ~ +10%	适度增加
Tier 3	0%	不调整
样本不足	-	启动，出价设为中位数

持平场景（maintain）

层级	出价调整幅度
所有层级	±3% 微调

Step 4: 出价边界检查

MIN_BID = 10  # 最低出价 0.10 元（10分）
MAX_BID = 10000  # 最高出价 100 元（10000分）

new_bid = int(current_bid * (1 + adjustment_ratio))
new_bid = max(MIN_BID, min(new_bid, MAX_BID))

if new_bid < MIN_BID:
    action = "pause"  # 低于最低出价，暂停广告

使用流程

1. 交互式使用

python examples/auto_put_ad/run.py
> 账户 123456 今日预算 10 万，帮我调整出价

系统行为：

1. Budget Agent 调用 budget_calculate_from_data
2. 输出分层出价调整方案
3. 等待用户确认
4. 用户确认后，调用 bid_adjustment_execute
5. 报告执行结果

2. 程序化调用

from examples.auto_put_ad.tools.budget_calc import (
    budget_calculate_from_data,
    bid_adjustment_execute
)

# 计算方案
result = await budget_calculate_from_data(
    account_id=123456,
    total_budget_yuan=100000
)

# 展示方案给用户
print(result.output)

# 用户确认后执行
if user_confirms:
    exec_result = await bid_adjustment_execute(
        adjustment_plan=result.data["adjustment_plan"],
        account_id=123456
    )
    print(exec_result.output)

输出示例

账户 123456 出价调整方案（缩量 42%，昨日消耗 173,765 元 → 今日预算 100,000 元）

策略：aggressive_scale_down（大幅缩量）

【Tier 1 核心广告 - 32个，保护资产，出价 -5%~0%】
ad_id: 90397405754 | 效率分: 8.93 | 昨日消耗: 3,100元
当前出价: 50分(0.50元) → 新出价: 48分(0.48元) | 动作: 调整

ad_id: 90397405755 | 效率分: 8.75 | 昨日消耗: 2,900元
当前出价: 52分(0.52元) → 新出价: 52分(0.52元) | 动作: 调整

【Tier 2 中部广告 - 76个，适度缩量，出价 -15%】
ad_id: 90397405800 | 效率分: 6.20 | 昨日消耗: 1,800元
当前出价: 45分(0.45元) → 新出价: 38分(0.38元) | 动作: 调整

【Tier 3 低效广告 - 108个，大幅削减，出价 -30%】
ad_id: 90397405900 | 效率分: 3.10 | 昨日消耗: 800元
当前出价: 40分(0.40元) → 新出价: 28分(0.28元) | 动作: 调整

【暂停广告 - 79个样本不足 + 15个出价过低】
ad_id: 90397406000 | 当前出价: 15分 → 低于最低出价 | 动作: 暂停

合计：预计消耗约 100,000 元（基于历史消耗和出价调整比例估算）

测试

运行单元测试

python3 examples/auto_put_ad/test_bid_adjustment_simple.py

测试覆盖：

• ✓ 策略判断逻辑
• ✓ 分层逻辑
• ✓ 出价调整计算
• ✓ 数据合并

测试结果示例

============================================================
测试 1: 策略判断
============================================================
✓ 缩量 50%: scale_ratio=0.50 → aggressive_scale_down
✓ 缩量 25%: scale_ratio=0.75 → moderate_scale_down
✓ 持平: scale_ratio=1.00 → maintain
✓ 扩量 50%: scale_ratio=1.50 → scale_up

============================================================
测试 2: 分层逻辑
============================================================
总广告数: 100
Tier 1 (Top 15%): 15 个，效率分范围 8.74 ~ 10.00
Tier 2 (中部 35%): 35 个，效率分范围 5.59 ~ 8.65
Tier 3 (尾部 50%): 50 个，效率分范围 1.09 ~ 5.50

技术要点

1. 数据合并逻辑

def _merge_efficiency_and_status(yesterday_data, current_status):
    """合并昨日效率数据和当前广告状态"""
    status_dict = {ad["ad_id"]: ad for ad in current_status}

    merged = []
    for yd in yesterday_data:
        ad_id = yd["ad_id"]
        if ad_id in status_dict:
            merged.append({**yd, **status_dict[ad_id]})

    return merged

2. 出价计算

# Tier 1: -5% ~ 0%
for ad in tier1:
    max_efficiency = tier1[0].get("efficiency", 1.0)
    adj_ratio = -0.05 if ad.get("efficiency", 0) < max_efficiency * 0.8 else 0
    new_bid = int(ad["bid_amount"] * (1 + adj_ratio))
    new_bid = max(new_bid, MIN_BID)

3. 批量执行

for item in adjustment_plan:
    if item["action"] == "pause":
        await ad_update(
            account_id=account_id,
            adgroup_id=item["ad_id"],
            configured_status="AD_STATUS_SUSPEND"
        )
    else:
        await ad_update(
            account_id=account_id,
            adgroup_id=item["ad_id"],
            bid_amount=item["new_bid"]
        )

配置参数

budget_calculate_from_data 参数

参数	类型	默认值	说明
account_id	int	必填	账户ID
total_budget_yuan	float	必填	今日总预算（元）
bizdate	str	"yesterday"	数据日期（YYYYMMDD）
strategy	str	"auto"	策略（auto/aggressive_scale_down/moderate_scale_down/scale_up/maintain）
tier1_ratio	float	0.15	Tier 1 占比
tier2_ratio	float	0.35	Tier 2 占比
min_bid_cents	int	10	最低出价（分）

注意事项

1. 调整节奏

• 单次调整后观察至少 2 小时再做下一次调整
• 避免频繁调整导致系统震荡

2. 出价边界

• 最低出价：10分（0.10元）
• 最高出价：10000分（100元）
• 低于最低出价的广告直接暂停

3. 样本量要求

• 有效数据门槛：首层小程序打开数 >= 100
• 样本不足的广告在缩量时暂停，扩量时启动

4. API 限制

• 单账户 QPS 限制 10
• 批量操作单次最多 50 条
• 出价和预算单位：分（1元 = 100分）

文件清单

核心文件

• tools/data_query.py：新增 get_ad_current_status 工具
• tools/budget_calc.py：重构 budget_calculate_from_data，新增 bid_adjustment_execute
• tools/ad_api.py：复用 ad_update 更新出价
• prompts/budget.prompt：更新为出价调整流程
• skills/budget_strategy.md：更新为多维出价调整策略

测试文件

• test_bid_adjustment.py：完整测试（需要 agent 框架）
• test_bid_adjustment_simple.py：简化测试（独立运行）

文档

• BID_ADJUSTMENT_README.md：本文档

未来扩展

多维属性增强（可选）

根据广告的定向、人群、转化目标进行微调：

def parse_targeting(targeting_json):
    """解析定向配置，提取渠道、人群包类型"""
    channel = "unknown"
    if "SITE_SET_MOMENTS" in str(targeting_json):
        channel = "moments"  # 朋友圈
    elif "SITE_SET_WECHAT_OFFICIAL_ACCOUNT" in str(targeting_json):
        channel = "official_account"  # 公众号
    return channel

# 在调整出价时考虑多维属性
for item in adjustment_plan:
    if item["action"] == "adjust":
        channel = parse_targeting(item.get("targeting"))
        optimization_goal = item.get("optimization_goal")

        # 高价值组合：朋友圈 + 关键页面浏览
        if channel == "moments" and optimization_goal == "OPTIMIZATIONGOAL_PROMOTION_VIEW_KEY_PAGE":
            # 缩量时保护，调整幅度减半
            item["adjustment_ratio"] *= 0.5
            item["new_bid"] = int(item["current_bid"] * (1 + item["adjustment_ratio"]))

版本历史

• v1.0 (2026-04-07)：初始实现
  • 基于裂变效率的出价调整
  • 多层级分层策略
  • 自动缩量/扩量判断
  • 批量执行功能

---

维护者： liulidong 最后更新： 2026-04-07