#!/usr/bin/env python3
"""
01_full_account_audits.py — Full performance audit across Google Ads & Meta Ads
via GoMarble MCP tools + Claude API.
Outputs: audit_campaigns.csv, audit_creatives.csv, audit_brief.txt
"""
# ┌──────────────────────────────────────────────────────────┐
# │ EDIT THESE VALUES BEFORE RUNNING │
# └──────────────────────────────────────────────────────────┘
ANTHROPIC_API_KEY = "your-anthropic-api-key-here" # Your Anthropic API key
GOMARBLE_API_KEY = "your-gomarble-api-key-here" # From GoMarble dashboard
META_AD_ACCOUNT = "" # e.g. "act_123456" (leave "" to skip Meta)
GOOGLE_ADS_CID = "" # e.g. "7556164258" (leave "" to skip Google)
# NOTE: At least one of META_AD_ACCOUNT or GOOGLE_ADS_CID must be set.
import csv, json, re, sys, time, subprocess
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"
# ── Helpers ──────────────────────────────────────────────────
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", "ads", "adsets"):
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")
# ── MCP / API layer ─────────────────────────────────────────
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=90)
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: Account Structure ───────────────────────────────
def step1_account_structure(meta_id, google_id):
banner(1, 7, "Account Structure")
results = {"meta": None, "google": None}
if meta_id:
print(f" Fetching Meta account: {meta_id}")
msg = (
f"Run facebook_get_details_of_ad_account with ad_account_id='{meta_id}'. "
"Return the raw JSON with account name, status, currency, timezone, spend cap, etc."
)
resp = mcp_request([{"role": "user", "content": msg}])
results["meta"] = get_text(resp)
print(f" Meta account info: {len(results['meta'])} chars")
if google_id:
print(f" Fetching Google account: {google_id}")
msg = (
f"Run google_ads_run_gaql with customer_id='{google_id}' and query:\n"
"SELECT customer.id, customer.descriptive_name, customer.currency_code, "
"customer.time_zone, metrics.impressions, metrics.clicks, metrics.cost_micros, "
"metrics.conversions, metrics.conversions_value "
"FROM customer WHERE segments.date DURING LAST_30_DAYS LIMIT 1\n\n"
"Return the raw JSON."
)
resp = mcp_request([{"role": "user", "content": msg}])
results["google"] = get_text(resp)
print(f" Google account info: {len(results['google'])} chars")
return results
# ── Step 2: Campaign Performance (30d) ──────────────────────
def step2_campaign_performance(meta_id, google_id):
banner(2, 7, "Campaign Performance (30d)")
campaigns = []
if meta_id:
print(" Fetching Meta campaign performance...")
msg = (
f"Run facebook_get_adaccount_insights with ad_account_id='{meta_id}', "
"level='campaign', date_preset='last_30d', "
"fields=['campaign_name','spend','impressions','clicks','ctr','cpc',"
"'purchase_roas','actions']. "
"Return the raw JSON."
)
resp = mcp_request([{"role": "user", "content": msg}])
text = get_text(resp)
for row in to_rows(text):
flat = flatten_row(row)
spend = safe_float(flat.get("spend"))
impressions = safe_float(flat.get("impressions"))
clicks = safe_float(flat.get("clicks"))
ctr = safe_float(flat.get("ctr"))
cpc = safe_float(flat.get("cpc"))
roas = safe_float(flat.get("purchase_roas", flat.get("purchase_roas_value")))
conversions = 0
actions = row.get("actions", [])
if isinstance(actions, list):
for a in actions:
if isinstance(a, dict) and a.get("action_type") in ("purchase", "offsite_conversion.fb_pixel_purchase"):
conversions = safe_float(a.get("value"))
break
flags = []
if spend > 100 and conversions == 0:
flags.append("ZERO_CONVERSIONS")
if spend > 200 and roas > 0 and roas < 1:
flags.append("LOW_ROAS")
if impressions > 0 and clicks == 0:
flags.append("NO_CLICKS")
campaigns.append({
"platform": "Meta", "campaign_name": flat.get("campaign_name", ""),
"spend": spend, "impressions": impressions, "clicks": clicks,
"ctr": ctr, "cpc": cpc, "roas": roas, "conversions": conversions,
"flags": "|".join(flags),
})
print(f" Meta: {len([c for c in campaigns if c['platform']=='Meta'])} campaigns")
if google_id:
print(" Fetching Google campaign performance...")
msg = (
f"Run google_ads_run_gaql with customer_id='{google_id}' and query:\n"
"SELECT campaign.id, campaign.name, campaign.status, "
"metrics.impressions, metrics.clicks, metrics.cost_micros, "
"metrics.conversions, metrics.conversions_value, metrics.ctr, "
"metrics.average_cpc "
"FROM campaign WHERE segments.date DURING LAST_30_DAYS "
"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}])
text = get_text(resp)
for row in to_rows(text):
flat = flatten_row(row)
# The API may return cost in micros (cost_micros) or dollars (cost/metrics_cost)
cost_micros = safe_float(flat.get("cost_micros", flat.get("metrics_cost_micros", 0)))
metrics_cost = safe_float(flat.get("cost", flat.get("metrics_cost", 0)))
if cost_micros > 0:
spend = cost_micros / 1_000_000 if cost_micros > 1000 else cost_micros
elif metrics_cost > 0:
spend = metrics_cost
else:
spend = 0.0
impressions = safe_float(flat.get("impressions", flat.get("metrics_impressions")))
clicks = safe_float(flat.get("clicks", flat.get("metrics_clicks")))
conversions = safe_float(flat.get("conversions", flat.get("metrics_conversions")))
conv_value = safe_float(flat.get("conversions_value", flat.get("metrics_conversions_value", flat.get("metrics_conversionsValue", 0))))
ctr = safe_float(flat.get("ctr", flat.get("metrics_ctr")))
avg_cpc_raw = safe_float(flat.get("average_cpc", flat.get("metrics_average_cpc", flat.get("metrics_averageCpc", 0))))
# averageCpc from GoMarble API appears to be in micros
cpc = avg_cpc_raw / 1_000_000 if avg_cpc_raw > 1000 else avg_cpc_raw
roas = conv_value / spend if spend > 0 else 0
flags = []
if spend > 100 and conversions == 0:
flags.append("ZERO_CONVERSIONS")
if spend > 200 and roas > 0 and roas < 1:
flags.append("LOW_ROAS")
if impressions > 0 and clicks == 0:
flags.append("NO_CLICKS")
campaigns.append({
"platform": "Google",
"campaign_name": flat.get("campaign_name", flat.get("name", flat.get("metrics_name", ""))),
"spend": round(spend, 2), "impressions": impressions, "clicks": clicks,
"ctr": ctr, "cpc": round(cpc, 2), "roas": round(roas, 2),
"conversions": conversions, "flags": "|".join(flags),
})
print(f" Google: {len([c for c in campaigns if c['platform']=='Google'])} campaigns")
flagged = [c for c in campaigns if c["flags"]]
if flagged:
print(f" Flagged campaigns: {len(flagged)}")
for c in flagged[:5]:
print(f" - {c['campaign_name']}: {c['flags']}")
return campaigns
# ── Step 3: Creative Fatigue ────────────────────────────────
def step3_creative_fatigue(meta_id, google_id):
banner(3, 7, "Creative Fatigue Detection")
creatives = []
if meta_id:
print(" Fetching Meta weekly ad-level data...")
msg = (
f"Run facebook_get_adaccount_insights with ad_account_id='{meta_id}', "
"level='ad', date_preset='last_30d', time_increment='7', "
"fields=['ad_name','campaign_name','impressions','clicks','ctr','frequency','spend']. "
"Return the raw JSON."
)
resp = mcp_request([{"role": "user", "content": msg}])
text = get_text(resp)
ad_weeks = {}
for row in to_rows(text):
flat = flatten_row(row)
ad_name = flat.get("ad_name", "")
if ad_name:
ad_weeks.setdefault(ad_name, []).append(flat)
for ad_name, weeks in ad_weeks.items():
if len(weeks) < 2:
continue
first, last = weeks[0], weeks[-1]
freq_first = safe_float(first.get("frequency"))
freq_last = safe_float(last.get("frequency"))
ctr_first = safe_float(first.get("ctr"))
ctr_last = safe_float(last.get("ctr"))
freq_change = ((freq_last - freq_first) / freq_first * 100) if freq_first > 0 else 0
ctr_change = ((ctr_last - ctr_first) / ctr_first * 100) if ctr_first > 0 else 0
fatigued = freq_change > 30 and ctr_change < -20
creatives.append({
"platform": "Meta", "ad_name": ad_name,
"campaign_name": last.get("campaign_name", ""),
"freq_week1": round(freq_first, 2), "freq_week4": round(freq_last, 2),
"freq_change_pct": round(freq_change, 1),
"ctr_week1": round(ctr_first, 4), "ctr_week4": round(ctr_last, 4),
"ctr_change_pct": round(ctr_change, 1),
"fatigued": "YES" if fatigued else "NO",
})
print(f" Meta creatives analyzed: {len(creatives)}")
if google_id:
print(" Fetching Google daily ad-level data...")
msg = (
f"Run google_ads_run_gaql with customer_id='{google_id}' and query:\n"
"SELECT ad_group_ad.ad.name, campaign.name, segments.date, "
"metrics.impressions, metrics.clicks, metrics.ctr, metrics.cost_micros "
"FROM ad_group_ad WHERE segments.date DURING LAST_30_DAYS "
"AND campaign.status = 'ENABLED' AND ad_group_ad.status = 'ENABLED' "
"ORDER BY segments.date ASC LIMIT 500\n\n"
"Return the raw JSON."
)
resp = mcp_request([{"role": "user", "content": msg}])
text = get_text(resp)
ad_days = {}
for row in to_rows(text):
flat = flatten_row(row)
ad_name = flat.get("ad_name", flat.get("name", flat.get("ad_group_ad_ad_name", "")))
if ad_name:
ad_days.setdefault(ad_name, []).append(flat)
for ad_name, days in ad_days.items():
n = len(days)
if n < 7:
continue
week_size = n // 4 or 1
week1, week4 = days[:week_size], days[-week_size:]
def avg_ctr(bucket):
imps = sum(safe_float(d.get("impressions", d.get("metrics_impressions"))) for d in bucket)
clicks = sum(safe_float(d.get("clicks", d.get("metrics_clicks"))) for d in bucket)
return clicks / imps if imps > 0 else 0
ctr_w1, ctr_w4 = avg_ctr(week1), avg_ctr(week4)
ctr_change = ((ctr_w4 - ctr_w1) / ctr_w1 * 100) if ctr_w1 > 0 else 0
imp_w1 = sum(safe_float(d.get("impressions", d.get("metrics_impressions"))) for d in week1) / len(week1)
imp_w4 = sum(safe_float(d.get("impressions", d.get("metrics_impressions"))) for d in week4) / len(week4)
imp_change = ((imp_w4 - imp_w1) / imp_w1 * 100) if imp_w1 > 0 else 0
fatigued = imp_change > 30 and ctr_change < -20
creatives.append({
"platform": "Google", "ad_name": ad_name,
"campaign_name": days[0].get("campaign_name", ""),
"freq_week1": round(imp_w1, 0), "freq_week4": round(imp_w4, 0),
"freq_change_pct": round(imp_change, 1),
"ctr_week1": round(ctr_w1, 4), "ctr_week4": round(ctr_w4, 4),
"ctr_change_pct": round(ctr_change, 1),
"fatigued": "YES" if fatigued else "NO",
})
print(f" Google creatives analyzed: {len([c for c in creatives if c['platform']=='Google'])}")
fatigued_count = len([c for c in creatives if c["fatigued"] == "YES"])
print(f" Fatigued creatives: {fatigued_count}")
return creatives
# ── Step 4: Audience Segments (Meta only) ────────────────────
def step4_audience_segments(meta_id):
banner(4, 7, "Audience Segments (Meta)")
if not meta_id:
print(" Skipped (no Meta account)")
return None
print(" Fetching user segment breakdown...")
msg = (
f"Run facebook_get_adaccount_insights with ad_account_id='{meta_id}', "
"level='campaign', date_preset='last_30d', "
"fields=['campaign_name','spend','impressions','clicks','ctr','purchase_roas','actions'], "
"breakdowns=['user_segment_key']. "
"Return the raw JSON."
)
resp = mcp_request([{"role": "user", "content": msg}])
text = get_text(resp)
rows = to_rows(text)
print(f" Audience segment rows: {len(rows)}")
segments = {}
for row in rows:
flat = flatten_row(row)
seg = flat.get("user_segment_key", flat.get("user_segment", "unknown"))
if seg not in segments:
segments[seg] = {"spend": 0, "impressions": 0, "clicks": 0}
segments[seg]["spend"] += safe_float(flat.get("spend"))
segments[seg]["impressions"] += safe_float(flat.get("impressions"))
segments[seg]["clicks"] += safe_float(flat.get("clicks"))
for seg, data in segments.items():
ctr = data["clicks"] / data["impressions"] * 100 if data["impressions"] > 0 else 0
print(f" {seg}: ${data['spend']:.0f} spend, {ctr:.2f}% CTR")
return {"raw_text": text, "segments": segments}
# ── Step 5: Budget Allocation ───────────────────────────────
def step5_budget_allocation(campaigns):
banner(5, 7, "Budget Allocation Analysis")
if not campaigns:
print(" No campaigns to analyze")
return []
total_spend = sum(safe_float(c["spend"]) for c in campaigns)
if total_spend == 0:
print(" No spend data")
return campaigns
for c in campaigns:
c["spend_share"] = safe_float(c["spend"]) / total_spend
roas_vals = sorted([safe_float(c["roas"]) for c in campaigns if safe_float(c["roas"]) > 0])
median_roas = roas_vals[len(roas_vals) // 2] if roas_vals else 0
spend_shares = sorted([c["spend_share"] for c in campaigns])
p70 = spend_shares[int(len(spend_shares) * 0.7)] if spend_shares else 0
p30 = spend_shares[int(len(spend_shares) * 0.3)] if spend_shares else 0
overfunded, underfunded = [], []
for c in campaigns:
roas = safe_float(c["roas"])
share = c["spend_share"]
if share >= p70 and roas < median_roas:
c["budget_flag"] = "OVERFUNDED"
overfunded.append(c["campaign_name"])
elif share <= p30 and roas > median_roas:
c["budget_flag"] = "UNDERFUNDED"
underfunded.append(c["campaign_name"])
else:
c["budget_flag"] = ""
print(f" Median ROAS: {median_roas:.2f}")
print(f" Overfunded ({len(overfunded)}): {', '.join(overfunded[:3]) or 'none'}")
print(f" Underfunded ({len(underfunded)}): {', '.join(underfunded[:3]) or 'none'}")
return campaigns
# ── Step 6: Generate Brief ──────────────────────────────────
def step6_generate_brief(account_info, campaigns, creatives, audience_data):
banner(6, 7, "Generate Audit Brief")
campaign_summary = json.dumps(campaigns[:30], indent=2, default=str)
fatigued = [c for c in creatives if c.get("fatigued") == "YES"]
fatigue_summary = json.dumps(fatigued[:20], indent=2, default=str) if fatigued else "No fatigued creatives detected."
audience_summary = audience_data.get("raw_text", "N/A")[:2000] if audience_data else "N/A (no Meta account)"
flagged = [c for c in campaigns if c.get("flags")]
flag_summary = json.dumps(flagged[:10], indent=2, default=str) if flagged else "No flags."
overfunded = [c for c in campaigns if c.get("budget_flag") == "OVERFUNDED"]
underfunded = [c for c in campaigns if c.get("budget_flag") == "UNDERFUNDED"]
budget_summary = (
f"Overfunded: {json.dumps([c['campaign_name'] for c in overfunded], default=str)}\n"
f"Underfunded: {json.dumps([c['campaign_name'] for c in underfunded], default=str)}"
)
by_roas = sorted(campaigns, key=lambda c: safe_float(c.get("roas")), reverse=True)
top5 = by_roas[:5]
bottom5 = by_roas[-5:] if len(by_roas) > 5 else []
prompt = f"""You are a senior paid media strategist. Write a concise performance audit brief.
Sections:
1. EXECUTIVE SUMMARY — 3-4 lines: overall health, total spend, blended ROAS, key takeaway
2. PERFORMANCE FLAGS — Each flagged campaign with the issue and recommended action
3. CREATIVE FATIGUE ALERTS — Fatigued ads with frequency/CTR trends and refresh recommendations
4. AUDIENCE SEGMENT SPLIT — (Meta only) New vs Engaged vs Existing spend/CTR breakdown
5. BUDGET REALLOCATION — Specific $ amounts to shift from overfunded to underfunded campaigns
6. TOP 5 CAMPAIGNS — By ROAS, with key metrics
7. BOTTOM 5 CAMPAIGNS — By ROAS, with recommended actions
Be specific with numbers. Use campaign names. Give actionable next steps.
### Account Info
Meta: {(account_info.get('meta') or 'N/A')[:1500]}
Google: {(account_info.get('google') or 'N/A')[:1500]}
### Campaign Performance (top 30)
{campaign_summary[:5000]}
### Performance Flags
{flag_summary[:2000]}
### Fatigued Creatives
{fatigue_summary[:3000]}
### Audience Segments
{audience_summary}
### Budget Analysis
{budget_summary}
### Top 5 by ROAS
{json.dumps(top5, indent=2, default=str)[:2000]}
### Bottom 5 by ROAS
{json.dumps(bottom5, indent=2, default=str)[:2000]}"""
resp = mcp_request([{"role": "user", "content": prompt}])
brief = get_text(resp)
print(f" Brief generated: {len(brief)} chars")
return brief
# ── Step 7: Output Files ────────────────────────────────────
def step7_output(campaigns, creatives, brief):
banner(7, 7, "Write Output Files")
campaign_fields = [
"platform", "campaign_name", "spend", "impressions", "clicks",
"ctr", "cpc", "roas", "conversions", "flags", "budget_flag", "spend_share",
]
campaign_rows = []
for c in campaigns:
row = {k: c.get(k, "") for k in campaign_fields}
if isinstance(row.get("spend_share"), float):
row["spend_share"] = f"{row['spend_share']:.1%}"
campaign_rows.append(row)
write_csv("audit_campaigns.csv", campaign_rows, campaign_fields)
creative_fields = [
"platform", "ad_name", "campaign_name",
"freq_week1", "freq_week4", "freq_change_pct",
"ctr_week1", "ctr_week4", "ctr_change_pct", "fatigued",
]
write_csv("audit_creatives.csv", creatives, creative_fields)
with open("audit_brief.txt", "w", encoding="utf-8") as f:
f.write("AD PERFORMANCE AUDIT\n")
f.write(f"Generated: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write("=" * 60 + "\n\n")
f.write(brief)
print(f" -> audit_brief.txt — {len(brief)} chars")
print(f"\n{'='*60}")
print(" AUDIT COMPLETE")
print(f"{'='*60}")
print(f" Campaigns analyzed: {len(campaigns)}")
print(f" Creatives analyzed: {len(creatives)}")
print(f" Flagged campaigns: {len([c for c in campaigns if c.get('flags')])}")
print(f" Fatigued creatives: {len([c for c in creatives if c.get('fatigued') == 'YES'])}")
print(f"\n Output files:")
print(f" audit_campaigns.csv")
print(f" audit_creatives.csv")
print(f" audit_brief.txt")
print(f"{'='*60}\n")
# ── Main ─────────────────────────────────────────────────────
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 of the file.")
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"\nAd Performance Audit")
if meta_id:
print(f" Meta account: {meta_id}")
if google_id:
print(f" Google account: {google_id}")
account_info = step1_account_structure(meta_id, google_id)
campaigns = step2_campaign_performance(meta_id, google_id)
creatives = step3_creative_fatigue(meta_id, google_id)
audience_data = step4_audience_segments(meta_id)
campaigns = step5_budget_allocation(campaigns)
brief = step6_generate_brief(account_info, campaigns, creatives, audience_data)
step7_output(campaigns, creatives, brief)
if __name__ == "__main__":
main()