如何通过单次Meta API调用批量更新多个广告组daily_budget

我尝试通过单次调用更新多个广告组的daily_budget参数，目前已实现批量调用方式，但根据Meta批量请求文档：

批量请求每个批次最多支持50个请求。批次中的每个调用会单独计入API调用限制和资源限制，例如包含10个API调用的批次会被算作10次调用，且每个调用对CPU资源限制的影响与单独调用一致。

这对大型账户来说是问题，因为可能需要更新大量广告组。因此我希望实现一个通过单次API调用完成该任务的函数，就像通过单次Meta API调用获取多个广告组daily_budget的方式一样（如下列代码所示）：

class MetaAPI:
    def __init__(
        self, access_token: str, ad_account_id_val: str, sqlapi: SQLAPI, test=False
    ):
        self.access_token = access_token
        self.ad_account_id = "act_" + ad_account_id_val
        self.sqlapi = sqlapi
        self.test = test

        self.root_url = "https://graph.facebook.com/v18.0/"
        self.campaigns_ext = "/campaigns"
        self.adsets_ext = "/adsets"
        self.insights_ext = "/insights"

    def fetch_group_adset_insights(
            self, adset_id_list: List[str], insights_list: List[str]
        ) -> Dict[str, Tuple[Dict[str, float], str]]:
            params_insights_get = {
                "access_token": self.access_token,
                "level": "adset",
                "filtering": str(
                    [
                        {
                            "field": "adset.id",
                            "operator": "IN",
                            "value": adset_id_list,
                        }
                    ]
                ),
                "fields": ",".join(
                    insights_list + ["optimization_goal", "adset_id"]
                ),  # 'objective,impressions',
                "time_range[since]": f"{self.date_start}",
                "time_range[until]": f"{self.date_end}",
            }

            response_insights_get = rq.get(
                self.root_url + self.ad_account_id + self.insights_ext,
                params=params_insights_get,
            )

我已实现单广告组更新的函数：

def update_object_budgets(self, object_id: str, budget: float):
        """
        Updates the budget of the object (adset or campaign) to the given value.
        """

        url = self.root_url + object_id
        params = {"access_token": self.access_token, "daily_budget": f"{budget}"}

        response = rq.post(url, data=params)

        return response.status_code == 200  # 200: success

但尝试扩展到多广告组场景时失败：

def update_objects_budgets(
        self, object_id_list: List[str], budget_list: List[float]
    ):
        """
        Updates the budget of the object (adset or campaign) to the given value.
        """

        url = self.root_url
        params = {
            "access_token": self.access_token,
            "adset_id": object_id_list,
            "daily_budget": budgets_list,
        }

        response = rq.post(url, data=params)

        return response.status_code == 200  # 200: success

调用该函数返回错误：

{'error': {'message': 'Unsupported post request. Please read the Graph API documentation at https://developers.facebook.com/docs/graph-api', 'type': 'GraphMethodException', 'code': 100, 'error_subcode': 33, 'fbtrace_id': 'APtBihvkXqpal9gGI9wCwXZ'}}

Meta Graph API不支持通过单次POST调用直接批量更新多个广告组的daily_budget，你尝试的传数组参数的方式不符合API规范，这也是报错的核心原因。针对大型账户批量更新需求，有两种可行方案：

方案1：优化批量请求逻辑 #

虽然批量请求按单个调用计数，但可以通过以下方式提升处理效率：

• 按50个广告组为一批次拆分请求，减少HTTP请求总次数
• 异步并发执行多个批次，缩短整体处理时间
• 加入重试机制，处理API调用失败的情况

示例优化后的批量更新代码：

import requests as rq
from typing import List, Dict
from concurrent.futures import ThreadPoolExecutor

class MetaAPI:
    # 保留原有__init__等方法...

    def update_object_budget(self, object_id: str, budget: float) -> bool:
        """单广告组预算更新逻辑"""
        url = self.root_url + object_id
        params = {"access_token": self.access_token, "daily_budget": f"{budget}"}
        response = rq.post(url, data=params)
        return response.status_code == 200

    def update_objects_budgets_batch(self, object_id_list: List[str], budget_list: List[float], batch_size: int = 50) -> Dict[str, bool]:
        """批量更新广告组预算，并发处理批次"""
        results = {}
        # 配对广告组ID与对应预算
        object_budget_pairs = list(zip(object_id_list, budget_list))
        
        # 拆分为50个一组的批次
        batches = [object_budget_pairs[i:i+batch_size] for i in range(0, len(object_budget_pairs), batch_size)]
        
        # 并发处理每个批次
        def process_batch(batch):
            batch_results = {}
            for obj_id, budget in batch:
                success = self.update_object_budget(obj_id, budget)
                batch_results[obj_id] = success
            return batch_results
        
        with ThreadPoolExecutor(max_workers=5) as executor:
            futures = [executor.submit(process_batch, batch) for batch in batches]
            for future in futures:
                results.update(future.result())
        
        return results

方案2：使用Meta Bulk API（批量操作API） #

Meta提供的Bulk API支持单次提交多个更新操作，且仅计为1次API调用（需注意该API有单独的调用限制和权限要求），适合大规模广告组更新场景。

使用Bulk API的核心步骤：

1. 构造包含多个更新操作的JSON payload，每个操作指定广告组ID和要更新的daily_budget
2. 向广告账户的/bulk端点发送POST请求

示例代码：

def update_objects_budgets_bulk(self, object_id_list: List[str], budget_list: List[float]) -> dict:
    """使用Bulk API批量更新广告组预算"""
    url = self.root_url + self.ad_account_id + "/bulk"
    
    # 构造批量操作条目
    entries = []
    for obj_id, budget in zip(object_id_list, budget_list):
        entries.append({
            "method": "POST",
            "relative_url": obj_id,
            "body": f"access_token={self.access_token}&daily_budget={budget}"
        })
    
    payload = {
        "access_token": self.access_token,
        "operations": entries
    }
    
    response = rq.post(url, json=payload)
    return response.json()

注意事项： #

• Bulk API需要ads_management等权限，需确保访问令牌具备对应权限
• Bulk API的响应是异步的，需通过返回的bulk_request_id查询最终操作状态
• 不同API版本对Bulk API的支持存在差异，建议使用最新稳定版

内容的提问来源于stack exchange，提问作者MercuryProbe