Casos de Uso y Automatizaciones con SP-API

Resumen rápido

SP-API permite construir desde automatizaciones simples (exportar pedidos a Google Sheets) hasta sistemas complejos (repricers, dashboards de análisis, pipelines de datos, notificaciones automáticas). Este documento recoge los casos de uso más comunes con su arquitectura recomendada.

---

Caso 1: Exportar pedidos a Google Sheets

Objetivo: Ver todos los pedidos de Amazon actualizados automáticamente en Google Sheets.

Arquitectura:

SP-API Orders API o Reports API
        ↓
Script Python (descarga pedidos)
        ↓
Google Sheets API (gspread)

Cuándo usar cada API:

• Pocos pedidos (<1.000/mes): Orders API directo.
• Muchos pedidos o historial: Reports API (GET_FLAT_FILE_ALL_ORDERS_DATA_BY_ORDER_DATE_GENERAL).

Implementación básica:

import gspread
from google.oauth2.service_account import Credentials
import requests

def sync_orders_to_sheets(
    access_token: str,
    endpoint: str,
    marketplace_id: str,
    spreadsheet_id: str
):
    # 1. Obtener pedidos de SP-API
    orders = get_orders_last_30_days(access_token, endpoint, marketplace_id)

    # 2. Conectar a Google Sheets
    creds = Credentials.from_service_account_file(
        "google-credentials.json",
        scopes=["https://spreadsheets.google.com/feeds",
                "https://www.googleapis.com/auth/drive"]
    )
    gc = gspread.authorize(creds)
    sheet = gc.open_by_key(spreadsheet_id).sheet1

    # 3. Escribir datos
    sheet.clear()
    sheet.append_row(["Order ID", "Estado", "Total", "Fecha", "Items"])
    for order in orders:
        sheet.append_row([
            order.get("AmazonOrderId"),
            order.get("OrderStatus"),
            order.get("OrderTotal", {}).get("Amount"),
            order.get("PurchaseDate"),
            order.get("NumberOfItemsShipped"),
        ])
    print(f"Sincronizados {len(orders)} pedidos")

Automatización: Ejecutar el script con un cron job diario o en AWS Lambda con un EventBridge trigger.

---

Caso 2: Dashboard de análisis de ventas

Objetivo: Panel web con métricas de ventas, inventario y rendimiento.

Arquitectura:

Reports API (pedidos, inventory, settlement)
        ↓
Base de datos (PostgreSQL, SQLite, BigQuery)
        ↓
Dashboard (Streamlit, Metabase, Power BI, Grafana)

APIs más útiles:

• GET_FLAT_FILE_ALL_ORDERS_DATA_BY_ORDER_DATE_GENERAL: Todos los pedidos
• GET_FBA_MYI_ALL_INVENTORY_DATA: Inventario FBA
• GET_V2_SETTLEMENT_REPORT_DATA_FLAT_FILE: Settlement (liquidaciones)
• Sales API → GetOrderMetrics: Métricas de ventas por periodo

Flujo recomendado:

1. Cada noche, ejecutar script que descarga informes del día.
2. Parsear TSV y guardar en base de datos.
3. Dashboard lee de la base de datos (no de la API directamente).

import pandas as pd
import sqlite3

def process_orders_report(csv_content: str, db_path: str):
    """Parsea el informe de pedidos y guarda en SQLite."""
    import io
    df = pd.read_csv(io.StringIO(csv_content), sep="\t")

    conn = sqlite3.connect(db_path)
    df.to_sql("orders", conn, if_exists="append", index=False)
    conn.close()
    print(f"Guardados {len(df)} pedidos en base de datos")

---

Caso 3: Repricer automático

Objetivo: Actualizar precios automáticamente en respuesta a cambios de competencia o reglas de negocio.

Arquitectura:

Notifications API (OFFER_CHANGED)
        ↓
SQS Queue
        ↓
Lambda / Worker (lógica de repricing)
        ↓
Listings API (actualizar precio) o Feeds API (si son muchos)

Lógica de repricing básica:

def calculate_new_price(
    buybox_price: float,
    my_current_price: float,
    min_price: float,
    target_margin: float = 0.15
) -> float:
    """
    Estrategia simple: si el buybox está por debajo de mi precio,
    igualo el buybox. Si ya tengo el buybox, subo un poco.
    Nunca bajo de min_price.
    """
    if my_current_price > buybox_price:
        # No tengo el buybox, igualo el precio
        new_price = max(buybox_price, min_price)
    elif my_current_price == buybox_price:
        # Tengo el buybox, intento subir ligeramente
        new_price = min(my_current_price * 1.02, buybox_price * 1.05)
    else:
        new_price = my_current_price

    return max(new_price, min_price)


def process_offer_changed_notification(notification: dict, access_token: str):
    """Procesa una notificación OFFER_CHANGED y ajusta el precio."""
    payload = notification.get("payload", {})
    asin = payload.get("ASIN")
    buybox_price = payload.get("BuyBoxPrice", {}).get("ListingPrice", {}).get("Amount")

    if not asin or not buybox_price:
        return

    # Obtener mis datos de listing para ese ASIN
    my_listings = get_my_listings_for_asin(access_token, asin)
    for listing in my_listings:
        sku = listing["sku"]
        my_price = listing["price"]
        min_price = listing["min_price"]

        new_price = calculate_new_price(float(buybox_price), my_price, min_price)
        if abs(new_price - my_price) > 0.01:
            update_listing_price(access_token, sku, new_price)
            print(f"SKU {sku}: precio actualizado de {my_price} a {new_price}")

