Building a Price Monitor with Python and a Scraping API

TL;DR

Build a price monitoring system by writing a Python function that calls a scraping API to fetch product pages, extracts the price field, and writes the result to a store. Schedule the function with cron or a lightweight scheduler to run as often as needed, and add basic error handling and alerts.

Introduction

Price monitoring is a common data pipeline: fetch a product page, pull the current price, compare it to previous values, and trigger an alert when a threshold is crossed. Doing this at scale requires handling anti‑bot measures, managing concurrency, and persisting results. A scraping API that abstracts proxy rotation, JavaScript rendering, and retry logic lets you focus on the business logic instead of infrastructure.

Architecture Overview

The system consists of four components:

1. Scraper function – calls the API, parses JSON, extracts the price.
2. Scheduler – triggers the scraper on a set interval (e.g., every 15 minutes).
3. Storage – appends each price reading with a timestamp to a CSV or time‑series DB.
4. Alert – checks the latest value against a rule and sends a notification (email, Slack, webhook).

Using an API removes the need to maintain a pool of headless browsers or rotate proxies yourself. The API returns clean JSON (or other formats) so you can avoid HTML parsing when possible.

Setting Up the Environment

First, install the AlterLab Python SDK and a scheduler. The SDK handles authentication and request formatting.

pip install alterlab schedule

Create a file monitor.py and add your API key. You can find your key in the dashboard after you sign up.

import os
from alterlab import AlterLabSync
import schedule
import time
import csv
from datetime import datetime

API_KEY = os.getenv("ALTERLAB_API_KEY")
client = AlterLabSync(api_key=API_KEY)

Writing the Scraper Function

The core function performs a single scrape, extracts the price, and appends it to a CSV. We request JSON output to avoid extra parsing.

def scrape_price(url: str) -> float | None:
    """Fetch a product page and return the price as a float."""
    try:
        resp = client.scrape(
            url,
            formats=["json"],          # ask for parsed JSON
            # optional: increase min_tier for JS‑heavy sites
            # min_tier=3
        )
        # Assuming the API returns a dict with a 'price' key
        price_str = resp.get("price")
        if price_str is None:
            return None
        # Strip currency symbols and commas
        return float(price_str.replace("$", "").replace(",", ""))
    except Exception as exc:
        print(f"Error scraping {url}: {exc}")
        return None

def log_price(url: str):
    price = scrape_price(url)
    if price is None:
        return
    timestamp = datetime.utcnow().isoformat()
    with open("prices.csv", "a", newline="") as f:
        writer = csv.writer(f)
        writer.writerow([timestamp, url, price])
    print(f"Logged {price} for {url} at {timestamp}")

The function uses the SDK’s scrape method. The formats=["json"] flag tells the API to return structured data when possible, reducing the need for custom HTML selectors.

cURL Equivalent

The same request can be made directly with curl if you prefer not to use the SDK.

curl -X POST https://api.alterlab.io/v1/scrape \
  -H "X-API-Key: $ALTERLAB_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"url": "https://example.com/product", "formats": ["json"]}'

Scheduling with Cron

A simple way to run the scraper repeatedly is to use the schedule library, which mimics cron syntax inside Python.

def job():
    log_price("https://example.com/product")

# Run every 15 minutes
schedule.every(15).minutes.do(job)

while True:
    schedule.run_pending()
    time.sleep(1)

If you prefer system‑level cron, add a line to your crontab:

*/15 * * * * /usr/bin/python3 /path/to/monitor.py >> /var/log/price_monitor.log 2>&1

Handling Data Storage and Alerts

For quick prototyping, a CSV file works. For longer retention and querying, consider appending to InfluxDB or Postgres. The alert logic can be as simple as checking the last two values.

def check_alert(url: str, threshold_pct: float = 5.0):
    if not os.path.exists("prices.csv"):
        return
    import pandas as pd
    df = pd.read_csv("prices.csv", header=None, names=["ts", "url", "price"])
    df = df[df["url"] == url]
    if len(df) < 2:
        return
    latest = df.iloc[-1]["price"]
    previous = df.iloc[-2]["price"]
    change_pct = (latest - previous) / previous * 100
    if abs(change_pct) >= threshold_pct:
        msg = f"Price changed {change_pct:.1f}% for {url}: {previous} → {latest}"
        # Replace with your preferred notification method
        print("ALERT:", msg)
        # Example: requests.post(SLACK_WEBHOOK, json={"text": msg})

Call check_alert after logging each price or run it separately in a nightly job.

Error Handling and Retries

The scraping API automatically retries failed requests with different proxy configurations. Still, wrap API calls in a try/except block and consider exponential backoff for transient errors. If you receive a 429 (rate‑limit) response, respect the Retry-After header or pause before retrying.

Best Practices and Legal Considerations

• Target only publicly accessible data that does not require login or bypassing a paywall.
• Respect rate limits – the API returns headers with your remaining quota; adjust your schedule accordingly.
• Identify your user agent if you make raw requests; the SDK does this automatically.
• Store data securely – price histories may contain sensitive business information.
• Review the site’s Terms of Service to ensure your use case is permitted.

Conclusion

By combining a lightweight Python script, a scheduling mechanism, and a scraping API that handles anti‑bot complexity, you can build a reliable price monitor with minimal operational overhead. The approach scales from a single product to hundreds by simply looping over a list of URLs and writing each result to a shared store.

Try it yourself with the interactive example below:

The typical workflow looks like this: