Ad Spend Anomaly Guard for Paid Campaigns

id: ad-spend-anomaly-guard
namespace: company.team
description: |
  Stop a runaway ad campaign from emptying its daily budget before anyone
  notices. A disabled daily schedule and a keyed on-demand webhook both start
  the same check: a three-campaign spend fixture is written with a heredoc,
  Python measures each campaign's spend today against its previous seven days
  to get a z-score, and anything over the daily cap or past the sigma
  threshold is flagged. Flagged campaigns are paused one at a time through the
  ads pause API, tracked in a GitHub issue and announced on Slack, while a
  healthy book of campaigns is logged as within budget. The verdict, the worst
  offending campaign, its spend and its deviation are published as outputs, and
  any failure escalates to Slack.

triggers:
  - id: on_demand_check
    type: io.kestra.plugin.core.trigger.Webhook
    description: POST here to run a spend guard on demand, for example when a flight
      launches early or a budget is topped up mid day.
    key: "{{ secret('ADS_WEBHOOK_KEY') }}"
  - id: daily_spend_watch
    type: io.kestra.plugin.core.trigger.Schedule
    description: Daily 06:00 UTC safety net, shipped disabled so no campaign is
      paused until an operator reviews the thresholds and switches it on.
    cron: "0 6 * * *"
    disabled: true

inputs:
  - id: daily_cap_usd
    type: FLOAT
    defaults: 500.0
    description: Hard per-campaign daily ceiling in US dollars, any campaign
      spending more than this today is paused regardless of its history.
  - id: sigma_threshold
    type: FLOAT
    defaults: 3.0
    description: How many standard deviations above its own seven day mean today's
      spend must sit before the campaign counts as a statistical anomaly.

