Price comparison tools and price drop alerts are one of the most practical applications for web scraping. Whether you’re tracking products across retailers for personal shopping, monitoring competitor pricing for a business, or building a deals aggregator, the core pattern is the same: extract prices reliably, store history, detect changes, and alert on meaningful drops.
This guide builds a complete price tracker in Python: async price extraction using UnWeb, SQLite-backed price history, change detection with configurable thresholds, and email/Slack alerting.
The Architecture
The system has four components:
- Price extraction: Fetch the product page, extract the price from structured Markdown
- Price history storage: SQLite database with price snapshots keyed by product + timestamp
- Change detection: Compare current price to previous price and historical minimum
- Alerting: Send notification when a price drops below a configured threshold
Data Models
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
@dataclass
class Product:
id: str # Unique identifier (e.g., "amazon-b08n5wrwnw")
name: str
url: str
alert_threshold: float # Alert when price drops below this (0.9 = 10% drop)
target_price: Optional[float] = None # Alert when price drops below this absolute value
@dataclass
class PriceSnapshot:
product_id: str
price: float
currency: str
captured_at: datetime
in_stock: bool
raw_price_text: str # Original price string before parsing
@dataclass
class PriceAlert:
product: Product
current_price: float
previous_price: float
historical_min: float
change_percent: float
alert_reason: str # "threshold_drop", "new_historical_low", "target_reached"
Price Extraction with UnWeb
E-commerce pages are among the hardest to scrape: heavy JavaScript, dynamic pricing, bot detection, and constantly changing layouts. UnWeb handles the rendering and returns clean Markdown — from which price extraction is far more reliable than parsing raw HTML:
import re
import httpx
from typing import Optional
UNWEB_API_KEY = "your_api_key"
async def fetch_page_markdown(url: str, client: httpx.AsyncClient) -> str:
response = await client.get(
"https://api.unweb.info/v1/convert",
params={"url": url},
headers={"Authorization": f"Bearer {UNWEB_API_KEY}"},
timeout=30.0,
)
response.raise_for_status()
return response.json()["markdown"]
def extract_price(markdown: str) -> Optional[tuple[float, str]]:
"""
Extract price from product page Markdown.
Returns (price_float, currency_symbol) or None.
"""
# Match common price patterns: $29.99, €149, £9.99, $1,299.00
patterns = [
r'[\$€£¥₹]\s*[\d,]+\.?\d*', # Symbol before number
r'[\d,]+\.?\d*\s*(?:USD|EUR|GBP|JPY)', # Number before currency code
]
for pattern in patterns:
match = re.search(pattern, markdown)
if match:
price_text = match.group(0).strip()
# Extract numeric value
numeric = re.sub(r'[^\d.]', '', price_text.replace(',', ''))
try:
price = float(numeric)
# Extract currency symbol
currency = re.sub(r'[\d.,\s]', '', price_text).strip()
return price, currency or "$"
except ValueError:
continue
return None
def extract_stock_status(markdown: str) -> bool:
"""Check if item is in stock based on Markdown content."""
out_of_stock_signals = [
'out of stock', 'currently unavailable', 'sold out',
'temporarily out of stock', 'not available',
]
lower = markdown.lower()
return not any(signal in lower for signal in out_of_stock_signals)
Price extraction reliability: The regex approach works for most standard product pages. For complex pages with multiple price variants (size/color selections), you may need LLM-assisted extraction. See our table extraction guide for the LLM pattern.
SQLite Price History
import sqlite3
from datetime import datetime
def init_db(db_path: str = "prices.db") -> sqlite3.Connection:
conn = sqlite3.connect(db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS products (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
url TEXT NOT NULL,
alert_threshold REAL DEFAULT 0.9,
target_price REAL,
created_at TEXT DEFAULT (datetime('now'))
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS price_snapshots (
id INTEGER PRIMARY KEY AUTOINCREMENT,
product_id TEXT NOT NULL,
price REAL NOT NULL,
currency TEXT NOT NULL,
in_stock INTEGER NOT NULL,
raw_price_text TEXT,
captured_at TEXT NOT NULL,
FOREIGN KEY (product_id) REFERENCES products (id)
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_snapshots_product_time
ON price_snapshots (product_id, captured_at DESC)
""")
conn.commit()
return conn
def save_snapshot(conn: sqlite3.Connection, snapshot: PriceSnapshot) -> None:
conn.execute(
"""INSERT INTO price_snapshots (product_id, price, currency, in_stock, raw_price_text, captured_at)
VALUES (?, ?, ?, ?, ?, ?)""",
(snapshot.product_id, snapshot.price, snapshot.currency,
int(snapshot.in_stock), snapshot.raw_price_text,
snapshot.captured_at.isoformat())
)
conn.commit()
def get_previous_price(conn: sqlite3.Connection, product_id: str) -> Optional[float]:
"""Get the most recent price before the current check."""
row = conn.execute(
"""SELECT price FROM price_snapshots
WHERE product_id = ? ORDER BY captured_at DESC LIMIT 1 OFFSET 1""",
(product_id,)
).fetchone()
return row[0] if row else None
def get_historical_min(conn: sqlite3.Connection, product_id: str) -> Optional[float]:
row = conn.execute(
"SELECT MIN(price) FROM price_snapshots WHERE product_id = ? AND in_stock = 1",
(product_id,)
).fetchone()
return row[0] if row and row[0] else None
def get_price_history(
conn: sqlite3.Connection,
product_id: str,
days: int = 30,
) -> list[tuple[str, float]]:
rows = conn.execute(
"""SELECT captured_at, price FROM price_snapshots
WHERE product_id = ?
AND captured_at >= datetime('now', ?)
ORDER BY captured_at""",
(product_id, f"-{days} days")
).fetchall()
return rows
Change Detection Logic
def detect_alert(
product: Product,
current_price: float,
previous_price: Optional[float],
historical_min: Optional[float],
) -> Optional[PriceAlert]:
"""
Return a PriceAlert if the current price warrants notification.
"""
if previous_price is None:
return None # First check, no comparison yet
change_percent = (current_price - previous_price) / previous_price
# Alert 1: Target price reached
if product.target_price and current_price <= product.target_price:
return PriceAlert(
product=product,
current_price=current_price,
previous_price=previous_price,
historical_min=historical_min or current_price,
change_percent=change_percent,
alert_reason="target_reached",
)
# Alert 2: New historical minimum (price is lower than ever recorded)
if historical_min and current_price < historical_min:
return PriceAlert(
product=product,
current_price=current_price,
previous_price=previous_price,
historical_min=historical_min,
change_percent=change_percent,
alert_reason="new_historical_low",
)
# Alert 3: Significant drop from previous check
if change_percent <= -(1.0 - product.alert_threshold):
return PriceAlert(
product=product,
current_price=current_price,
previous_price=previous_price,
historical_min=historical_min or current_price,
change_percent=change_percent,
alert_reason="threshold_drop",
)
return None
Email and Slack Alerts
import smtplib
from email.mime.text import MIMEText
import httpx
async def send_email_alert(alert: PriceAlert, to_email: str, smtp_config: dict) -> None:
change_pct = abs(alert.change_percent) * 100
reason_text = {
"target_reached": f"Target price reached!",
"new_historical_low": f"New historical low!",
"threshold_drop": f"Price dropped {change_pct:.1f}%!",
}[alert.alert_reason]
body = f"""
{reason_text}
Product: {alert.product.name}
URL: {alert.product.url}
Current Price: ${alert.current_price:.2f}
Previous Price: ${alert.previous_price:.2f}
Change: {alert.change_percent * 100:+.1f}%
Historical Low: ${alert.historical_min:.2f}
""".strip()
msg = MIMEText(body)
msg['Subject'] = f"[Price Alert] {alert.product.name} — ${alert.current_price:.2f}"
msg['From'] = smtp_config['from']
msg['To'] = to_email
with smtplib.SMTP_SSL(smtp_config['host'], smtp_config['port']) as server:
server.login(smtp_config['user'], smtp_config['password'])
server.send_message(msg)
async def send_slack_alert(alert: PriceAlert, webhook_url: str) -> None:
change_pct = abs(alert.change_percent) * 100
async with httpx.AsyncClient() as client:
await client.post(webhook_url, json={
"text": (
f"*Price Alert: {alert.product.name}*\n"
f"${alert.current_price:.2f} (was ${alert.previous_price:.2f}, "
f"{alert.change_percent * 100:+.1f}%)\n"
f"<{alert.product.url}|View product>"
)
})
The Main Tracker Loop
import asyncio
import schedule
import time
from datetime import datetime
PRODUCTS = [
Product(
id="product-1",
name="Sony WH-1000XM5 Headphones",
url="https://www.amazon.com/dp/B09XS7JWHH",
alert_threshold=0.9, # Alert on 10%+ drop
target_price=249.99,
),
Product(
id="product-2",
name="Apple AirPods Pro (2nd Gen)",
url="https://www.amazon.com/dp/B0CHWRXH8B",
alert_threshold=0.85, # Alert on 15%+ drop
target_price=189.99,
),
]
async def check_prices(conn: sqlite3.Connection) -> None:
async with httpx.AsyncClient() as client:
for product in PRODUCTS:
try:
markdown = await fetch_page_markdown(product.url, client)
price_result = extract_price(markdown)
if price_result is None:
print(f"[{product.id}] Could not extract price")
continue
price, currency = price_result
in_stock = extract_stock_status(markdown)
snapshot = PriceSnapshot(
product_id=product.id,
price=price,
currency=currency,
captured_at=datetime.utcnow(),
in_stock=in_stock,
raw_price_text=f"{currency}{price:.2f}",
)
save_snapshot(conn, snapshot)
previous_price = get_previous_price(conn, product.id)
historical_min = get_historical_min(conn, product.id)
alert = detect_alert(product, price, previous_price, historical_min)
if alert:
print(f"[ALERT] {product.name}: ${price:.2f} ({alert.alert_reason})")
await send_slack_alert(alert, SLACK_WEBHOOK_URL)
else:
print(f"[OK] {product.name}: ${price:.2f} (prev: ${previous_price:.2f if previous_price else 'N/A'})")
# Polite delay between products
await asyncio.sleep(2.0)
except Exception as e:
print(f"[ERROR] {product.id}: {e}")
def run_checks(conn: sqlite3.Connection) -> None:
asyncio.run(check_prices(conn))
# Run every 6 hours
conn = init_db()
schedule.every(6).hours.do(run_checks, conn=conn)
# Run once immediately at startup
run_checks(conn)
while True:
schedule.run_pending()
time.sleep(60)
Viewing Price History
The SQLite database is queryable directly, but a quick visualization function helps understand trends:
def print_price_history(conn: sqlite3.Connection, product_id: str, days: int = 30) -> None:
history = get_price_history(conn, product_id, days)
if not history:
print("No price history found")
return
prices = [row[1] for row in history]
min_price = min(prices)
max_price = max(prices)
latest = prices[-1]
print(f"\nPrice history ({days} days):")
print(f" Current: ${latest:.2f}")
print(f" Min: ${min_price:.2f}")
print(f" Max: ${max_price:.2f}")
print(f" Range: {(max_price - min_price) / min_price * 100:.1f}%")
# Simple ASCII sparkline
if max_price != min_price:
normalized = [(p - min_price) / (max_price - min_price) for p in prices[-20:]]
chars = ' ▁▂▃▄▅▆▇█'
sparkline = ''.join(chars[min(int(v * 8), 8)] for v in normalized)
print(f" Trend: {sparkline}")
Extract prices from JS-heavy e-commerce pages
UnWeb converts any page — including dynamically rendered e-commerce sites — to clean Markdown for reliable price extraction. Free tier available.
Extending the Tracker
- Multiple retailers: Add more products with different URLs pointing to the same item across Amazon, Best Buy, Target, etc.
- Price trend analysis: Query 30-day rolling average to filter out temporary price spikes
- Dashboard: Serve price history from SQLite via a simple FastAPI endpoint and display with Chart.js
- Deal scoring: Combine current price vs. 30-day average vs. historical minimum to compute a “deal score” and rank products