用 Python 监控 Kalshi 和 Polymarket

Kalshi 和 Polymarket 各自发布独立的目录和定价模型，并各有自己的标识符。一个有用的跨平台监控程序，必须先确认两条挂牌指向同一个主张，才能把它们的价格放进同一行。Dino 负责处理这一匹配步骤，返回一个同时包含两个平台价格的 Market 对象，因此 Python 程序可以专注于筛选和展示。

本教程使用官方的 dino-markets 包构建一个小型终端监控程序。它会对 Free 套餐提供的延迟 REST 快照中已开放的匹配市场进行排序。最后一节会加入实时数据流，实现事件发生时的即时更新。

安装客户端并创建密钥

创建一个免费 API 密钥，并将它保存到环境变量中。然后安装 Python 包：

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

客户端会自动读取 DINO_API_KEY。第一个请求会拉取已开放且实时的 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 分钟。这使得结果适合用于目录浏览和研究。它也可以在数据流更新到达之前，为本地视图提供一个起点。

在每个结果上展示两个平台的价格

每个 Market 在体育和非体育类别中都保持相同的 outcomes[] 结构。每个结果都带有两个平台的价格，以及一个 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 中各个结果的一种观测值，并不意味着存在可执行的套利。已确认的 Opportunity 位于单独的 client.find_arbitrage() 资源中，包含已选定的环节、费用、规模和快照新鲜度。

在使用某一行之前检查结算信息

列表响应是精简的。当监控程序需要平台条款和结算风险时，可以通过稳定的 id 拉取单个 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 的稳定标识符，是缓存行的正确键。当某条挂牌被重新分析时，展示标题和平台标识符可能会发生变化。当 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 对象参考文档涵盖了这里用到的每个字段。

完成后的监控程序会读取匹配数据并展示价格差异。它不会在任何一个平台上下单。这是开发者数据，不构成交易或金融建议。创建一个免费密钥，先运行 REST 版本。