#!/usr/bin/env python3 """negatives_audit.py — audit your Amazon Sponsored Products negative keywords, and start the archive that makes next quarter's audit possible. https://www.databaaba.com/free/blocked-negatives/ WHAT THIS ANSWERS "Are any of my negative keywords blocking search terms that convert?" For most of your negatives the honest answer is: nobody can tell you, including Amazon. Two walls make it impossible. 1. THE BLACKOUT. A negative does not just block a search term, it erases the evidence. The moment the negative goes live the term stops appearing in your search-term report. There is nothing left to measure. 2. THE RETENTION WALL. Search-term data expires. Ask for a report older than the limit and the API tells you so itself: startDate (2026-05-20) must be equal to or after report type data retention start date (2026-05-28) That was 95 days on the account I probed it on. This script measures YOUR wall rather than trusting mine — Amazon has not committed to the number. Together: a negative can only be tested if it was created INSIDE the retention window, because that is the only period where the term it blocks still has visible history. On my own account that was 588 of 1,821 live negatives. The other 1,233 are not hard to check. They are unanswerable. So this script does two things: - audits the slice that is still auditable, honestly - ARCHIVES your search-term reports, so the wall stops moving on you The archive is the part that matters. Run it on a schedule and in a year you can answer the question for real. Skip it and you are back here next quarter with a bigger unauditable number. WHAT IT DOES NOT DO It never writes to your account. Every call is a list or a report read. It does not delete or modify a single negative. What to remove is your call. CREDENTIALS — yours, in your shell, never leaving your machine export ADS_CLIENT_ID='amzn1.application-oa2-client.xxxx' export ADS_CLIENT_SECRET='xxxx' export ADS_REFRESH_TOKEN='Atzr|xxxx' USAGE python3 negatives_audit.py --profiles python3 negatives_audit.py --profile 1234567890 --region NA python3 negatives_audit.py --profile 1234567890 --region NA --archive-only Standard library only. Python 3.8+. """ import argparse import collections import csv import datetime as dt import gzip import json import os import string import sys import time import urllib.error import urllib.parse import urllib.request LWA = "https://api.amazon.com/auth/o2/token" HOSTS = {"NA": "advertising-api.amazon.com", "EU": "advertising-api-eu.amazon.com", "FE": "advertising-api-fe.amazon.com"} # endpoint -> (vendor resource name, the key rows actually come back under). # These keys are not guessable from the path: /sp/negativeTargets/list answers # under "negativeTargetingClauses". Reading the wrong key gives a silent empty # list with HTTP 200, which reads as "no negatives" — the worst possible bug in # a script whose whole job is counting negatives. EP = { "campaigns": ("spCampaign", "campaigns"), "adGroups": ("spAdGroup", "adGroups"), "negativeKeywords": ("spNegativeKeyword", "negativeKeywords"), "campaignNegativeKeywords": ("spCampaignNegativeKeyword", "campaignNegativeKeywords"), "negativeTargets": ("spNegativeTargetingClause", "negativeTargetingClauses"), "campaignNegativeTargets": ("spCampaignNegativeTargetingClause", "campaignNegativeTargetingClauses"), } PUNCT = str.maketrans({c: " " for c in string.punctuation}) def die(msg): print(f"error: {msg}", file=sys.stderr) sys.exit(1) def env(name): v = os.environ.get(name) if not v: die(f"{name} is not set. See the header of this file.") return v def token(): body = urllib.parse.urlencode({ "grant_type": "refresh_token", "refresh_token": env("ADS_REFRESH_TOKEN"), "client_id": env("ADS_CLIENT_ID"), "client_secret": env("ADS_CLIENT_SECRET"), }).encode() req = urllib.request.Request( LWA, data=body, headers={"Content-Type": "application/x-www-form-urlencoded"}) try: with urllib.request.urlopen(req, timeout=30) as r: return json.load(r)["access_token"] except urllib.error.HTTPError as e: die(f"login failed ({e.code}): {e.read().decode()[:300]}") class Api: def __init__(self, profile, region): self.host = HOSTS[region] self.profile = profile self.tok = token() def call(self, method, path, body=None, ctype=None): h = {"Authorization": f"Bearer {self.tok}", "Amazon-Advertising-API-ClientId": env("ADS_CLIENT_ID"), "Amazon-Advertising-API-Scope": str(self.profile), "Content-Type": ctype or "application/json", "Accept": ctype or "application/json"} data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(f"https://{self.host}{path}", data=data, method=method, headers=h) try: with urllib.request.urlopen(req, timeout=60) as r: raw = r.read().decode() return r.status, (json.loads(raw) if raw else {}) except urllib.error.HTTPError as e: raw = e.read().decode() try: return e.code, json.loads(raw) except json.JSONDecodeError: return e.code, raw def page(self, name): """Every row of an SP v3 list endpoint, following nextToken. includeExtendedDataFields is what returns creationDateTime, and without that date none of this works: a negative with no birthday cannot be placed on either side of the retention wall. """ vendor, key = EP[name] ct = f"application/vnd.{vendor}.v3+json" path = f"/sp/{name}/list" rows, nxt = [], None while True: body = {"maxResults": 1000, "includeExtendedDataFields": True} if nxt: body["nextToken"] = nxt status, d = self.call("POST", path, body, ct) if status != 200: if not rows: print(f" ! {name}: HTTP {status} {str(d)[:160]}", file=sys.stderr) break rows += d.get(key) or [] nxt = d.get("nextToken") if not nxt: break time.sleep(0.2) return rows def norm(s): return " ".join((s or "").lower().translate(PUNCT).split()) def blocks(neg_text, match_type, term): """Does this negative literally block this search term? NEGATIVE_EXACT -> the whole term, identically NEGATIVE_PHRASE -> the negative appears as a run of consecutive words Deliberately literal. Amazon ALSO blocks close variants and plurals, so real conflicts are at least what this reports, never fewer. A scanner that guesses at variants would hand you false positives, and a false positive here costs you a negative you should have kept. """ n, t = norm(neg_text).split(), norm(term).split() if not n: return False if match_type == "NEGATIVE_EXACT": return n == t return any(t[i:i + len(n)] == n for i in range(len(t) - len(n) + 1)) def find_wall(api): """Ask Amazon where its own retention wall is, by walking into it. The 400 names the date. That is more trustworthy than any documented number and it is per-account, so measure it rather than assume it. """ probe = (dt.date.today() - dt.timedelta(days=400)).isoformat() status, d = api.call("POST", "/reporting/reports", { "name": "retention probe", "startDate": probe, "endDate": probe, "configuration": {"adProduct": "SPONSORED_PRODUCTS", "groupBy": ["searchTerm"], "columns": ["searchTerm", "impressions", "date"], "reportTypeId": "spSearchTerm", "timeUnit": "DAILY", "format": "GZIP_JSON"}}, "application/vnd.createasyncreportrequest.v3+json") msg = json.dumps(d) if isinstance(d, dict) else str(d) # The date arrives wrapped in parentheses inside the message string: # "... data retention start date (2026-05-28)" # so the brackets have to come off before splitting or every candidate is # 11 characters and nothing matches. cleaned = msg.replace('"', " ").replace("(", " ").replace(")", " ") for token_ in cleaned.split(): if len(token_) == 10 and token_[4] == "-" and token_[7] == "-": try: cand = dt.date.fromisoformat(token_) except ValueError: continue if cand != dt.date.fromisoformat(probe): return cand, msg[:300] return None, msg[:300] def fetch_reports(api, start, end, out_dir): """Search-term reports in <=31-day slices, saved to disk as they arrive. Saving each slice as it lands is the whole point of the archive: this data expires and cannot be re-requested once it falls behind the wall. """ slices, cur = [], start while cur <= end: stop = min(cur + dt.timedelta(days=30), end) slices.append((cur, stop)) cur = stop + dt.timedelta(days=1) jobs = [] for a, b in slices: status, d = api.call("POST", "/reporting/reports", { "name": f"negatives_audit {a}..{b}", "startDate": a.isoformat(), "endDate": b.isoformat(), "configuration": {"adProduct": "SPONSORED_PRODUCTS", "groupBy": ["searchTerm"], "columns": ["campaignId", "adGroupId", "keywordId", "targeting", "searchTerm", "impressions", "clicks", "cost", "purchases7d", "sales7d", "date"], "reportTypeId": "spSearchTerm", "timeUnit": "DAILY", "format": "GZIP_JSON"}}, "application/vnd.createasyncreportrequest.v3+json") if status in (200, 202) and isinstance(d, dict) and d.get("reportId"): jobs.append((d["reportId"], a, b)) print(f" requested {a}..{b}") else: print(f" ! {a}..{b} rejected: {str(d)[:180]}") time.sleep(0.4) rows, waited = [], 0 while jobs and waited < 900: for job in list(jobs): rid, a, b = job status, d = api.call("GET", f"/reporting/reports/{rid}", None, "application/vnd.createasyncreportrequest.v3+json") state = d.get("status") if isinstance(d, dict) else None if state == "COMPLETED": with urllib.request.urlopen(d["url"], timeout=180) as r: blob = r.read() got = json.loads(gzip.decompress(blob).decode()) name = os.path.join(out_dir, f"searchterms_{a}_{b}.json") with open(name, "w") as f: json.dump(got, f) print(f" archived {os.path.basename(name)} {len(got)} rows") rows += got jobs.remove(job) elif state in ("FAILED", "CANCELLED"): print(f" ! {a}..{b} failed: {d.get('failureReason')}") jobs.remove(job) if jobs: time.sleep(20) waited += 20 if jobs: print(f" ! {len(jobs)} report(s) still pending after 15 minutes; re-run to collect") return rows def audit(negs, camps, ags, rows, wall): """Split negatives into testable and unauditable, then test the testable.""" camp_state = {str(c["campaignId"]): c.get("state") for c in camps} ag_state = {str(a["adGroupId"]): a.get("state") for a in ags} live, testable = [], [] for n in negs: if n.get("state") != "ENABLED": continue cid = str(n.get("campaignId") or "") if camp_state.get(cid) == "ARCHIVED": continue live.append(n) born = ((n.get("extendedData") or {}).get("creationDateTime") or "")[:10] if born and wall and born >= wall.isoformat(): testable.append(n) by_ag, by_camp = collections.defaultdict(list), collections.defaultdict(list) for n in testable: if n.get("adGroupId"): by_ag[str(n["adGroupId"])].append(n) else: by_camp[str(n["campaignId"])].append(n) agg = collections.defaultdict(lambda: {"purchases": 0, "sales": 0.0, "clicks": 0, "cost": 0.0, "dates": []}) for r in rows: k = (str(r["campaignId"]), str(r["adGroupId"]), r["searchTerm"]) a = agg[k] a["purchases"] += r.get("purchases7d") or 0 a["sales"] += r.get("sales7d") or 0.0 a["clicks"] += r.get("clicks") or 0 a["cost"] += r.get("cost") or 0.0 if (r.get("purchases7d") or 0) > 0: a["dates"].append(r["date"]) conflicts, suspect, checked = [], [], 0 for (cid, agid, term), a in agg.items(): if a["purchases"] < 1 or a["sales"] <= 0: continue if camp_state.get(cid) == "ARCHIVED" or ag_state.get(agid) == "ARCHIVED": continue checked += 1 hits = [] for scope, pool in (("ad group", by_ag.get(agid, [])), ("campaign", by_camp.get(cid, []))): for n in pool: if blocks(n.get("keywordText"), n.get("matchType"), term): hits.append({"scope": scope, "negative": n.get("keywordText"), "matchType": n.get("matchType"), "keywordId": n.get("keywordId"), "created": ((n.get("extendedData") or {}) .get("creationDateTime") or "")[:10]}) if not hits: continue born = sorted(h["created"] for h in hits if h["created"]) rec = {"searchTerm": term, "campaignId": cid, "adGroupId": agid, "orders": a["purchases"], "sales": round(a["sales"], 2), "clicks": a["clicks"], "cost": round(a["cost"], 2), "lastConverted": max(a["dates"]), "blockedBy": hits, "negativeAdded": born[0] if born else None} # The causality test. If the term still converted AFTER the negative went # live, the negative is not what stopped it — so this rule over-matched # and the row is not a finding. Excluding these is what stops the script # telling you to delete a negative that is doing its job. if born and max(a["dates"]) < born[0]: conflicts.append(rec) else: suspect.append(rec) conflicts.sort(key=lambda r: -r["sales"]) return live, testable, conflicts, suspect, checked def list_profiles(): tok = token() for region, host in HOSTS.items(): req = urllib.request.Request( f"https://{host}/v2/profiles", headers={"Authorization": f"Bearer {tok}", "Amazon-Advertising-API-ClientId": env("ADS_CLIENT_ID")}) try: with urllib.request.urlopen(req, timeout=30) as r: data = json.load(r) except urllib.error.HTTPError as e: print(f"{region}: HTTP {e.code}") continue print(f"\n{region}: {len(data)} profile(s)") for p in data: info = p.get("accountInfo", {}) print(f" {p.get('profileId')} {p.get('countryCode')} " f"{info.get('name')!r} {info.get('type')}") print("\nUK/GB is in EU, not NA. That catches everyone once.") def main(): ap = argparse.ArgumentParser( description="Audit Amazon SP negative keywords and archive your search terms.") ap.add_argument("--profiles", action="store_true", help="list your profile IDs and exit") ap.add_argument("--profile", help="profile ID to audit") ap.add_argument("--region", default="NA", choices=list(HOSTS)) ap.add_argument("--days", type=int, default=95, help="days of search-term history to request (default 95)") ap.add_argument("--archive-only", action="store_true", help="just save the reports; skip the audit") ap.add_argument("--out", default="negatives-archive", help="directory for the archive and the report (default negatives-archive)") a = ap.parse_args() if a.profiles: list_profiles() return if not a.profile: die("--profile is required. Run --profiles to find yours.") os.makedirs(a.out, exist_ok=True) api = Api(a.profile, a.region) print("Finding your retention wall...") wall, raw = find_wall(api) if wall: age = (dt.date.today() - wall).days print(f" Amazon will not serve search-term data before {wall} " f"— {age} days back.\n") else: wall = dt.date.today() - dt.timedelta(days=a.days) print(f" could not read the wall from the API; assuming {wall}.") print(f" response was: {raw}\n") end = dt.date.today() - dt.timedelta(days=2) start = max(wall, end - dt.timedelta(days=a.days)) print(f"Archiving search terms {start}..{end} into {a.out}/") rows = fetch_reports(api, start, end, a.out) print(f" {len(rows)} rows archived\n") if a.archive_only: print("Archive written. That is the part that compounds — schedule it.") return print("Reading your negatives...") negs = (api.page("negativeKeywords") + api.page("campaignNegativeKeywords")) ntargets = (api.page("negativeTargets") + api.page("campaignNegativeTargets")) camps, ags = api.page("campaigns"), api.page("adGroups") print(f" {len(negs)} negative keywords, {len(ntargets)} negative product targets, " f"{len(camps)} campaigns\n") live, testable, conflicts, suspect, checked = audit(negs, camps, ags, rows, wall) dark = len(live) - len(testable) bar = "=" * 66 print(bar) print(" NEGATIVE KEYWORD AUDIT") print(bar) print(f" live negatives {len(live):6}") print(f" testable (created after {wall}) {len(testable):6}") print(f" UNAUDITABLE {dark:6}" f" ({100 * dark / max(len(live), 1):.0f}%)") print() print(f" converting search terms checked {checked:6}") print(f" conflicts found {len(conflicts):6}") print(f" excluded, still converting {len(suspect):6}") print(bar) if conflicts: print("\n Terms that converted, then got blocked:\n") for c in conflicts[:25]: b = c["blockedBy"][0] print(f" {c['sales']:10.2f} {c['orders']:3} order(s) {c['searchTerm'][:40]}") print(f"{'':14}blocked by {b['scope']} {b['matchType']} " f"{b['negative']!r} added {b['created']}") print(f"{'':14}last converted {c['lastConverted']}") else: print("\n No conflicts in the testable slice. That is a real result — the") print(" negatives you added recently are not blocking anything that converts.") if dark: print(f"\n But {dark} of your negatives could not be tested at all. Each one") print(" blocked its search term, the term stopped reporting that day, and the") print(" history behind it has expired. No tool can recover that, including") print(" Amazon's. The only fix is forward: keep the archive running.") out = os.path.join(a.out, "audit.json") with open(out, "w") as f: json.dump({"profile": a.profile, "region": a.region, "runAt": dt.datetime.now().isoformat(timespec="seconds"), "retentionWall": wall.isoformat(), "liveNegatives": len(live), "testable": len(testable), "unauditable": dark, "convertingTermsChecked": checked, "conflicts": conflicts, "excludedStillConverting": suspect}, f, indent=1) if conflicts: csv_path = os.path.join(a.out, "conflicts.csv") with open(csv_path, "w", newline="") as f: w = csv.writer(f) w.writerow(["searchTerm", "orders", "sales", "clicks", "cost", "lastConverted", "negative", "matchType", "scope", "negativeAdded", "campaignId", "adGroupId"]) for c in conflicts: b = c["blockedBy"][0] w.writerow([c["searchTerm"], c["orders"], c["sales"], c["clicks"], c["cost"], c["lastConverted"], b["negative"], b["matchType"], b["scope"], b["created"], c["campaignId"], c["adGroupId"]]) print(f"\n wrote {csv_path}") print(f" wrote {out}") print(f"\n Re-run monthly. The archive in {a.out}/ is the only thing that makes") print(" next year's version of this question answerable.") if __name__ == "__main__": main()