Web Scraping for Price Monitoring: Build an AI-Powered Price Tracker in 2026

March 10, 2026 ยท 14 min read Price Monitoring E-Commerce AI Agents

Your competitor just dropped their price by 15%. You find out three days later when a customer mentions it on a sales call. By then, you've already lost deals you didn't even know were at risk.

Price monitoring used to mean hiring a VA to check competitor websites manually, or paying $500+/month for a SaaS tool that breaks every time a competitor redesigns their pricing page. In 2026, there's a better way: AI-powered price monitoring agents that understand pricing pages the way a human does โ€” and never miss a change.

In this guide, you'll build a complete price monitoring system using Python and the Mantis WebPerception API that scrapes competitor pricing pages, extracts structured price data with AI, detects changes, and sends you alerts โ€” all running on autopilot.

Why Traditional Price Monitoring Breaks

If you've tried monitoring competitor prices before, you know the pain:

AI extraction solves all of these. Instead of writing brittle CSS selectors for each competitor, you tell the AI what you want โ€” plan names, prices, features, limits โ€” and it figures out where on the page that data lives.

Architecture Overview

Here's what we're building:

  1. Scrape โ€” Fetch competitor pricing pages (handles JavaScript rendering)
  2. Extract โ€” Use AI to pull structured pricing data from any layout
  3. Store โ€” Save pricing snapshots to a database with timestamps
  4. Compare โ€” Detect price changes, new plans, removed features
  5. Alert โ€” Send notifications via Slack, email, or webhook when something changes
  6. Analyze โ€” Use an LLM to interpret what the changes mean strategically

Step 1: Define Your Pricing Schema

First, define the structure you want to extract from every pricing page. This schema works across most SaaS competitors:

from pydantic import BaseModel
from typing import Optional

class PricingTier(BaseModel):
    plan_name: str
    monthly_price: Optional[float]
    annual_price: Optional[float]
    currency: str = "USD"
    features: list[str]
    limits: dict[str, str]  # e.g., {"API calls": "10,000/mo", "users": "5"}
    is_enterprise: bool = False
    cta_text: str  # "Start Free", "Contact Sales", etc.

class CompetitorPricing(BaseModel):
    company_name: str
    url: str
    scraped_at: str
    tiers: list[PricingTier]
    has_free_tier: bool
    has_enterprise_tier: bool
    billing_options: list[str]  # ["monthly", "annual"]

Step 2: Scrape and Extract Pricing Data

The Mantis WebPerception API handles JavaScript rendering and AI extraction in a single call. No headless browser setup, no Puppeteer configs:

import requests
import json
from datetime import datetime

MANTIS_API_KEY = "your-api-key"

def extract_pricing(url: str) -> dict:
    """Scrape a pricing page and extract structured data with AI."""
    response = requests.post(
        "https://api.mantisapi.com/v1/extract",
        headers={"Authorization": f"Bearer {MANTIS_API_KEY}"},
        json={
            "url": url,
            "prompt": """Extract all pricing tiers from this page. For each tier, get:
            - Plan name
            - Monthly price (null if not shown)
            - Annual price (null if not shown)
            - Currency
            - List of features included
            - Usage limits (API calls, users, storage, etc.)
            - Whether it's an enterprise/custom tier
            - CTA button text
            Also note: does the page have a free tier? Enterprise tier? 
            What billing options are available (monthly, annual, etc.)?""",
            "schema": {
                "type": "object",
                "properties": {
                    "company_name": {"type": "string"},
                    "tiers": {
                        "type": "array",
                        "items": {
                            "type": "object",
                            "properties": {
                                "plan_name": {"type": "string"},
                                "monthly_price": {"type": "number", "nullable": True},
                                "annual_price": {"type": "number", "nullable": True},
                                "currency": {"type": "string"},
                                "features": {"type": "array", "items": {"type": "string"}},
                                "limits": {"type": "object"},
                                "is_enterprise": {"type": "boolean"},
                                "cta_text": {"type": "string"}
                            }
                        }
                    },
                    "has_free_tier": {"type": "boolean"},
                    "has_enterprise_tier": {"type": "boolean"},
                    "billing_options": {"type": "array", "items": {"type": "string"}}
                }
            }
        }
    )
    
    data = response.json()
    data["url"] = url
    data["scraped_at"] = datetime.utcnow().isoformat()
    return data

