Salve preços da API de odds da Pinnacle em CSV com Python

O quickstart de Python deste site faz polling na API de odds da Pinnacle e imprime a linha money line. A próxima pergunta que a maioria dos leitores faz é como guardar o que buscou. Um arquivo CSV basta para um primeiro armazenamento: uma linha por seleção por mudança, anexada a cada poll, legível pelo Excel, pandas ou uma importação de banco de dados depois.

O que uma linha deve conter

Guarde a observação, não só o preço. Uma linha útil nomeia o evento (id, liga, times), as coordenadas do mercado (período, mercado, lado) e a própria observação: o preço decimal e a hora em que você o viu. Some o cursor last da resposta se quiser rastrear qual poll gravou a linha. Os preços chegam como odds decimais, então guarde-os exatamente como servidos e converta depois, se precisar de outro formato.

Escreva o timestamp em UTC e em um formato estrito. Um preço que você não consegue posicionar no tempo é difícil de comparar com um preço de fechamento depois, e fusos misturados corrompem uma semana de registros cuidadosos em uma tarde.

Anexe na mudança, não no poll

O cursor since já limita cada resposta aos eventos que mudaram, mas um evento alterado ainda contém muitas seleções sem mudança. Mantenha um pequeno dicionário do último preço gravado por seleção e escreva uma linha só quando o valor diferir. Sem essa regra, o arquivo cresce com a sua taxa de poll em vez do movimento do mercado, e um loop de dia inteiro grava centenas de linhas idênticas.

Um detalhe: o dicionário começa vazio após um reinício. Se você quer deduplicação limpa entre reinícios, alimente-o na inicialização lendo o arquivo e guardando a última linha de cada chave de seleção.

O loop, estendido

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)

A pausa de 900 segundos mantém o dia inteiro dentro das 100 chamadas da chave gratuita, o mesmo orçamento que o quickstart de Python usa para um teste de dia inteiro. O nível gratuito aparece como 20 chamadas por minuto e 100 por dia na página de preços, verificado em 2026-09-29. Num plano pago, o mesmo script roda sem mudanças com uma pausa menor.

Onde isso para

CSV é um bom primeiro armazenamento e um mau armazenamento de longo prazo. Não tem transações, não tem deduplicação entre reinícios a menos que você recarregue os últimos preços do arquivo, e não tem compressão. Quando o arquivo passar de algumas centenas de milhares de linhas, ou quando mais de um processo precisar lê-lo, mova a mesma lógica de mudança para uma tabela. Até lá, um arquivo simples que você abre numa planilha é o jeito mais rápido de ver o seu próprio polling funcionando.