Build a multi-marketplace price tracker that queries Amazon and Walmart daily for your target products, stores price history, detects price changes, and sends alerts when prices drop below your threshold. Price tracking across marketplaces reveals arbitrage opportunities, seasonal patterns, and competitor pricing strategies. This tracker runs as a daily cron job and maintains a JSON-based price history that grows over time, making trend analysis possible without a database.
Prerequisites
- Python 3.8+ installed
- requests library installed
- A Scavio API key from scavio.dev
- A list of products to track
Walkthrough
Step 1: Define products to track
Configure the products and price alert thresholds.
Python
import os, requests, json, re, datetime
API_KEY = os.environ['SCAVIO_API_KEY']
TRACKED_PRODUCTS = [
{'name': 'Apple AirPods Pro 2', 'alert_below': 200},
{'name': 'Sony WH-1000XM5', 'alert_below': 280},
{'name': 'Samsung Galaxy S24 case', 'alert_below': 15},
]
PLATFORMS = ['amazon', 'walmart']
HISTORY_FILE = 'price_history.json'Step 2: Daily cron query both platforms
Search each product on both platforms and extract the current price from the top result.
Python
def parse_price(price_str: str) -> float:
if not price_str:
return 0.0
cleaned = re.sub(r'[^\d.]', '', str(price_str))
try:
return float(cleaned)
except ValueError:
return 0.0
def fetch_prices(product_name: str) -> list:
prices = []
for platform in PLATFORMS:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY},
json={'platform': platform, 'query': product_name}, timeout=15)
results = resp.json().get('organic_results', [])
if results:
top = results[0]
price = parse_price(top.get('price', ''))
prices.append({
'platform': platform,
'price': price,
'title': top.get('title', ''),
'url': top.get('link', ''),
})
return prices
prices = fetch_prices(TRACKED_PRODUCTS[0]['name'])
for p in prices:
print(f"{p['platform']}: ${p['price']}")Step 3: Store price history
Append daily prices to a JSON history file.
Python
def store_prices(product_name: str, prices: list):
history = []
try:
with open(HISTORY_FILE) as f:
history = json.load(f)
except FileNotFoundError:
pass
entry = {
'date': datetime.date.today().isoformat(),
'product': product_name,
'prices': prices,
}
history.append(entry)
with open(HISTORY_FILE, 'w') as f:
json.dump(history, f, indent=2)
def daily_track():
for product in TRACKED_PRODUCTS:
prices = fetch_prices(product['name'])
store_prices(product['name'], prices)
for p in prices:
print(f"{product['name']} on {p['platform']}: ${p['price']}")
daily_track()Step 4: Detect price changes
Compare today's prices with the previous day to detect drops and spikes.
Python
def detect_changes(product_name: str) -> list:
try:
with open(HISTORY_FILE) as f:
history = json.load(f)
except FileNotFoundError:
return []
product_entries = [h for h in history if h['product'] == product_name]
if len(product_entries) < 2:
return []
prev = {p['platform']: p['price'] for p in product_entries[-2].get('prices', [])}
curr = {p['platform']: p['price'] for p in product_entries[-1].get('prices', [])}
changes = []
for platform in curr:
if platform in prev and curr[platform] > 0 and prev[platform] > 0:
diff = curr[platform] - prev[platform]
pct = round(diff / prev[platform] * 100, 1)
if abs(pct) > 1: # Only report >1% changes
changes.append({
'platform': platform,
'prev': prev[platform],
'curr': curr[platform],
'change': round(diff, 2),
'pct': pct,
})
return changes
for product in TRACKED_PRODUCTS:
changes = detect_changes(product['name'])
for c in changes:
direction = 'dropped' if c['change'] < 0 else 'increased'
print(f"{product['name']} {direction} ${abs(c['change'])} on {c['platform']}")Step 5: Alert on price drops
Check if any tracked product has dropped below the alert threshold and generate notifications.
Python
def check_alerts(products: list) -> list:
alerts = []
for product in products:
try:
with open(HISTORY_FILE) as f:
history = json.load(f)
except FileNotFoundError:
continue
entries = [h for h in history if h['product'] == product['name']]
if not entries:
continue
latest = entries[-1]
for p in latest.get('prices', []):
if p['price'] > 0 and p['price'] < product['alert_below']:
alert = {
'product': product['name'],
'platform': p['platform'],
'price': p['price'],
'threshold': product['alert_below'],
'url': p.get('url', ''),
}
alerts.append(alert)
print(f'ALERT: {product["name"]} is ${p["price"]} on {p["platform"]} (threshold: ${product["alert_below"]})')
if not alerts:
print('No price alerts triggered')
return alerts
check_alerts(TRACKED_PRODUCTS)Python Example
Python
import requests, os, re
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
def track_price(product, platforms=['amazon', 'walmart']):
prices = {}
for p in platforms:
data = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': p, 'query': product}).json()
top = (data.get('organic_results', []) or [{}])[0]
raw = re.sub(r'[^\d.]', '', top.get('price', '0'))
prices[p] = float(raw) if raw else 0
return prices
print(track_price('Apple AirPods Pro 2'))JavaScript Example
JavaScript
const H = {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
async function trackPrice(product) {
const prices = {};
for (const p of ['amazon', 'walmart']) {
const r = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: H, body: JSON.stringify({platform: p, query: product})
});
const top = ((await r.json()).organic_results || [])[0] || {};
prices[p] = parseFloat((top.price || '0').replace(/[^\d.]/g, '')) || 0;
}
return prices;
}
trackPrice('Apple AirPods Pro 2').then(console.log);Expected Output
JSON
A daily-running price tracker that monitors products across Amazon and Walmart, stores price history, detects changes, and alerts when prices drop below configured thresholds.