PythonでKalshiとPolymarketを監視する

KalshiとPolymarketは、それぞれ独自の識別子を持つ別々のカタログとプライシングモデルを公開しています。有用なvenue間モニターは、2つの掲載が同じ主張を指していることを確認してから、価格を1行にまとめる必要があります。Dinoはこのマッチング処理を担い、両venueの価格を持つ1つのMarketオブジェクトを返すため、Pythonプログラムはフィルタリングと表示に専念できます。

このチュートリアルでは、公式の dino-markets パッケージを使って小さなターミナルモニターを構築します。Freeプランで利用できる遅延RESTスナップショットから、オープン中のマッチング済みmarketをランキング表示します。最後のセクションでは、発生した時点で更新を受け取るリアルタイムストリームを追加します。

クライアントをインストールしてキーを作成する

無料APIキーを作成し、環境変数に保存します。次にPythonパッケージをインストールします。

export DINO_API_KEY=sk_live_...
pip install dino-markets

クライアントは DINO_API_KEY を自動的に読み込みます。最初のリクエストでは、観測されたvenue間価格差が大きい順に並んだ、オープンおよびライブのMarketカタログを取得します。

from dino_markets import Dino

client = Dino()

catalog = client.markets(
    status="open,live",
    sort="-spread_pts",
    limit=25,
)

ranked = [
    market
    for market in catalog["markets"]
    if market["spread_pts"] is not None
]
if not ranked:
    raise SystemExit("No priced matched markets are available right now.")

print("as returned by the delayed REST snapshot")
for market in ranked:
    print(
        f'{market["spread_pts"]:>5.1f} pts  '
        f'{market["title"]:<42} '
        f'{market["category"]}/{market["market_type"]}'
    )

RESTの価格は、どのプランでもライブから約2分遅れて保持されます。そのため、この結果はカタログの閲覧やリサーチに適しています。ストリームの更新が届く前に、ローカルビューを構築する起点としても使えます。

outcomeごとに両venueの価格を表示する

すべてのMarketは、スポーツ・非スポーツのカテゴリを問わず同じ outcomes[] の形式を保ちます。各outcomeは両venueの価格に加えて cheaper フィールドを持ちます。価格が欠けている場合は None になるため、差分を計算する前に明示的にフィルタリングします。

def priced_outcomes(market):
    return [
        outcome
        for outcome in market["outcomes"]
        if outcome["kalshi"] is not None
        and outcome["polymarket"] is not None
    ]

for market in ranked[:10]:
    print(f'\n{market["title"]}')
    for outcome in priced_outcomes(market):
        gap = abs(outcome["kalshi"] - outcome["polymarket"]) * 100
        print(
            f'  {outcome["name"]:<24} '
            f'Kalshi {outcome["kalshi"]:.2f}  '
            f'Polymarket {outcome["polymarket"]:.2f}  '
            f'gap {gap:.1f} pts'
        )

spread_pts はMarketのoutcome間の観測値です。実行可能なアービトラージが存在することを意味するわけではありません。確定済みのOpportunityは別の client.find_arbitrage() リソースにあり、選択されたレグ、手数料、サイズ、スナップショットの鮮度を持ちます。

行を使う前に解決を確認する

リスト応答は最小限の情報のみです。モニターがvenueの規約や解決のリスクを必要とする場合は、安定した id を使って1つのMarketを取得します。

selected = ranked[0]
detail = client.market(selected["id"])

print(detail["title"], detail["match"], detail["signal"])
for risk in detail.get("settlement", {}).get("risks", []):
    print(risk["severity"], risk["event"], risk["note"])

id はDinoの安定した識別子であり、キャッシュされた行のキーとして適切です。表示タイトルとvenueの識別子は、掲載が再分析されると変わることがあります。updated_at が変わったら詳細を再取得し、更新された識別情報や解決の開示内容をモニターに反映させます。

リアルタイム更新を追加する

オプションのストリーム依存パッケージをインストールし、SDKの再接続ヘルパーを使います。

pip install "dino-markets[stream]"

import asyncio
from dino_markets import Dino, watch

client = Dino()

def on_publication(channel, frame):
    if frame.get("type") != "price":
        return
    print(
        channel,
        frame["id"],
        frame.get("spread_pts"),
        frame.get("priced_at"),
    )

asyncio.run(watch(client, on_publication))

このヘルパーは接続の試行ごとに新しい短命チケットを発行し、短時間の切断後には直近の配信内容を復元します。Freeのキーはキュレーションされた sample チャンネルを受け取ります。有料プランはすべて、Marketストリーム全体を受け取ります。

WebSocket更新を追加した後も、プログラムにRESTでの起点取得処理は残しておきます。これによりモニターは開始時点のスナップショットを得られ、ストリームのフレームはキャッシュされたMarketのidに対して現在の価格変化を適用します。WebSocketリファレンスは再接続の挙動を、Marketオブジェクトリファレンスはここで使われるすべてのフィールドを解説しています。

完成したモニターはマッチング済みデータを読み取り、価格差を表示します。どちらのvenueにも発注は行いません。これは開発者向けデータであり、投資助言や取引助言ではありません。まずは無料キーを作成して、REST版を実行してください。