Get alerted when campaign conversions drop

from collections import defaultdict
from datetime import date, timedelta
from clickruntime import on_trigger, credentials, http, notify

CUSTOMER_ID = "1234567890"

def gaql(query):
    headers = {"Authorization": f"Bearer {credentials.connection('google', 'ads')}",
               "developer-token": credentials.secret("GOOGLE_ADS_DEVELOPER_TOKEN")}
    batches = http.post(f"https://googleads.googleapis.com/v21/customers/{CUSTOMER_ID}/googleAds:searchStream",
                        headers=headers, json={"query": query}).json()
    return [row for b in batches for row in b.get("results", [])]


@on_trigger
def run(event):
    start, end = date.today() - timedelta(days=8), date.today() - timedelta(days=1)
    rows = gaql("SELECT campaign.name, segments.date, metrics.conversions FROM campaign "
                f"WHERE segments.date BETWEEN '{start}' AND '{end}' AND campaign.status = 'ENABLED'")
    conv = defaultdict(dict)
    for r in rows:
        conv[r["campaign"]["name"]][r["segments"]["date"]] = float(r["metrics"].get("conversions", 0))
    alerts = []
    for name, days in conv.items():
        y = days.pop(str(end), 0)
        avg = sum(days.values()) / 7
        if avg >= 2 and y < avg * 0.5:
            alerts.append(f"{name}: {y:.0f} conversions yesterday vs {avg:.1f}/day")
    if alerts:
        notify.slack("Google Ads conversions dropped\n" + "\n".join(alerts), channel="#marketing")
    return {"alerts": len(alerts)}