Build a Price Monitoring System

Create a complete price monitoring system from scratch to track price changes on e-commerce sites like Amazon. This step-by-step guide walks you through building an automated system that scrapes, stores, and monitors price data.

Who Is this for?

This guide is designed for developers who want to build a price monitoring system from the ground up using web scraping. No existing price monitoring infrastructure required.

What you’ll learn

Extract product price data from an Amazon product page using ZenRows.
Clean the extracted raw data and structure it for storage and price monitoring.
Store price history with timestamps.
Schedule automated scraping at regular intervals.
Set up price change notifications.
Optimize performance and manage costs.

Step 1: Set Up Data Extraction

Extract specific product data related to pricing, availability, and social cues using ZenRows css_extractor. Stringify the css_extractor for compatibility:

Python

import json

# specify the CSS selectors for extracting data
css_extractor = json.dumps(
    {
        "name": "span#productTitle",
        "price": "span.aok-offscreen",
        "ratings": "#acrPopover",
        "reviewCount": "#acrCustomerReviewText",
        "demandHistory": "#social-proofing-faceout-title-tk_bought",
        "availability": "#availability",
    }
)

Define a scraper function to extract data from a product URL based on the css_extractor.

CSS selectors can change when websites update their code. To maintain a reliable scraper, monitor your selectors regularly and update them as needed. Learn more about CSS selectors here.

The js_render parameter handles the site’s dynamic rendering, while premium_proxy routes the request through premium proxies to avoid blocking. proxy_country is set to us to send requests from US-based IP addresses:

Python

pip3 install requests
#...
import requests
# define the scraper function
def scraper(url, css_extractor):
    apikey = "YOUR_ZENROWS_API_KEY"
    params = {
        "url": url,
        "apikey": apikey,
        "js_render": "true",
        "premium_proxy": "true",
        "proxy_country": "us",
        "css_extractor": css_extractor,
    }
    response = requests.get("https://api.zenrows.com/v1/", params=params)
    return response.json()

The proxy_country parameter is optional. If not specified, ZenRows will use a random IP address worldwide. See more about geolocation here.

Execute the scraper function and print the extracted raw data. Here’s the updated code:

Python

# pip3 install requests
import requests
import json

# specify the CSS selectors for extracting data
css_extractor = json.dumps(
    {
        "name": "span#productTitle",
        "price": "span.aok-offscreen",
        "ratings": "#acrPopover",
        "reviewCount": "#acrCustomerReviewText",
        "demandHistory": "#social-proofing-faceout-title-tk_bought",
        "availability": "#availability",
    }
)


# define the scraper function
def scraper(url, css_extractor):
    apikey = "YOUR_ZENROWS_API_KEY"
    params = {
        "url": url,
        "apikey": apikey,
        "js_render": "true",
        "premium_proxy": "true",
        "proxy_country": "us",
        "css_extractor": css_extractor,
    }
    response = requests.get("https://api.zenrows.com/v1/", params=params)
    return response.json()


product_url = "https://www.amazon.com/dp/B001CXYMFS"

raw_data = scraper(product_url, css_extractor)

print(raw_data)

The code returns the required data from the product page. Here’s what it looks like:

JSON response

{
    "availability": [
        'In Stock                 {"isInternal":false,"isRobot":false,"showFaceout":true,"merchantId":"ATVPDKIKX0DER","availableBadges":"","loggedIn":false,"asin":"B001CXYMFS","showBadge":false,"availableFaceouts":"TK_BOUGHT"}',
        "Only 1 left in stock - order soon.",
    ],
    "demandHistory": "500+ bought in past month",
    "name": "Thrustmaster T-Flight Hotas X (Compatible with PC)",
    "price": ["$69.99", "$65.79 with 6 percent savings", "New Condition Price: $69.99"],
    "ratings": ["4.4", "4.4"],
    "reviewCount": ["8,067 ratings", "8,067 ratings"],
}