# Monitor multiple competitors
competitors = [
    "https://competitor-a.com/pricing",
    "https://competitor-b.com/pricing",
    "https://competitor-c.com/pricing",
]

results = []
for url in competitors:
    pricing = extract_pricing(url)
    results.append(pricing)
    print(f"โœ… Extracted {len(pricing.get('tiers', []))} tiers from {pricing.get('company_name', url)}")

Step 3: Store Pricing Snapshots

Store each scrape as a timestamped snapshot so you can track changes over time:

import sqlite3
from datetime import datetime

def init_db(db_path: str = "price_monitor.db"):
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS pricing_snapshots (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            company_name TEXT NOT NULL,
            url TEXT NOT NULL,
            scraped_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            data JSON NOT NULL,
            checksum TEXT NOT NULL
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS price_changes (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            company_name TEXT NOT NULL,
            detected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            change_type TEXT NOT NULL,
            description TEXT NOT NULL,
            old_value TEXT,
            new_value TEXT,
            severity TEXT DEFAULT 'medium'
        )
    """)
    conn.commit()
    return conn

def store_snapshot(conn, pricing_data: dict):
    """Store a pricing snapshot and return True if data changed."""
    import hashlib
    
    data_str = json.dumps(pricing_data, sort_keys=True)
    checksum = hashlib.sha256(data_str.encode()).hexdigest()
    
    # Check if this is different from the last snapshot
    cursor = conn.execute(
        "SELECT checksum FROM pricing_snapshots WHERE url = ? ORDER BY scraped_at DESC LIMIT 1",
        (pricing_data["url"],)
    )
    last = cursor.fetchone()
    
    conn.execute(
        "INSERT INTO pricing_snapshots (company_name, url, data, checksum) VALUES (?, ?, ?, ?)",
        (pricing_data.get("company_name", "Unknown"), pricing_data["url"], data_str, checksum)
    )
    conn.commit()
    
    if last and last[0] != checksum:
        return True  # Data changed!
    return False

Step 4: Detect and Classify Changes

This is where AI shines. Instead of writing brittle diff logic, use an LLM to understand what changed and why it matters:

from openai import OpenAI

openai_client = OpenAI()

def analyze_price_change(old_data: dict, new_data: dict) -> dict:
    """Use AI to analyze what changed and assess impact."""
    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "system",
            "content": """You are a competitive intelligence analyst. Compare two pricing 
            snapshots and identify all changes. Classify each change as:
            - PRICE_INCREASE / PRICE_DECREASE
            - NEW_PLAN / REMOVED_PLAN
            - FEATURE_ADDED / FEATURE_REMOVED
            - LIMIT_CHANGED
            - POSITIONING_SHIFT (e.g., renamed plans, changed CTAs)
            
            Rate severity: critical (price war / major competitive shift), 
            important (notable change worth watching), minor (cosmetic or small adjustment).
            
            Provide strategic analysis: what might this change signal about the competitor's strategy?"""
        }, {
            "role": "user",
            "content": f"OLD PRICING:\n{json.dumps(old_data, indent=2)}\n\nNEW PRICING:\n{json.dumps(new_data, indent=2)}"
        }],
        response_format={"type": "json_object"}
    )
    
    return json.loads(response.choices[0].message.content)

