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를 자동으로 읽습니다. 첫 번째 요청은 관측된 거래소 간 가격 차이가 가장 큰 순서로 정렬된 open 및 live 상태의 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 버전을 먼저 실행하려면 무료 키 생성하기.