How to Build a CPA Tracking and Alert System

I had a lead generation campaign running at $45 CPA for two weeks. Then a creative refresh dropped performance and CPA crept to $67 over three days. I didn't notice until the weekly check-in. By then I'd spent $340 above what a real-time cpa tracking alert system automated would have saved. The campaign needed one tweak. I caught it four days late.

Real-time CPA monitoring is table stakes for any account spending more than $50/day. The math is simple: if your target CPA is $50 and you're running at $80 for 24 hours without intervention, that's the cost of not having alerts. This system checks CPA every hour and fires a Slack message the moment it drifts past your threshold.

What You Need Before Starting

• Python 3.10+
• Meta Ads API access (user token or system user token)
• SQLite for local data storage
• requests library
• A Slack webhook URL for alert delivery

Step 1: Define Your CPA Targets

CPA targets live in a config file, not hardcoded in scripts. This lets you update thresholds without touching code.

import json
from pathlib import Path

CPA_CONFIG_PATH = "cpa_targets.json"

# Write this config once; update targets as campaigns scale
CPA_CONFIG = {
    "campaigns": {
        "lead_gen_uk_broad": {
            "target_cpa":    45.00,
            "alert_at_pct":  1.3,      # alert at 130% of target = $58.50
            "kill_at_pct":   1.8,      # pause at 180% of target = $81.00
            "min_spend_usd": 30.00,    # don't alert until this spend threshold
            "conversion_event": "lead"
        },
        "purchase_us_retarget": {
            "target_cpa":    120.00,
            "alert_at_pct":  1.25,
            "kill_at_pct":   2.0,
            "min_spend_usd": 60.00,
            "conversion_event": "purchase"
        }
    }
}

if not Path(CPA_CONFIG_PATH).exists():
    Path(CPA_CONFIG_PATH).write_text(json.dumps(CPA_CONFIG, indent=2))

def load_config() -> dict:
    return json.loads(Path(CPA_CONFIG_PATH).read_text())

Step 2: Pull Live CPA Data from Meta Ads API

Query the Insights API for spend and conversions over your chosen window.

import requests
import os
from datetime import date, timedelta

META_TOKEN = os.getenv("META_ACCESS_TOKEN")
META_ACCOUNT_ID = os.getenv("META_ACCOUNT_ID")  # act_XXXXXXXXXX

def get_campaign_insights(campaign_name_filter: str,
                            date_preset: str = "today") -> dict:
    """
    Returns spend and conversion data for campaigns matching the name filter.
    date_preset: "today", "last_3d", "last_7d"
    """
    url = f"https://graph.facebook.com/v19.0/{META_ACCOUNT_ID}/insights"
    
    params = {
        "access_token": META_TOKEN,
        "level":        "campaign",
        "date_preset":  date_preset,
        "fields":       "campaign_name,spend,actions,cost_per_action_type",
        "filtering":    json.dumps([{
            "field": "campaign.name",
            "operator": "CONTAIN",
            "value": campaign_name_filter
        }])
    }
    
    resp = requests.get(url, params=params, timeout=30)
    resp.raise_for_status()
    data = resp.json().get("data", [])
    
    results = {}
    for row in data:
        name = row.get("campaign_name", "")
        spend = float(row.get("spend", 0))
        
        actions = {a["action_type"]: int(a["value"])
                   for a in row.get("actions", [])}
        
        results[name] = {
            "spend": spend,
            "leads":     actions.get("lead", 0),
            "purchases": actions.get("purchase", 0),
            "all_actions": actions
        }
    
    return results

def calculate_cpa(spend: float, conversions: int) -> float | None:
    if conversions == 0:
        return None
    return round(spend / conversions, 2)

Step 3: Build the CPA Check Function

Compare actual CPA against targets and classify the status.

@dataclass
class CPAStatus:
    campaign: str
    spend: float
    conversions: int
    actual_cpa: float | None
    target_cpa: float
    pct_of_target: float | None
    status: str   # "ok", "warning", "critical", "no_conversions", "below_min_spend"
    alert_message: str

def check_cpa_status(campaign_key: str, insights: dict,
                      config: dict) -> CPAStatus:
    cfg = config["campaigns"].get(campaign_key)
    if not cfg:
        raise KeyError(f"No config for campaign: {campaign_key}")
    
    data = insights.get(campaign_key, {"spend": 0, "leads": 0, "purchases": 0})
    spend = data["spend"]
    conv_event = cfg["conversion_event"]
    conversions = data.get(f"{conv_event}s", 0) or data.get(conv_event, 0)
    
    target = cfg["target_cpa"]
    actual = calculate_cpa(spend, conversions)
    
    if spend < cfg["min_spend_usd"]:
        return CPAStatus(campaign_key, spend, conversions, actual, target,
                          None, "below_min_spend",
                          f"Spend ${spend:.2f} below minimum ${cfg['min_spend_usd']:.2f}")
    
    if actual is None:
        return CPAStatus(campaign_key, spend, conversions, actual, target,
                          None, "no_conversions",
                          f"${spend:.2f} spent, 0 {conv_event}s")
    
    pct = actual / target
    
    if pct >= cfg["kill_at_pct"]:
        status = "critical"
        msg = (f"CPA ${actual:.2f} is {pct*100:.0f}% of target ${target:.2f}. "
               f"KILL threshold reached. Consider pausing.")
    elif pct >= cfg["alert_at_pct"]:
        status = "warning"
        msg = (f"CPA ${actual:.2f} is {pct*100:.0f}% of target ${target:.2f}. "
               f"Needs attention.")
    else:
        status = "ok"
        msg = f"CPA ${actual:.2f} — on target (${target:.2f})"
    
    return CPAStatus(campaign_key, spend, conversions, actual, target, pct, status, msg)