def detect_changes(conn, pricing_data: dict) -> list:
    """Compare current pricing with last snapshot and detect changes."""
    cursor = conn.execute(
        """SELECT data FROM pricing_snapshots 
           WHERE url = ? ORDER BY scraped_at DESC LIMIT 1 OFFSET 1""",
        (pricing_data["url"],)
    )
    previous = cursor.fetchone()
    
    if not previous:
        return []  # First snapshot, nothing to compare
    
    old_data = json.loads(previous[0])
    changes = analyze_price_change(old_data, pricing_data)
    
    # Store detected changes
    for change in changes.get("changes", []):
        conn.execute(
            """INSERT INTO price_changes 
               (company_name, change_type, description, old_value, new_value, severity)
               VALUES (?, ?, ?, ?, ?, ?)""",
            (
                pricing_data.get("company_name"),
                change.get("type"),
                change.get("description"),
                change.get("old_value"),
                change.get("new_value"),
                change.get("severity", "medium")
            )
        )
    conn.commit()
    
    return changes.get("changes", [])

Step 5: Send Alerts

When a change is detected, send an alert with context:

def send_slack_alert(changes: list, company_name: str, webhook_url: str):
    """Send a formatted Slack alert for price changes."""
    severity_emoji = {
        "critical": "๐Ÿšจ",
        "important": "โš ๏ธ",
        "minor": "โ„น๏ธ"
    }
    
    blocks = [{
        "type": "header",
        "text": {"type": "plain_text", "text": f"๐Ÿ’ฐ Price Change Detected: {company_name}"}
    }]
    
    for change in changes:
        emoji = severity_emoji.get(change.get("severity"), "โ„น๏ธ")
        blocks.append({
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"{emoji} *{change['type']}* ({change['severity']})\n{change['description']}"
            }
        })
    
    # Add strategic analysis if present
    if changes and "strategic_analysis" in changes[0]:
        blocks.append({
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": f"๐Ÿ“Š *Strategic Analysis:*\n{changes[0]['strategic_analysis']}"
            }
        })
    
    requests.post(webhook_url, json={"blocks": blocks})

def send_email_alert(changes: list, company_name: str, recipients: list):
    """Send email alerts for critical and important changes."""
    critical_changes = [c for c in changes if c.get("severity") in ("critical", "important")]
    if not critical_changes:
        return
    
    # Use your preferred email service (Resend, SendGrid, SES, etc.)
    subject = f"๐Ÿšจ {company_name} pricing changed โ€” {len(critical_changes)} notable changes"
    body = "\n\n".join([
        f"[{c['severity'].upper()}] {c['type']}: {c['description']}"
        for c in critical_changes
    ])
    
    print(f"Would send email to {recipients}: {subject}")
    # Add your email sending logic here

Step 6: Run on a Schedule

Put it all together in a monitoring loop that runs daily (or hourly for fast-moving markets):

import schedule
import time

SLACK_WEBHOOK = "https://hooks.slack.com/services/YOUR/WEBHOOK/URL"

competitors = {
    "ScrapingBee": "https://www.scrapingbee.com/pricing/",
    "Apify": "https://apify.com/pricing",
    "Browserless": "https://www.browserless.io/pricing",
    "Firecrawl": "https://www.firecrawl.dev/pricing",
}

def run_price_check():
    """Run a full price check across all competitors."""
    conn = init_db()
    
    for name, url in competitors.items():
        try:
            print(f"๐Ÿ” Checking {name}...")
            pricing = extract_pricing(url)
            changed = store_snapshot(conn, pricing)
            
            if changed:
                changes = detect_changes(conn, pricing)
                if changes:
                    print(f"๐Ÿšจ {len(changes)} changes detected for {name}!")
                    send_slack_alert(changes, name, SLACK_WEBHOOK)
                else:
                    print(f"๐Ÿ“Š Data changed but no significant pricing changes for {name}")
            else:
                print(f"โœ… No changes for {name}")
                
        except Exception as e:
            print(f"โŒ Error checking {name}: {e}")
    
    conn.close()

# Run daily at 9 AM
schedule.every().day.at("09:00").do(run_price_check)

# Or run every 6 hours for more frequent monitoring
# schedule.every(6).hours.do(run_price_check)

print("๐Ÿƒ Price monitor started. Checking competitors on schedule...")
while True:
    schedule.run_pending()
    time.sleep(60)

Advanced: Historical Price Analysis

Once you've been collecting data for a while, you can analyze trends:

