#!/usr/bin/env python3
"""
04_budget_pacing_overspend_forecasting.py — Track budget pacing, detect
overspend risk, and forecast month-end spend across Google Ads & Meta Ads
via GoMarble MCP tools + Claude API.
Outputs: budget_pacing.csv, budget_forecast_brief.txt
"""
# ┌──────────────────────────────────────────────────────────┐
# │ EDIT THESE VALUES BEFORE RUNNING │
# └──────────────────────────────────────────────────────────┘
ANTHROPIC_API_KEY = "sk-ant-api03-..." # Your Anthropic API key
GOMARBLE_API_KEY = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx" # From GoMarble dashboard
META_AD_ACCOUNT = "" # e.g. "act_123456" (leave "" to skip Meta)
GOOGLE_ADS_CID = "" # e.g. "7556164258" (leave "" to skip Google)
MONTHLY_BUDGET = 0 # Total monthly budget in $ (0 = auto-detect)
# NOTE: At least one of META_AD_ACCOUNT or GOOGLE_ADS_CID must be set.
import csv, json, re, sys, time, subprocess
from datetime import datetime, timedelta
try:
import requests
except ImportError:
subprocess.check_call([sys.executable, "-m", "pip", "install", "requests"])
import requests
API_URL = "https://api.anthropic.com/v1/messages"
MODEL = "claude-sonnet-4-20250514"
def safe_float(val, default=0.0):
if val is None: return default
try:
s = str(val).replace(",", "").replace("$", "").replace("%", "").strip()
return float(s) if s else default
except (ValueError, TypeError): return default
def try_parse_json(text):
text = re.sub(r"\`\`\`(?:json)?\s*", "", text)
for open_ch, close_ch in [("{", "}"), ("[", "]")]:
pos = 0
while pos < len(text):
start = text.find(open_ch, pos)
if start == -1: break
depth = 0
for i in range(start, len(text)):
if text[i] == open_ch: depth += 1
elif text[i] == close_ch: depth -= 1
if depth == 0:
try: return json.loads(text[start : i + 1])
except json.JSONDecodeError: pass
break
pos = start + 1
return None
def to_rows(text):
data = try_parse_json(text)
if data is None: return []
if isinstance(data, list): return data
if isinstance(data, dict):
for key in ("results", "rows", "data", "campaigns"):
if key in data and isinstance(data[key], list): return data[key]
return [data]
return []
def flatten_row(row):
if not isinstance(row, dict): return {}
flat = {}
def _flatten(obj, prefix=""):
if isinstance(obj, dict):
for k, v in obj.items(): _flatten(v, f"{prefix}_{k}" if prefix else k)
elif not isinstance(obj, list): flat[prefix] = obj
_flatten(row)
return flat
def banner(step, total, title):
print(f"\n{'─'*60}\n Step {step}/{total} │ {title}\n{'─'*60}")
def write_csv(path, rows, fields):
if not rows: print(f" (no data for {path})"); return
with open(path, "w", newline="", encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
w.writeheader(); w.writerows(rows)
print(f" -> {path} — {len(rows)} rows")
def mcp_request(messages, system=None, max_tokens=16000):
headers = {"x-api-key": ANTHROPIC_API_KEY, "anthropic-version": "2023-06-01",
"anthropic-beta": "mcp-client-2025-04-04", "content-type": "application/json"}
payload = {"model": MODEL, "max_tokens": max_tokens, "messages": messages,
"mcp_servers": [{"type": "url", "url": "https://apps.gomarble.ai/mcp-api/sse",
"name": "gomarble", "authorization_token": GOMARBLE_API_KEY}]}
if system: payload["system"] = system
for attempt in range(3):
try:
resp = requests.post(API_URL, headers=headers, json=payload, timeout=120)
if resp.status_code != 200: print(f" API error {resp.status_code}: {resp.text[:300]}"); resp.raise_for_status()
return resp.json()
except requests.exceptions.ReadTimeout:
print(f" Timeout (attempt {attempt+1}/3), retrying...")
if attempt == 2: raise
except requests.exceptions.ConnectionError:
print(f" Connection error (attempt {attempt+1}/3), retrying..."); time.sleep(5)
if attempt == 2: raise
def get_text(response):
parts = []
for block in response.get("content", []):
if block.get("type") == "text": parts.append(block["text"])
elif block.get("type") == "mcp_tool_result":
content = block.get("content", "")
if isinstance(content, list):
for c in content:
if isinstance(c, dict) and c.get("text"): parts.append(c["text"])
elif isinstance(content, str): parts.append(content)
return "\n".join(parts)
# ── Step 1: Month-to-Date Spend ───────────────────────────────
def step1_mtd_spend(meta_id, google_id):
banner(1, 4, "Month-to-Date Spend by Campaign")
meta_text, google_text = None, None
if meta_id:
print(" Fetching Meta MTD spend...")
msg = (
f"Run facebook_get_adaccount_insights with ad_account_id='{meta_id}', "
"level='campaign', date_preset='this_month', "
"fields=['campaign_name','spend','impressions','clicks','ctr','cpc',"
"'purchase_roas','actions','daily_budget','lifetime_budget']. "
"Return the raw JSON."
)
resp = mcp_request([{"role": "user", "content": msg}])
meta_text = get_text(resp)
print(f" Meta campaigns: {len(to_rows(meta_text))}")
if google_id:
print(" Fetching Google MTD spend...")
msg = (
f"Run google_ads_run_gaql with customer_id='{google_id}' and query:\n"
"SELECT campaign.name, campaign.status, campaign.campaign_budget, "
"metrics.cost_micros, metrics.impressions, metrics.clicks, "
"metrics.conversions, metrics.conversions_value, metrics.ctr "
"FROM campaign WHERE segments.date DURING THIS_MONTH "
"AND campaign.status = 'ENABLED' "
"ORDER BY metrics.cost_micros DESC LIMIT 50\n\n"
"Return the raw JSON."
)
resp = mcp_request([{"role": "user", "content": msg}])
google_text = get_text(resp)
print(f" Google campaigns: {len(to_rows(google_text))}")
return meta_text, google_text
# ── Step 2: Daily Spend Trend ─────────────────────────────────
def step2_daily_spend(meta_id, google_id):
banner(2, 4, "Daily Spend Trend (this month)")
meta_daily, google_daily = None, None
if meta_id:
msg = (
f"Run facebook_get_adaccount_insights with ad_account_id='{meta_id}', "
"level='account', date_preset='this_month', time_increment='1', "
"fields=['spend','impressions','clicks','purchase_roas']. "
"Return the raw JSON."
)
resp = mcp_request([{"role": "user", "content": msg}])
meta_daily = get_text(resp)
print(f" Meta daily points: {len(to_rows(meta_daily))}")
if google_id:
msg = (
f"Run google_ads_run_gaql with customer_id='{google_id}' and query:\n"
"SELECT segments.date, metrics.cost_micros, metrics.impressions, "
"metrics.clicks, metrics.conversions "
"FROM customer WHERE segments.date DURING THIS_MONTH "
"ORDER BY segments.date ASC\n\nReturn the raw JSON."
)
resp = mcp_request([{"role": "user", "content": msg}])
google_daily = get_text(resp)
print(f" Google daily points: {len(to_rows(google_daily))}")
return meta_daily, google_daily
# ── Step 3: Generate Pacing Report ────────────────────────────
def step3_generate_report(meta_campaign, google_campaign, meta_daily, google_daily, monthly_budget):
banner(3, 4, "Generate Budget Pacing Report")
today = datetime.now()
days_in_month = (today.replace(month=today.month % 12 + 1, day=1) - timedelta(days=1)).day if today.month < 12 else 31
days_elapsed = today.day
days_remaining = days_in_month - days_elapsed
pct_through = days_elapsed / days_in_month * 100
prompt = f"""You are a media buying budget analyst. Write a budget pacing and overspend forecast report.
Today is day {days_elapsed} of {days_in_month} ({pct_through:.0f}% through the month).
Days remaining: {days_remaining}.
Monthly budget target: ${monthly_budget if monthly_budget > 0 else 'Auto-detect from campaign budgets'}.
Sections:
1. PACING DASHBOARD — For each platform and campaign:
- MTD spend vs expected spend (at {pct_through:.0f}% pacing)
- Pacing status: ON TRACK / UNDERPACING / OVERPACING
- Projected month-end spend at current rate
2. OVERSPEND RISK — Campaigns at risk of exceeding budget:
- Current daily run rate vs daily budget
- Projected overspend amount
- Risk level (LOW / MEDIUM / HIGH / CRITICAL)
3. UNDERSPEND OPPORTUNITIES — Campaigns with budget headroom:
- How much unspent budget is available
- Recommendations to capture it
4. DAILY SPEND TREND — Is spend accelerating or decelerating?
5. FORECAST — Three scenarios for month-end:
- Conservative (reduce daily spend by 20%)
- Current trajectory
- Aggressive (increase daily spend by 20%)
6. BUDGET REALLOCATION — Move budget from underpacing to overpacing campaigns
7. ACTION ITEMS — Specific budget adjustments to make today
Use exact dollar amounts and campaign names throughout.
### Meta Campaign MTD Spend
{meta_campaign[:4000] if meta_campaign else 'N/A'}
### Google Campaign MTD Spend
{google_campaign[:4000] if google_campaign else 'N/A'}
### Meta Daily Spend
{meta_daily[:3000] if meta_daily else 'N/A'}
### Google Daily Spend
{google_daily[:3000] if google_daily else 'N/A'}"""
resp = mcp_request([{"role": "user", "content": prompt}])
report = get_text(resp)
print(f" Report: {len(report)} chars")
return report
# ── Step 4: Output ────────────────────────────────────────────
def step4_output(meta_campaign, google_campaign, report):
banner(4, 4, "Write Output Files")
rows = []
if meta_campaign:
for row in to_rows(meta_campaign):
flat = flatten_row(row)
flat["platform"] = "Meta"
rows.append(flat)
if google_campaign:
for row in to_rows(google_campaign):
flat = flatten_row(row)
flat["platform"] = "Google"
rows.append(flat)
if rows:
fields = ["platform"] + [k for k in rows[0].keys() if k != "platform"][:10]
write_csv("budget_pacing.csv", rows, fields)
with open("budget_forecast_brief.txt", "w", encoding="utf-8") as f:
f.write("BUDGET PACING & OVERSPEND FORECAST\n")
f.write(f"Generated: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write("=" * 60 + "\n\n")
f.write(report)
print(f" -> budget_forecast_brief.txt — {len(report)} chars")
print(f"\n{'='*60}\n BUDGET PACING COMPLETE\n{'='*60}")
def main():
meta_id = META_AD_ACCOUNT.strip() or None
google_id = GOOGLE_ADS_CID.strip().replace("-", "") or None
if not meta_id and not google_id:
sys.exit("Set at least one of META_AD_ACCOUNT or GOOGLE_ADS_CID at the top.")
if "..." in ANTHROPIC_API_KEY or "xxxx" in GOMARBLE_API_KEY:
sys.exit("Edit ANTHROPIC_API_KEY and GOMARBLE_API_KEY at the top of the file.")
print(f"\nBudget Pacing & Overspend Forecasting")
if meta_id: print(f" Meta: {meta_id}")
if google_id: print(f" Google: {google_id}")
if MONTHLY_BUDGET > 0: print(f" Monthly budget: ${MONTHLY_BUDGET:,.0f}")
meta_campaign, google_campaign = step1_mtd_spend(meta_id, google_id)
meta_daily, google_daily = step2_daily_spend(meta_id, google_id)
report = step3_generate_report(meta_campaign, google_campaign, meta_daily, google_daily, MONTHLY_BUDGET)
step4_output(meta_campaign, google_campaign, report)
if __name__ == "__main__":
main()