Building a Real-Time Price Intelligence Pipeline for Under $15/Month

#webdev #python #dataengineering #tutorial

Price intelligence tools like Prisync cost $99-$299/month. You can build the same pipeline for $5/month. Here is exactly how.

The Architecture

Playwright scraper -> PostgreSQL -> dbt transforms -> BI dashboard
                                 -> Price alerts (Telegram)

Four components. Total cost: ~$5/month on a VPS.

Component 1: The Scraper

import asyncio
from playwright.async_api import async_playwright
from datetime import datetime
import re

async def scrape_product(page, url: str, price_selector: str) -> dict:
    await page.goto(url, wait_until='networkidle', timeout=20000)
    price_text = await page.locator(price_selector).first.text_content()
    match = re.search(r'[\d.]+', price_text.replace(',', '.'))
    return {
        'url': url,
        'price': float(match.group()) if match else None,
        'scraped_at': datetime.utcnow().isoformat()
    }

Component 2: PostgreSQL Schema

CREATE TABLE price_snapshots (
    id SERIAL PRIMARY KEY,
    competitor VARCHAR(50),
    product_url TEXT NOT NULL,
    price DECIMAL(10,2),
    scraped_at TIMESTAMPTZ DEFAULT NOW()
);

-- Current prices view
CREATE VIEW current_prices AS
SELECT DISTINCT ON (product_url, competitor)
    competitor, product_url, price, scraped_at
FROM price_snapshots
ORDER BY product_url, competitor, scraped_at DESC;

Component 3: Alerting on Price Changes

import httpx, os

async def check_and_alert(pool, new_data: dict):
    async with pool.acquire() as conn:
        prev = await conn.fetchrow(
            'SELECT price FROM price_snapshots WHERE product_url = $1 ORDER BY scraped_at DESC LIMIT 1',
            new_data['url']
        )

        if prev and new_data['price']:
            change = abs(new_data['price'] - prev['price']) / prev['price'] * 100
            if change >= 5:  # Alert on 5%+ changes
                await send_telegram(
                    f"Price change: {change:.1f}%",
                    f"Old: {prev['price']} -> New: {new_data['price']}"
                )

async def send_telegram(title: str, message: str):
    bot_token = os.environ['TELEGRAM_BOT_TOKEN']
    chat_id = os.environ['TELEGRAM_CHAT_ID']
    async with httpx.AsyncClient() as client:
        await client.post(
            f'https://api.telegram.org/bot{bot_token}/sendMessage',
            json={'chat_id': chat_id, 'text': f'*{title}*\n{message}', 'parse_mode': 'Markdown'}
        )

Component 4: Scheduling

# crontab -e - run every 4 hours
0 */4 * * * cd /opt/price-monitor && python3 scraper.py >> logs/scraper.log 2>&1

Cost Breakdown

Component	Cost
VPS (2GB RAM)	$5/month
PostgreSQL (self-hosted)	$0
Playwright scraper	$0
Total	$5/month

Commercial alternative (Prisync): $99-$299/month.

What You Can Track

Competitor product prices (Amazon, Shopify, any e-commerce)
SERP rankings for target keywords
Job posting volumes (competitor hiring signals)
LinkedIn follower counts

The Business Case

Price intelligence at $5/month vs $299/month lets you:

Undercut competitors by 1-2% on key items
Raise prices when competitors raise theirs (missed revenue otherwise)
Identify when competitors are clearing stock

Need pre-built scrapers for Amazon, Google Shopping, and 10 more e-commerce sites?

Apify Scrapers Bundle - 29 EUR -> https://vhubster3.gumroad.com/l/fjmtqn

DEV Community