Monitor MAP violations on Google Shopping

Monitoring MAP violations on Google Shopping comes down to three jobs: see every seller offering each of your products, compare each seller’s advertised price with your minimum advertised price, and keep a dated record of what you saw. This guide builds a Python script that does all three for a list of products, writes violations to a CSV with the evidence attached, and prints a daily report grouped by seller.

It uses the CrawlerData Google Shopping Sellers API, which returns every seller Google Shopping lists for a product in one request, with no pagination, for 1 credit per 10 sellers returned (rounded up, minimum 1). Field names and behaviour below match responses we received from the live API on 12 October 2026.

What you are comparing

For each product, the sellers endpoint returns a list of sellers. Here is one seller from a real response, trimmed, to show the fields. It is not a MAP violation: we have no MAP for this product.

{
  "position": 4,
  "merchant": "Abt Electronics & Appliances",
  "merchant_rating": 4.7,
  "offer_title": "Sony Silver Over-Ear WH-1000XM5 Wireless Industry Leading Noise Canceling Headphones WH1000XM5/S at ABT",
  "price": { "value": 198, "currency": "USD", "raw": "198.00 USD" },
  "shipping": { "value": 0, "currency": "USD", "raw": "0.00 USD" },
  "total_price": { "value": 198, "currency": "USD", "raw": "198.00 USD" },
  "delivery": "Free delivery",
  "availability": "In Stock",
  "returns": "Free 30-day returns",
  "link": null
}

The fields that matter for MAP:

• price is the item price the seller advertises. This is normally what you compare with MAP.
• shipping and total_price are kept separately. Most MAP policies treat shipping on its own terms, so record it but do not fold it into the comparison unless your policy says so. shipping can be null when Google shows no shipping figure.
• merchant is the seller’s name as Google Shopping shows it. Marketplace sellers include the marketplace, such as eBay - firesale-deals.
• offer_title is how the seller names the product. It is where you spot offers that may fall outside your policy, such as used or open-box items.

The response is sorted by total_price, and sellers_count tells you how many sellers there are.

Step 1: list products and MAP prices

Put each product’s Google Shopping product_id, the country, your MAP for that country and a label in map_prices.csv:

product_id,country,map_price,label
YOUR_PRODUCT_ID,us,249.99,Your product name

You get each product_id once, from a google-shopping-search request for the product’s name or model number. Two things make this step worth doing carefully:

• One model can have several product IDs. Different listings of the same model appear under different IDs, each with its own sellers. A seller below MAP on a listing you are not checking will not show up. Search by model number and brand and add every listing that is really your product.
• Set MAP in local currency. Prices come back in the currency of the country you request, such as USD for us or EUR for de. Put a MAP per country in the file rather than converting one figure.

Step 2: fetch every seller, with retries

import csv
import hashlib
import os
import pathlib
import time

import requests

API_URL = "https://api.crawlerdata.com/v1/jobs"
HEADERS = {"Authorization": f"Bearer {os.environ['CRAWLERDATA_API_KEY']}"}
EVIDENCE_DIR = pathlib.Path("evidence")
FIELDS = ["checked_at", "job_id", "product_id", "country", "label", "map_price", "merchant",
          "offer_title", "price", "shipping", "total_price", "currency", "availability",
          "below_map_by", "evidence_file", "evidence_sha256"]


def check_product(product_id, country, attempts=3):
    """Return the parsed response (job + result) and the exact bytes received."""
    body = {"endpoint": "google-shopping-product",
            "params": {"product_id": product_id, "country": country}, "queue": "live"}
    for attempt in range(attempts):
        try:
            resp = requests.post(API_URL, json=body, headers=HEADERS, timeout=200)
        except requests.RequestException:
            resp = None
        if resp is not None and resp.status_code in (400, 401, 402):
            raise RuntimeError(f"HTTP {resp.status_code}: {resp.json()['error']['message']}")
        if resp is not None and resp.ok and resp.json()["job"]["status"] == "completed":
            return resp.json(), resp.content
        time.sleep(10 * (attempt + 1))  # failed jobs are not charged, so retrying is free
    raise RuntimeError(f"{product_id} ({country}): no result after {attempts} attempts")

A sellers request loads the full Google Shopping product page and collects every seller on it, so the client timeout is generous (200 seconds). If a job fails, it comes back with "credits": 0, so the loop retries at no cost. A 400, 401 or 402 (bad parameters, bad key, no credits) will not get better on retry, so it stops.

Step 3: save the evidence before you judge it

If you may ever need to show a seller what you saw, the record has to be complete and unchanged. The script saves the exact bytes the API returned, one file per check, and computes a SHA-256 hash of them:

def save_evidence(response, raw):
    """Store the response bytes unchanged and return the file path and SHA-256."""
    job = response["job"]
    day = job["completedAt"][:10]
    folder = EVIDENCE_DIR / day
    folder.mkdir(parents=True, exist_ok=True)
    path = folder / f'{job["params"]["product_id"]}_{job["params"]["country"]}_{job["id"]}.json'
    path.write_bytes(raw)
    return str(path), hashlib.sha256(raw).hexdigest()

Each saved file holds the job record as well as the result:

{
  "job": {
    "id": "6acc6befe93571bd329d8652",
    "endpoint": "google-shopping-product",
    "queue": "live",
    "status": "completed",
    "params": { "product_id": "11726664398501807381", "country": "us", "language": "en" },
    "credits": 3,
    "latencyMs": 31094,
    "error": null,
    "createdAt": "2026-10-12T05:11:11.926Z",
    "completedAt": "2026-10-12T05:11:43.090Z"
  },
  "result": { "search_parameters": { "...": "..." }, "product": { "...": "..." }, "sellers_count": 22, "sellers": [ "..." ] }
}

