#!/usr/bin/env python3 """ selfcomp_scan.py — find the places your own Amazon Sponsored Products campaigns compete with each other, and price what the overlap costs. Python 3.8+. Standard library only. Read-only: it never writes to your account. export ADS_CLIENT_ID=amzn1.application-oa2-client.xxxx export ADS_CLIENT_SECRET=xxxx export ADS_REFRESH_TOKEN=Atzr|xxxx python3 selfcomp_scan.py --profiles # list your profiles python3 selfcomp_scan.py --profile 1234567890 --region NA python3 selfcomp_scan.py --profile 1234567890 --region NA --structure-only What it reports --------------- STRUCTURE (your account as it stands today) tier 1 the same keyword + match type in two ad groups on the same ASIN tier 2 an exact keyword in one campaign that a phrase/broad keyword in another campaign would also match — checked against your existing negatives, so anything you already blocked is not counted against you tier 3 an auto campaign running on an ASIN your manual campaigns target tier 4 the same keyword on two different ASINs of yours SPEND (what actually happened, from your search-term reports) contested a search term that took clicks in 2+ of your own campaigns inside the same 30-day window premium clicks x (this campaign's CPC - the cheapest campaign's CPC) for the same term in the same window Read the premium carefully. It is arithmetic, not a promise. It assumes the cheaper campaign could have won those clicks at its own price, which is an upper bound. The script therefore prints a second figure that ignores any baseline campaign with fewer than --floor clicks, because one click at $0.80 is not evidence that 700 clicks were available at $0.80. Trust the second number. Nothing here proves two of your ads entered the same auction. Amazon serves one ad per advertiser per query. What this measures is how much of your spend lands on terms several of your campaigns chase, and how far apart the prices are. """ import argparse import collections import csv import datetime as dt import gzip import itertools import json import os 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, key the rows come back under). # /sp/targets/list answers under "targetingClauses" — reading "targets" gives a # silent empty list with HTTP 200, which reads as "no product targeting". EP = { "campaigns": ("spCampaign", "campaigns"), "adGroups": ("spAdGroup", "adGroups"), "productAds": ("spProductAd", "productAds"), "keywords": ("spKeyword", "keywords"), "targets": ("spTargetingClause", "targetingClauses"), "negativeKeywords": ("spNegativeKeyword", "negativeKeywords"), "campaignNegativeKeywords": ("spCampaignNegativeKeyword", "campaignNegativeKeywords"), } 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, region, profile=None): self.host = HOSTS[region] self.profile = profile self.tok = token() self.minted = time.time() def call(self, method, path, body=None, headers=None): if time.time() - self.minted > 2700: # access tokens last an hour self.tok, self.minted = token(), time.time() h = {"Authorization": f"Bearer {self.tok}", "Amazon-Advertising-API-ClientId": env("ADS_CLIENT_ID"), "Content-Type": "application/json"} if self.profile: h["Amazon-Advertising-API-Scope"] = str(self.profile) h.update(headers or {}) 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 None) 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, kind): res, key = EP[kind] ct = f"application/vnd.{res}.v3+json" rows, nxt, guard = [], None, 0 while True: body = {"maxResults": 500} if nxt: body["nextToken"] = nxt st, d = self.call("POST", f"/sp/{kind}/list", body=body, headers={"Content-Type": ct, "Accept": ct}) if st == 429: time.sleep(3) continue if st != 200: print(f" ! {kind}: HTTP {st} {json.dumps(d)[:160]}", file=sys.stderr) return rows rows += d.get(key, []) nxt, guard = d.get("nextToken"), guard + 1 if not nxt or guard > 80: return rows # ---------------------------------------------------------------- structure def toks(s): return [t for t in "".join(c if c.isalnum() else " " for c in s.lower()).split() if t] def subsumes(wide_text, match_type, exact_text): """Would a phrase/broad keyword also match what this exact keyword matches?""" w, e = toks(wide_text.replace("+", " ")), toks(exact_text) if not w or not e: return False if match_type == "PHRASE": return any(e[i:i + len(w)] == w for i in range(len(e) - len(w) + 1)) if match_type == "BROAD": return set(w) <= set(e) return False def blocked_by(term, campaign_id, ad_group_id, negatives): t = toks(term) for n in negatives: if n["campaignId"] != campaign_id: continue if n.get("adGroupId") and ad_group_id and n["adGroupId"] != ad_group_id: continue nt, mt = toks(n.get("keywordText", "")), n.get("matchType") if mt == "NEGATIVE_EXACT" and nt == t: return n["keywordText"] if mt == "NEGATIVE_PHRASE" and nt and any( t[i:i + len(nt)] == nt for i in range(len(t) - len(nt) + 1)): return n["keywordText"] return None def structure(api): raw = {k: api.page(k) for k in EP} camps = {c["campaignId"]: c for c in raw["campaigns"]} ags = {a["adGroupId"]: a for a in raw["adGroups"]} live_c = {i: c for i, c in camps.items() if c.get("state") == "ENABLED"} live_ag = {i: a for i, a in ags.items() if a.get("state") == "ENABLED" and a.get("campaignId") in live_c} ag_asins = collections.defaultdict(set) for p in raw["productAds"]: if p.get("state") == "ENABLED" and p.get("adGroupId") in live_ag: ag_asins[p["adGroupId"]].add(p.get("asin") or f"sku:{p.get('sku')}") by_asin = collections.defaultdict(set) for ag, asins in ag_asins.items(): for a in asins: by_asin[a].add(ag) negs = [n for n in raw["negativeKeywords"] + raw["campaignNegativeKeywords"] if n.get("state") == "ENABLED"] live_kw = [k for k in raw["keywords"] if k.get("state") == "ENABLED" and k.get("adGroupId") in live_ag] def cname(ag): return live_c[live_ag[ag]["campaignId"]].get("name") def cid(ag): return live_ag[ag]["campaignId"] out = {"live": {"campaigns": len(live_c), "adGroups": len(live_ag), "keywords": len(live_kw), "enabledNegatives": len(negs)}, "asins": {}, "tier1": [], "tier2": [], "tier3": [], "tier4": []} for asin, ags_ in sorted(by_asin.items(), key=lambda x: -len(x[1])): out["asins"][asin] = {"campaigns": len({cid(g) for g in ags_}), "adGroups": len(ags_)} kws = [k for k in live_kw if k["adGroupId"] in ags_] buckets = collections.defaultdict(list) for k in kws: buckets[(k["keywordText"].lower().strip(), k["matchType"])].append(k) for (text, mt), grp in buckets.items(): if len({g["adGroupId"] for g in grp}) > 1: out["tier1"].append({"asin": asin, "term": text, "matchType": mt, "campaigns": [cname(g["adGroupId"]) for g in grp], "bids": [g.get("bid") for g in grp]}) exacts = [k for k in kws if k["matchType"] == "EXACT"] wides = [k for k in kws if k["matchType"] in ("PHRASE", "BROAD")] for e, w in itertools.product(exacts, wides): if e["adGroupId"] == w["adGroupId"]: continue if not subsumes(w["keywordText"], w["matchType"], e["keywordText"]): continue out["tier2"].append({ "asin": asin, "term": e["keywordText"], "exactCampaign": cname(e["adGroupId"]), "exactBid": e.get("bid"), "wideCampaign": cname(w["adGroupId"]), "wideKeyword": w["keywordText"], "wideMatchType": w["matchType"], "wideBid": w.get("bid"), "blockedBy": blocked_by(e["keywordText"], cid(w["adGroupId"]), w["adGroupId"], negs)}) for c in {cid(g) for g in ags_ if live_c[cid(g)].get("targetingType") == "AUTO"}: reach = [e["keywordText"] for e in exacts if not blocked_by(e["keywordText"], c, None, negs)] out["tier3"].append({"asin": asin, "campaign": live_c[c].get("name"), "manualExactTermsNotNegated": sorted(reach)}) one_asin = {g: sorted(a)[0] for g, a in ag_asins.items() if len(a) == 1} spread = collections.defaultdict(list) for k in live_kw: if k["adGroupId"] in one_asin: spread[(k["keywordText"].lower().strip(), k["matchType"])].append(k) for (text, mt), grp in spread.items(): asins = {one_asin[g["adGroupId"]] for g in grp} if len(asins) > 1: out["tier4"].append({"term": text, "matchType": mt, "asins": sorted(asins), "campaigns": [cname(g["adGroupId"]) for g in grp], "bids": [g.get("bid") for g in grp]}) return out # ------------------------------------------------------------------- spend def search_terms(api, days): """Search-term rows for the last `days` days, in 30-day windows.""" ct = "application/vnd.createasyncreportrequest.v3+json" cols = ["campaignId", "campaignName", "adGroupName", "keyword", "matchType", "searchTerm", "impressions", "clicks", "cost"] today = dt.date.today() windows, cursor = [], today - dt.timedelta(days=days) while cursor < today: end = min(cursor + dt.timedelta(days=29), today - dt.timedelta(days=1)) windows.append((cursor.isoformat(), end.isoformat())) cursor = end + dt.timedelta(days=1) jobs = [] for start, end in windows: st, d = api.call("POST", "/reporting/reports", headers={"Content-Type": ct, "Accept": ct}, body={"name": f"selfcomp-{start}", "startDate": start, "endDate": end, "configuration": {"adProduct": "SPONSORED_PRODUCTS", "groupBy": ["searchTerm"], "columns": cols, "reportTypeId": "spSearchTerm", "timeUnit": "SUMMARY", "format": "GZIP_JSON"}}) if st in (200, 202): jobs.append({"id": d["reportId"], "window": start}) print(f" requested {start}..{end}") else: print(f" ! report {start}: HTTP {st} {json.dumps(d)[:200]}", file=sys.stderr) rows = [] while jobs: time.sleep(20) for j in list(jobs): st, d = api.call("GET", f"/reporting/reports/{j['id']}", headers={"Accept": ct}) status = d.get("status") if isinstance(d, dict) else st if status == "COMPLETED": blob = urllib.request.urlopen(d["url"], timeout=180).read() got = json.loads(gzip.decompress(blob).decode()) for r in got: r["_window"] = j["window"] rows += got print(f" {j['window']}: {len(got)} rows") jobs.remove(j) elif status in ("FAILURE", "CANCELLED"): print(f" ! {j['window']}: {status} {d.get('failureReason')}", file=sys.stderr) jobs.remove(j) return rows def contested(rows, floor): agg = collections.defaultdict(lambda: {"clicks": 0, "cost": 0.0, "kw": set(), "name": None}) total = {"clicks": 0, "cost": 0.0} for r in rows: a = agg[(r["_window"], r["searchTerm"], r["campaignId"])] a["clicks"] += r.get("clicks", 0) or 0 a["cost"] += r.get("cost", 0) or 0 a["name"] = r.get("campaignName") if r.get("keyword"): a["kw"].add(f"{r['keyword']} [{r.get('matchType')}]") total["clicks"] += r.get("clicks", 0) or 0 total["cost"] += r.get("cost", 0) or 0 grouped = collections.defaultdict(dict) for (win, term, cid), a in agg.items(): if a["clicks"] > 0: grouped[(win, term)][cid] = a out = [] for (win, term), camps in grouped.items(): if len(camps) < 2: continue cr = sorted(({"campaign": a["name"], "clicks": a["clicks"], "cost": round(a["cost"], 2), "cpc": round(a["cost"] / a["clicks"], 4), "keywords": sorted(a["kw"])} for a in camps.values()), key=lambda r: r["cpc"]) thick = [r for r in cr if r["clicks"] >= floor] out.append({ "window": win, "term": term, "campaigns": len(cr), "clicks": sum(r["clicks"] for r in cr), "cost": round(sum(r["cost"] for r in cr), 2), "cpcLow": cr[0]["cpc"], "cpcHigh": cr[-1]["cpc"], "premium": round(sum(r["clicks"] * (r["cpc"] - cr[0]["cpc"]) for r in cr[1:]), 2), "premiumRobust": round( sum(r["clicks"] * (r["cpc"] - thick[0]["cpc"]) for r in thick[1:]), 2) if len(thick) >= 2 else 0.0, "rows": cr}) out.sort(key=lambda c: -c["premiumRobust"] or -c["premium"]) return out, total # ------------------------------------------------------------------ output def report(struct, cases, total, floor, profile, out_dir): t2_open = [x for x in struct["tier2"] if not x["blockedBy"]] print("\n" + "=" * 70) print("STRUCTURE — your account as it stands today") print("=" * 70) L = struct["live"] print(f" live campaigns {L['campaigns']} ad groups {L['adGroups']} " f"keywords {L['keywords']} enabled negatives {L['enabledNegatives']}") print(f"\n ASINs by how many live campaigns advertise them:") for asin, v in list(struct["asins"].items())[:10]: bar = "#" * min(50, v["campaigns"]) print(f" {asin:14} {v['campaigns']:>3} campaigns {bar}") print(f"\n tier 1 same keyword + match type, same ASIN : {len(struct['tier1'])}") print(f" tier 2 exact also matched by phrase/broad : {len(struct['tier2'])}" f" ({len(t2_open)} not already negated)") print(f" tier 3 auto campaign over your manual terms : {len(struct['tier3'])}") print(f" tier 4 same keyword, different ASIN : {len(struct['tier4'])}") for x in t2_open[:12]: print(f" ! {x['term']!r}") print(f" exact {x['exactBid']} {x['exactCampaign']}") print(f" {x['wideMatchType'].lower():6} {x['wideBid']} {x['wideCampaign']}" f" via {x['wideKeyword']!r}") if cases is None: print("\n(structure only — rerun without --structure-only to price it)") return print("\n" + "=" * 70) print("SPEND — what actually happened") print("=" * 70) cc = sum(c["cost"] for c in cases) share = (100 * cc / total["cost"]) if total["cost"] else 0 print(f" {total['clicks']:,} clicks, {total['cost']:,.2f} spend") print(f" contested term-windows {len(cases)}") print(f" distinct contested terms {len({c['term'] for c in cases})}") print(f" most campaigns on one term {max((c['campaigns'] for c in cases), default=0)}") print(f" contested spend {cc:,.2f} ({share:.1f}% of all spend)") print(f" premium, naive {sum(c['premium'] for c in cases):,.2f}") print(f" premium, baseline needs {floor}+ clicks " f"{sum(c['premiumRobust'] for c in cases):,.2f} <- trust this one") solid = [c for c in cases if c["premiumRobust"] > 0] if not solid: print(f"\n No term had two campaigns each clearing {floor} clicks in one window.") print(" The overlap is real but too thin to price. That is a finding, not a failure.") else: print(f"\n Worth your attention, most expensive first:") for c in solid[:10]: print(f"\n {c['term']!r} [{c['window']}] {c['campaigns']} campaigns") for r in c["rows"]: mark = " " if r["clicks"] >= floor else "." print(f" {mark} {r['cpc']:>7.2f} cpc {r['clicks']:>5} clicks " f"{r['cost']:>9.2f} {r['campaign']}") print(f" difference on the priced clicks: {c['premiumRobust']:.2f}") os.makedirs(out_dir, exist_ok=True) jf = os.path.join(out_dir, f"selfcomp-{profile}.json") with open(jf, "w") as f: json.dump({"structure": struct, "contested": cases, "totals": total}, f, indent=2) cf = os.path.join(out_dir, f"selfcomp-{profile}-contested.csv") with open(cf, "w", newline="") as f: w = csv.writer(f) w.writerow(["window", "searchTerm", "campaign", "keyword", "clicks", "cost", "cpc", "campaignsOnTerm", "premiumRobust"]) for c in cases: for r in c["rows"]: w.writerow([c["window"], c["term"], r["campaign"], "; ".join(r["keywords"]), r["clicks"], r["cost"], r["cpc"], c["campaigns"], c["premiumRobust"]]) print(f"\n wrote {jf}") print(f" wrote {cf}") def list_profiles(): ct = "application/vnd.listaccountsresource.v1+json" for region in HOSTS: api = Api(region) st, d = api.call("POST", "/adsAccounts/list", body={}, headers={"Content-Type": ct, "Accept": ct}) if st != 200: print(f"{region}: HTTP {st} {json.dumps(d)[:160]}") continue for a in (d or {}).get("adsAccounts", []): print(f"\n{region} {a.get('accountName')!r} status={a.get('status')}") for alt in a.get("alternateIds", []): if alt.get("profileId"): print(f" profile {alt['profileId']} {alt.get('countryCode')}") break # /adsAccounts/list returns the whole global account from any host def main(): ap = argparse.ArgumentParser( description="Find where your own Sponsored Products campaigns compete with each other.") ap.add_argument("--profiles", action="store_true", help="list your profile IDs and exit") ap.add_argument("--profile", help="profile ID to scan") ap.add_argument("--region", default="NA", choices=list(HOSTS), help="NA (US/CA/MX/BR), EU (UK/DE/FR/IT/ES/NL/SE/PL/BE/IE/AE/SA), FE (JP/AU)") ap.add_argument("--days", type=int, default=90, help="days of search-term history (default 90)") ap.add_argument("--floor", type=int, default=20, help="minimum clicks for a campaign to be used as a price baseline (default 20)") ap.add_argument("--structure-only", action="store_true", help="skip the reports — fast, and uses no report quota") ap.add_argument("--out", default=".", help="directory for the JSON and CSV") a = ap.parse_args() if a.profiles: list_profiles() return if not a.profile: die("give --profile (run --profiles to find it)") api = Api(a.region, a.profile) print(f"scanning profile {a.profile} in {a.region} ...") struct = structure(api) cases = None tot = {"clicks": 0, "cost": 0.0} if not a.structure_only: print(f"\nrequesting {a.days} days of search-term reports " f"(a few minutes; Amazon generates them asynchronously)") rows = search_terms(api, a.days) cases, tot = contested(rows, a.floor) report(struct, cases, tot, a.floor, a.profile, a.out) if __name__ == "__main__": main()