Systems Library / Marketing Automation / How to Automate Competitor Ad Monitoring
Marketing Automation paid advertising

How to Automate Competitor Ad Monitoring

Track competitor ads automatically using Meta Ad Library and Google Transparency.

Jay Banlasan

Jay Banlasan

The AI Systems Guy

You can automate competitor ad monitoring using the Meta Ad Library API and Google Ads Transparency Center. I run this weekly for every client. Instead of manually browsing the ad library, a script pulls new ads, flags changes, and sends a summary to Slack.

Knowing what competitors run tells you what angles they are testing and what they stopped running. Both are valuable signals.

What You Need Before Starting

Step 1: Pull Ads from Meta Ad Library

import requests
import os
from dotenv import load_dotenv

load_dotenv()

def fetch_competitor_ads(page_id, country="GB", limit=50):
    token = os.getenv("META_ACCESS_TOKEN")
    
    url = "https://graph.facebook.com/v19.0/ads_archive"
    params = {
        "access_token": token,
        "search_page_ids": page_id,
        "ad_reached_countries": f'["{country}"]',
        "ad_active_status": "ACTIVE",
        "fields": "id,ad_creative_bodies,ad_creative_link_titles,ad_creative_link_captions,ad_delivery_start_time,page_name",
        "limit": limit,
    }
    
    resp = requests.get(url, params=params)
    return resp.json().get("data", [])

Step 2: Store and Track Changes

import sqlite3
from datetime import datetime

def store_competitor_ads(db_path, ads, competitor_name):
    conn = sqlite3.connect(db_path)
    conn.execute("""CREATE TABLE IF NOT EXISTS competitor_ads (
        id INTEGER PRIMARY KEY, ad_id TEXT UNIQUE, competitor TEXT,
        body TEXT, headline TEXT, first_seen TEXT, last_seen TEXT,
        status TEXT DEFAULT 'active'
    )""")
    
    new_ads = 0
    for ad in ads:
        body = ad.get("ad_creative_bodies", [""])[0] if ad.get("ad_creative_bodies") else ""
        headline = ad.get("ad_creative_link_titles", [""])[0] if ad.get("ad_creative_link_titles") else ""
        ad_id = ad.get("id", "")
        
        existing = conn.execute("SELECT id FROM competitor_ads WHERE ad_id = ?", (ad_id,)).fetchone()
        
        if existing:
            conn.execute("UPDATE competitor_ads SET last_seen = ? WHERE ad_id = ?",
                        (datetime.now().isoformat(), ad_id))
        else:
            conn.execute("""INSERT INTO competitor_ads (ad_id, competitor, body, headline, first_seen, last_seen)
                VALUES (?,?,?,?,?,?)""",
                (ad_id, competitor_name, body, headline, datetime.now().isoformat(), datetime.now().isoformat()))
            new_ads += 1
    
    conn.commit()
    conn.close()
    return new_ads

Step 3: Detect New and Stopped Ads

def detect_changes(db_path, competitor_name, days_inactive=7):
    conn = sqlite3.connect(db_path)
    
    new_ads = conn.execute("""
        SELECT body, headline, first_seen FROM competitor_ads
        WHERE competitor = ? AND first_seen >= DATE('now', '-7 days')
        ORDER BY first_seen DESC
    """, (competitor_name,)).fetchall()
    
    stopped_ads = conn.execute("""
        SELECT body, headline, last_seen FROM competitor_ads
        WHERE competitor = ? AND last_seen < DATE('now', ? || ' days') AND status = 'active'
    """, (competitor_name, f"-{days_inactive}")).fetchall()
    
    # Mark stopped ads
    conn.execute("""UPDATE competitor_ads SET status = 'stopped'
        WHERE competitor = ? AND last_seen < DATE('now', ? || ' days') AND status = 'active'
    """, (competitor_name, f"-{days_inactive}"))
    conn.commit()
    conn.close()
    
    return {"new": new_ads, "stopped": stopped_ads}

Step 4: Generate AI Analysis

import anthropic

def analyze_competitor_ads(changes, competitor_name):
    client = anthropic.Anthropic()
    
    new_text = "\n".join([f"- Headline: {a[1]}\n  Body: {a[0][:200]}" for a in changes["new"]])
    stopped_text = "\n".join([f"- Headline: {a[1]}\n  Body: {a[0][:200]}" for a in changes["stopped"]])
    
    prompt = f"""Analyze this competitor's ad changes this week.

COMPETITOR: {competitor_name}

NEW ADS LAUNCHED:
{new_text or "None"}

ADS STOPPED:
{stopped_text or "None"}

Provide:
1. What angles are they testing?
2. What did they stop (possible losers)?
3. Any new offers or messaging shifts?
4. Opportunities for us based on gaps they are not covering
5. Threats or angles we should test before they saturate

Keep it brief and actionable."""

    resp = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=600,
        messages=[{"role": "user", "content": prompt}]
    )
    return resp.content[0].text

Step 5: Weekly Report to Slack

def send_weekly_report(competitors):
    for name, page_id in competitors.items():
        ads = fetch_competitor_ads(page_id)
        new_count = store_competitor_ads("competitors.db", ads, name)
        changes = detect_changes("competitors.db", name)
        
        if changes["new"] or changes["stopped"]:
            analysis = analyze_competitor_ads(changes, name)
            message = f"*Competitor Update: {name}*\nNew ads: {len(changes['new'])} | Stopped: {len(changes['stopped'])}\n\n{analysis}"
        else:
            message = f"*Competitor Update: {name}*\nNo changes this week."
        
        # Send to Slack
        requests.post(os.getenv("SLACK_WEBHOOK_URL"), json={"text": message})

# Run weekly
competitors = {
    "Competitor A": "123456789",
    "Competitor B": "987654321",
}
send_weekly_report(competitors)

Run this every Monday morning. You start the week knowing exactly what competitors changed, what they tested, and where the gaps are.

What to Build Next

Add screenshot capture of competitor ads using a headless browser. Then build a swipe file database that categorizes competitor creatives by angle, format, and hook type.

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