tasks:
  # --- 1. campaign spend workspace ---
  - id: campaign_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: One shared directory so the spend fixture and the anomaly analysis
      read and write the same file for the whole run.
    tasks:
      # --- 2. spend fixture ---
      - id: write_campaign_fixture
        type: io.kestra.plugin.scripts.shell.Commands
        description: Write a deterministic three-campaign fixture with a heredoc, one
          ordinary day, one sharp spike above the cap and one steady
          under-spender, then size the file for the log.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          entryPoint: []
        containerImage: alpine:3.20
        outputFiles:
          - campaigns.json
        commands:
          - |
            set -u
            cat > campaigns.json <<'JSON'
            {
              "campaigns": [
                {"id": "cmp-brand-search", "name": "Brand Search US", "history": [180.0, 210.0, 195.0, 205.0, 190.0, 200.0, 215.0], "today": 205.0},
                {"id": "cmp-summer-prospecting", "name": "Summer Prospecting EU", "history": [460.0, 500.0, 475.0, 490.0, 465.0, 495.0, 475.0], "today": 544.2},
                {"id": "cmp-retargeting", "name": "Retargeting US", "history": [295.0, 305.0, 300.0, 310.0, 298.0, 302.0, 300.0], "today": 310.0}
              ]
            }
            JSON
            echo "wrote $(wc -c < campaigns.json) bytes of campaign spend data"

      # --- 3. anomaly analysis ---
      - id: analyze_spend
        type: io.kestra.plugin.scripts.python.Script
        description: Read the campaign fixture, compute the mean, the sample standard
          deviation and the z-score of today's spend per campaign, flag a breach
          on cap or sigma, and emit the verdict through the outputs protocol.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.12-slim
        env:
          DAILY_CAP_USD: "{{ inputs.daily_cap_usd }}"
          SIGMA_THRESHOLD: "{{ inputs.sigma_threshold }}"
        script: |
          import json
          import os
          import statistics

          CAMPAIGN_FILE = "campaigns.json"


          def as_float(value, fallback=0.0):
              try:
                  return float(value)
              except (TypeError, ValueError):
                  return fallback


          def load_campaigns(path):
              if not os.path.exists(path) or os.path.getsize(path) == 0:
                  return []
              try:
                  with open(path, "r", encoding="utf-8") as handle:
                      document = json.load(handle)
              except ValueError:
                  return []
              if isinstance(document, dict):
                  document = document.get("campaigns")
              if not isinstance(document, list):
                  return []
              return [row for row in document if isinstance(row, dict)]


          cap = as_float(os.environ.get("DAILY_CAP_USD"), 500.0)
          sigma = as_float(os.environ.get("SIGMA_THRESHOLD"), 3.0)

          records = []
          for row in load_campaigns(CAMPAIGN_FILE):
              history = []
              raw_history = row.get("history")
              if isinstance(raw_history, list):
                  for value in raw_history:
                      try:
                          history.append(float(value))
                      except (TypeError, ValueError):
                          continue
              if not history:
                  continue
              try:
                  today = float(row.get("today"))
              except (TypeError, ValueError):
                  continue
              campaign_id = str(row.get("id") or row.get("name") or "unknown-campaign")
              campaign_name = str(row.get("name") or campaign_id)
              mean = statistics.fmean(history)
              spread = statistics.stdev(history) if len(history) > 1 else 0.0
              if spread > 0.0:
                  deviation = (today - mean) / spread
              else:
                  deviation = 0.0 if today <= mean else 999.0
              over_cap = today > cap
              over_sigma = deviation > sigma
              reasons = []
              if over_cap:
                  reasons.append("daily cap")
              if over_sigma:
                  reasons.append("sigma threshold")
              records.append({
                  "id": campaign_id,
                  "name": campaign_name,
                  "spend": round(today, 2),
                  "deviation": round(deviation, 2),
                  "mean": round(mean, 2),
                  "breach": bool(reasons),
                  "reason": " and ".join(reasons),
              })

          breached = sorted(
              [record for record in records if record["breach"]],
              key=lambda record: (-record["deviation"], -record["spend"]),
          )
          reviewable = [{"id": r["id"], "name": r["name"], "spend": r["spend"],
                         "deviation": r["deviation"], "reason": r["reason"]}
                        for r in breached]

          if breached:
              verdict = "PAUSE_REQUIRED"
              worst = breached[0]
              campaign = worst["id"]
              spend = float(worst["spend"])
              deviation = float(worst["deviation"])
          elif records:
              verdict = "WITHIN_BUDGET"
              campaign = "none"
              spend = float(max(record["spend"] for record in records))
              deviation = float(max(record["deviation"] for record in records))
          else:
              verdict = "NO_DATA"
              campaign = "none"
              spend = 0.0
              deviation = 0.0

          print("reviewed=%d breached=%d cap=%.2f sigma=%.2f verdict=%s" % (
              len(records), len(reviewable), cap, sigma, verdict))
          print("::" + json.dumps({"outputs": {
              "verdict": verdict,
              "campaign": campaign,
              "spend": spend,
              "deviation": deviation,
              "breached": reviewable,
          }}) + "::")

  # --- 4. pause gate ---
  - id: pause_gate
    type: io.kestra.plugin.core.flow.If
    description: Branch on whether any campaign breached the cap or the sigma
      threshold, so only breached campaigns are paused and escalated.
    condition: "{{ outputs.analyze_spend.vars.breached | length > 0 }}"
    then:
      # --- 5. pause every breached campaign ---
      - id: pause_campaigns
        type: io.kestra.plugin.core.flow.Loop
        description: Walk the breached campaigns one at a time and pause each through
          the ads pause endpoint, keeping the API calls in lockstep instead of a
          burst.
        values: "{{ outputs.analyze_spend.vars.breached }}"
        concurrencyLimit: 1
        tasks:
          - id: pause_campaign
            type: io.kestra.plugin.core.http.Request
            description: POST the pause command for the current campaign to the ads platform
              with a bearer token, so spend stops before the budget empties.
            method: POST
            uri: "https://ads.example.com/v1/campaigns/{{ item.value.id }}/pause"
            headers:
              Authorization: "Bearer {{ secret('ADS_API_TOKEN') }}"
              Content-Type: application/json
            body: |
              {"campaign_id": "{{ item.value.id }}", "reason": "{{ item.value.reason }}", "source": "{{ flow.id }}"}
            retry:
              type: exponential
              interval: PT2S
              maxInterval: PT10S
              maxAttempts: 3

      # --- 6. track the pause ---
      - id: open_pause_issue
        type: io.kestra.plugin.github.issues.Create
        description: Open a labeled GitHub issue listing every paused campaign with its
          spend and deviation, so the marketing team can review creative or
          bidding after the pause.
        oauthToken: "{{ secret('GITHUB_TOKEN') }}"
        repository: your-org/marketing-ops
        title: "Ad spend guard paused {{ outputs.analyze_spend.vars.campaign }}"
        body: |
          ## Ad spend anomaly detected

          Campaign **{{ outputs.analyze_spend.vars.campaign }}** was paused after today's
          spend reached **${{ outputs.analyze_spend.vars.spend }}** against a daily cap of
          ${{ inputs.daily_cap_usd }}, a deviation of **{{ outputs.analyze_spend.vars.deviation }}**
          sigma with a threshold of {{ inputs.sigma_threshold }}.

          Verdict `{{ outputs.analyze_spend.vars.verdict }}`, execution `{{ execution.id }}`.

          ### Paused campaigns

          ```json
          {{ outputs.analyze_spend.vars.breached | toJson }}
          ```

          ### Next steps

          - Check bids, budgets and audience expansion on the paused campaigns.
          - Lower the daily cap or the sigma threshold if this campaign should never have reached this spend.
          - Resume the campaign from the ads console once the cause is understood.
        labels:
          - ad-spend
          - anomaly

      # --- 7. channel alert ---
      - id: alert_pause
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Tell the marketing ops channel how many campaigns were paused, the
          worst offender and its deviation, right after the pauses and the issue
          exist.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "Ad spend guard paused {{ outputs.analyze_spend.vars.breached | length }} campaign(s). Worst offender {{ outputs.analyze_spend.vars.campaign }} spent ${{ outputs.analyze_spend.vars.spend }} today against a ${{ inputs.daily_cap_usd }} cap, a deviation of {{ outputs.analyze_spend.vars.deviation }} sigma (threshold {{ inputs.sigma_threshold }}). Issue opened in your-org/marketing-ops. Execution {{ execution.id }}."
          }
    else:
      # --- 8. all clear ---
      - id: log_within_budget
        type: io.kestra.plugin.core.log.Log
        description: Record the within budget outcome so the execution history still
          proves which day was checked.
        message: "Spend guard clear - verdict {{ outputs.analyze_spend.vars.verdict }},
          highest campaign spend ${{ outputs.analyze_spend.vars.spend }} under
          the ${{ inputs.daily_cap_usd }} cap and max deviation {{
          outputs.analyze_spend.vars.deviation }} sigma under {{
          inputs.sigma_threshold }}, no campaign paused."

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate any failure to the ops channel with the execution url.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: verdict
    type: STRING
    description: PAUSE_REQUIRED when at least one campaign breached, WITHIN_BUDGET
      when every campaign stayed inside both thresholds, otherwise NO_DATA.
    value: "{{ outputs.analyze_spend.vars.verdict }}"
  - id: campaign
    type: STRING
    description: Identifier of the worst offending campaign, for example
      cmp-summer-prospecting, or none when nothing breached.
    value: "{{ outputs.analyze_spend.vars.campaign }}"
  - id: spend
    type: FLOAT
    description: Today's spend in US dollars for the worst offending campaign, or
      the highest observed spend when every campaign is within budget.
    value: "{{ outputs.analyze_spend.vars.spend }}"
  - id: deviation
    type: FLOAT
    description: Z-score of today's spend against the campaign's own seven day
      history, for example 4.2 for a sharp spike.
    value: "{{ outputs.analyze_spend.vars.deviation }}"

