Ad Fatigue Detector

import os, requests
from datetime import datetime, timedelta

META = os.environ["META_ACCESS_TOKEN"]
ACCT = os.environ["META_AD_ACCOUNT_ID"]
MV = os.environ["MAVERA_API_KEY"]
GRAPH = "https://graph.facebook.com/v24.0"
MB = "https://app.mavera.io/api/v1"
MH = {"Authorization": f"Bearer {MV}", "Content-Type": "application/json"}

FREQUENCY_THRESHOLD = 3.0
CTR_DROP_PCT = 0.25

# 1. Pull daily performance per ad (last 14 days)
insights = requests.get(
    f"{GRAPH}/{ACCT}/insights",
    params={
        "access_token": META,
        "fields": "ad_id,ad_name,frequency,ctr,cpc,impressions,clicks,spend",
        "level": "ad",
        "time_increment": 1,
        "date_preset": "last_14d",
        "limit": 500,
    },
).json().get("data", [])

# 2. Group by ad and analyze trend
from collections import defaultdict
ad_series = defaultdict(list)
for row in insights:
    ad_series[row["ad_id"]].append({
        "date": row.get("date_start"),
        "frequency": float(row.get("frequency", 0)),
        "ctr": float(row.get("ctr", 0)),
        "cpc": float(row.get("cpc", 0)),
        "spend": float(row.get("spend", 0)),
        "name": row.get("ad_name", ""),
    })

fatigued = []
for ad_id, series in ad_series.items():
    series.sort(key=lambda x: x["date"])
    if len(series) < 3:
        continue

    first_3_ctr = sum(d["ctr"] for d in series[:3]) / 3
    last_3_ctr = sum(d["ctr"] for d in series[-3:]) / 3
    latest_freq = series[-1]["frequency"]
    total_spend = sum(d["spend"] for d in series)

    ctr_decline = (first_3_ctr - last_3_ctr) / max(first_3_ctr, 0.01)

    if latest_freq > FREQUENCY_THRESHOLD and ctr_decline > CTR_DROP_PCT:
        fatigued.append({
            "ad_id": ad_id,
            "name": series[0]["name"],
            "frequency": latest_freq,
            "ctr_initial": first_3_ctr,
            "ctr_current": last_3_ctr,
            "ctr_decline": ctr_decline,
            "total_spend": total_spend,
        })

print(f"Fatigued ads: {len(fatigued)} / {len(ad_series)} active")

if not fatigued:
    print("No fatigued ads detected.")
else:
    # 3. Build context for Mave
    fatigue_report = "\n".join(
        f"- \"{f['name']}\": freq={f['frequency']:.1f}, CTR dropped {f['ctr_decline']:.0%} "
        f"({f['ctr_initial']:.2f}% → {f['ctr_current']:.2f}%), spend=${f['total_spend']:.0f}"
        for f in fatigued
    )

    # 4. Get fresh angles from Mave
    research = requests.post(f"{MB}/mave/chat", headers=MH, json={
        "message": f"""These Meta ads are showing fatigue — frequency is high and CTR is declining.
Research fresh creative angles to replace or refresh them.

FATIGUED ADS:
{fatigue_report}

For each fatigued ad:
1. Why the current angle is likely fatiguing (audience saturation, message wear-out, etc.)
2. 3 fresh creative angles that maintain the core value prop but use new hooks
3. Recommended format changes (video length, static vs carousel, UGC vs polished)
4. Targeting adjustments to reduce frequency without losing quality"""
    }).json()

    print("\n=== Ad Fatigue Analysis & Recommendations ===")
    print(research.get("content", ""))