api de cuotas de pinnacle / python / requests / websockets

API de cuotas de Pinnacle en Python

GET/kit/v1/markets?sport_id=2&since=141

solo eventos cambiados200 OK

Tres scripts cortos de Python para la API de cuotas de Pinnacle: un bucle de sondeo que trae solo lo que cambió, un oyente de caídas de precio enviadas y un lector WebSocket que ve cada frame. Configura las variables de entorno y ejecútalos sin cambios.

Crear una clave gratuita

Sondea con un cursor

El bucle guarda last de cada respuesta y lo devuelve como since, así que tras la primera llamada solo llegan eventos cambiados. Un 429 lleva Retry-After, y el bucle duerme exactamente ese tiempo en lugar de fallar.

A una llamada cada tres segundos, las 100 llamadas diarias de la clave gratuita duran cinco minutos. Sondea cada 15 minutos para una prueba de todo el día, o pásate a poll para 10 llamadas por segundo.

poll.py

import os
import time
import 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


def money_line(ev):
    p = ev.get("periods", {})
    return p.get("num_0", {}).get(
        "money_line")


def poll(sport_id, every=3.0):
    since = None
    while True:
        q = {"sport_id": sport_id}
        if since is not None:
            q["since"] = since
        r = s.get(URL, params=q,
                  timeout=10)
        if r.status_code == 429:
            wait = r.headers.get(
                "Retry-After", "5")
            time.sleep(int(wait))
            continue
        r.raise_for_status()
        data = r.json()
        since = data["last"]
        for ev in data["events"]:
            ml = money_line(ev)
            if ml:
                print(ev["home"], "v",
                      ev["away"], ml)
        time.sleep(every)


poll(sport_id=2)  # 2 = tennis

GET/odds-drop?min_drop=5

a medida que caen los preciosdata: [alerts]

Escucha las caídas

Una solicitud HTTP que permanece abierta. El servidor escribe una línea data: por cada lote de alertas. Un objeto JSON en lugar de una lista es un mensaje de control: el saludo connected, o plan_lacks_sse cuando el plan de la clave no tiene stream.

El stream no tiene filtro de deporte, así que comprueba sport en el bucle si solo quieres algunos deportes. min_drop baja hasta 1.

Necesita push, poll + push o bulk.

drops.py

import json
import os
import requests

BASE = os.environ["ODDS_BASE_URL"]
KEY = os.environ["ODDS_API_KEY"]
HDR = {"Accept": "text/event-stream"}

r = requests.get(
    BASE + "/odds-drop",
    params={"min_drop": 5, "key": KEY},
    headers=HDR,
    stream=True,
    timeout=(10, None),
)
r.raise_for_status()

for raw in r.iter_lines():
    if not raw.startswith(b"data:"):
        continue
    msg = json.loads(raw[5:])
    if isinstance(msg, dict):
        # connected, or an error such
        # as plan_lacks_sse
        print("control:", msg)
        continue
    for a in msg:
        print(a["sport"], a["home"],
              a["outcome"],
              a["from_price"], "->",
              a["to_price"],
              "fair", a["nvp"])

WSS/ws/feed?key=<your key>

subscribeen menos de 5 s

luego actualizacionessnapshot

Lee cada frame

El socket espera un mensaje de suscripción en los cinco segundos siguientes a la conexión. El primer frame de cada stream es un snapshot; cada frame posterior actualiza esa línea base. Responde a cada ping con un pong.

Los frames perdidos no se repiten. Tras una desconexión, conecta de nuevo y empieza desde el snapshot nuevo. Puedes suscribirte por deporte o por hasta 200 IDs de evento por stream.

Necesita el complemento WebSocket en poll, poll + push o bulk.

ws.py

import asyncio
import json
import os
import websockets

WS = os.environ["ODDS_WS_URL"]
KEY = os.environ["ODDS_API_KEY"]
PONG = json.dumps({"type": "pong"})
SUB = json.dumps({
    "type": "subscribe",
    "streams": ["live"],
    "sport_ids": [1, 2],
})


async def handle(ws, f):
    kind = f.get("type")
    if kind == "ping":
        await ws.send(PONG)
    elif kind == "snapshot":
        ev = f.get("events", [])
        print("baseline", len(ev))
    else:
        print(kind, f)


async def main():
    url = f"{WS}?key={KEY}"
    ws = await websockets.connect(url)
    await ws.send(SUB)  # within 5 s
    async for raw in ws:
        f = json.loads(raw)
        await handle(ws, f)


asyncio.run(main())

antes de ejecutarlos

Dos librerías, tres variables

Crea una clave. La gratuita basta para poll.py.
Instala requests y websockets, luego exporta la clave y las dos direcciones que lista la documentación.
Ejecuta un script. Cada uno imprime en la terminal; sustituye el print por tu propia lógica de almacenamiento o de señal.

terminal

pip install requests websockets
export ODDS_API_KEY="your key"
export ODDS_BASE_URL="https://..."
export ODDS_WS_URL="wss://.../ws/feed"
python poll.py

Qué plan necesita cada script.

Script	Funciona en
poll.py	Todos los planes, incluida la clave gratuita
drops.py	push, poll + push, bulk
ws.py	poll, poll + push o bulk con el complemento WebSocket

Script

Funciona en

poll.py

Todos los planes, incluida la clave gratuita

drops.py

push, poll + push, bulk

ws.py

poll, poll + push o bulk con el complemento WebSocket

SIGUIENTEpython poll.py

Pega, exporta, ejecuta

La clave gratuita ejecuta hoy el script de sondeo. Cuando las caídas o el socket se ganen su sitio en tu código, la página de precios muestra qué plan los añade.

Crear una clave gratuita Lee la documentación de la API