That gives you, for each observation: when the check ran (completedAt, in UTC), which product and country it was for, the job ID, and every seller with price, shipping and total. Together, merchant, offer_title and the timestamp identify each offer. The hash in the violations CSV lets you show later that the file has not been edited.

Step 4: flag sellers below MAP

def find_violations(row, response, evidence_file, digest):
    job, result = response["job"], response["result"]
    map_price = float(row["map_price"])
    found = []
    for s in result["sellers"]:
        if not s["price"] or s["price"]["value"] >= map_price:
            continue
        found.append({
            "checked_at": job["completedAt"],
            "job_id": job["id"],
            "product_id": row["product_id"],
            "country": row["country"],
            "label": row["label"],
            "map_price": map_price,
            "merchant": s["merchant"],
            "offer_title": s["offer_title"],
            "price": s["price"]["value"],
            "shipping": s["shipping"]["value"] if s["shipping"] else None,
            "total_price": s["total_price"]["value"] if s["total_price"] else None,
            "currency": s["price"]["currency"],
            "availability": s["availability"],
            "below_map_by": round(map_price - s["price"]["value"], 2),
            "evidence_file": evidence_file,
            "evidence_sha256": digest,
        })
    return found


def main():
    with open("map_prices.csv", newline="", encoding="utf-8") as f:
        products = list(csv.DictReader(f))

    new_file = not os.path.exists("violations.csv")
    with open("violations.csv", "a", newline="", encoding="utf-8") as out:
        writer = csv.DictWriter(out, fieldnames=FIELDS)
        if new_file:
            writer.writeheader()
        for row in products:
            try:
                response, raw = check_product(row["product_id"], row["country"])
            except RuntimeError as exc:
                print(f'SKIPPED {row["label"]}: {exc}')
                continue
            if response["result"].get("error"):
                print(f'SKIPPED {row["label"]}: {response["result"]["error"]}')
                continue
            evidence_file, digest = save_evidence(response, raw)
            violations = find_violations(row, response, evidence_file, digest)
            writer.writerows(violations)
            print(f'{row["label"]} ({row["country"]}): {response["result"]["sellers_count"]} sellers, '
                  f'{len(violations)} below MAP')


if __name__ == "__main__":
    main()

The result.get("error") check handles product IDs Google does not recognise. Those jobs complete with "error": "Product not found" and an empty seller list, and are charged the 1-credit minimum like any completed request, so fix or remove the ID.

We ran the script against a real sellers response saved from the live API, with a made-up MAP, to check the logic end to end. It wrote one CSV row per offer below the MAP, each pointing at the same evidence file and hash.

Step 5: a daily report by seller

Enforcement usually happens per seller, not per product, so the report groups the latest day’s violations by merchant:

import csv
from collections import defaultdict

with open("violations.csv", newline="", encoding="utf-8") as f:
    rows = list(csv.DictReader(f))

latest_day = max(r["checked_at"][:10] for r in rows)
today = [r for r in rows if r["checked_at"].startswith(latest_day)]

by_merchant = defaultdict(list)
for r in today:
    by_merchant[r["merchant"]].append(r)

print(f"MAP report for {latest_day}: {len(today)} offers below MAP from {len(by_merchant)} sellers\n")
for merchant, offers in sorted(by_merchant.items(), key=lambda kv: -len(kv[1])):
    print(f"{merchant}: {len(offers)} product(s)")
    for o in sorted(offers, key=lambda o: -float(o["below_map_by"])):
        print(f'  {o["label"]} ({o["country"]}): {o["price"]} {o["currency"]} vs MAP {o["map_price"]}, '
              f'checked {o["checked_at"]}, job {o["job_id"]}')

Each line carries the timestamp and job ID, so anyone reading the report can open the matching evidence file.

Before you act on a flag

A row in violations.csv means “Google Shopping showed this price at this time”. Before treating it as a breach of your policy, check:

• Is it the same product and condition? Read offer_title. Used, refurbished, open-box or bundled offers may be listed under your product, and your policy may treat them differently.
• Is it a one-off? A price can change between checks. Many teams only act when the same seller is below MAP on two consecutive checks.
• Is it a promotion you approved? When many sellers drop to the same price at once, check for an authorised promotion before contacting anyone.
• Is it your MAP for that country and currency?

Running it on a schedule and what it costs

Run the script once or twice a day from cron or any scheduler. For large catalogs, submit jobs on the fast or regular queue and poll GET /v1/jobs/{id}, or pass a postbackUrl for a webhook, instead of holding a live connection open per product. All queues cost the same.

Each check returns every seller in one request and costs 1 credit per 10 sellers returned, rounded up, with a minimum of 1: the product above, with 22 sellers, costs 3 credits per check. Failed jobs cost nothing. Assuming about 21 to 30 sellers per product (3 credits per check), 500 products checked twice a day is 500 × 2 × 30 × 3 = 90,000 credits a month: $90 at the Starter pack rate of $1.00 per 1,000 (in practice 2 Starter packs, 100,000 credits, for $100), and less on larger packs. Checking a second country doubles it. Pack prices are on the pricing page.

For an overview of the use case and a cost example, see MAP monitoring. If you also watch competitors rather than resellers, competitor price monitoring uses the same endpoint. For how seller-level pricing compares between APIs, read Google Shopping API pricing compared.