DEV Community

Brad
Brad

Posted on

Python Competitor Price Monitoring: Get Alerts When Rivals Change Prices

Knowing when your competitors change prices can be the difference between winning and losing deals. Here's a Python system that monitors competitor prices 24/7 and alerts you immediately.

Why Price Monitoring Matters

  • Competitors adjusting prices signal demand shifts, supplier changes, or promotions
  • Manual checking is unreliable — you'll always be behind
  • Automated monitoring catches changes within hours, not days

The Price Monitor Architecture

import httpx
import sqlite3
import smtplib
import re
import time
from datetime import datetime
from email.mime.text import MIMEText

# Products to monitor (competitor URLs)
COMPETITORS = [
    {
        "name": "Widget Pro",
        "our_price": 49.99,
        "competitors": [
            {"name": "CompetitorA", "url": "https://competitor-a.com/widget-pro"},
            {"name": "CompetitorB", "url": "https://competitor-b.com/widgets/pro"},
        ]
    }
]

ALERT_EMAIL = "you@yourdomain.com"
SMTP_PASSWORD = "your-app-password"
Enter fullscreen mode Exit fullscreen mode

Extracting Prices from Pages

def extract_price(html):
    """Extract price from HTML using common patterns."""

    # Pattern 1: JSON-LD structured data (most reliable)
    matches = re.findall(r'"price":\s*"?(\d+\.?\d*)"?', html)
    if matches:
        return float(matches[0])

    # Pattern 2: Common price patterns
    price_patterns = [
        r'\$\s*(\d+\.?\d*)',           # $49.99
        r'USD\s*(\d+\.?\d*)',           # USD 49.99
        r'data-price="(\d+\.?\d*)"',   # data-price="49.99"
    ]

    for pattern in price_patterns:
        for m in re.findall(pattern, html):
            price = float(m)
            if 0.01 < price < 100000:
                return price

    return None

def fetch_price(url):
    """Fetch a page and extract the price."""
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
        'Accept': 'text/html,application/xhtml+xml',
    }

    try:
        resp = httpx.get(url, headers=headers, follow_redirects=True, timeout=15)
        if resp.status_code == 200:
            return extract_price(resp.text)
    except Exception as e:
        print(f"  Error fetching {url}: {e}")

    return None
Enter fullscreen mode Exit fullscreen mode

Storing Price History

def init_price_db():
    conn = sqlite3.connect('price_monitor.db')
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS price_history (
        id INTEGER PRIMARY KEY,
        product_name TEXT,
        competitor_name TEXT,
        url TEXT,
        price REAL,
        our_price REAL,
        recorded_at TEXT
    )''')
    conn.commit()
    return conn

def save_price(conn, product_name, competitor_name, url, price, our_price):
    c = conn.cursor()
    c.execute('''INSERT INTO price_history
                 (product_name, competitor_name, url, price, our_price, recorded_at)
                 VALUES (?, ?, ?, ?, ?, ?)''',
              (product_name, competitor_name, url, price, our_price,
               datetime.now().isoformat()))
    conn.commit()

def get_last_price(conn, product_name, competitor_name):
    c = conn.cursor()
    c.execute('''SELECT price FROM price_history
                 WHERE product_name = ? AND competitor_name = ?
                 ORDER BY recorded_at DESC LIMIT 1''',
              (product_name, competitor_name))
    row = c.fetchone()
    return row[0] if row else None
Enter fullscreen mode Exit fullscreen mode

Price Change Detection and Alerts

def send_price_alert(changes):
    """Email alert when a competitor changes their price."""
    if not changes:
        return

    body = "COMPETITOR PRICE CHANGES DETECTED\n\n"

    for change in changes:
        direction = "DOWN" if change['new_price'] < change['old_price'] else "UP"
        pct_change = ((change['new_price'] - change['old_price']) / change['old_price']) * 100

        body += f"Product: {change['product']}\n"
        body += f"Competitor: {change['competitor']}\n"
        body += f"Price: ${change['old_price']:.2f} -> ${change['new_price']:.2f} ({direction} {pct_change:+.1f}%)\n"
        body += f"Our Price: ${change['our_price']:.2f}\n"

        if change['new_price'] < change['our_price']:
            body += "WARNING: Their price is now lower than ours\n"
        else:
            body += "We are still cheaper\n"

        body += f"URL: {change['url']}\n\n"

    msg = MIMEText(body)
    msg['Subject'] = f"Price Alert: {len(changes)} competitor price changes"
    msg['From'] = ALERT_EMAIL
    msg['To'] = ALERT_EMAIL

    with smtplib.SMTP_SSL('smtp.gmail.com', 465) as server:
        server.login(ALERT_EMAIL, SMTP_PASSWORD)
        server.send_message(msg)

    print(f"Alert sent for {len(changes)} price changes")
Enter fullscreen mode Exit fullscreen mode

The Main Monitoring Loop

def run_price_monitor():
    conn = init_price_db()
    changes = []

    print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M')}] Checking competitor prices...")

    for product in COMPETITORS:
        product_name = product['name']
        our_price = product['our_price']

        for competitor in product['competitors']:
            comp_name = competitor['name']
            url = competitor['url']

            print(f"  Checking {comp_name} for {product_name}...")
            new_price = fetch_price(url)

            if new_price is None:
                print(f"    Could not extract price")
                continue

            old_price = get_last_price(conn, product_name, comp_name)
            save_price(conn, product_name, comp_name, url, new_price, our_price)

            if old_price and abs(new_price - old_price) > 0.01:
                print(f"    Price: ${new_price:.2f} (was ${old_price:.2f}) - CHANGED!")
                changes.append({
                    'product': product_name,
                    'competitor': comp_name,
                    'url': url,
                    'old_price': old_price,
                    'new_price': new_price,
                    'our_price': our_price
                })
            else:
                print(f"    Price: ${new_price:.2f} (unchanged)")

            time.sleep(2)  # Be polite to servers

    if changes:
        send_price_alert(changes)

    conn.close()
    print(f"Done. {len(changes)} changes detected.")

if __name__ == '__main__':
    run_price_monitor()
Enter fullscreen mode Exit fullscreen mode

Cron Schedule

# Check every 6 hours
0 */6 * * * /usr/bin/python3 /path/to/price_monitor.py >> /var/log/price_monitor.log 2>&1
Enter fullscreen mode Exit fullscreen mode

Extending the Monitor

  • More products: Extend the COMPETITORS list
  • Telegram alerts: Use Telegram Bot API for instant mobile notifications
  • Historical charts: Plot price trends with matplotlib
  • Amazon monitoring: Adapt to scrape comparison sites
  • Price rules: Auto-update your own prices based on competitor changes

This price monitor — plus 20 more business automation scripts — is available as a complete toolkit: https://lukassbrad.gumroad.com/l/ugeka


What products are you tracking? Share your use case in the comments.

Top comments (0)