pinnacle odds api / python / requests / websockets

Pinnacle odds API in Python

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

changed events only200 OK

Three short Python scripts for the Pinnacle odds API: a polling loop that fetches only what changed, a listener for pushed price drops, and a WebSocket reader that sees every frame. Set the environment variables and run them unchanged.

Create a free key

Poll with a cursor

The loop keeps last from each response and sends it back as since, so after the first call only changed events arrive. A 429 carries Retry-After, and the loop sleeps exactly that long instead of crashing.

At one call every three seconds the free key's 100 daily calls last five minutes. Poll every 15 minutes for an all-day test, or move to poll for 10 calls a second.

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

as prices falldata: [alerts]

Listen for drops

One HTTP request that stays open. The server writes a data: line for each batch of alerts. A JSON object instead of a list is a control message: the connected greeting, or plan_lacks_sse when the key's plan has no stream.

The stream has no sport filter, so check sport in the loop if you only want some sports. min_drop goes down to 1.

Needs push, poll + push or 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>

subscribewithin 5 s

then updatessnapshot

Read every frame

The socket expects a subscribe message within five seconds of connecting. The first frame for each stream is a snapshot; every frame after it updates that baseline. Answer each ping with a pong.

Missed frames are not replayed. After a disconnect, connect again and start from the new snapshot. You can subscribe by sport or by up to 200 event IDs per stream.

Needs the WebSocket add-on on poll, poll + push or 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())

before you run them

Two libraries, three variables

Create a key. The free one is enough for poll.py.
Install requests and websockets, then export the key and the two addresses listed in the docs.
Run a script. Each prints to the terminal; replace the print with your own storage or signal logic.

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

Which plan each script needs.

Script	Works on
poll.py	Every plan, including the free key
drops.py	push, poll + push, bulk
ws.py	poll, poll + push or bulk with the WebSocket add-on

Script

Works on

poll.py

Every plan, including the free key

drops.py

push, poll + push, bulk

ws.py

poll, poll + push or bulk with the WebSocket add-on

NEXTpython poll.py

Paste, export, run

The free key runs the polling script today. When the drops or the socket earn their place in your code, the pricing page shows which plan adds them.

Create a free key Read the API docs