Competitive Intelligence: Automating Competitor UI Monitoring

Competitor products change constantly: pricing adjusts, features launch, messaging shifts, CTAs move. Most teams find out via Twitter or a customer mentioning it in a call. By then the change is old news. A screenshot-based monitoring pipeline puts you on the same schedule as your competitors' deploy pipeline — you see changes within hours, not weeks.

This isn't about scraping competitor data. It's about monitoring visual changes to public-facing pages you're already looking at manually anyway. The automation just makes it systematic.

What to Monitor

Prioritize pages that change meaningfully and matter for your competitive position:

Start with 5-10 URLs maximum. Monitoring everything creates noise. Monitoring the pages that drive your own sales conversations creates signal.

Building the Monitoring Pipeline

import requests
import sqlite3
import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional

SCREENSHOT_API_KEY = "your-api-key"
SCREENSHOT_API_URL = "https://hermesforge.dev/api/screenshot"
DB_PATH = "./competitor_monitor.db"
SCREENSHOTS_DIR = "./competitor_screenshots"


def init_db():
    """Initialize SQLite database for tracking competitor page state."""
    conn = sqlite3.connect(DB_PATH)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS monitored_pages (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            url TEXT NOT NULL,
            competitor TEXT NOT NULL,
            label TEXT NOT NULL,
            active INTEGER DEFAULT 1,
            created_at TEXT DEFAULT (datetime('now'))
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS snapshots (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            page_id INTEGER NOT NULL,
            captured_at TEXT NOT NULL,
            file_path TEXT NOT NULL,
            content_hash TEXT NOT NULL,
            changed INTEGER DEFAULT 0,
            FOREIGN KEY (page_id) REFERENCES monitored_pages(id)
        )
    """)
    conn.commit()
    conn.close()


def add_page(url: str, competitor: str, label: str) -> int:
    """Register a competitor page for monitoring."""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.execute(
        "INSERT INTO monitored_pages (url, competitor, label) VALUES (?, ?, ?)",
        (url, competitor, label),
    )
    page_id = cursor.lastrowid
    conn.commit()
    conn.close()
    return page_id


def capture_page(url: str, page_id: int) -> Optional[dict]:
    """Capture a screenshot and return metadata."""
    response = requests.get(
        SCREENSHOT_API_URL,
        params={
            "url": url,
            "format": "png",
            "width": 1280,
            "full_page": "true",
            "wait": "networkidle",
            "block_ads": "true",  # Consistent captures without ad variation
        },
        headers={"X-API-Key": SCREENSHOT_API_KEY},
        timeout=45,
    )

    if response.status_code != 200:
        print(f"  Capture failed: HTTP {response.status_code}")
        return None

    # Hash the image bytes for change detection
    content_hash = hashlib.sha256(response.content).hexdigest()[:16]
    timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")

    # Save screenshot
    output_dir = Path(SCREENSHOTS_DIR) / str(page_id)
    output_dir.mkdir(parents=True, exist_ok=True)
    file_path = output_dir / f"{timestamp}_{content_hash}.png"
    file_path.write_bytes(response.content)

    return {
        "file_path": str(file_path),
        "content_hash": content_hash,
        "captured_at": datetime.now(timezone.utc).isoformat(),
    }


def check_for_change(page_id: int, new_hash: str) -> bool:
    """Return True if this hash differs from the most recent snapshot."""
    conn = sqlite3.connect(DB_PATH)
    row = conn.execute(
        "SELECT content_hash FROM snapshots WHERE page_id = ? ORDER BY id DESC LIMIT 1",
        (page_id,),
    ).fetchone()
    conn.close()
    if row is None:
        return False  # First capture — not a change
    return row[0] != new_hash


def record_snapshot(page_id: int, capture: dict, changed: bool):
    """Store snapshot metadata in the database."""
    conn = sqlite3.connect(DB_PATH)
    conn.execute(
        """INSERT INTO snapshots (page_id, captured_at, file_path, content_hash, changed)
           VALUES (?, ?, ?, ?, ?)""",
        (
            page_id,
            capture["captured_at"],
            capture["file_path"],
            capture["content_hash"],
            1 if changed else 0,
        ),
    )
    conn.commit()
    conn.close()

Running the Monitoring Loop

def run_monitoring_cycle() -> list[dict]:
    """
    Capture all active monitored pages and detect changes.
    Returns list of changed pages for alerting.
    """
    conn = sqlite3.connect(DB_PATH)
    pages = conn.execute(
        "SELECT id, url, competitor, label FROM monitored_pages WHERE active = 1"
    ).fetchall()
    conn.close()

    changes = []
    for page_id, url, competitor, label in pages:
        print(f"Checking {competitor} — {label}...")
        capture = capture_page(url, page_id)
        if capture is None:
            continue

        changed = check_for_change(page_id, capture["content_hash"])
        record_snapshot(page_id, capture, changed)

        if changed:
            print(f"  CHANGED: {competitor} — {label}")
            changes.append({
                "competitor": competitor,
                "label": label,
                "url": url,
                "screenshot": capture["file_path"],
                "detected_at": capture["captured_at"],
            })
        else:
            print(f"  No change")

    return changes


def get_recent_changes(days: int = 7) -> list[dict]:
    """Query recent changes from the database."""
    conn = sqlite3.connect(DB_PATH)
    rows = conn.execute(
        """
        SELECT p.competitor, p.label, p.url, s.captured_at, s.file_path
        FROM snapshots s
        JOIN monitored_pages p ON s.page_id = p.id
        WHERE s.changed = 1
          AND s.captured_at > datetime('now', ?)
        ORDER BY s.captured_at DESC
        """,
        (f"-{days} days",),
    ).fetchall()
    conn.close()

    return [
        {
            "competitor": r[0],
            "label": r[1],
            "url": r[2],
            "detected_at": r[3],
            "screenshot": r[4],
        }
        for r in rows
    ]

Visual Diff Generation

A hash change tells you something changed. A visual diff shows you what changed — which is the part that matters for competitive analysis.

from PIL import Image, ImageChops, ImageFilter
import numpy as np


def generate_visual_diff(before_path: str, after_path: str, diff_path: str) -> dict:
    """
    Generate a visual diff highlighting what changed between two screenshots.
    Returns change metadata.
    """
    before = Image.open(before_path).convert("RGB")
    after = Image.open(after_path).convert("RGB")

    # Align sizes
    if before.size != after.size:
        after = after.resize(before.size, Image.LANCZOS)

    # Compute pixel difference
    diff = ImageChops.difference(before, after)
    diff_array = np.array(diff)

    # Threshold: ignore sub-pixel rendering noise
    significant = diff_array > 15
    change_mask = significant.any(axis=2)
    change_pct = change_mask.mean() * 100

    # Create highlighted diff image
    # Red channel = changes, others dimmed
    highlighted = np.array(after).copy()
    highlighted[change_mask] = [255, 80, 80]  # Red highlight on changed pixels

    # Blend: 70% after, 30% highlight
    after_array = np.array(after)
    blended = (after_array * 0.7 + highlighted * 0.3).astype(np.uint8)
    Image.fromarray(blended).save(diff_path)

    # Find bounding box of changes (for summary)
    rows = np.any(change_mask, axis=1)
    cols = np.any(change_mask, axis=0)
    if rows.any():
        rmin, rmax = np.where(rows)[0][[0, -1]]
        cmin, cmax = np.where(cols)[0][[0, -1]]
        change_region = {"top": int(rmin), "bottom": int(rmax), "left": int(cmin), "right": int(cmax)}
    else:
        change_region = None

    return {
        "change_pct": round(float(change_pct), 2),
        "change_region": change_region,
        "diff_path": diff_path,
        "significant": change_pct > 2.0,
    }


def diff_most_recent_change(page_id: int) -> Optional[dict]:
    """
    Generate a diff between the last two snapshots for a page.
    Call this after detecting a change.
    """
    conn = sqlite3.connect(DB_PATH)
    rows = conn.execute(
        "SELECT file_path FROM snapshots WHERE page_id = ? ORDER BY id DESC LIMIT 2",
        (page_id,),
    ).fetchall()
    conn.close()

    if len(rows) < 2:
        return None

    after_path, before_path = rows[0][0], rows[1][0]
    diff_path = after_path.replace(".png", "_diff.png")

    return generate_visual_diff(before_path, after_path, diff_path)

Alert Delivery

When a change is detected, the diff screenshot is the artifact. Send it via email or Slack:

import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.image import MIMEImage


def send_change_alert(change: dict, diff_info: Optional[dict] = None):
    """
    Send email alert with screenshot diff attached.
    Configure SMTP_* environment variables.
    """
    import os

    msg = MIMEMultipart("related")
    msg["Subject"] = f"Competitor change: {change['competitor']} — {change['label']}"
    msg["From"] = os.environ["ALERT_FROM"]
    msg["To"] = os.environ["ALERT_TO"]

    change_summary = ""
    if diff_info:
        change_summary = f"\nChange magnitude: {diff_info['change_pct']}% of pixels affected"
        if diff_info.get("change_region"):
            r = diff_info["change_region"]
            change_summary += f"\nChange region: top={r['top']}px, left={r['left']}px"

    body = f"""Competitor page change detected.

Competitor: {change['competitor']}
Page: {change['label']}
URL: {change['url']}
Detected at: {change['detected_at']}{change_summary}

Screenshot diff attached.
"""
    msg.attach(MIMEText(body, "plain"))

    # Attach diff image if available
    diff_path = diff_info.get("diff_path") if diff_info else change.get("screenshot")
    if diff_path and Path(diff_path).exists():
        with open(diff_path, "rb") as f:
            img = MIMEImage(f.read(), name=Path(diff_path).name)
            msg.attach(img)

    with smtplib.SMTP_SSL(os.environ["SMTP_HOST"], 465) as smtp:
        smtp.login(os.environ["SMTP_USER"], os.environ["SMTP_PASS"])
        smtp.send_message(msg)

Complete Monitoring Script

def main():
    init_db()

    # Register pages (run once during setup)
    # Uncomment and customize for your competitors:
    # add_page("https://competitor.com/pricing", "Competitor A", "Pricing")
    # add_page("https://competitor.com/features", "Competitor A", "Features")
    # add_page("https://other-competitor.com/pricing", "Competitor B", "Pricing")

    print(f"Running monitoring cycle at {datetime.now(timezone.utc).isoformat()}")
    changes = run_monitoring_cycle()

    if changes:
        print(f"\n{len(changes)} change(s) detected:")
        for change in changes:
            print(f"  - {change['competitor']}: {change['label']}")

            # Generate visual diff
            page_id = sqlite3.connect(DB_PATH).execute(
                "SELECT id FROM monitored_pages WHERE url = ?", (change["url"],)
            ).fetchone()[0]
            diff_info = diff_most_recent_change(page_id)

            # Send alert
            send_change_alert(change, diff_info)
    else:
        print("No changes detected.")


if __name__ == "__main__":
    main()

Scheduling

Run as a daily cron job during business hours:

# Check competitors at 08:00 UTC every weekday
0 8 * * 1-5 cd /path/to/monitor && python3 competitor_monitor.py >> logs/monitor.log 2>&1

Rate limit planning: Each monitored page = 1 API call per run. Monitoring 10 competitor pages daily = 10 calls/day. At the Starter tier (200/day), you have headroom for 20 competitors with daily checks, or 10 competitors with morning + evening checks. Pro tier (1000/day) supports comprehensive monitoring across multiple competitors with multiple daily checks.

Reducing False Positives

Hash-based change detection is binary — it doesn't distinguish between a pricing change and an A/B test rotating a banner image. Reduce noise by:

Blocking dynamic elements via CSS injection:

STABLE_CSS = """
    [data-testid="banner"], .ad-container, .cookie-notice,
    [class*="notification"], [class*="announcement-bar"] {
        display: none !important;
    }
"""
# Add inject_css=STABLE_CSS to your API params

Setting a change threshold: Filter changes below 2% pixel difference — these are usually rendering artifacts, not real content changes. See significant in the diff output above.

Grouping by time window: If 8 of 10 monitored pages change on the same day, it's likely a site-wide redesign, not 8 separate feature announcements.

hermesforge.dev — screenshot API for monitoring, testing, and automation. Free: 10/day. Starter: $4/30 days (200/day). Pro: $9 (1000/day). Business: $29 (5000/day).