Lección 25 de 29•Módulo 5: Web Scraping y APIs
25. Proyecto: Extraer Datos de Competidores
Proyecto real: monitorea precios y disponibilidad de competidores
30 minutos
En esta lección construirás un proyecto completo de web scraping para monitorear precios y disponibilidad de productos de competidores. Este tipo de análisis competitivo es crucial para e-commerce, retail y estrategias de pricing.
¿Por qué monitorear competidores?
El análisis competitivo automatizado te permite:
- Ajustar precios dinámicamente: detecta cuando competidores bajan precios
- Identificar oportunidades: productos que están agotados en competencia pero tú tienes
- Análisis de tendencias: qué productos lanzan tus competidores y cuándo
- Validar estrategias: compara tu catálogo vs. el mercado
- Tomar decisiones data-driven: pricing basado en datos reales, no intuición
Caso real: E-commerce de electrónica
Imagina que vendes laptops online. Necesitas saber:
- ¿Cuánto cuestan los mismos modelos en 5 competidores principales?
- ¿Qué competidor tiene el precio más bajo?
- ¿Hay productos agotados en competencia que tú podrías capitalizar?
- ¿Cómo varían los precios por día de la semana?
Construirás un scraper que responda estas preguntas automáticamente.
Arquitectura del proyecto
Componentes del sistema
# Estructura del proyecto
competitor_monitoring/
├── scraper.py # Lógica de extracción
├── data_processor.py # Limpieza y transformación
├── alerts.py # Sistema de alertas
├── config.py # Configuración y URLs
├── data/
│ └── competitors.csv # Datos históricos
└── reports/
└── daily_report.html
Workflow completo
- Scraping: Extrae datos de 3-5 competidores
- Limpieza: Normaliza precios, nombres, disponibilidad
- Comparación: Calcula precio promedio, mínimo, máximo
- Alertas: Notifica cambios significativos (>10% de variación)
- Reporte: Genera CSV con historial de precios
- Visualización: Gráfico de evolución de precios
Implementación paso a paso
Paso 1: Configuración y estructura base
# config.py
"""
Configuración del scraper de competidores
"""
# Competidores a monitorear
COMPETITORS = {
'TechMart': {
'url': 'https://techmart.com/laptops',
'selector_producto': 'div.product-card',
'selector_nombre': 'h3.product-name',
'selector_precio': 'span.price',
'selector_disponible': 'button.add-to-cart'
},
'ElectroShop': {
'url': 'https://electroshop.com/laptops',
'selector_producto': 'article.item',
'selector_nombre': 'h2.title',
'selector_precio': 'div.price-box span',
'selector_disponible': 'span.stock-status'
},
'MegaStore': {
'url': 'https://megastore.com/computers',
'selector_producto': 'div.product',
'selector_nombre': 'a.product-title',
'selector_precio': 'p.price-current',
'selector_disponible': 'div.availability'
}
}
# Productos a monitorear (modelos específicos)
TARGET_PRODUCTS = [
'MacBook Pro 14" M3',
'Dell XPS 15',
'Lenovo ThinkPad X1',
'HP Spectre x360',
'ASUS ZenBook 14'
]
# Configuración de alertas
ALERT_THRESHOLD_PERCENTAGE = 10 # Alertar si precio cambia >10%
ALERT_EMAIL = 'analytics@tuempresa.com'
# Rate limiting (no sobrecargar servidores)
REQUEST_DELAY_SECONDS = 2
Paso 2: Scraper principal con manejo robusto de errores
# scraper.py
"""
Scraper de precios de competidores
"""
import requests
from bs4 import BeautifulSoup
import pandas as pd
from datetime import datetime
import time
import re
from config import COMPETITORS, TARGET_PRODUCTS, REQUEST_DELAY_SECONDS
class CompetitorScraper:
"""Scraper robusto para monitoreo de competidores"""
def __init__(self):
self.session = requests.Session()
# User-Agent realista para evitar bloqueos
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})
self.resultados = []
def limpiar_precio(self, texto_precio):
"""Extrae número de precio de texto"""
# Ejemplos: "$1,299.99" -> 1299.99, "S/. 4,500" -> 4500
texto_limpio = re.sub(r'[^\d.]', '', texto_precio)
try:
return float(texto_limpio)
except ValueError:
return None
def limpiar_nombre_producto(self, nombre):
"""Normaliza nombre de producto"""
# Eliminar espacios extras, convertir a minúsculas
return ' '.join(nombre.strip().lower().split())
def producto_coincide(self, nombre_scrapeado, producto_objetivo):
"""Verifica si producto scrapeado coincide con target"""
nombre_norm = self.limpiar_nombre_producto(nombre_scrapeado)
objetivo_norm = self.limpiar_nombre_producto(producto_objetivo)
# Coincidencia si contiene palabras clave principales
palabras_objetivo = objetivo_norm.split()
coincidencias = sum(1 for palabra in palabras_objetivo if palabra in nombre_norm)
# Requiere al menos 70% de coincidencia
return coincidencias >= len(palabras_objetivo) * 0.7
def scrapear_competidor(self, nombre_competidor, config):
"""Scrapea un competidor específico"""
print(f"\n🔍 Scrapeando {nombre_competidor}...")
try:
# Request con timeout
response = self.session.get(config['url'], timeout=10)
response.raise_for_status()
soup = BeautifulSoup(response.content, 'html.parser')
productos = soup.select(config['selector_producto'])
print(f" Encontrados {len(productos)} productos")
for producto in productos:
try:
# Extraer datos
nombre_elem = producto.select_one(config['selector_nombre'])
precio_elem = producto.select_one(config['selector_precio'])
disponible_elem = producto.select_one(config['selector_disponible'])
if not nombre_elem or not precio_elem:
continue
nombre = nombre_elem.text.strip()
precio_texto = precio_elem.text.strip()
precio = self.limpiar_precio(precio_texto)
# Verificar disponibilidad
disponible = disponible_elem is not None
if disponible_elem:
texto_disp = disponible_elem.text.lower()
disponible = 'agotado' not in texto_disp and 'out of stock' not in texto_disp
# Verificar si es producto de interés
for producto_objetivo in TARGET_PRODUCTS:
if self.producto_coincide(nombre, producto_objetivo):
self.resultados.append({
'fecha': datetime.now().strftime('%Y-%m-%d %H:%M'),
'competidor': nombre_competidor,
'producto': producto_objetivo,
'nombre_exacto': nombre,
'precio': precio,
'disponible': disponible,
'url': config['url']
})
print(f" ✓ {producto_objetivo}: ${precio} ({'Disponible' if disponible else 'Agotado'})")
except Exception as e:
print(f" ⚠ Error procesando producto: {e}")
continue
# Rate limiting - esperar entre requests
time.sleep(REQUEST_DELAY_SECONDS)
except requests.RequestException as e:
print(f" ❌ Error scrapeando {nombre_competidor}: {e}")
except Exception as e:
print(f" ❌ Error inesperado en {nombre_competidor}: {e}")
def scrapear_todos(self):
"""Scrapea todos los competidores configurados"""
print("="*60)
print("🎯 INICIANDO MONITOREO DE COMPETIDORES")
print("="*60)
for nombre, config in COMPETITORS.items():
self.scrapear_competidor(nombre, config)
print("\n" + "="*60)
print(f"✅ Scraping completado: {len(self.resultados)} productos encontrados")
print("="*60)
return pd.DataFrame(self.resultados)
Paso 3: Análisis y comparación de precios
# data_processor.py
"""
Procesa y analiza datos de competidores
"""
import pandas as pd
import numpy as np
class DataProcessor:
"""Analiza datos de competidores"""
def __init__(self, df_nuevo):
self.df_nuevo = df_nuevo
self.df_historico = self.cargar_historico()
def cargar_historico(self):
"""Carga datos históricos si existen"""
try:
return pd.read_csv('data/competitors.csv')
except FileNotFoundError:
return pd.DataFrame()
def guardar_datos(self):
"""Guarda datos actualizados"""
if self.df_historico.empty:
df_completo = self.df_nuevo
else:
df_completo = pd.concat([self.df_historico, self.df_nuevo], ignore_index=True)
df_completo.to_csv('data/competitors.csv', index=False)
print("💾 Datos guardados en data/competitors.csv")
def analizar_precios(self):
"""Calcula estadísticas de precios por producto"""
if self.df_nuevo.empty:
print("⚠ No hay datos para analizar")
return None
# Filtrar solo productos disponibles para comparación
df_disponibles = self.df_nuevo[self.df_nuevo['disponible'] == True].copy()
# Agrupar por producto
analisis = df_disponibles.groupby('producto').agg({
'precio': ['mean', 'min', 'max', 'std', 'count'],
'competidor': lambda x: list(x)
}).round(2)
analisis.columns = ['precio_promedio', 'precio_min', 'precio_max',
'desviacion_std', 'num_competidores', 'competidores']
# Calcular rango de variación
analisis['variacion_%'] = (
(analisis['precio_max'] - analisis['precio_min']) /
analisis['precio_min'] * 100
).round(1)
return analisis
def detectar_cambios_significativos(self, threshold=10):
"""Detecta cambios de precio >threshold%"""
if self.df_historico.empty:
return []
cambios = []
# Últimos precios por producto y competidor
df_historico_ultimo = self.df_historico.sort_values('fecha').groupby(
['producto', 'competidor']
).last().reset_index()
# Comparar con precios actuales
for _, row_nuevo in self.df_nuevo.iterrows():
match = df_historico_ultimo[
(df_historico_ultimo['producto'] == row_nuevo['producto']) &
(df_historico_ultimo['competidor'] == row_nuevo['competidor'])
]
if not match.empty:
precio_anterior = match.iloc[0]['precio']
precio_actual = row_nuevo['precio']
if precio_anterior and precio_actual:
cambio_pct = ((precio_actual - precio_anterior) / precio_anterior) * 100
if abs(cambio_pct) >= threshold:
cambios.append({
'producto': row_nuevo['producto'],
'competidor': row_nuevo['competidor'],
'precio_anterior': precio_anterior,
'precio_actual': precio_actual,
'cambio_%': round(cambio_pct, 1),
'tipo': 'SUBIDA' if cambio_pct > 0 else 'BAJADA'
})
return cambios
def generar_reporte(self):
"""Genera reporte de análisis"""
print("\n" + "="*60)
print("📊 REPORTE DE ANÁLISIS COMPETITIVO")
print("="*60)
analisis = self.analizar_precios()
if analisis is not None and not analisis.empty:
print("\n🏆 ANÁLISIS DE PRECIOS POR PRODUCTO:\n")
print(analisis.to_string())
# Encontrar mejores precios
print("\n💰 MEJORES PRECIOS ENCONTRADOS:\n")
for producto in analisis.index:
row = analisis.loc[producto]
precio_min = row['precio_min']
# Encontrar competidor con precio mínimo
comp_min = self.df_nuevo[
(self.df_nuevo['producto'] == producto) &
(self.df_nuevo['precio'] == precio_min)
]['competidor'].values[0]
print(f"{producto}:")
print(f" → Mejor precio: ${precio_min} en {comp_min}")
print(f" → Promedio mercado: ${row['precio_promedio']}")
print(f" → Variación: {row['variacion_%']}%\n")
# Detectar cambios
cambios = self.detectar_cambios_significativos()
if cambios:
print("🚨 CAMBIOS SIGNIFICATIVOS DE PRECIO:\n")
for cambio in cambios:
emoji = "📉" if cambio['tipo'] == 'BAJADA' else "📈"
print(f"{emoji} {cambio['producto']} en {cambio['competidor']}:")
print(f" Antes: ${cambio['precio_anterior']} → Ahora: ${cambio['precio_actual']}")
print(f" Cambio: {cambio['cambio_%']:+.1f}%\n")
else:
print("✅ No se detectaron cambios significativos de precio\n")
print("="*60)
Paso 4: Sistema de ejecución completo
# main.py
"""
Script principal de monitoreo de competidores
"""
from scraper import CompetitorScraper
from data_processor import DataProcessor
from datetime import datetime
import os
def crear_estructura_directorios():
"""Crea directorios necesarios"""
os.makedirs('data', exist_ok=True)
os.makedirs('reports', exist_ok=True)
def main():
"""Ejecución principal"""
crear_estructura_directorios()
print(f"\n🕐 Inicio: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
# 1. Scrapear competidores
scraper = CompetitorScraper()
df_resultados = scraper.scrapear_todos()
if df_resultados.empty:
print("⚠ No se obtuvieron datos. Verifica la configuración.")
return
# 2. Procesar y analizar
processor = DataProcessor(df_resultados)
processor.generar_reporte()
processor.guardar_datos()
print(f"\n🕐 Fin: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
print("\n✅ Proceso completado exitosamente")
if __name__ == "__main__":
main()
Consideraciones éticas y legales
Buenas prácticas de web scraping
- Respetar robots.txt
# Verificar robots.txt antes de scrapear
from urllib.robotparser import RobotFileParser
def verificar_robots_txt(url):
"""Verifica si el scraping está permitido"""
rp = RobotFileParser()
rp.set_url(f"{url}/robots.txt")
rp.read()
return rp.can_fetch("*", url)
- Rate limiting agresivo
- Espera 2-5 segundos entre requests
- Evita scraping durante horas pico
- Limita requests totales por día
- Identificación honesta
- Usa User-Agent descriptivo
- Incluye información de contacto
- No simules ser navegador real si no lo eres
- Cumplir términos de servicio
- Lee y respeta ToS de cada sitio
- Usa APIs oficiales cuando existan
- No scrapees datos personales
Alternativas más éticas
- APIs de precio: Servicios como PriceAPI, Rainforest API
- Feeds RSS: Muchos e-commerce tienen feeds públicos
- Partnerships: Acuerdos de intercambio de datos
Ejecución y mantenimiento
Automatizar con cron (veremos en lección 28)
# Ejecutar diariamente a las 6 AM
0 6 * * * /usr/bin/python3 /ruta/al/main.py >> /ruta/logs/scraper.log 2>&1
Monitorear errores
# Agregar logging robusto
import logging
logging.basicConfig(
filename='logs/scraper.log',
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logging.info("Scraping iniciado")
logging.error(f"Error en scraping: {error}")
Casos de uso reales
1. E-commerce de moda
Monitorea precios de 50 competidores, 500 productos. Ajusta precios automáticamente para ser 5% más barato que la competencia.
2. Agencia de viajes
Scrapea precios de vuelos y hoteles de 10 competidores cada 6 horas. Genera alertas cuando encuentra ofertas para capitalizar.
3. Retail electrónica
Monitorea disponibilidad de consolas y GPUs. Cuando competencia se agota, aumenta marketing de productos en stock.
Resumen de la lección
✅ Construiste un scraper completo para monitoreo de competidores ✅ Implementaste limpieza y normalización de datos scrapeados ✅ Creaste sistema de alertas para cambios significativos ✅ Aprendiste buenas prácticas éticas de web scraping ✅ Desarrollaste análisis competitivo automatizado real
🎯 Próxima lección: 26. Automatización de Reportes - Genera reportes en PDF, Excel y HTML automáticamente
En la siguiente lección aprenderás a transformar estos datos en reportes profesionales que se generan y distribuyen automáticamente. ¡El scraping es solo el primer paso!
¿Completaste esta lección?
Marca esta lección como completada. Tu progreso se guardará en tu navegador.