api de odds da pinnacle / python / requests / websockets

API de odds da Pinnacle em Python

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

só os eventos alterados200 OK

Três scripts Python curtos para a API de odds da Pinnacle: um loop de polling que busca só o que mudou, um ouvinte de quedas de preço enviadas e um leitor WebSocket que vê cada frame. Defina as variáveis de ambiente e rode-os sem alterações.

Criar uma chave gratuita

Consulte com um cursor

O loop guarda last de cada resposta e o devolve como since, então depois da primeira chamada só chegam eventos alterados. Um 429 traz Retry-After, e o loop dorme exatamente esse tempo em vez de travar.

Com uma chamada a cada três segundos, as 100 chamadas diárias da chave gratuita duram cinco minutos. Consulte a cada 15 minutos para um teste de dia inteiro, ou passe para o poll com 10 chamadas 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

conforme os preços caemdata: [alerts]

Escute as quedas

Uma requisição HTTP que fica aberta. O servidor escreve uma linha data: para cada lote de alertas. Um objeto JSON em vez de uma lista é uma mensagem de controle: a saudação connected, ou plan_lacks_sse quando o plano da chave não tem stream.

O stream não tem filtro de esporte, então verifique sport no loop se quiser só alguns esportes. min_drop vai até 1.

Requer push, poll + push ou 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>

subscribeem até 5 s

depois atualizaçõessnapshot

Leia cada frame

O socket espera uma mensagem de subscribe em até cinco segundos após a conexão. O primeiro frame de cada stream é um snapshot; cada frame seguinte atualiza essa baseline. Responda a cada ping com um pong.

Frames perdidos não são reenviados. Após uma desconexão, conecte de novo e comece do novo snapshot. Você pode se inscrever por esporte ou por até 200 IDs de evento por stream.

Requer o complemento WebSocket em poll, poll + push ou 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 executá-los

Duas bibliotecas, três variáveis

Crie uma chave. A gratuita basta para o poll.py.
Instale requests e websockets, depois exporte a chave e os dois endereços listados na documentação.
Execute um script. Cada um imprime no terminal; substitua o print pela sua própria lógica de armazenamento ou de sinal.

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

De qual plano cada script precisa.

Script	Funciona em
poll.py	Todos os planos, incluindo a chave gratuita
drops.py	push, poll + push, bulk
ws.py	poll, poll + push ou bulk com o complemento WebSocket

Script

Funciona em

poll.py

Todos os planos, incluindo a chave gratuita

drops.py

push, poll + push, bulk

ws.py

poll, poll + push ou bulk com o complemento WebSocket

PRÓXIMOpython poll.py

Cole, exporte, execute

A chave gratuita roda o script de polling hoje. Quando as quedas ou o socket conquistarem seu lugar no seu código, a página de preços mostra qual plano os adiciona.

Criar uma chave gratuita Leia a documentação da API