Build an AI Price Tracker That Finds Deals Automatically

Everyone knows prices fluctuate online. Amazon changes prices millions of times per day. Airlines adjust fares based on demand, day of week, and how desperately you seem to need that flight. Electronics prices drop predictably after product launches. Yet most people buy at whatever price they happen to see.

An AI price tracker changes this. It monitors prices continuously, identifies patterns, predicts when prices will drop, and alerts you at the optimal buying moment. This tutorial builds one from scratch — real-time monitoring, AI-powered analysis, and smart alerts.

What We’re Building

A price tracking system that:

1. Monitors product prices across multiple websites
2. Stores historical price data
3. Uses AI to identify pricing patterns and predict drops
4. Sends alerts when prices hit target thresholds or AI predicts a drop
5. Provides a dashboard showing price history and predictions

Tech Stack

• Python 3.11+
• Playwright for web scraping (handles JavaScript-rendered pages)
• SQLite for price history storage
• Claude API for price analysis and prediction
• Streamlit for the dashboard
• APScheduler for periodic price checks

Step 1: Price Scraping Engine

from playwright.async_api import async_playwright
from dataclasses import dataclass
from datetime import datetime
import re

@dataclass
class PriceResult:
    url: str
    product_name: str
    price: float
    currency: str
    timestamp: datetime
    in_stock: bool

async def scrape_price(url: str) -> PriceResult:
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()

        # Set a realistic user agent
        await page.set_extra_http_headers({
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)"
        })

        await page.goto(url, wait_until="domcontentloaded", timeout=30000)
        await page.wait_for_timeout(2000)

        # Get page content for AI extraction
        content = await page.content()
        title = await page.title()

        # Extract visible text (limit to reduce token usage)
        text = await page.evaluate("""
            () => document.body.innerText.substring(0, 3000)
        """)

        await browser.close()

        # Use AI to extract price from page text
        price_data = await extract_price_with_ai(text, url)
        return PriceResult(
            url=url,
            product_name=price_data.get("name", title),
            price=price_data["price"],
            currency=price_data.get("currency", "USD"),
            timestamp=datetime.now(),
            in_stock=price_data.get("in_stock", True)
        )

Step 2: AI Price Extraction

from anthropic import Anthropic
import json

client = Anthropic()

async def extract_price_with_ai(page_text: str, url: str) -> dict:
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=256,
        system="""Extract product pricing info from webpage text.
Return JSON: {"name": "product name", "price": 29.99, "currency": "USD", "in_stock": true}
Price should be a number, not a string. Use the current/sale price if multiple shown.""",
        messages=[{"role": "user", "content": f"URL: {url}\n\nPage text:\n{page_text[:2000]}"}]
    )
    text = response.content[0].text
    if "```json" in text:
        text = text.split("```json")[1].split("```")[0]
    return json.loads(text.strip())

Step 3: Price History Database

import sqlite3
from datetime import datetime

def init_db():
    conn = sqlite3.connect("prices.db")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS products (
            id INTEGER PRIMARY KEY,
            url TEXT UNIQUE,
            name TEXT,
            target_price REAL
        )
    """)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS price_history (
            id INTEGER PRIMARY KEY,
            product_id INTEGER,
            price REAL,
            currency TEXT,
            in_stock BOOLEAN,
            timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY(product_id) REFERENCES products(id)
        )
    """)
    conn.commit()
    return conn

def add_product(url: str, name: str, target_price: float = None):
    conn = sqlite3.connect("prices.db")
    conn.execute(
        "INSERT OR IGNORE INTO products (url, name, target_price) VALUES (?, ?, ?)",
        (url, name, target_price)
    )
    conn.commit()

def record_price(url: str, price: float, currency: str, in_stock: bool):
    conn = sqlite3.connect("prices.db")
    product = conn.execute(
        "SELECT id FROM products WHERE url = ?", (url,)
    ).fetchone()
    if product:
        conn.execute(
            "INSERT INTO price_history (product_id, price, currency, in_stock) VALUES (?, ?, ?, ?)",
            (product[0], price, currency, in_stock)
        )
        conn.commit()

