How to Build an Ad Fatigue Detection System

An automated ad fatigue detection system catches declining performance before it costs you money. I run this on every active account. It tracks three signals: rising frequency, falling CTR, and increasing CPA. When two out of three trigger, the ad gets flagged.

Most advertisers only notice fatigue when performance has already dropped 30%+. This system catches it at 10%.

What You Need Before Starting

• Historical ad performance data (daily granularity)
• Python 3.8+ with sqlite3 and requests
• At least 7 days of data per ad
• Slack webhook for alerts

Step 1: Define Fatigue Signals

FATIGUE_CONFIG = {
    "frequency_warning": 2.5,
    "frequency_critical": 4.0,
    "ctr_decline_warning": 0.15,
    "ctr_decline_critical": 0.30,
    "cpa_increase_warning": 0.20,
    "cpa_increase_critical": 0.40,
    "min_spend_to_evaluate": 20,
    "lookback_days": 14,
    "comparison_window": 7,
}

Step 2: Calculate Fatigue Metrics

import sqlite3
from datetime import datetime, timedelta

def calculate_fatigue_metrics(db_path, ad_id):
    conn = sqlite3.connect(db_path)
    config = FATIGUE_CONFIG
    
    rows = conn.execute("""
        SELECT date, spend, impressions, clicks, reach, leads
        FROM ad_daily
        WHERE ad_id = ? AND date >= DATE('now', ? || ' days')
        ORDER BY date
    """, (ad_id, f"-{config['lookback_days']}")).fetchall()
    conn.close()
    
    if len(rows) < config["comparison_window"]:
        return None
    
    midpoint = len(rows) // 2
    first_half = rows[:midpoint]
    second_half = rows[midpoint:]
    
    def avg_metric(data, idx):
        vals = [r[idx] for r in data if r[idx] and r[idx] > 0]
        return sum(vals) / len(vals) if vals else 0
    
    first_impressions = sum(r[2] for r in first_half)
    first_clicks = sum(r[3] for r in first_half)
    first_reach = sum(r[4] for r in first_half) or 1
    first_spend = sum(r[1] for r in first_half)
    first_leads = sum(r[5] for r in first_half)
    
    second_impressions = sum(r[2] for r in second_half)
    second_clicks = sum(r[3] for r in second_half)
    second_reach = sum(r[4] for r in second_half) or 1
    second_spend = sum(r[1] for r in second_half)
    second_leads = sum(r[5] for r in second_half)
    
    first_ctr = (first_clicks / first_impressions * 100) if first_impressions > 0 else 0
    second_ctr = (second_clicks / second_impressions * 100) if second_impressions > 0 else 0
    
    first_freq = first_impressions / first_reach
    second_freq = second_impressions / second_reach
    
    first_cpa = first_spend / first_leads if first_leads > 0 else 999
    second_cpa = second_spend / second_leads if second_leads > 0 else 999
    
    ctr_change = (second_ctr - first_ctr) / first_ctr if first_ctr > 0 else 0
    cpa_change = (second_cpa - first_cpa) / first_cpa if first_cpa > 0 and first_cpa < 999 else 0
    
    return {
        "ad_id": ad_id,
        "current_frequency": round(second_freq, 2),
        "ctr_change": round(ctr_change, 3),
        "cpa_change": round(cpa_change, 3),
        "first_ctr": round(first_ctr, 2),
        "second_ctr": round(second_ctr, 2),
        "first_cpa": round(first_cpa, 2),
        "second_cpa": round(second_cpa, 2),
        "total_spend": first_spend + second_spend,
    }

Step 3: Score Fatigue Level

