Guarda las cuotas de la API de Pinnacle en CSV con Python

El quickstart de Python de este sitio sondea la API de cuotas de Pinnacle e imprime la línea money line. La siguiente pregunta que la mayoría de los lectores hace es cómo guardar lo que obtuvo. Un archivo CSV basta para un primer almacén: una fila por selección por cambio, añadida en cada sondeo, legible por Excel, pandas o una importación a base de datos después.

Qué debe contener una fila

Guarda la observación, no solo el precio. Una fila útil nombra el evento (id, liga, equipos), las coordenadas del mercado (período, mercado, lado) y la propia observación: el precio decimal y la hora en que lo viste. Añade el cursor last de la respuesta si quieres rastrear qué sondeo escribió la fila. Los precios llegan como cuotas decimales, así que guárdalos exactamente como se sirven y convierte después si necesitas otro formato.

Escribe la marca de tiempo en UTC y en un formato estricto. Un precio que no puedes situar en el tiempo es difícil de comparar con un precio de cierre después, y las zonas horarias mezcladas corrompen una semana de registros cuidadosos en una tarde.

Añade en el cambio, no en el sondeo

El cursor since ya limita cada respuesta a los eventos que cambiaron, pero un evento cambiado aún contiene muchas selecciones sin cambios. Mantén un pequeño diccionario del último precio escrito por selección y escribe una fila solo cuando el valor difiera. Sin esa regla, el archivo crece con tu ritmo de sondeo en lugar del movimiento del mercado, y un bucle de todo el día escribe cientos de filas idénticas.

Un detalle: el diccionario empieza vacío tras un reinicio. Si quieres una deduplicación limpia entre reinicios, aliméntalo al arrancar leyendo el archivo y guardando la última fila de cada clave de selección.

El bucle, extendido

import csv, os, time, requests

BASE = os.environ["ODDS_BASE_URL"]
KEY = os.environ["ODDS_API_KEY"]
URL = BASE + "/kit/v1/markets"
s = requests.Session()
s.headers["x-portal-apikey"] = KEY

last_seen = {}   # (event_id, side) -> price
since = None

def rows(ev, seen_at):
    ml = ev.get("periods", {}).get("num_0", {}).get("money_line", {})
    for side in ("home", "draw", "away"):
        price = ml.get(side)
        if price is None:
            continue
        key = (ev["id"], side)
        if last_seen.get(key) != price:
            last_seen[key] = price
            yield [seen_at, ev["id"], ev.get("league_name", ""),
                   ev["home"], ev["away"], side, price]

with open("odds.csv", "a", newline="") as f:
    w = csv.writer(f)
    if f.tell() == 0:
        w.writerow(["observed_at", "event_id", "league", "home",
                    "away", "side", "price"])
    while True:
        q = {"sport_id": 2}
        if since is not None:
            q["since"] = since
        r = s.get(URL, params=q, timeout=10)
        if r.status_code == 429:
            time.sleep(int(r.headers.get("Retry-After", "5")))
            continue
        r.raise_for_status()
        data = r.json()
        since = data["last"]
        seen_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
        for ev in data["events"]:
            w.writerows(rows(ev, seen_at))
        f.flush()
        time.sleep(900)

La pausa de 900 segundos mantiene todo el día dentro de las 100 llamadas de la clave gratuita, el mismo presupuesto que el quickstart de Python usa para una prueba de todo el día. El nivel gratuito aparece como 20 llamadas por minuto y 100 al día en la página de precios, verificado el 2026-09-29. En un plan de pago, el mismo script corre sin cambios con una pausa más corta.

Dónde termina esto

CSV es un buen primer almacén y uno malo a largo plazo. No tiene transacciones, no tiene deduplicación entre reinicios a menos que recargues los últimos precios del archivo, y no tiene compresión. Cuando el archivo pase de unos cientos de miles de filas, o cuando más de un proceso necesite leerlo, mueve la misma lógica de solo cambios a una tabla. Hasta entonces, un archivo plano que abres en una hoja de cálculo es la forma más rápida de ver tu propio sondeo funcionando.