How to Monitor Competitor Website Changes Without the Noise

The whole job is a loop that runs once a day and a report that runs once a week. Each stage hands a smaller, cleaner object to the next one:

watchlist.yaml ──> fetch (Scraping API or proxy) ──> extract CSS regions
      ──> normalize text ──> compare with last snapshot (SQLite)
      ──> unchanged: touch checked_at, stop
      ──> changed:   classify ──> screenshot + HTML diff report ──> changes table
                                   └─ high severity (price, terms): alert now
Monday 07:30 ──> weekly digest email built from the changes table

Three design decisions carry most of the weight. First, you never compare full HTML. You compare named regions ("pricing cards", "hero headline", "terms body") pulled with CSS selectors, so markup churn outside those regions can't trigger anything. Second, every region passes through a normalizer before comparison, because even inside a pricing card the text can wobble (a countdown timer, a non-breaking space, "$1,299" today and "$1299.00" tomorrow). Third, detection and notification are split. Price and terms changes page someone the same day; headline tweaks and new FAQ entries wait for Monday's digest. An alert for every copy edit quickly becomes an alert nobody opens.

Everything below runs on Python 3.10 or newer with pip install requests beautifulsoup4 pyyaml playwright.

If you already run a large scraper and want the HTTP side of change detection (ETag, If-None-Match, sitemap lastmod), that layer is covered in our guide to incremental web scraping and change detection. Competitor pages rarely send useful validators for the parts you care about, so this pipeline works one level up, on extracted text.

Start with a short list. Most founders who try this watch the homepage and nothing else, which catches rebrands and misses the pricing change that cost them a deal. A more useful starting set per competitor is four to six URLs:

Write the watchlist as data, not code, so a marketer can add a URL without touching Python. Each region gets a CSS selector, a kind that drives severity, and an ordered flag (explained in the normalization section):

# watchlist.yaml
- competitor: rival-one
  url: https://www.sparkproxy.io/pricing      # replace with the competitor URL
  country: US                                 # optional: price as seen from the US
  regions:
    plans:      {css: ".pricing-card",        list: true,  kind: pricing, ordered: true}
    plan_grid:  {css: "table.compare tr",     list: true,  kind: pricing, ordered: true}
- competitor: rival-one
  url: https://www.sparkproxy.io/terms
  regions:
    terms_body: {css: "main article",         list: false, kind: legal}
- competitor: rival-one
  url: https://www.sparkproxy.io/
  regions:
    hero:       {css: "header h1, header h1 + p", list: true, kind: messaging, ordered: true}
    logos:      {css: ".customer-logos img",  list: true,  kind: messaging, ordered: false, attr: alt}

Pick selectors that describe meaning rather than layout. .pricing-card survives a redesign more often than div:nth-child(3) > div > span, and a data-* attribute or an id beats both when the site has one. Open DevTools on each page, right-click the region, and test your selector in the console with document.querySelectorAll(".pricing-card").length before it goes in the file.

You have two ways to fetch, and most teams end up using both for different pages.