Step 2: Clean and Structure the Data

The returned data is unsuitable for price monitoring in its current state since it contains undesired strings. Since some fields are returned as a list, define a consistency_handler to handle them as a list before cleaning:

Python

# treat specific fields as a list for consistency
def consistency_handler(field):
    if not field == "":
        if isinstance(field, str):
            return [field]
        elif isinstance(field, list):
            return field
    else:
        return []

Define a function to obtain the current price (discounted), the original price, and the discount percentage from the price field:

Python

# ...
import re

# parse price information from the price list
def parse_price_info(price_list):
    # apply the consistency_handler function to treat as a list
    price_list = consistency_handler(price_list)
    original_price = None
    current_price = None
    discount_percent = None
    currency = "USD"

    # obtain price and discount information
    for item in price_list:
        price_match = re.search(r"\$(\d+[\.,]?\d*)", item)
        percent_match = re.search(r"(\d+)\s*percent", item)
        if "with" in item and price_match and percent_match:
            current_price = float(price_match.group(1).replace(",", ""))
            discount_percent = int(percent_match.group(1))
        elif "New Condition Price" in item and price_match:
            original_price = float(price_match.group(1).replace(",", ""))
        elif price_match and not current_price:
            original_price = float(price_match.group(1).replace(",", ""))
    if current_price is None:
        current_price = original_price
    return original_price, current_price, discount_percent, currency

Clean the reviewCount field by extracting the count integer from the list of strings:

Python

# parse review count
def parse_review_count(review_list):
    # apply the consistency_handler function to treat as a list
    review_list = consistency_handler(review_list)
    if review_list:
        count_str = review_list[0].split()[0].replace(",", "")
        return int(count_str)
    return None

Obtain the average_rating as a float from the raw data:

Python

# parse average rating
def parse_rating(ratings_list):
    # apply the consistency_handler function to treat as a list
    ratings_list = consistency_handler(ratings_list)
    if ratings_list:
        try:
            return float(ratings_list[0])
        except Exception:
            return None
    return None

Similarly, parse the estimated number of units sold from the demandHistory field:

Python

# parse units sold from demand history
def parse_units_sold(demand_history):
    if not demand_history:
        return None
    # apply the consistency_handler function to treat as a list
    demand_history = consistency_handler(demand_history)
    match = re.search(r"(\d+[+]?)(?= bought)", demand_history[0])
    if match:
        return match.group(1)
    return None

Clean the availability field by extracting the In Stock string from the strings:

Python

# parse availability
def parse_availability(availability):
    # apply the consistency_handler function to treat as a list
    availability = consistency_handler(availability)
    for item in availability:
        if "In Stock" in item:
            return "In Stock"
    return None

Finally, parse the cleaned data as a separate dictionary. Use datetime and timestamp to get the timestamp of the scraping operation to track historical data:

Python

# ...
from datetime import datetime, timezone

# parse the cleaned data
def parse_cleaned_data(raw_data):
    original_price, current_price, discount_percent, currency = parse_price_info(
        raw_data.get("price", [])
    )
    average_rating = parse_rating(raw_data.get("ratings", []))
    review_count = parse_review_count(raw_data.get("reviewCount", []))
    units_sold = parse_units_sold(raw_data.get("demandHistory", ""))
    availability = parse_availability(raw_data.get("availability", []))
    # get the scraping timestamp
    scrape_timestamp = datetime.now(timezone.utc).isoformat()

    cleaned_data = {
        "name": raw_data.get("name", ""),
        "original_price": original_price,
        "current_price": current_price,
        "discount_percent": discount_percent,
        "currency": currency,
        "average_rating": average_rating,
        "review_count": review_count,
        "units_sold": units_sold,
        "availability": availability,
        "scrape_timestamp": scrape_timestamp,
    }
    return cleaned_data

Apply the parse_cleaned_data to the raw_data to get a cleaned version with a recorded timestamp:

Python

