25. Proyecto: Extraer Datos de Competidores

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!