Cree un rastreador de precios en múltiples mercados que consulte diariamente a Amazon y Walmart sobre sus productos objetivo, almacene el historial de precios, detecte cambios de precios y envíe alertas cuando los precios caigan por debajo de su umbral. El seguimiento de precios en los mercados revela oportunidades de arbitraje, patrones estacionales y estrategias de precios de la competencia. Este rastreador se ejecuta como un trabajo cron diario y mantiene un historial de precios basado en JSON que crece con el tiempo, lo que hace posible el análisis de tendencias sin una base de datos.
Requisitos previos
- Python 3.8+ instalado
- solicita biblioteca instalada
- Una clave API de Scavio de scavio.dev
- Una lista de productos para rastrear
Guia paso a paso
Paso 1: Definir productos a seguir
Configurar los productos y umbrales de alerta de precios.
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'Paso 2: Consulta cron diaria en ambas plataformas
Busque cada producto en ambas plataformas y extraiga el precio actual del resultado superior.
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']}")Paso 3: Historial de precios de la tienda
Agregue precios diarios a un archivo de historial JSON.
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()Paso 4: Detectar cambios de precios
Compare los precios de hoy con los del día anterior para detectar caídas y picos.
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']}")Paso 5: Alerta sobre bajadas de precios
Compruebe si algún producto rastreado ha caído por debajo del umbral de alerta y genere notificaciones.
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)Ejemplo en Python
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'))Ejemplo en JavaScript
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);Salida esperada
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.