ガイド · Python · Node

PythonとNodeでピナクルのオッズをWebSocketからリアルタイム取得

更新日 2026年11月9日 · PinnFeed · 約3分で読めます

このガイドでは、ピナクルの価格の最新コピーをメモリ上に保持するクライアントを作ります。接続、購読、スナップショットの読み込み、各デルタの適用、ハートビートへの応答、自動再接続までを行います。ほかのガイド（スチームのアラート、バリューベット、Telegramボット）はこのクライアントを土台にしています。

WebSocketプランであるBuilder（月額99ユーロ、3日間無料）のPinnFeedキーが必要です。FreeまたはHobbyのキーでは、ドキュメント（英語）のRESTの例をお試しいただけます。

ステップ 1 — キーを取得して環境変数に設定する#

ダッシュボードでキーを作成し、コードに直接書き込まないよう環境変数としてエクスポートします。

export PINNFEED_KEY="ps_live_..."

ステップ 2 — フレームを理解する#

接続して subscribe を送ると、サーバーは次のものを送信します。

subscribed、続いてフィルターに一致するすべてのイベントとマーケットを含む、分割されたスナップショット（part、parts、final）。
デルタフレーム。それぞれに changes のリストがあり、upsert は新しいオブジェクト全体を、remove はそのIDだけを含みます。
20秒ごとのping。{"type":"pong"} で応答してください。60秒間応答がないと、サーバーはソケットを閉じます。

マーケットは id（eventId:key）で識別されるため、状態の管理には辞書が1つあれば十分です。各結果には decimal、american、fair（マージンなしのフェアオッズ）、fairProb が含まれます。すべてのフィールドはWebSocketリファレンス（英語）に記載しています。

ステップ 3 — Pythonのクライアント#

唯一の依存関係を pip install websockets（Python 3.9以上）でインストールします。pinnfeed_stream.py として保存してください。スチームとTelegramのガイドでこのファイルをインポートします。

# pinnfeed_stream.py: PinnFeedのWebSocketから、ピナクルのオッズブックの最新コピーを保持します。
# pip install websockets   (Python 3.9+)
import asyncio, json, os, random
import websockets

WS_URL = os.environ.get("PINNFEED_WS", "wss://pinnfeed.io/v1/stream")
KEY = os.environ["PINNFEED_KEY"]
SIDES = {"home", "away", "draw", "over", "under"}


class Book:
    """ID別のイベントとマーケット。スナップショットとデルタのフレームで更新します。"""

    def __init__(self):
        self.events, self.markets = {}, {}

    def apply(self, msg, on_market=None):
        if msg["type"] == "snapshot":
            for e in msg["events"]:
                self.events[e["id"]] = e
            for m in msg["markets"]:
                self.markets[m["id"]] = m
        elif msg["type"] == "delta":
            for c in msg["changes"]:
                if c["kind"] == "event" and c["op"] == "upsert":
                    self.events[c["event"]["id"]] = c["event"]
                elif c["kind"] == "event" and c["op"] == "remove":
                    self.events.pop(c["id"], None)
                    for mid in [k for k, m in self.markets.items() if m["eventId"] == c["id"]]:
                        del self.markets[mid]
                elif c["kind"] == "market" and c["op"] == "upsert":
                    old = self.markets.get(c["market"]["id"])
                    self.markets[c["market"]["id"]] = c["market"]
                    if on_market:
                        on_market(old, c["market"], self.events.get(c["market"]["eventId"]))
                elif c["kind"] == "market" and c["op"] == "remove":
                    self.markets.pop(c["id"], None)


async def stream(subscription, on_market=None, on_ready=None):
    """接続・購読してオッズブックを最新に保ち、バックオフ付きで無期限に再接続します。"""
    delay = 1
    while True:
        book = Book()
        try:
            async with websockets.connect(WS_URL, additional_headers={"x-api-key": KEY}, max_size=None) as ws:
                await ws.send(json.dumps({"type": "subscribe", **subscription}))
                async for raw in ws:
                    msg = json.loads(raw)
                    if msg["type"] == "ping":
                        await ws.send('{"type":"pong"}')
                    elif msg["type"] == "error":
                        print("server error:", msg["code"], msg["message"])
                    elif msg["type"] in ("snapshot", "delta"):
                        book.apply(msg, on_market if msg["type"] == "delta" else None)
                        if msg["type"] == "snapshot" and msg.get("final") and on_ready:
                            on_ready(book)
                        delay = 1
        except (OSError, websockets.ConnectionClosed) as e:
            print(f"disconnected ({e}); reconnecting in {delay}s")
        await asyncio.sleep(delay + random.random())
        delay = min(delay * 2, 30)


if __name__ == "__main__":
    def ready(book):
        print(f"book ready: {len(book.events)} events, {len(book.markets)} markets")

    def moved(old, new, event):
        if old and event:
            for o_old, o_new in zip(old["outcomes"], new["outcomes"]):
                if o_new["name"] in SIDES and o_old["decimal"] != o_new["decimal"]:
                    print(f'{event["home"]} v {event["away"]} {new["type"]} {o_new["name"]}: '
                          f'{o_old["decimal"]} -> {o_new["decimal"]} (fair {o_new["fair"]})')

    asyncio.run(stream({"sports": ["soccer"], "markets": ["moneyline"], "main": True, "specials": False}, moved, ready))