def get_price_history(conn, company_name: str, plan_name: str = None) -> list:
    """Get historical pricing data for a competitor."""
    cursor = conn.execute(
        """SELECT scraped_at, data FROM pricing_snapshots 
           WHERE company_name = ? ORDER BY scraped_at ASC""",
        (company_name,)
    )
    
    history = []
    for row in cursor.fetchall():
        data = json.loads(row[1])
        for tier in data.get("tiers", []):
            if plan_name and tier.get("plan_name") != plan_name:
                continue
            history.append({
                "date": row[0],
                "plan": tier.get("plan_name"),
                "monthly_price": tier.get("monthly_price"),
                "annual_price": tier.get("annual_price"),
            })
    
    return history

def generate_competitive_report(conn) -> str:
    """Generate a weekly competitive pricing report using AI."""
    # Get all changes from the last 7 days
    cursor = conn.execute(
        """SELECT company_name, change_type, description, severity, detected_at 
           FROM price_changes WHERE detected_at > datetime('now', '-7 days')
           ORDER BY detected_at DESC"""
    )
    recent_changes = cursor.fetchall()
    
    # Get current pricing for all competitors
    companies = conn.execute(
        """SELECT DISTINCT company_name, data FROM pricing_snapshots 
           WHERE scraped_at = (SELECT MAX(scraped_at) FROM pricing_snapshots ps2 
           WHERE ps2.company_name = pricing_snapshots.company_name)"""
    ).fetchall()
    
    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "system",
            "content": """Generate a concise weekly competitive pricing report. Include:
            1. Summary of all changes this week
            2. Current competitive landscape (who's cheapest, best value, etc.)
            3. Trends to watch
            4. Recommended pricing actions for our team"""
        }, {
            "role": "user",
            "content": f"Changes this week:\n{json.dumps(recent_changes)}\n\nCurrent pricing:\n{json.dumps([{'company': c[0], 'data': json.loads(c[1])} for c in companies])}"
        }]
    )
    
    return response.choices[0].message.content

Deployment Options

MethodBest ForCostComplexity
Cron job (Linux server)Simple daily checks~$5/mo (VPS)Low
AWS Lambda + EventBridgeServerless, cost-optimized~$2/moMedium
GitHub Actions (scheduled)Free tier, git-based workflowFreeLow
Docker + Docker ComposeSelf-hosted, full controlVariesMedium

Cost Optimization

A typical price monitoring setup costs surprisingly little:

ComponentTraditional ApproachAI Agent Approach
Price monitoring SaaS$200-500/mo$0
Mantis API (extraction)โ€”$29/mo (5K calls)
OpenAI API (analysis)โ€”~$5/mo
HostingIncluded in SaaS$0-5/mo
Total$200-500/mo~$34/mo

Monitoring 20 competitors daily = ~600 API calls/month. Well within the Starter plan. And unlike SaaS tools, your AI agent adapts to any pricing page layout without configuration.

Real-World Use Cases

E-Commerce Price Matching

Retailers track competitor prices across thousands of SKUs and automatically adjust their own prices to stay competitive. An AI agent can extract prices from any product page โ€” even ones with complex variant selectors and dynamic pricing.

SaaS Competitive Intelligence

Track when competitors add features, change pricing tiers, or adjust their positioning. The AI analysis layer catches subtle signals like CTA text changes ("Start Free" โ†’ "Book a Demo") that indicate strategy shifts.

Travel and Hospitality

Monitor hotel rates, flight prices, and package deals across booking platforms. AI extraction handles the complex layouts of travel sites that break traditional scrapers.

Financial Markets

Track commodity prices, interest rates, and financial product terms from institutional websites that don't offer APIs. Historical tracking enables trend analysis.

Best Practices

Start Monitoring Competitor Prices Today

The Mantis WebPerception API handles JavaScript rendering, AI extraction, and screenshots in one call. No headless browser setup required.

Get Your Free API Key โ†’

What's Next

Price monitoring is just one piece of competitive intelligence. Combine it with:

Build the price monitoring agent first. When your competitor drops their price next Tuesday, you'll know about it in minutes โ€” not days.