How it works

on_demand_check (io.kestra.plugin.core.trigger.Webhook, keyed with the ADS_WEBHOOK_KEY secret) runs the guard on demand, for example right after a flight launches or a budget is raised. daily_spend_watch (io.kestra.plugin.core.trigger.Schedule, cron 0 6 * * *) is the daily safety net and ships disabled: true until an operator turns it on.
campaign_workspace (io.kestra.plugin.core.flow.WorkingDirectory) opens one shared filesystem so the fixture and the analysis see the same file.
write_campaign_fixture (io.kestra.plugin.scripts.shell.Commands on the io.kestra.plugin.scripts.runner.docker.Docker runner, image alpine:3.20) writes campaigns.json with a heredoc. The fixture holds three campaigns, each with a seven day history and today's spend, one ordinary day, one sharp spike above the cap and one steady under-spender. entryPoint: [] clears the image entrypoint so the shell script runs instead of the default command.
analyze_spend (io.kestra.plugin.scripts.python.Script on python:3.12-slim) reads the fixture, computes the mean and the sample standard deviation of each history, then the z-score of today's spend. A campaign breaches when today > daily_cap_usd or z > sigma_threshold; the worst offender is the breached campaign with the highest z-score. The verdict, the campaign, the spend, the deviation and the breached list are printed as a single ::{"outputs": ...}:: line.
pause_gate (io.kestra.plugin.core.flow.If) branches on breached | length > 0. Above zero, pause_campaigns (io.kestra.plugin.core.flow.Loop, concurrencyLimit: 1) walks the breached list and pause_campaign (io.kestra.plugin.core.http.Request) POSTs https://ads.example.com/v1/campaigns/{id}/pause with a Bearer token from the ADS_API_TOKEN secret and retries transient failures. open_pause_issue (io.kestra.plugin.github.issues.Create) files a labeled issue with the paused list, and alert_pause posts the counts to Slack. At zero, log_within_budget records the quiet outcome.
The flow level errors handler (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) reports any failure with the execution id, because a guard that dies silently looks exactly like a quiet day.

