Systems Library / Marketing Automation / How to Build a ROAS Calculator and Alert System
Marketing Automation paid advertising

How to Build a ROAS Calculator and Alert System

Calculate real-time ROAS and get alerts when campaigns drop below targets.

Jay Banlasan

Jay Banlasan

The AI Systems Guy

A roas calculator with automated alerts tells you the second a campaign stops making money. I built this after a client lost $800 in a weekend because nobody checked ROAS on Saturday. The system runs every few hours and pings you when any campaign drops below your target.

ROAS is the one metric that ties ad spend directly to revenue. Everything else is a proxy.

What You Need Before Starting

Step 1: Pull Revenue Data Alongside Spend

import requests
import os
from datetime import datetime, timedelta

def fetch_campaign_roas(account_id, date=None):
    if not date:
        date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
    
    token = os.getenv("META_ACCESS_TOKEN")
    url = f"https://graph.facebook.com/v19.0/{account_id}/insights"
    params = {
        "access_token": token,
        "fields": "campaign_name,campaign_id,spend,actions,action_values",
        "level": "campaign",
        "time_range": f'{{"since":"{date}","until":"{date}"}}',
    }
    
    resp = requests.get(url, params=params)
    campaigns = []
    
    for row in resp.json().get("data", []):
        spend = float(row.get("spend", 0))
        revenue = 0
        purchases = 0
        
        for av in row.get("action_values", []):
            if av["action_type"] == "purchase":
                revenue = float(av["value"])
        
        for a in row.get("actions", []):
            if a["action_type"] == "purchase":
                purchases = int(a["value"])
        
        roas = revenue / spend if spend > 0 else 0
        
        campaigns.append({
            "campaign_id": row.get("campaign_id"),
            "campaign_name": row.get("campaign_name"),
            "date": date,
            "spend": spend,
            "revenue": revenue,
            "purchases": purchases,
            "roas": round(roas, 2),
        })
    
    return campaigns

Step 2: Set ROAS Targets by Campaign

ROAS_TARGETS = {
    "JB | Purchase | US | Book": {"target": 3.0, "min_acceptable": 2.0, "lookback_days": 7},
    "JB | Purchase | UK | Course": {"target": 4.0, "min_acceptable": 2.5, "lookback_days": 7},
}

DEFAULT_TARGET = {"target": 2.5, "min_acceptable": 1.5, "lookback_days": 7}

Step 3: Calculate Rolling ROAS

import sqlite3

def store_roas_data(db_path, campaigns):
    conn = sqlite3.connect(db_path)
    conn.execute("""CREATE TABLE IF NOT EXISTS roas_daily (
        id INTEGER PRIMARY KEY, campaign_id TEXT, campaign_name TEXT,
        date TEXT, spend REAL, revenue REAL, purchases INTEGER, roas REAL,
        fetched_at DATETIME DEFAULT CURRENT_TIMESTAMP
    )""")
    
    for c in campaigns:
        conn.execute("INSERT INTO roas_daily (campaign_id, campaign_name, date, spend, revenue, purchases, roas) VALUES (?,?,?,?,?,?,?)",
            (c["campaign_id"], c["campaign_name"], c["date"], c["spend"], c["revenue"], c["purchases"], c["roas"]))
    conn.commit()
    conn.close()

def get_rolling_roas(db_path, campaign_name, days=7):
    conn = sqlite3.connect(db_path)
    row = conn.execute("""
        SELECT SUM(spend), SUM(revenue), SUM(purchases),
            CASE WHEN SUM(spend) > 0 THEN SUM(revenue)/SUM(spend) ELSE 0 END
        FROM roas_daily
        WHERE campaign_name = ? AND date >= DATE('now', ? || ' days')
    """, (campaign_name, f"-{days}")).fetchone()
    conn.close()
    
    return {
        "spend": row[0] or 0,
        "revenue": row[1] or 0,
        "purchases": row[2] or 0,
        "rolling_roas": round(row[3] or 0, 2),
    }

Step 4: Build the Alert Engine

def check_roas_alerts(db_path):
    conn = sqlite3.connect(db_path)
    campaigns = conn.execute("SELECT DISTINCT campaign_name FROM roas_daily").fetchall()
    conn.close()
    
    alerts = []
    for (campaign_name,) in campaigns:
        target_config = ROAS_TARGETS.get(campaign_name, DEFAULT_TARGET)
        rolling = get_rolling_roas(db_path, campaign_name, target_config["lookback_days"])
        
        if rolling["spend"] < 10:
            continue
        
        if rolling["rolling_roas"] < target_config["min_acceptable"]:
            alerts.append({
                "level": "critical",
                "campaign": campaign_name,
                "message": f"ROAS {rolling['rolling_roas']}x below minimum {target_config['min_acceptable']}x. Spend: ${rolling['spend']:.2f}, Revenue: ${rolling['revenue']:.2f}",
            })
        elif rolling["rolling_roas"] < target_config["target"]:
            alerts.append({
                "level": "warning",
                "campaign": campaign_name,
                "message": f"ROAS {rolling['rolling_roas']}x below target {target_config['target']}x. Trending down.",
            })
    
    return alerts

Step 5: Send and Schedule

def send_roas_alerts(alerts):
    webhook = os.getenv("SLACK_WEBHOOK_URL")
    for alert in alerts:
        icon = "🔴" if alert["level"] == "critical" else "🟡"
        requests.post(webhook, json={"text": f"{icon} ROAS Alert: *{alert['campaign']}*\n{alert['message']}"})

if __name__ == "__main__":
    account_id = os.getenv("META_AD_ACCOUNT_ID")
    campaigns = fetch_campaign_roas(account_id)
    store_roas_data("roas.db", campaigns)
    alerts = check_roas_alerts("roas.db")
    if alerts:
        send_roas_alerts(alerts)
    print(f"Checked {len(campaigns)} campaigns, {len(alerts)} alerts sent")
0 8,14,20 * * * cd /path/to/project && python3 roas_checker.py

Three checks per day catches drops before they cost serious money.

What to Build Next

Add a ROAS trend chart that shows 7-day and 30-day rolling ROAS by campaign. Then build an automatic budget reduction trigger for campaigns that stay below minimum ROAS for 3+ consecutive days.

Related Reading

Want this system built for your business?

Get a free assessment. We will map every system your business needs and show you the ROI.

Get Your Free Assessment

Related Systems