python pinnfeed_stream.py で実行すると、book ready: N events, M markets と表示され、その後は価格が動くたびに1行ずつ出力されます。

ステップ 4 — Nodeのクライアント#

同じ処理をNode 18以上と npm install ws で実装します。stream.mjs として保存し、node stream.mjs を実行します。

// stream.mjs: PinnFeedのWebSocketから取得するピナクルのライブオッズブック（Node 18以上）。
// npm install ws
import WebSocket from 'ws';

const WS_URL = process.env.PINNFEED_WS || 'wss://pinnfeed.io/v1/stream';
const KEY = process.env.PINNFEED_KEY;
const SIDES = new Set(['home', 'away', 'draw', 'over', 'under']);
const events = new Map();
const markets = new Map();

function connect(delay = 1000) {
  const ws = new WebSocket(WS_URL, { headers: { 'x-api-key': KEY }, perMessageDeflate: true });
  ws.on('open', () => {
    delay = 1000;
    ws.send(JSON.stringify({ type: 'subscribe', sports: ['soccer', 'tennis'], markets: ['moneyline'], main: true, specials: false }));
  });
  ws.on('message', (raw) => {
    const msg = JSON.parse(raw);
    if (msg.type === 'ping') return ws.send('{"type":"pong"}');
    if (msg.type === 'error') return console.error('server error:', msg.code, msg.message);
    if (msg.type === 'snapshot') {
      if (msg.part === 1) { events.clear(); markets.clear(); }
      msg.events.forEach((e) => events.set(e.id, e));
      msg.markets.forEach((m) => markets.set(m.id, m));
      if (msg.final) console.log(`book ready: ${events.size} events, ${markets.size} markets`);
    }
    if (msg.type === 'delta') {
      for (const c of msg.changes) {
        if (c.kind === 'event') c.op === 'upsert' ? events.set(c.event.id, c.event) : events.delete(c.id);
        if (c.kind === 'market' && c.op === 'remove') markets.delete(c.id);
        if (c.kind === 'market' && c.op === 'upsert') {
          const old = markets.get(c.market.id);
          markets.set(c.market.id, c.market);
          const ev = events.get(c.market.eventId);
          if (!old || !ev) continue;
          c.market.outcomes.forEach((o, i) => {
            const prev = old.outcomes[i];
            if (prev && SIDES.has(o.name) && prev.decimal !== o.decimal) console.log(`${ev.home} v ${ev.away} ${o.name}: ${prev.decimal} -> ${o.decimal} (fair ${o.fair})`);
          });
        }
      }
    }
  });
  ws.on('close', (code) => {
    const wait = Math.min(delay * 2, 30000);
    console.log(`closed (${code}); reconnecting in ${wait / 1000}s`);
    setTimeout(() => connect(wait), wait + Math.random() * 1000);
  });
  ws.on('error', (e) => console.error('ws error:', e.message));
}
connect();

ステップ 5 — 購読を絞り込む#

フィルターはサーバー側で適用されるため、指定したものだけを受信・処理します。

{ "type": "subscribe", "sports": ["tennis"], "live": true, "markets": ["moneyline", "total"], "main": true }

sports：スラッグまたはID。leagues と events：IDのリスト。
live：true でライブのみ、false でプレマッチのみ。省略すると両方です。
main: true にすると、各イベントの主要なマネーライン、ハンディキャップ、トータルだけに絞られます。
subscribe を再送すると購読内容が置き換わり、新しいスナップショットが送られます。

本番運用のチェックリスト#

バックオフとジッターで再接続する（どちらのクライアントも1秒、2秒、4秒…最大30秒で実装済み）。新しいスナップショットから再構築します。
圧縮を有効にする：Pythonの websockets はデフォルトで permessage-deflate をネゴシエートします。Nodeでは perMessageDeflate: true を指定してください。スナップショットは5〜10分の1に小さくなります。
素早く読み取る。 処理の遅いクライアントに16MBを超えるデータが溜まると、サーバーはコード 4008 で切断します。重い処理はソケットのループの外で行ってください。
seq はログ用に使い、欠落の検出には使わない：購読していないバッチはフィルターによってスキップされるためです。

よくある質問#

WebSocketにはどのプランが必要ですか？#

Builder（月額99ユーロ）以上です。FreeとHobbyのキーではWebSocketで 403 plan_lacks_ws が返されますが、RESTはご利用いただけます。

接続は何本まで開けますか？#

Builderでは1アカウントにつきWebSocket接続3本、Proでは10本です。

一部の結果にhomeやawayではなく参加者IDが付いているのはなぜですか？#

スペシャルや一部の長期ベットでは、ホームとアウェイではなく参加者が一覧になります。試合のラインだけが必要な場合は、例と同じように "specials": false で購読し、home、away、draw、over、under で絞り込んでください。