# ...
product_url = "https://www.amazon.com/dp/B001CXYMFS"
raw_data = scraper(product_url, css_extractor)
cleaned_data = parse_cleaned_data(raw_data)
print(cleaned_data)

Here’s a sample of the cleaned data:

JSON

{
    "name": "Thrustmaster T-Flight Hotas X (Compatible with PC)",
    "original_price": 69.99,
    "current_price": 65.79,
    "discount_percent": 6,
    "currency": "USD",
    "average_rating": 4.4,
    "review_count": 8067,
    "units_sold": "500+",
    "availability": "In Stock",
    "scrape_timestamp": "2025-08-08T06:11:35.662685+00:00",
}

Step 3: Store the Price History Data

Since the scraping will be scheduled, start the storage procedure by setting up logging to track the extraction process in real-time:

Python

# ...
import logging

# set up logging
logging.basicConfig(filename="scraper.log", level=logging.INFO)

Define a function to create a new price record for each scraping request and store it in a product_history.json file. Index the stored data by product name:

Python

#...
import os

# update the price history for each product in the JSON file
def update_and_save_price_history(new_product, filename="product_history.json"):
    # load existing history
    if os.path.exists(filename):
        with open(filename, "r", encoding="utf-8") as json_file:
            product_history = json.load(json_file)
    else:
        product_history = []

    # index products by name for easy lookup
    history_dict = {product["name"]: product for product in product_history}

    # create a new price record
    price_record = {
        "timestamp": new_product["scrape_timestamp"],
        "current_price": new_product["current_price"],
        "original_price": new_product["original_price"],
        "discount_percent": new_product["discount_percent"],
        "currency": new_product["currency"],
    }

Set up an alert trigger (alert_triggered=False) and check for price changes by comparing the previously scraped price to the current one. Set alert_triggered to True if a price change is detected. Then, log the price difference. Append new price records with existing ones in the JSON file. If a product doesn’t exist in the JSON file, create a new record for it:

Python

# ...
def update_and_save_price_history(new_product, filename="product_history.json"):
    # ...

    # create a flag to indicate if an alert was triggered
    alert_triggered = False

    # check for price changes
    if new_product["name"] in history_dict:
        price_history = history_dict[new_product["name"]]["price_history"]
        if price_history:
            last_price = price_history[-1]["current_price"]
            if new_product["current_price"] != last_price:
                alert_triggered = True
                logging.info(
                    f"ALERT: Price changed for {new_product['name']} from {last_price} to {new_product['current_price']}"
                )

        # append to existing price history
        price_history.append(price_record)
        # update ratings, review count, and units sold
        history_dict[new_product["name"]]["average_rating"] = new_product.get(
            "average_rating"
        )
        history_dict[new_product["name"]]["review_count"] = new_product.get(
            "review_count"
        )
        history_dict[new_product["name"]]["units_sold"] = new_product.get("units_sold")
        history_dict[new_product["name"]]["availability"] = new_product.get(
            "availability"
        )
    else:
        # create a new product entry with price history
        history_dict[new_product["name"]] = {
            "name": new_product["name"],
            "price_history": [price_record],
            "average_rating": new_product.get("average_rating"),
            "review_count": new_product.get("review_count"),
            "units_sold": new_product.get("units_sold"),
        }

Finally, update the existing records with the new pricing records and return the alert status:

Python

def update_and_save_price_history(new_product, filename="product_history.json"):
    # ...

    # write updated history back to the JSON file
    with open(filename, "w", encoding="utf-8") as f:
        json.dump(list(history_dict.values()), f, ensure_ascii=False, indent=2)

    return alert_triggered

Here’s the complete update_and_save_price_history function:

Python