The SparkProxy Scraping API (https://scrape.sparkproxy.io/api/v1, authenticated with an X-API-Key header) renders the page in headless Chromium, rotates the exit IP for you, and can return just the regions you asked for. You pay in credits per request, and new accounts get 1,000 free credits. The raw proxy gateway gives you rotating datacenter IPs to plug into your own Playwright or requests code. Plans are sold by concurrent threads with unlimited bandwidth, so a page that weighs 4 MB costs the same as one that weighs 40 KB. On 2026-10-30 the SparkProxy homepage listed the Starter plan at $75/month (USD, excluding VAT) for 100 threads, with a 24-hour free trial and no credit card required.

Inside the proxy option, the session mode matters:

Concurrency barely matters at this scale. Forty pages a day is nothing for a 100-thread plan; run four to eight workers and spread the start times so a competitor never sees a burst of 6 requests in the same second. The point of rotation here isn't raw throughput. It's that a pricing page checked every morning at 06:00 from one office IP is the easiest scraper on earth to spot, and the cheapest one to feed a decoy page to. For a deeper look at proxy types for competitive data, see our piece on competitor analysis proxies.

The Scraping API's extract_rules parameter runs your CSS selectors inside the rendered page and returns an extracted object instead of HTML. That means the noise never leaves the competitor's server: no scripts, no nonces, no ad slots, only the text of the regions you named. A plain selector string returns the element's text content, and {"selector": ..., "type": "list"} returns one string per matching element, which is exactly the shape you want for plan cards and feature lists.

import os
import requests

API = "https://scrape.sparkproxy.io/api/v1"
HEADERS = {"X-API-Key": os.environ["SPARK_API_KEY"], "Content-Type": "application/json"}


class RateLimited(Exception):
    def __init__(self, retry_after: int):
        super().__init__(f"retry after {retry_after}s")
        self.retry_after = retry_after


class ScrapeFailed(Exception):
    def __init__(self, body: dict):
        super().__init__(body.get("error", "scrape failed"))
        self.body = body


def build_rules(regions: dict) -> dict:
    rules = {}
    for name, r in regions.items():
        if r.get("attr"):
            continue  # extract_rules reads text, href or src only; parse these locally
        rules[name] = {"selector": r["css"], "type": "list"} if r.get("list") else r["css"]
    return rules


def fetch_sections(page: dict) -> dict:
    payload = {
        "url": page["url"],
        "render_js": True,          # extract_rules needs the browser: 5 credits
        "block_ads": True,          # ad networks never load, so never reach the text
        "extract_rules": build_rules(page["regions"]),
        "tag": f"compwatch/{page['competitor']}",
    }
    if page.get("country"):
        payload["country_code"] = page["country"]   # +5 credits
    r = requests.post(API, headers=HEADERS, json=payload, timeout=200)
    if r.status_code == 429:
        raise RateLimited(int(r.json().get("retry_after_seconds", 30)))
    if r.status_code == 530:
        raise ScrapeFailed(r.json())
    r.raise_for_status()
    return r.json()["extracted"]

Two contract details trip people up. extract_rules replaces the normal response, so don't add json_response=true to the same call; the envelope and the extraction are different response shapes. And the documented return types are text (the default), list, href and src. There's no way to ask for an image's alt attribute, which is why build_rules skips any region with an attr key; the logo wall in the watchlist goes through the local parsing path below, where you control exactly what you pull.

Plenty of competitor pages are server-rendered, and those don't need a browser at all. Batch mode fetches several URLs in one call with render_js=false, and the whole batch costs 1 credit. You then select regions locally with BeautifulSoup:

import requests
from bs4 import BeautifulSoup


def batch_fetch(urls: list[str]) -> dict:
    # Batch mode requires render_js=false. URLs containing commas can't be batched.
    r = requests.get(API, headers={"X-API-Key": HEADERS["X-API-Key"]},
                     params={"url": ",".join(urls), "render_js": "false"}, timeout=120)
    r.raise_for_status()
    return {item["url"]: item for item in r.json()["results"]}


def select_regions(html: str, regions: dict) -> dict:
    soup = BeautifulSoup(html, "html.parser")
    out = {}
    for name, r in regions.items():
        nodes = soup.select(r["css"])
        if r.get("attr"):
            values = [n.get(r["attr"], "") for n in nodes]
        else:
            values = [n.get_text(" ", strip=True) for n in nodes]
        out[name] = values if r.get("list") else " ".join(values)
    return out

How do you know which tier a page belongs in? Run it once through batch_fetch and select_regions. If every region comes back non-empty and matches what you see in the browser, the page is server-rendered and can live in the 1-credit batch forever. If the pricing cards come back empty, the prices are drawn by JavaScript after load and the page needs extract_rules with rendering. We sort each new URL this way on day one; it's the single biggest cost lever in the whole pipeline.

Section extraction removes most of the noise. Normalization removes the rest, and it's where you encode your own definition of "the same". Every region goes through one function that returns a stable, line-oriented string:

import re
import unicodedata

INVISIBLE = dict.fromkeys(map(ord, "​‌‍⁠­"), None)
VOLATILE = [
    re.compile(r"\b(updated|posted|edited)\s+\d+\s+(seconds?|minutes?|hours?|days?)\s+ago\b", re.I),
    re.compile(r"\b(offer|sale|deal)\s+ends\s+in\b[^.]*", re.I),
    re.compile(r"\b\d{1,2}:\d{2}(:\d{2})?\s*(am|pm)?\b", re.I),
    re.compile(r"\b\d[\d,]*\s+(people|users|teams)\s+(viewing|joined)\s+(now|today)\b", re.I),
]
PRICE = re.compile(r"([$€£])\s?(\d{1,3}(?:,\d{3})+|\d+)(?:\.(\d{1,2}))?")


def _price(m: re.Match) -> str:
    whole = int(m.group(2).replace(",", ""))
    cents = (m.group(3) or "0").ljust(2, "0")
    return f"{m.group(1)}{whole}.{cents}"


def normalize_text(text: str) -> str:
    t = unicodedata.normalize("NFKC", text or "").translate(INVISIBLE)
    for rx in VOLATILE:
        t = rx.sub("", t)
    t = PRICE.sub(_price, t)
    t = re.sub(r"\s+", " ", t).strip()
    # One sentence per line, so a diff points at the sentence that moved.
    return "\n".join(s for s in re.split(r"(?<=[.!?])\s+", t) if s)


def normalize_region(value, ordered: bool = True) -> str:
    if isinstance(value, list):
        items = [normalize_text(v).replace("\n", " ") for v in value]
        items = [i for i in items if i]
        return "\n".join(items if ordered else sorted(items))
    return normalize_text(value)

A few of these rules earn their place the hard way:

• NFKC plus invisible characters. CMSs swap a normal space for a non-breaking one, or insert soft hyphens, after an unrelated edit. Visually identical, byte-different.
• Canonical prices. "$1,299", "$1299" and "$1,299.00" all become $1299.00, so a reformatting deploy doesn't look like a price change, and a real change shows up as a clean token swap the classifier can read.
• Sentence per line. extract_rules returns text content, which is one long line per element. Splitting at sentence ends turns a 900-word terms page into a diff that highlights the one changed clause rather than "line 1 changed".
• The ordered flag. Logo walls, testimonial carousels and "featured integrations" are often shuffled server-side on every request. If order carries no meaning, sorting the items before comparison kills a whole class of daily false alarms. Keep ordered: true for plan cards, where moving a plan from third to first position is itself news.

Test the normalizer the boring way: fetch the same page twice, ten minutes apart, and confirm every region produces an identical string. Any region that differs between two back-to-back fetches has noise you haven't handled yet, and it will alert daily until you do.

Store the normalized string and its SHA-256 per URL and region. On each run, compare hashes first; only when they differ do you compute a line diff and classify it.

import difflib
import hashlib
import json
import sqlite3
from datetime import datetime, timezone

PRICE_TOKEN = re.compile(r"[$€£]\d+\.\d{2}")

db = sqlite3.connect("compwatch.db")
db.executescript("""
CREATE TABLE IF NOT EXISTS latest (
  url TEXT, region TEXT, content TEXT, sha256 TEXT, checked_at TEXT,
  PRIMARY KEY (url, region));
CREATE TABLE IF NOT EXISTS changes (
  id INTEGER PRIMARY KEY, competitor TEXT, url TEXT, region TEXT,
  kind TEXT, severity TEXT, removed TEXT, added TEXT,
  before TEXT, after TEXT, screenshot TEXT, report TEXT,
  detected_at TEXT, alerted INTEGER DEFAULT 0);
""")


def line_diff(old: str, new: str) -> tuple[list, list]:
    a, b = old.splitlines(), new.splitlines()
    removed, added = [], []
    for op, i1, i2, j1, j2 in difflib.SequenceMatcher(None, a, b, autojunk=False).get_opcodes():
        if op in ("replace", "delete"):
            removed += a[i1:i2]
        if op in ("replace", "insert"):
            added += b[j1:j2]
    return removed, added


def classify(region_kind: str, removed: list, added: list) -> tuple[str, str]:
    old_prices = set(PRICE_TOKEN.findall(" ".join(removed)))
    new_prices = set(PRICE_TOKEN.findall(" ".join(added)))
    if region_kind == "pricing" and old_prices != new_prices:
        return "price", "high"
    if region_kind == "legal":
        return "terms", "high"
    if added and not removed:
        return "added", "medium" if region_kind == "pricing" else "low"
    if removed and not added:
        return "removed", "medium"
    return "copy", "medium" if region_kind == "messaging" else "low"


def compare_and_store(page: dict, name: str, content: str):
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    sha = hashlib.sha256(content.encode("utf-8")).hexdigest()
    row = db.execute("SELECT content, sha256 FROM latest WHERE url=? AND region=?",
                     (page["url"], name)).fetchone()
    db.execute("INSERT INTO latest VALUES (?,?,?,?,?) ON CONFLICT(url, region) DO UPDATE SET "
               "content=excluded.content, sha256=excluded.sha256, checked_at=excluded.checked_at",
               (page["url"], name, content, sha, now))
    if row is None or row[1] == sha:          # first sighting is a baseline, not a change
        db.commit()
        return None
    removed, added = line_diff(row[0], content)
    kind, severity = classify(page["regions"][name]["kind"], removed, added)
    cur = db.execute("INSERT INTO changes (competitor, url, region, kind, severity, removed, added, "
                     "before, after, detected_at) VALUES (?,?,?,?,?,?,?,?,?,?)",
                     (page["competitor"], page["url"], name, kind, severity,
                      json.dumps(removed), json.dumps(added), row[0], content, now))
    db.commit()
    return cur.lastrowid

The latest table holds one row per region, so it stays small no matter how long you run; history lives in changes, which only grows when something actually moved. If you'd rather keep a full time series of every snapshot, the schema and upsert trade-offs are in our guide to storing scraped data.

One rule worth copying exactly: the first time a region is seen, store it and report nothing. Otherwise adding a new competitor dumps 20 "changes" into the next digest and teaches everyone to skim it.

A diff line saying $49.00 became $59.00 is accurate, but the person reading Monday's digest will ask "show me". Capture visual evidence only when a change is detected, so you pay for screenshots a few times a month rather than daily.

The Scraping API returns a full-page JPEG (quality 80) when you set format=screenshot. It needs render_js=true and costs 10 credits in total: 5 for the render and 5 for the screenshot add-on.

from pathlib import Path


def page_screenshot(url: str, out_path: Path, country: str | None = None) -> Path:
    payload = {"url": url, "render_js": True, "format": "screenshot",
               "block_ads": True, "window_width": 1366, "window_height": 900}
    if country:
        payload["country_code"] = country
    r = requests.post(API, headers=HEADERS, json=payload, timeout=200)
    r.raise_for_status()
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_bytes(r.content)           # the body is the JPEG itself
    return out_path

A full page is heavy for a digest. If you want a crop of just the changed region, run Playwright yourself through a Sticky Sessions exit, so every asset of that page view loads from the same IP, and screenshot the element:

import os
from playwright.sync_api import sync_playwright

PROXY = {"server": "http://gateway.sparkproxy.io:11002",
         "username": os.environ["SPARK_USER"], "password": os.environ["SPARK_PASS"]}


def region_screenshot(url: str, css: str, out_path: str) -> str:
    with sync_playwright() as p:
        browser = p.chromium.launch(proxy=PROXY)
        page = browser.new_page(viewport={"width": 1366, "height": 900})
        page.goto(url, wait_until="networkidle", timeout=60_000)
        page.locator(css).first.screenshot(path=out_path)
        browser.close()
    return out_path

The current Playwright for Python release on PyPI was 1.64.1 (uploaded 2026-10-22) when we checked on 2026-10-30. Then pair the image with a side-by-side text diff. Python's standard library already ships one in difflib.HtmlDiff:

import difflib
import html
from pathlib import Path


def write_report(change_id: int, title: str, before: str, after: str,
                 shot: str, out_dir: Path) -> Path:
    table = difflib.HtmlDiff(wrapcolumn=90).make_file(
        before.splitlines(), after.splitlines(), "previous", "current",
        context=True, numlines=2)
    header = (f"<h2>{html.escape(title)}</h2>"
              f'<p><img src="{html.escape(shot)}" style="max-width:100%;border:1px solid #ccc"></p>')
    report = table.replace("<body>", "<body>" + header, 1)
    out_dir.mkdir(parents=True, exist_ok=True)
    path = out_dir / f"change-{change_id}.html"
    path.write_text(report, encoding="utf-8")
    return path

With context=True the report shows only the changed lines plus two lines either side, so a single new refund clause in a long terms page reads as a short, highlighted excerpt instead of a wall of text. A small helper ties the two together and writes both paths back to the changes row, so the digest can link straight to them:

def attach_evidence(change_id: int, page: dict, region: str):
    out_dir = Path("evidence")
    shot = page_screenshot(page["url"], out_dir / f"change-{change_id}.jpg", page.get("country"))
    before, after = db.execute("SELECT before, after FROM changes WHERE id=?",
                               (change_id,)).fetchone()
    report = write_report(change_id, f"{page['competitor']} / {region}",
                          before, after, shot.name, out_dir)
    db.execute("UPDATE changes SET screenshot=?, report=? WHERE id=?",
               (str(shot), str(report), change_id))
    db.commit()

Two cron entries cover it. Pick an odd minute and add jitter inside the script, because 40 requests that land at exactly 06:00:00 every day form a signature of their own:

# m  h  dom mon dow  command
17   5  *   *   *    cd /opt/compwatch && .venv/bin/python run.py daily  >> logs/daily.log 2>&1
30   7  *   *   1    cd /opt/compwatch && .venv/bin/python run.py digest >> logs/digest.log 2>&1

The daily runner sleeps a random interval per page, fetches in a small thread pool, and fires an immediate alert only for high-severity rows:

import random
import time
from concurrent.futures import ThreadPoolExecutor

import yaml


def fetch_page(page: dict):
    time.sleep(random.uniform(0, 900))                 # spread starts over 15 minutes
    try:
        return page, fetch_sections(page), None
    except (RateLimited, ScrapeFailed, requests.RequestException) as exc:
        return page, None, exc


def run_daily():
    with open("watchlist.yaml", encoding="utf-8") as fh:
        pages = yaml.safe_load(fh)
    found = 0
    with ThreadPoolExecutor(max_workers=6) as pool:
        for page, sections, exc in pool.map(fetch_page, pages):
            if exc is not None:
                print(f"FAILED {page['url']}: {exc}")  # handled in the failures section
                continue
            present, missing = split_missing(sections)
            for name, value in present.items():
                content = normalize_region(value, page["regions"][name].get("ordered", True))
                change_id = compare_and_store(page, name, content)
                if change_id:
                    attach_evidence(change_id, page, name)
                    found += 1
    for cid, comp, region, kind in db.execute(
            "SELECT id, competitor, region, kind FROM changes "
            "WHERE severity='high' AND alerted=0").fetchall():
        send_email(f"[compwatch] {comp}: {kind} change", f"{region} changed, see evidence/change-{cid}.html")
        db.execute("UPDATE changes SET alerted=1 WHERE id=?", (cid,))
    db.commit()
    print(f"{len(pages)} pages checked, {found} changes")

Fetching happens in the pool, but every database write happens on the main thread, because a Python sqlite3 connection raises ProgrammingError when another thread touches it (the check_same_thread default). Pages are also fetched one per call here; regions sharing a URL ride in the same extract_rules request. Scheduling options beyond cron (systemd timers, GitHub Actions, queues) and retry policy are covered in how to schedule and automate web scrapers.

The weekly digest is a grouped query and an email. It orders by severity so price and terms changes sit at the top, and groups by competitor so each rival reads as a short story of their week:

import os
import smtplib
from collections import defaultdict
from datetime import datetime, timedelta, timezone
from email.message import EmailMessage

SEVERITY_ORDER = "CASE severity WHEN 'high' THEN 0 WHEN 'medium' THEN 1 ELSE 2 END"


def weekly_digest(db) -> str:
    since = (datetime.now(timezone.utc) - timedelta(days=7)).isoformat(timespec="seconds")
    rows = db.execute(
        f"SELECT competitor, url, region, kind, severity, removed, added, report "
        f"FROM changes WHERE detected_at >= ? ORDER BY competitor, {SEVERITY_ORDER}, detected_at",
        (since,)).fetchall()
    by_rival = defaultdict(list)
    for comp, url, region, kind, sev, removed, added, report in rows:
        gone = "; ".join(json.loads(removed))[:200]
        new = "; ".join(json.loads(added))[:200]
        by_rival[comp].append(f"[{sev.upper()}] {kind} in {region} ({url})\n"
                              f"    was: {gone or '-'}\n    now: {new or '-'}\n"
                              f"    report: {report or 'n/a'}")
    if not by_rival:
        return "No competitor changes this week."
    parts = [f"{comp} ({len(items)} change{'s' if len(items) != 1 else ''})\n" + "\n".join(items) for comp, items in by_rival.items()]
    return "\n\n".join(parts)


def send_email(subject: str, body: str):
    msg = EmailMessage()
    msg["Subject"] = subject
    msg["From"] = "compwatch@sparkproxy.io"
    msg["To"] = "growth@sparkproxy.io"
    msg.set_content(body)
    with smtplib.SMTP(os.environ["SMTP_HOST"], 587) as s:
        s.starttls()
        s.login(os.environ["SMTP_USER"], os.environ["SMTP_PASS"])
        s.send_message(msg)

run.py digest then needs one line: send_email(f"Competitor changes, week of {date.today()}", weekly_digest(db)). Keep the digest plain. A line that says "rival-one: price in plans, was $49.00, now $59.00" gets read; a dashboard link gets bookmarked and forgotten.

Every failure in this pipeline has to be told apart from a change, or your digest fills with fiction. Work through these in order, cheapest check first:

The empty-region case is the one that bites. If a redesign renames .pricing-card, a naive pipeline stores an empty string, reports "all plans removed", and then reports "all plans added" once you fix the selector. Guard it before comparison:

def split_missing(sections: dict) -> tuple[dict, list]:
    present, missing = {}, []
    for name, value in sections.items():
        if value is None or value == "" or value == [] or value == [""]:
            missing.append(name)            # never overwrite a baseline with nothing
        else:
            present[name] = value
    return present, missing


def probe_status(url: str) -> int:
    # transparent_status_code only works with render_js=false; headless calls
    # return 200 on success or 530 on failure, never the target's own 404.
    r = requests.get(API, headers={"X-API-Key": HEADERS["X-API-Key"]},
                     params={"url": url, "render_js": "false",
                             "transparent_status_code": "true"}, timeout=60)
    return r.status_code

Run probe_status when every region on a page comes back missing. A 404 or 410 means the competitor killed the page, which is worth knowing (a pricing page that disappears is often a move to "contact sales"). A 200 with missing regions means your selectors are stale, so the alert goes to whoever maintains the watchlist, not to the growth team. For retry timing on the transient errors, our guide to retry and backoff strategies has the jitter formulas.

Be clear about what proxies can't solve. Rotation stops a competitor from rate-limiting or fingerprinting one IP across your daily checks. It does nothing for a stale selector, a page behind a login you don't have, or a price that only appears after entering a postcode, which needs a js_scenario (+5 credits, render_js=true, up to 50 steps) to fill the form first.

Take a realistic setup: 8 competitors, 5 URLs each, 40 pages checked once a day for 30 days, so 1,200 checks a month. Credit figures below come from the SparkProxy Scraping API credit table as published on 2026-10-30; the arithmetic is ours.

Two things stand out. The 1,000 free credits cover about five days of the all-rendered setup (200 credits a day), which is enough to tune selectors and the normalizer before you spend anything. And the tier sorting from the sections chapter dominates the bill: if half your 40 pages turn out to be server-rendered, moving them into batch mode cuts the rendered spend from 6,000 to 3,000 credits a month while the batch half costs about 30.

If you'd rather not write code at all, the open-source changedetection.io project does visual selectors, CSS and XPath filters, screenshots and per-watch proxies. Read on 2026-10-30, its GitHub repository showed 35,393 stars, an Apache-2.0 licence, release 0.60.10 published 2026-10-26 and a commit on 2026-10-30, so it is actively maintained. Its README also advertises a hosted plan at $8.99/month (USD, the project's own figure). It's a reasonable choice for a handful of URLs; the pipeline above makes more sense once you want classification, severity routing and a digest shaped around your own competitors. If pricing pages are your main target, our post on proxies for SaaS pricing research covers sampling A/B-tested price pages, which a once-a-day check can misread as a change.