#!/usr/bin/env python3
"""
03_anomaly_detection_alerting.py — Detect anomalies in ad account metrics
(spend spikes, CTR drops, ROAS crashes) and generate alerts
via GoMarble MCP tools + Claude API.
Outputs: anomaly_daily_data.csv, anomaly_alerts.csv, anomaly_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)
# NOTE: At least one of META_AD_ACCOUNT or GOOGLE_ADS_CID must be set.
import csv, json, re, sys, time, subprocess, statistics
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"):
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: Fetch Daily Data ──────────────────────────────────
def step1_daily_data(meta_id, google_id):
banner(1, 4, "Fetch Daily Account Data (30d)")
all_daily = []
if meta_id:
print(" Fetching Meta daily metrics...")
msg = (
f"Run facebook_get_adaccount_insights with ad_account_id='{meta_id}', "
"level='account', date_preset='last_30d', time_increment='1', "
"fields=['spend','impressions','clicks','ctr','cpc','cpm',"
"'purchase_roas','actions','frequency']. "
"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)
all_daily.append({
"platform": "Meta",
"date": flat.get("date_start", flat.get("date", "")),
"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")),
"cpm": safe_float(flat.get("cpm")),
"roas": safe_float(flat.get("purchase_roas", flat.get("purchase_roas_value"))),
})
print(f" Meta: {len([d for d in all_daily if d['platform']=='Meta'])} days")
if google_id:
print(" Fetching Google daily metrics...")
msg = (
f"Run google_ads_run_gaql with customer_id='{google_id}' and query:\n"
"SELECT segments.date, metrics.impressions, metrics.clicks, "
"metrics.cost_micros, metrics.conversions, metrics.conversions_value, "
"metrics.ctr, metrics.average_cpc "
"FROM customer WHERE segments.date DURING LAST_30_DAYS "
"ORDER BY segments.date ASC\n\nReturn the raw JSON."
)
resp = mcp_request([{"role": "user", "content": msg}])
text = get_text(resp)
for row in to_rows(text):
flat = flatten_row(row)
cost_micros = safe_float(flat.get("cost_micros", flat.get("metrics_cost_micros")))
spend = cost_micros / 1_000_000 if cost_micros > 100 else cost_micros
all_daily.append({
"platform": "Google",
"date": flat.get("date", flat.get("segments_date", "")),
"spend": round(spend, 2),
"impressions": safe_float(flat.get("impressions", flat.get("metrics_impressions"))),
"clicks": safe_float(flat.get("clicks", flat.get("metrics_clicks"))),
"ctr": safe_float(flat.get("ctr", flat.get("metrics_ctr"))),
"cpc": 0, "cpm": 0,
"roas": 0,
})
print(f" Google: {len([d for d in all_daily if d['platform']=='Google'])} days")
return all_daily
# ── Step 2: Detect Anomalies ─────────────────────────────────
def step2_detect_anomalies(daily_data):
banner(2, 4, "Detect Anomalies (Z-score)")
alerts = []
for platform in ("Meta", "Google"):
pdata = [d for d in daily_data if d["platform"] == platform]
if len(pdata) < 7:
continue
for metric in ("spend", "ctr", "cpc", "cpm", "roas"):
vals = [d[metric] for d in pdata if d[metric] > 0]
if len(vals) < 7:
continue
mean = statistics.mean(vals)
stdev = statistics.stdev(vals) if len(vals) > 1 else 0
if stdev == 0:
continue
for d in pdata:
val = d[metric]
if val == 0:
continue
z = (val - mean) / stdev
if abs(z) >= 2.0:
direction = "SPIKE" if z > 0 else "DROP"
pct_change = ((val - mean) / mean * 100)
severity = "HIGH" if abs(z) >= 3.0 else "MEDIUM"
alerts.append({
"platform": platform,
"date": d["date"],
"metric": metric,
"value": round(val, 4),
"mean": round(mean, 4),
"z_score": round(z, 2),
"direction": direction,
"pct_from_mean": round(pct_change, 1),
"severity": severity,
})
alerts.sort(key=lambda a: abs(a["z_score"]), reverse=True)
print(f" Anomalies detected: {len(alerts)}")
if alerts:
print(f" HIGH severity: {len([a for a in alerts if a['severity']=='HIGH'])}")
for a in alerts[:5]:
print(f" {a['severity']} | {a['platform']} {a['date']}: {a['metric']} "
f"{a['direction']} ({a['pct_from_mean']:+.1f}% from mean, z={a['z_score']})")
return alerts
# ── Step 3: Generate Alert Brief ──────────────────────────────
def step3_generate_brief(daily_data, alerts):
banner(3, 4, "Generate Anomaly Brief")
alert_summary = json.dumps(alerts[:30], indent=2, default=str)
daily_summary = json.dumps(daily_data[-14:], indent=2, default=str)
prompt = f"""You are a paid media monitoring specialist. Analyze the detected anomalies and write an alert report.
Sections:
1. ALERT SUMMARY — Total anomalies, severity breakdown, affected platforms
2. CRITICAL ALERTS — Each HIGH severity anomaly with:
- What happened (metric, date, magnitude)
- Likely cause
- Recommended immediate action
3. WARNING ALERTS — MEDIUM severity anomalies grouped by pattern
4. TREND CONTEXT — Are these anomalies isolated or part of a trend?
5. ROOT CAUSE HYPOTHESES — Most likely explanations ranked by probability:
- Budget changes, bid strategy shifts
- Creative fatigue, audience saturation
- Competitive pressure, seasonality
- Tracking/attribution issues
6. ACTION PLAN — Prioritized list of actions to take today
7. MONITORING THRESHOLDS — Recommended alert thresholds for ongoing monitoring
Use exact dates, metrics, and percentage changes.
### Detected Anomalies (top 30)
{alert_summary[:6000]}
### Recent 14-Day Daily Data
{daily_summary[:5000]}"""
resp = mcp_request([{"role": "user", "content": prompt}])
brief = get_text(resp)
print(f" Brief: {len(brief)} chars")
return brief
# ── Step 4: Output ────────────────────────────────────────────
def step4_output(daily_data, alerts, brief):
banner(4, 4, "Write Output Files")
daily_fields = ["platform", "date", "spend", "impressions", "clicks", "ctr", "cpc", "cpm", "roas"]
write_csv("anomaly_daily_data.csv", daily_data, daily_fields)
alert_fields = ["severity", "platform", "date", "metric", "direction",
"value", "mean", "pct_from_mean", "z_score"]
write_csv("anomaly_alerts.csv", alerts, alert_fields)
with open("anomaly_brief.txt", "w", encoding="utf-8") as f:
f.write("ANOMALY DETECTION & ALERTING REPORT\n")
f.write(f"Generated: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
f.write("=" * 60 + "\n\n")
f.write(brief)
print(f" -> anomaly_brief.txt — {len(brief)} chars")
print(f"\n{'='*60}\n ANOMALY DETECTION COMPLETE\n{'='*60}")
print(f" Days analyzed: {len(daily_data)}")
print(f" Anomalies found: {len(alerts)}")
print(f" HIGH alerts: {len([a for a in alerts if a['severity']=='HIGH'])}")
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"\nAnomaly Detection & Alerting")
if meta_id: print(f" Meta: {meta_id}")
if google_id: print(f" Google: {google_id}")
daily_data = step1_daily_data(meta_id, google_id)
alerts = step2_detect_anomalies(daily_data)
brief = step3_generate_brief(daily_data, alerts)
step4_output(daily_data, alerts, brief)
if __name__ == "__main__":
main()