Consideraciones:

• El repricing agresivo puede llevar a "race conditions" de precios (todos bajan hasta el mínimo).
• Implementa siempre un min_price por producto.
• Las actualizaciones de precio con Listings API son casi inmediatas.

---

Caso 4: Sistema de notificaciones de pedidos nuevos

Objetivo: Recibir una notificación por email/Slack cuando llega un pedido nuevo.

Arquitectura:

ORDER_CHANGE notification
        ↓
SQS Queue
        ↓
Lambda (obtiene detalles del pedido)
        ↓
Email (SES) / Slack (webhook) / Telegram Bot

import boto3
import json
import requests

def lambda_handler(event, context):
    """Lambda disparada por mensaje SQS de nueva orden."""
    for record in event["Records"]:
        body = json.loads(record["body"])

        if body.get("notificationType") != "ORDER_CHANGE":
            continue

        order_id = body["payload"]["orderId"]

        # Obtener detalles del pedido (1 sola llamada)
        order = get_order_details(order_id)

        # Enviar notificación a Slack
        send_slack_notification(
            webhook_url=SLACK_WEBHOOK,
            message=f"Nuevo pedido: {order_id}\n"
                    f"Total: {order['OrderTotal']['Amount']} {order['OrderTotal']['CurrencyCode']}\n"
                    f"Items: {order['NumberOfItemsShipped']}"
        )


def send_slack_notification(webhook_url: str, message: str):
    requests.post(webhook_url, json={"text": message})

---

Caso 5: Pipeline de datos para BI (Power BI / Tableau)

Objetivo: Alimentar una herramienta de Business Intelligence con datos de Amazon actualizados diariamente.

Arquitectura:

Reports API (noche)
        ↓
Storage (S3, Google Drive, Azure Blob)
        ↓
Conector BI (Direct Query o programado)

Informes clave para BI:

• GET_FLAT_FILE_ALL_ORDERS_DATA_BY_ORDER_DATE_GENERAL: Pedidos
• GET_FBA_MYI_ALL_INVENTORY_DATA: Inventario FBA
• GET_V2_SETTLEMENT_REPORT_DATA_FLAT_FILE: Finanzas
• GET_FLAT_FILE_RETURNS_DATA_BY_RETURN_DATE: Devoluciones

Consideración importante: Para Power BI o similares, guarda los TSV en un bucket de S3 y conéctalo como data source. Evita que Power BI llame a SP-API directamente (causaría polling continuo y fees elevados).

---

Caso 6: Reconciliación contable automática

Objetivo: Cruzar facturas, pedidos y settlement reports para la contabilidad.

Arquitectura:

Reports API → Settlement Report (quincenal, automático)
Reports API → Orders Report (diario)
        ↓
Script Python (cross-reference)
        ↓
CSV para contabilidad / ERP

---

Arquitecturas de referencia por tipo de app

Tipo de app	APIs principales	Patrón de datos
Dashboard de ventas	Reports + Sales API	Batch nocturno
Sistema de notificaciones	Notifications + Orders	Push (event-driven)
Repricer	Notifications + Catalog + Listings	Reactivo (push)
Sincronización de inventario	Feeds + Listings	Programado
Herramienta de análisis de catálogo	Catalog Items	Bajo demanda
Exportación a Google Sheets	Orders / Reports	Programado diario

---

Errores comunes en automatizaciones

• No implementar reintentos: Las llamadas a API pueden fallar puntualmente. Sin reintentos, la automatización falla.
• Guardar access tokens en variables de entorno en producción: Usa gestores de secretos (AWS Secrets Manager, Azure Key Vault, etc.).
• No gestionar la expiración del access token: El token dura 1 hora. El script debe obtener uno nuevo automáticamente.
• No manejar la paginación: Si tienes más de 100 pedidos, GetOrders devuelve solo 100 y un NextToken. Sin manejar la paginación, pierdes datos.
• Ejecutar scripts de larga duración en entornos serverless sin manejar timeouts: Lambda tiene un límite de 15 minutos. Para jobs largos, usa ECS, EC2 o divide en pasos.

---

Qué debo saber antes de programarlo

• Para automatizaciones simples (exportar a Google Sheets), no necesitas infraestructura compleja. Un script Python en tu máquina con un cron job es suficiente para empezar.
• Para producción seria (repricer, notificaciones en tiempo real), invierte en arquitectura serverless (Lambda + SQS) desde el principio. Es más barato y escalable.
• Siempre separa la lógica de autenticación de la lógica de negocio. El refresh del access token debería ser transparente.

---

Pendiente de revisar

• ¿Hay conectores nativos de SP-API para Power BI o Tableau sin código intermedio?
• ¿Qué librerías de Python para Google Sheets son más estables y recomendadas en 2026?
• ¿Cómo gestionar múltiples cuentas de vendedor en una sola automatización?