# update the price history for each product in the JSON file
def update_and_save_price_history(new_product, filename="product_history.json"):
    # load existing history
    if os.path.exists(filename):
        with open(filename, "r", encoding="utf-8") as json_file:
            product_history = json.load(json_file)
    else:
        product_history = []

    # index products by name for easy lookup
    history_dict = {product["name"]: product for product in product_history}

    # create a new price record
    price_record = {
        "timestamp": new_product["scrape_timestamp"],
        "current_price": new_product["current_price"],
        "original_price": new_product["original_price"],
        "discount_percent": new_product["discount_percent"],
        "currency": new_product["currency"],
    }

    # create a flag to indicate if an alert was triggered
    alert_triggered = False

    # check for price changes
    if new_product["name"] in history_dict:
        price_history = history_dict[new_product["name"]]["price_history"]
        if price_history:
            last_price = price_history[-1]["current_price"]
            if new_product["current_price"] != last_price:
                alert_triggered = True
                logging.info(
                    f"ALERT: Price changed for {new_product['name']} from {last_price} to {new_product['current_price']}"
                )

        # append to existing price history
        price_history.append(price_record)
        # update ratings, review count, and units sold
        history_dict[new_product["name"]]["average_rating"] = new_product.get(
            "average_rating"
        )
        history_dict[new_product["name"]]["review_count"] = new_product.get(
            "review_count"
        )
        history_dict[new_product["name"]]["units_sold"] = new_product.get("units_sold")
        history_dict[new_product["name"]]["availability"] = new_product.get(
            "availability"
        )
    else:
        # create a new product entry with price history
        history_dict[new_product["name"]] = {
            "name": new_product["name"],
            "price_history": [price_record],
            "average_rating": new_product.get("average_rating"),
            "review_count": new_product.get("review_count"),
            "units_sold": new_product.get("units_sold"),
        }

    # write updated history back to the JSON file
    with open(filename, "w", encoding="utf-8") as f:
        json.dump(list(history_dict.values()), f, ensure_ascii=False, indent=2)

    return alert_triggered

Step 4: Create a Scraping Job

Create a job function to execute the extraction, data cleaning, and storage logics. This function logs the scraping timestamps and price change notifications:

Python

## define a scraping job
def job():
    logging.info(f"scraping job started at {datetime.now(timezone.utc).isoformat()}")

    product_url = "https://www.amazon.com/dp/B001CXYMFS"

    raw_data = scraper(product_url, css_extractor)
    result = parse_cleaned_data(raw_data)
    alert_triggered = update_and_save_price_history(result)

    logging.info(f"scraping job stopped at {datetime.now(timezone.utc).isoformat()}")
    if alert_triggered:
        logging.info("ALERT: Price change detected!")
    else:
        logging.info("No price change detected.")

Step 5: Schedule the Scraping Job

Schedule the extraction process using Python’s schedule module. The code below runs the scraping job every 30 seconds and checks for jobs every second:

Python

# ...
# pip3 install schedule 
import schedule
import time

# schedule the job to run every 2 minutes
schedule.every(2).minutes.do(job)

# run the scheduler
while True:
    schedule.run_pending()
    # check for pending jobs every second
    time.sleep(60)

Step 6: Put Everything Together

Here’s the complete code:

Python

# pip3 install requests, schedule
import requests
import json
import re
from datetime import datetime, timezone
import logging
import os
import schedule
import time

# specify the CSS selectors for extracting data
css_extractor = json.dumps(
    {
        "name": "span#productTitle",
        "price": "span.aok-offscreen",
        "ratings": "#acrPopover",
        "reviewCount": "#acrCustomerReviewText",
        "demandHistory": "#social-proofing-faceout-title-tk_bought",
        "availability": "#availability",
    }
)

# define the scraper function
def scraper(url, css_extractor):
    apikey = "YOUR_ZENROWS_API_KEY"
    params = {
        "url": url,
        "apikey": apikey,
        "js_render": "true",
        "premium_proxy": "true",
        "proxy_country": "us",
        "css_extractor": css_extractor,
    }
    response = requests.get("https://api.zenrows.com/v1/", params=params)
    return response.json()