def score_fatigue(metrics):
    if not metrics or metrics["total_spend"] < FATIGUE_CONFIG["min_spend_to_evaluate"]:
        return {"level": "insufficient_data", "score": 0, "signals": []}
    
    signals = []
    score = 0
    
    # Frequency signal
    if metrics["current_frequency"] >= FATIGUE_CONFIG["frequency_critical"]:
        signals.append(f"Frequency critical: {metrics['current_frequency']}")
        score += 3
    elif metrics["current_frequency"] >= FATIGUE_CONFIG["frequency_warning"]:
        signals.append(f"Frequency warning: {metrics['current_frequency']}")
        score += 1
    
    # CTR decline signal
    if metrics["ctr_change"] <= -FATIGUE_CONFIG["ctr_decline_critical"]:
        signals.append(f"CTR dropped {abs(metrics['ctr_change'])*100:.0f}% ({metrics['first_ctr']}% to {metrics['second_ctr']}%)")
        score += 3
    elif metrics["ctr_change"] <= -FATIGUE_CONFIG["ctr_decline_warning"]:
        signals.append(f"CTR declining {abs(metrics['ctr_change'])*100:.0f}%")
        score += 1
    
    # CPA increase signal
    if metrics["cpa_change"] >= FATIGUE_CONFIG["cpa_increase_critical"]:
        signals.append(f"CPA up {metrics['cpa_change']*100:.0f}% (${metrics['first_cpa']} to ${metrics['second_cpa']})")
        score += 3
    elif metrics["cpa_change"] >= FATIGUE_CONFIG["cpa_increase_warning"]:
        signals.append(f"CPA rising {metrics['cpa_change']*100:.0f}%")
        score += 1
    
    if score >= 5:
        level = "critical"
    elif score >= 3:
        level = "warning"
    elif score >= 1:
        level = "watch"
    else:
        level = "healthy"
    
    return {"level": level, "score": score, "signals": signals}

Step 4: Run Detection Across All Active Ads

def detect_fatigue_all(db_path):
    conn = sqlite3.connect(db_path)
    active_ads = conn.execute("""
        SELECT DISTINCT ad_id, ad_name FROM ad_daily
        WHERE date >= DATE('now', '-3 days') AND spend > 0
    """).fetchall()
    conn.close()
    
    results = {"critical": [], "warning": [], "watch": [], "healthy": []}
    
    for ad_id, ad_name in active_ads:
        metrics = calculate_fatigue_metrics(db_path, ad_id)
        fatigue = score_fatigue(metrics)
        fatigue["ad_name"] = ad_name
        fatigue["ad_id"] = ad_id
        results[fatigue["level"]].append(fatigue)
    
    return results

Step 5: Alert and Report

import requests
import os

def send_fatigue_report(results):
    critical = results.get("critical", [])
    warning = results.get("warning", [])
    
    if not critical and not warning:
        return
    
    message = "*Ad Fatigue Report*\n\n"
    
    if critical:
        message += "*CRITICAL (rotate now):*\n"
        for ad in critical:
            message += f"  {ad['ad_name']}: {', '.join(ad['signals'])}\n"
    
    if warning:
        message += "\n*WARNING (prepare replacement):*\n"
        for ad in warning:
            message += f"  {ad['ad_name']}: {', '.join(ad['signals'])}\n"
    
    requests.post(os.getenv("SLACK_WEBHOOK_URL"), json={"text": message})

if __name__ == "__main__":
    results = detect_fatigue_all("meta_ads.db")
    send_fatigue_report(results)
    print(f"Critical: {len(results['critical'])}, Warning: {len(results['warning'])}, Healthy: {len(results['healthy'])}")

Run this daily. Critical ads need replacement within 48 hours. Warning ads need a backup creative in the queue within a week.

What to Build Next

Connect this to your creative rotation system so fatigued ads get swapped automatically. Then build a fatigue prediction model that estimates when currently healthy ads will start declining.

Related Reading

• AI for Creative Strategy and Testing - managing the creative lifecycle
• How Systems Compound Over Time - building detection systems that improve
• The Feedback Loop That Powers Everything - using performance data to drive creative decisions