def get_price_history(url: str, days: int = 90):
    conn = sqlite3.connect("prices.db")
    return conn.execute("""
        SELECT ph.price, ph.timestamp
        FROM price_history ph
        JOIN products p ON ph.product_id = p.id
        WHERE p.url = ? AND ph.timestamp > datetime('now', ?)
        ORDER BY ph.timestamp
    """, (url, f"-{days} days")).fetchall()

Step 4: AI Price Analysis and Prediction

def analyze_price_trend(url: str) -> dict:
    history = get_price_history(url)
    if len(history) < 5:
        return {"prediction": "insufficient_data"}

    price_data = "\n".join([f"{ts}: ${price:.2f}" for price, ts in history])

    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=512,
        system="""Analyze price history data and provide:
1. Current trend (rising, falling, stable, cyclical)
2. Price prediction for next 7 days
3. Buy recommendation (buy_now, wait, set_alert)
4. Confidence level (high, medium, low)
Return JSON.""",
        messages=[{"role": "user", "content": f"Price history:\n{price_data}"}]
    )
    text = response.content[0].text
    if "```json" in text:
        text = text.split("```json")[1].split("```")[0]
    return json.loads(text.strip())

Step 5: Alert System

import smtplib
from email.mime.text import MIMEText

def check_alerts(url: str, current_price: float):
    conn = sqlite3.connect("prices.db")
    product = conn.execute(
        "SELECT name, target_price FROM products WHERE url = ?", (url,)
    ).fetchone()

    if product and product[1] and current_price <= product[1]:
        send_price_alert(product[0], current_price, product[1], url)

    # AI-based alert
    analysis = analyze_price_trend(url)
    if analysis.get("recommendation") == "buy_now" and analysis.get("confidence") == "high":
        send_price_alert(
            product[0], current_price, None, url,
            reason=f"AI predicts price will rise. Current trend: {analysis.get('trend')}"
        )

def send_price_alert(name, current_price, target_price, url, reason=None):
    msg = f"Price Alert: {name}\nCurrent: ${current_price:.2f}"
    if target_price:
        msg += f"\nTarget: ${target_price:.2f}"
    if reason:
        msg += f"\nReason: {reason}"
    msg += f"\nURL: {url}"
    print(msg)  # Replace with email/Slack notification

Step 6: Scheduled Monitoring

from apscheduler.schedulers.asyncio import AsyncIOScheduler
import asyncio

async def check_all_prices():
    conn = sqlite3.connect("prices.db")
    products = conn.execute("SELECT url FROM products").fetchall()

    for (url,) in products:
        try:
            result = await scrape_price(url)
            record_price(url, result.price, result.currency, result.in_stock)
            check_alerts(url, result.price)
        except Exception as e:
            print(f"Error checking {url}: {e}")
        await asyncio.sleep(5)  # Rate limiting

scheduler = AsyncIOScheduler()
scheduler.add_job(check_all_prices, "interval", hours=6)

Step 7: Dashboard

Build a Streamlit dashboard showing price history charts with Plotly, current prices for all tracked products, AI analysis and buy recommendations, and an interface to add new products to track.

Step 8: Deployment and Monitoring

Deploy on Railway or a VPS with a cron job. Key considerations:

• Respect robots.txt and rate limits
• Rotate user agents to avoid blocks
• Handle CAPTCHAs gracefully (skip and retry later)
• Store raw page snapshots for debugging extraction failures
• Monitor scraping success rates

Legal Considerations

Web scraping legality varies by jurisdiction. Generally:

• Scraping publicly available pricing data for personal use is broadly accepted
• Respect Terms of Service
• Don’t overload servers with requests
• Don’t circumvent access controls

The Bottom Line

A personal price tracker saves real money. Users tracking electronics purchases report saving 15-30% by buying at optimal times. The AI analysis layer adds prediction capability that simple threshold alerts can’t match.

Build time: 4-5 hours. Ongoing cost: $5-10/month for hosting and API calls. Potential savings: significant on any purchase over $100.