# treat specific fields as a list for consistency
def consistency_handler(field):
    if not field == "":
        if isinstance(field, str):
            return [field]
        elif isinstance(field, list):
            return field
    else:
        return []

# parse price information from the price list
def parse_price_info(price_list):
    # apply the consistency_handler function to treat as a list
    price_list = consistency_handler(price_list)
    original_price = None
    current_price = None
    discount_percent = None
    currency = "USD"
    # obtain price and discount information
    for item in price_list:
        price_match = re.search(r"\$(\d+[\.,]?\d*)", item)
        percent_match = re.search(r"(\d+)\s*percent", item)
        if "with" in item and price_match and percent_match:
            current_price = float(price_match.group(1).replace(",", ""))
            discount_percent = int(percent_match.group(1))
        elif "New Condition Price" in item and price_match:
            original_price = float(price_match.group(1).replace(",", ""))
        elif price_match and not current_price:
            original_price = float(price_match.group(1).replace(",", ""))
    if current_price is None:
        current_price = original_price
    return original_price, current_price, discount_percent, currency

# parse review count
def parse_review_count(review_list):
    # apply the consistency_handler function to treat as a list
    review_list = consistency_handler(review_list)
    if review_list:
        # Remove any non-digit characters (parentheses, commas, etc.)
        import re
        count_str = re.sub(r"[^\d]", "", review_list[0])
        if count_str:
            return int(count_str)
    return None

# parse average rating
def parse_rating(ratings_list):
    # apply the consistency_handler function to treat as a list
    ratings_list = consistency_handler(ratings_list)
    if ratings_list:
        try:
            return float(ratings_list[0])
        except Exception:
            return None
    return None

# parse units sold from demand history
def parse_units_sold(demand_history):
    if not demand_history:
        return None
    # apply the consistency_handler function to treat as a list
    demand_history = consistency_handler(demand_history)
    match = re.search(r"(\d+[+]?)(?= bought)", demand_history[0])
    if match:
        return match.group(1)
    return None

# parse availability
def parse_availability(availability):
    # apply the consistency_handler function to treat as a list
    availability = consistency_handler(availability)
    for item in availability:
        if "In Stock" in item:
            return "In Stock"
    return None

# parse the cleaned data
def parse_cleaned_data(raw_data):
    original_price, current_price, discount_percent, currency = parse_price_info(
        raw_data.get("price", [])
    )
    average_rating = parse_rating(raw_data.get("ratings", []))
    review_count = parse_review_count(raw_data.get("reviewCount", []))
    units_sold = parse_units_sold(raw_data.get("demandHistory", ""))
    availability = parse_availability(raw_data.get("availability", []))
    # get the scraping timestamp
    scrape_timestamp = datetime.now(timezone.utc).isoformat()

    cleaned_data = {
        "name": raw_data.get("name", ""),
        "original_price": original_price,
        "current_price": current_price,
        "discount_percent": discount_percent,
        "currency": currency,
        "average_rating": average_rating,
        "review_count": review_count,
        "units_sold": units_sold,
        "availability": availability,
        "scrape_timestamp": scrape_timestamp,
    }
    return cleaned_data

# set up logging
logging.basicConfig(filename="scraper.log", level=logging.INFO)