Step 4: Build the Alert Dispatcher

Send alerts to Slack with context and a recommended action.

SLACK_WEBHOOK = os.getenv("SLACK_WEBHOOK_URL")

STATUS_EMOJI = {
    "ok":              ":white_check_mark:",
    "warning":         ":warning:",
    "critical":        ":rotating_light:",
    "no_conversions":  ":hourglass:",
    "below_min_spend": ":information_source:"
}

def send_cpa_alert(status: CPAStatus):
    if status.status in ("ok", "below_min_spend"):
        return  # don't alert on good news or insufficient data
    
    emoji = STATUS_EMOJI.get(status.status, ":question:")
    
    text = (
        f"{emoji} *CPA Alert: {status.campaign}*\n"
        f"> Status: {status.status.upper()}\n"
        f"> Spend today: ${status.spend:.2f}\n"
        f"> Conversions: {status.conversions}\n"
        f"> Actual CPA: ${status.actual_cpa:.2f}\n"
        f"> Target CPA: ${status.target_cpa:.2f}\n"
        f"> {status.alert_message}"
    )
    
    if status.status == "critical":
        text += f"\n> *Recommended action: Review and consider pausing.*"
    
    requests.post(SLACK_WEBHOOK, json={"text": text}, timeout=10)

def run_cpa_checks():
    config = load_config()
    
    for campaign_key in config["campaigns"]:
        try:
            # Use campaign key as filter (requires it to match part of campaign name)
            insights = get_campaign_insights(campaign_key, date_preset="today")
            status = check_cpa_status(campaign_key, insights, config)
            
            print(f"{status.campaign}: {status.status} — {status.alert_message}")
            send_cpa_alert(status)
            log_cpa_event(status)
        
        except Exception as e:
            print(f"Error checking {campaign_key}: {e}")

Step 5: Log CPA History for Trend Analysis

Track every check so you can see CPA trends over time.

import sqlite3
from datetime import datetime

def log_cpa_event(status: CPAStatus):
    conn = sqlite3.connect("cpa_log.db")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS cpa_log (
            ts TEXT, campaign TEXT, spend REAL, conversions INTEGER,
            actual_cpa REAL, target_cpa REAL, pct_of_target REAL, status TEXT
        )
    """)
    conn.execute("INSERT INTO cpa_log VALUES (?,?,?,?,?,?,?,?)", (
        datetime.utcnow().isoformat(), status.campaign, status.spend,
        status.conversions, status.actual_cpa, status.target_cpa,
        status.pct_of_target, status.status
    ))
    conn.commit()
    conn.close()

def cpa_trend(campaign: str, days: int = 7) -> list:
    conn = sqlite3.connect("cpa_log.db")
    rows = conn.execute("""
        SELECT DATE(ts) as day, AVG(actual_cpa) as avg_cpa,
               MIN(actual_cpa) as min_cpa, MAX(actual_cpa) as max_cpa,
               COUNT(*) as checks
        FROM cpa_log
        WHERE campaign = ? AND ts >= datetime('now', ?) AND actual_cpa IS NOT NULL
        GROUP BY DATE(ts)
        ORDER BY day
    """, (campaign, f'-{days} days')).fetchall()
    conn.close()
    return [{"day": r[0], "avg_cpa": round(r[1], 2),
             "min": round(r[2], 2), "max": round(r[3], 2), "checks": r[4]}
            for r in rows]

Step 6: Schedule Hourly Checks

# crontab — check every hour during business hours
0 8-22 * * * python /scripts/cpa_monitor.py >> /var/log/cpa_monitor.log 2>&1

# cpa_monitor.py
if __name__ == "__main__":
    print(f"CPA check at {datetime.utcnow().isoformat()}")
    run_cpa_checks()
    print("Done")

What to Build Next

• Add automatic pausing via the Meta Ads API when a campaign hits the kill threshold, with a Slack confirmation message before pausing
• Build a 3-day rolling CPA view in your daily brief so you can spot slow-building trends before they hit alert thresholds
• Add a re-alert suppression so you don't get the same Slack alert every hour once a campaign is already flagged

Related Reading

• How to Automate Daily Meta Ads Reporting - CPA alerts are the real-time layer; daily reporting is the recap layer
• How to Build an AI Ad Headline Generator - when CPA spikes, new headlines are often the first thing to test
• How to Automate Ad Account Health Monitoring - CPA tracking is one signal in a broader account health picture