What you get

A pause before the budget empties: breached campaigns are stopped through the ads API the moment they cross either threshold.
Two thresholds working together, a hard dollar cap and a statistical z-score, so a small budget that suddenly triples is caught even when the absolute number looks modest.
Gateable results: verdict, campaign, spend and deviation are flow outputs a downstream budget review or dashboard can branch on.
Escalation where the work happens, a labeled GitHub issue carrying the paused list plus a Slack ping with the counts, and a quiet log when everything is healthy.

Who it's for

Performance marketers who own paid search and paid social budgets and cannot watch the ad consoles all day.
Growth and finance partners who need a hard daily ceiling that nobody can silently overspend.
Marketing ops teams who want paused campaigns to arrive as tickets and channel messages instead of a surprise invoice.

Why orchestrate this with Kestra

An ad platform's own budget cap is a single number with no history behind it, and it tells you nothing about a campaign that usually spends forty dollars and suddenly spends five hundred. Kestra gives the check a cadence with a disabled daily schedule as the safety net, a keyed webhook for on demand runs, a fixture driven analysis that z-scores today against the campaign's own past week, a conditional branch that separates the pause from the all clear, a loop that calls the pause endpoint once per breached campaign, and routing into GitHub and Slack from one version controlled file. The same file carries the thresholds, the escalation policy and the execution history that proves the guard actually ran.

Prerequisites

A worker with Docker access, the fixture is written in alpine:3.20 and the analysis runs in python:3.12-slim.
An ads platform account whose pause endpoint matches https://ads.example.com/v1/campaigns/{id}/pause, or a stub of that API while you trial the flow.
A GitHub token that can open issues in the repository named on open_pause_issue (classic PAT scope repo, or fine-grained Issues: write).
A Slack incoming webhook for the pause alert and for failure alerts.

Secrets

ADS_WEBHOOK_KEY: key segment guarding the on demand webhook URL, anyone holding it can trigger a spend check.
ADS_API_TOKEN: bearer token used to call the ads platform pause endpoint.
GITHUB_TOKEN: token used to open the pause review issue.
SLACK_WEBHOOK_URL: Slack incoming webhook URL for pause and failure alerts.

Quick start

Add the four secrets to your Kestra namespace.
Run the flow once with the default inputs and confirm verdict reads PAUSE_REQUIRED, campaign reads cmp-summer-prospecting, spend reads 544.2 and deviation reads about 4.2.
Point open_pause_issue.repository at a repository where issues are welcome, then re-run with sigma_threshold at 10 to watch the sigma branch go quiet while the cap still pauses the spike.
Trigger on_demand_check at /api/v1/{tenant}/executions/webhook/{namespace}/ad-spend-anomaly-guard/{ADS_WEBHOOK_KEY}, or set disabled: false on daily_spend_watch once the token and webhook are in place.

How to extend

Replace the heredoc fixture with a io.kestra.plugin.core.http.Request that pulls yesterday's spend from the ads reporting API, keeping the fixture as the offline fallback.
Add a third threshold on week-to-date spend with another If inside pause_gate so long flights are caught before the daily cap trips.
Resume campaigns automatically with a second POST to the resume endpoint once tomorrow's spend normalizes.
Fan the breached list out to a subflow per campaign with io.kestra.plugin.core.flow.ForEachItem when pausing across several ad networks at once.

Pause Runaway Ad Campaigns Before They Empty the Daily Budget

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Pause Runaway Ad Campaigns Before They Empty the Daily Budget

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Business Blueprints

AI Customer Call Compliance and Regulatory Disclosure Auditor

Adanos Asset Sentiment Drift Sentinel

AI Contract Clause Risk Analyzer and Legal Sign-Off Gate

Read out an A/B test with a sample ratio check and a guardrail metric

Automate Customer Feedback Triage, Sentiment Analysis, and Escalation with Google Gemini and Slack

Generate AI Blog Posts in Your Brand Voice and Publish to WordPress as Draft

New to Kestra?