# update the price history for each product in the JSON file
def update_and_save_price_history(new_product, filename="product_history.json"):
    # load existing history
    if os.path.exists(filename):
        with open(filename, "r", encoding="utf-8") as json_file:
            product_history = json.load(json_file)
    else:
        product_history = []

    # index products by name for easy lookup
    history_dict = {product["name"]: product for product in product_history}

    # create a new price record
    price_record = {
        "timestamp": new_product["scrape_timestamp"],
        "current_price": new_product["current_price"],
        "original_price": new_product["original_price"],
        "discount_percent": new_product["discount_percent"],
        "currency": new_product["currency"],
    }

    # create a flag to indicate if an alert was triggered
    alert_triggered = False

    # check for price changes
    if new_product["name"] in history_dict:
        price_history = history_dict[new_product["name"]]["price_history"]
        if price_history:
            last_price = price_history[-1]["current_price"]
            if new_product["current_price"] != last_price:
                alert_triggered = True
                logging.info(
                    f"ALERT: Price changed for {new_product['name']} from {last_price} to {new_product['current_price']}"
                )

        # append to existing price history
        price_history.append(price_record)
        # update ratings, review count, and units sold
        history_dict[new_product["name"]]["average_rating"] = new_product.get(
            "average_rating"
        )
        history_dict[new_product["name"]]["review_count"] = new_product.get(
            "review_count"
        )
        history_dict[new_product["name"]]["units_sold"] = new_product.get("units_sold")
        history_dict[new_product["name"]]["availability"] = new_product.get(
            "availability"
        )
    else:
        # create a new product entry with price history
        history_dict[new_product["name"]] = {
            "name": new_product["name"],
            "price_history": [price_record],
            "average_rating": new_product.get("average_rating"),
            "review_count": new_product.get("review_count"),
            "units_sold": new_product.get("units_sold"),
        }

    # write updated history back to the JSON file
    with open(filename, "w", encoding="utf-8") as f:
        json.dump(list(history_dict.values()), f, ensure_ascii=False, indent=2)

    return alert_triggered

# schedule the scraping job
def job():
    logging.info(f"scraping job started at {datetime.now(timezone.utc).isoformat()}")
    product_url = "https://www.amazon.com/dp/B001CXYMFS"

    raw_data = scraper(product_url, css_extractor)
    result = parse_cleaned_data(raw_data)
    alert_triggered = update_and_save_price_history(result)
    logging.info(f"scraping job stopped at {datetime.now(timezone.utc).isoformat()}")
    if alert_triggered:
        logging.info("ALERT: Price change detected!")
    else:
        logging.info("No price change detected.")

# schedule the job to run every 2 minutes
schedule.every(2).minutes.do(job)

# run the scheduler
while True:
    schedule.run_pending()
    # check for pending jobs every second
    time.sleep(60)

The code initiates a scheduled scraping process, extracts price data every 30 seconds, and mocks an alert for price changes. Below is a sample of data for two schedules from the price_history JSON file:

JSON

[
    {
        "name": "Thrustmaster T-Flight Hotas X (Compatible with PC)",
        "price_history": [
            {
                "timestamp": "2025-08-08T08:37:59.510891+00:00",
                "current_price": 65.79,
                "original_price": 69.99,
                "discount_percent": 6,
                "currency": "USD",
            },
            {
                "timestamp": "2025-08-08T09:07:59.510891+00:00",
                "current_price": 65.79,
                "original_price": 69.99,
                "discount_percent": 6,
                "currency": "USD",
            },
        ],
        "average_rating": 4.4,
        "review_count": 8067,
        "units_sold": "500+",
        "availability": "In Stock",
    }
]

Here’s a sample log to track price changes and scraping intervals for the two records:

INFO:root:scraping job started at 2025-08-08T08:37:37.287507+00:00
INFO:root:scraping job stopped at 2025-08-08T08:37:59.511894+00:00
INFO:root:No price change detected.

INFO:root:scraping job started at 2025-08-08T09:07:37.287507+00:00
INFO:root:scraping job stopped at 2025-08-08T09:07:59.511894+00:00
INFO:root:No price change detected.

Congratulations! 🎉 You’ve built a price monitoring system using ZenRows as the scraping solution.

Next Steps: Enhance Your System

Add Email Notifications

Set up email alerts when prices change:

Python

import smtplib
from email.mime.text import MIMEText

