Inteligencia de Precios & Pipelines ETL en Python

Fijar precios en un mercado B2B o B2C basándose únicamente en la intuición o en revisar manualmente tres páginas web una vez al año es una receta segura para perder cuota de mercado o regalar margen neto. En sectores dinámicos —como el comercio electrónico, la distribución de suministros industriales o las plataformas Web-to-Print de artes gráficas—, las tarifas cambian semanalmente según el coste de la celulosa, la ocupación de las máquinas o las campañas estacionales.

En esta guía práctica de ingeniería de datos aprenderás a diseñar un pipeline ETL (Extract, Transform, Load) en Python capaz de consolidar catálogos públicos y feeds de datos, normalizar unidades dispares (como escalados por tirada o precios por metro cuadrado), filtrar valores atípicos mediante estadística intercuartílica en Pandas y conectar el histórico SQL resultante con Power BI.

1. Arquitectura de un Pipeline ETL de Precios Ético y Escalable

Antes de escribir una sola línea de código, todo sistema de inteligencia competitiva profesional debe regirse por tres principios arquitectónicos:

• Respeto estricto de políticas y robots.txt: Prioriza siempre APIs públicas, feeds XML/JSON de catálogo (Schema.org/Product incrustado en el HTML) y aplica pausas de cortesía (Rate Limiting) para no generar carga en servidores terceros.
• Desacoplamiento entre Extracción y Transformación (Capa Bronze / Silver / Gold): Guarda primero la respuesta en bruto con su marca de tiempo; aplica después la limpieza con Pandas en una tabla normalizada (Silver) y sirve finalmente las métricas agregadas a Power BI (Gold).
• Homogeneización de Especificaciones (Product Matching): No puedes comparar un flyer A5 en estucado de 135 g/m² frente a uno de 300 g/m² laminado mate. Cada registro debe mapearse a una clave maestra estandarizada (sku_canonico).

2. Extracción de Datos Estructurados JSON-LD de Producto con Python

Hoy en día, más del 75% de las tiendas online y portales de producto incluyen metadatos Schema.org JSON-LD dentro de la etiqueta <script type="application/ld+json"> para posicionar en Google Shopping. Extraer este bloque JSON es mucho más limpio y estable que depender de selectores CSS frágiles que cambian cada vez que se rediseña una web:

import json
import time
import urllib.request
from bs4 import BeautifulSoup
from datetime import date

def extraer_schema_producto(url_producto):
    headers = {
        "User-Agent": "Mozilla/5.0 (compatible; DataAnalyticsBot/1.0; +https://portfolio.xuangfm.com)"
    }
    req = urllib.request.Request(url_producto, headers=headers)
    
    with urllib.request.urlopen(req, timeout=10) as resp:
        html = resp.read().decode("utf-8", errors="ignore")
        
    soup = BeautifulSoup(html, "html.parser")
    bloques_json = soup.find_all("script", type="application/ld+json")
    
    for bloque in bloques_json:
        try:
            datos = json.loads(bloque.string)
            if isinstance(datos, dict) and datos.get("@type") == "Product":
                oferta = datos.get("offers", {})
                if isinstance(oferta, list):
                    oferta = oferta[0]
                return {
                    "fecha_captura": str(date.today()),
                    "nombre": datos.get("name"),
                    "sku_origen": datos.get("sku"),
                    "precio": float(oferta.get("price", 0)),
                    "moneda": oferta.get("priceCurrency", "EUR"),
                    "disponibilidad": oferta.get("availability", "").split("/")[-1]
                }
        except (json.JSONDecodeError, ValueError, TypeError):
            continue
    return None

3. Transformación con Pandas: Normalización de Escalados y Filtro IQR

En sectores industriales y de impresión, el precio unitario cae drásticamente con el volumen (economías de escala al amortizar el coste fijo de planchas y arreglo de máquina). Además, es habitual encontrar errores en los datos de origen (por ejemplo, un accesorio de 2 € mezclado dentro de la categoría de expositores Roll-up de 65 €).

Con Pandas calculamos el coste por millar normalizado y aplicamos el criterio estadístico del Rango Intercuartílico ($IQR = Q_3 - Q_1$) agrupado por categoría y tramo de cantidad para neutralizar cualquier valor atípico (outlier):

import pandas as pd
import sqlite3

def transformar_y_limpiar_precios(df_bruto):
    df = df_bruto.copy()
    
    # 1. Calcular coste normalizado por cada 1.000 unidades (Coste por Millar)
    df["precio_por_millar"] = (df["precio_total_lote"] / df["cantidad_lote"]) * 1000.0
    
    # 2. Detección estadística de Outliers mediante Rango Intercuartílico (IQR) por grupo
    grupos = df.groupby(["sku_canonico", "cantidad_lote"])["precio_por_millar"]
    q1 = grupos.transform(lambda x: x.quantile(0.25))
    q3 = grupos.transform(lambda x: x.quantile(0.75))
    iqr = q3 - q1
    
    limite_inferior = q1 - 1.5 * iqr
    limite_superior = q3 + 1.5 * iqr
    
    df["es_outlier"] = (df["precio_por_millar"] < limite_inferior) | (df["precio_por_millar"] > limite_superior)
    
    # 3. Filtrar registros limpios y calcular posición respecto a la mediana del mercado
    df_limpio = df[~df["es_outlier"]].copy()
    df_limpio["mediana_mercado_millar"] = (
        df_limpio.groupby(["sku_canonico", "cantidad_lote"])["precio_por_millar"]
        .transform("median")
    )
    
    return df_limpio

def guardar_historico_sql(df_limpio, ruta_db="pricing_warehouse.db"):
    with sqlite3.connect(ruta_db) as conn:
        df_limpio.to_sql("fact_historico_precios", conn, if_exists="append", index=False)

4. Matriz de Posicionamiento Competitivo en Power BI (Medidas DAX)

Una vez conectada nuestra tabla SQL fact_historico_precios al modelo en estrella de Power BI (consulta nuestra guía de arquitectura de dashboards en Power BI), definimos el Índice de Precios de Mercado (Price Index):

La medida DAX para evaluar este índice dinámicamente por categoría, fecha o tramo de tirada es:

Indice_Posicionamiento_Precio = 
VAR MiPrecio = 
    CALCULATE(
        AVERAGE(Fact_Historico_Precios[precio_por_millar]),
        Fact_Historico_Precios[competidor] = "Propia_Empresa"
    )
VAR MedianaCompetencia = 
    CALCULATE(
        MEDIAN(Fact_Historico_Precios[precio_por_millar]),
        Fact_Historico_Precios[competidor] <> "Propia_Empresa"
    )
RETURN
    DIVIDE(MiPrecio, MedianaCompetencia, 0) * 100

5. Preguntas Frecuentes sobre Análisis de Precios con Python y BI