Tutorial

How to Build a Stable ASIN Monitoring Pipeline

Build a daily ASIN monitoring pipeline that tracks Amazon product prices, ratings, and availability without scraper maintenance. Stable and cost-effective.

Monitoring Amazon ASINs daily for price changes, rating shifts, and stock availability is essential for e-commerce intelligence. Scraper-based solutions require constant maintenance as Amazon changes page layouts and tightens bot detection. This tutorial builds a stable monitoring pipeline using the Scavio API that checks a list of ASINs on a schedule, stores historical data, detects meaningful changes, and sends alerts. Each ASIN check costs $0.005 with zero maintenance overhead.

Prerequisites

  • Python 3.9+ installed
  • requests library installed
  • A Scavio API key from scavio.dev
  • A list of ASINs to monitor

Walkthrough

Step 1: Define the ASIN watchlist and storage

Set up the list of ASINs to monitor and a JSON-based storage system for historical tracking.

Python
import os, requests, json, time
from datetime import datetime

SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}
URL = 'https://api.scavio.dev/api/v1/search'

WATCHLIST = ['B09G9FPHY6', 'B07FZ8S74R', 'B08N5WRWNW', 'B0BSHF7WHW']
HISTORY_FILE = 'asin_history.json'

def load_history() -> dict:
    if os.path.exists(HISTORY_FILE):
        with open(HISTORY_FILE) as f:
            return json.load(f)
    return {}

def save_history(history: dict):
    with open(HISTORY_FILE, 'w') as f:
        json.dump(history, f, indent=2)

Step 2: Fetch current product data for each ASIN

Query the Scavio API for each ASIN and extract price, rating, review count, and availability.

Python
def fetch_asin(asin: str) -> dict:
    resp = requests.post(URL, headers=H,
        json={'platform': 'amazon', 'query': asin, 'marketplace': 'US'})
    resp.raise_for_status()
    data = resp.json()
    product = data.get('product', data)
    price_str = product.get('price', '')
    price = None
    if price_str:
        price = float(price_str.replace('$', '').replace(',', ''))
    return {
        'asin': asin,
        'title': product.get('title', ''),
        'price': price,
        'rating': product.get('rating', ''),
        'reviews': product.get('reviews_count', 0),
        'availability': product.get('availability', ''),
        'timestamp': datetime.now().isoformat(),
    }

for asin in WATCHLIST[:2]:
    data = fetch_asin(asin)
    print(f'{asin}: ${data["price"]} | {data["rating"]} stars | {data["title"][:40]}')
    time.sleep(0.3)

Step 3: Detect changes and generate alerts

Compare current data with previous snapshots to detect price drops, rating changes, and stock issues.

Python
def check_changes(asin: str, current: dict, history: dict) -> list:
    alerts = []
    prev_records = history.get(asin, [])
    if not prev_records:
        return alerts
    prev = prev_records[-1]
    # Price change
    if current['price'] and prev.get('price'):
        pct_change = (current['price'] - prev['price']) / prev['price'] * 100
        if abs(pct_change) > 5:
            direction = 'dropped' if pct_change < 0 else 'increased'
            alerts.append(f'PRICE {direction} {abs(pct_change):.1f}%: ${prev["price"]} -> ${current["price"]}')
    # Rating change
    if current['rating'] and prev.get('rating') and current['rating'] != prev['rating']:
        alerts.append(f'RATING changed: {prev["rating"]} -> {current["rating"]}')
    # Stock status
    if 'out of stock' in (current.get('availability', '') or '').lower():
        alerts.append('OUT OF STOCK')
    return alerts

def run_monitoring():
    history = load_history()
    for asin in WATCHLIST:
        current = fetch_asin(asin)
        alerts = check_changes(asin, current, history)
        if asin not in history:
            history[asin] = []
        history[asin].append(current)
        status = f'[ALERT] {", ".join(alerts)}' if alerts else '[OK]'
        print(f'{status} {asin}: ${current["price"]} | {current["title"][:35]}')
        time.sleep(0.3)
    save_history(history)
    print(f'\nMonitored {len(WATCHLIST)} ASINs | Cost: ${len(WATCHLIST) * 0.005:.3f}')

run_monitoring()

Python Example

Python
import os, requests, json, time
from datetime import datetime

SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}

def monitor_asins(asins):
    for asin in asins:
        resp = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
            json={'platform': 'amazon', 'query': asin, 'marketplace': 'US'})
        p = resp.json().get('product', resp.json())
        print(f'{asin}: {p.get("price", "N/A")} | {p.get("rating", "N/A")} | {p.get("title", "")[:40]}')
        time.sleep(0.3)
    print(f'Cost: ${len(asins) * 0.005:.3f}')

monitor_asins(['B09G9FPHY6', 'B07FZ8S74R'])

JavaScript Example

JavaScript
const SCAVIO_KEY = process.env.SCAVIO_API_KEY;

async function monitorAsins(asins) {
  for (const asin of asins) {
    const resp = await fetch('https://api.scavio.dev/api/v1/search', {
      method: 'POST',
      headers: { 'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json' },
      body: JSON.stringify({ platform: 'amazon', query: asin, marketplace: 'US' })
    });
    const p = (await resp.json()).product || {};
    console.log(`${asin}: ${p.price || 'N/A'} | ${p.rating || 'N/A'} | ${(p.title || '').slice(0, 40)}`);
  }
  console.log(`Cost: $${(asins.length * 0.005).toFixed(3)}`);
}

monitorAsins(['B09G9FPHY6', 'B07FZ8S74R']);

Expected Output

JSON
[OK] B09G9FPHY6: $29.99 | Echo Dot (5th Gen) | Smart speake
[ALERT] PRICE dropped 12.3%: $28.49 -> $24.99 B07FZ8S74R: $24.99 | Fire TV Stick 4K with Alexa
[OK] B08N5WRWNW: $89.99 | Kindle Paperwhite (11th Generatio
[OK] B0BSHF7WHW: $39.99 | Fire TV Stick 4K Max (2nd Gen)

Monitored 4 ASINs | Cost: $0.020

Related Tutorials

Frequently Asked Questions

Most developers complete this tutorial in 15 to 30 minutes. You will need a Scavio API key (free tier works) and a working Python or JavaScript environment.

Python 3.9+ installed. requests library installed. A Scavio API key from scavio.dev. A list of ASINs to monitor. A Scavio API key gives you 250 free credits per month.

Yes. The free tier includes 250 credits per month, which is more than enough to complete this tutorial and prototype a working solution.

Scavio has a native LangChain package (langchain-scavio), an MCP server, and a plain REST API that works with any HTTP client. This tutorial uses the raw REST API, but you can adapt to your framework of choice.

Start Building

Build a daily ASIN monitoring pipeline that tracks Amazon product prices, ratings, and availability without scraper maintenance. Stable and cost-effective.