def send_price_alert(product_name, old_price, new_price):
    """Send email notification for price changes"""
    msg = MIMEText(f"Price changed for {product_name}: ${old_price} → ${new_price}")
    msg['Subject'] = f'Price Alert: {product_name}'
    msg['From'] = 'your-email@gmail.com'
    msg['To'] = 'recipient@gmail.com'
    
    # Configure your email server
    server = smtplib.SMTP('smtp.gmail.com', 587)
    server.starttls()
    server.login('your-email@gmail.com', 'your-password')
    server.send_message(msg)
    server.quit()

Monitor Multiple Products

Track multiple products by modifying the monitoring function:

Python

def monitor_multiple_products():
    """Monitor multiple products"""
    products = [
        "https://www.amazon.com/dp/B001CXYMFS",
        "https://www.amazon.com/dp/B08N5WRWNW",
        # Add more product URLs
    ]
    
    for url in products:
        try:
            raw_data = scrape_product_data(url, css_extractor)
            cleaned_data = clean_product_data(raw_data)
            price_changed = save_price_history(cleaned_data)
            
            if price_changed:
                logging.info(f"Price alert for {cleaned_data['name']}")
        except Exception as e:
            logging.error(f"Failed to monitor {url}: {e}")

Database Integration

Replace JSON storage with a database for better performance:

Python

import sqlite3

def init_database():
    """Initialize SQLite database"""
    conn = sqlite3.connect('price_monitor.db')
    cursor = conn.cursor()
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS products (
            id INTEGER PRIMARY KEY,
            name TEXT UNIQUE,
            current_price REAL,
            average_rating REAL,
            review_count INTEGER,
            availability TEXT,
            last_updated TIMESTAMP
        )
    ''')
    
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS price_history (
            id INTEGER PRIMARY KEY,
            product_id INTEGER,
            price REAL,
            timestamp TIMESTAMP,
            FOREIGN KEY (product_id) REFERENCES products (id)
        )
    ''')
    
    conn.commit()
    conn.close()

Troubleshooting

Missing Data

Enable js_render for dynamic content
Increase wait time for slow-loading pages or consider switching to wait_for
Verify if the CSS selectors are correct

Rate Limiting

Use premium_proxy for residential IPs
Add delays between requests
Monitor your API usage

Selector Changes

Test selectors regularly
Use more stable selectors when possible
Implement fallback selectors

Best Practices

Monitor selector stability: Check if your CSS selectors still work monthly
Handle errors gracefully: Always include try-catch blocks for network requests
Log everything: Comprehensive logging helps debug issues
Start small: Begin with one product before scaling to multiple products
Respect rate limits: Don’t overwhelm target websites with too many requests

Price Monitoring

Real Estate

General Topics

Build a Price Monitoring System

Who Is this for?

What you’ll learn

Step 1: Set Up Data Extraction

Step 2: Clean and Structure the Data

Step 3: Store the Price History Data

Step 4: Create a Scraping Job

Step 5: Schedule the Scraping Job

Step 6: Put Everything Together

Next Steps: Enhance Your System

Add Email Notifications

Monitor Multiple Products

Database Integration

Troubleshooting

Missing Data

Rate Limiting

Selector Changes

Best Practices

​Who Is this for?

​What you’ll learn

​Step 1: Set Up Data Extraction

​Step 2: Clean and Structure the Data

​Step 3: Store the Price History Data

​Step 4: Create a Scraping Job

​Step 5: Schedule the Scraping Job

​Step 6: Put Everything Together

​Next Steps: Enhance Your System

​Add Email Notifications

​Monitor Multiple Products

​Database Integration

​Troubleshooting

​Missing Data

​Rate Limiting

​Selector Changes

​Best Practices

Who Is this for?

What you’ll learn

Step 1: Set Up Data Extraction

Step 2: Clean and Structure the Data

Step 3: Store the Price History Data

Step 4: Create a Scraping Job

Step 5: Schedule the Scraping Job

Step 6: Put Everything Together

Next Steps: Enhance Your System

Add Email Notifications

Monitor Multiple Products

Database Integration

Troubleshooting

Missing Data

Rate Limiting

Selector Changes

Best Practices