Cross-Platform Price Monitor
Compare live best bids across Kalshi, Polymarket, and Outpoll
What this script does
#!/usr/bin/env python3
"""Compare best bids across Kalshi, Polymarket, and Outpoll in real time."""
import asyncio, json, time
import websockets
from cryptography.hazmat.primitives import serialization, hashes
from cryptography.hazmat.primitives.asymmetric import padding
import base64
# ─── Configuration ───────────────────────────────────────────────────
# Kalshi (requires RSA key pair — see https://docs.kalshi.com)
KALSHI_KEY_ID = "your_kalshi_api_key_id"
KALSHI_PRIVATE_KEY_PATH = "path/to/kalshi_private_key.pem"
KALSHI_MARKET_TICKER = "KXBTC-25DEC31-T100000" # example: BTC $100k by end 2025
KALSHI_WS = "wss://api.elections.kalshi.com/trade-api/ws/v2"
# Polymarket (no auth for market data)
POLYMARKET_ASSET_ID = "12345..." # YES token asset_id from gamma-api.polymarket.com/markets
POLYMARKET_WS = "wss://ws-subscriptions-clob.polymarket.com/ws/market"
# Outpoll (no auth for order book)
OUTPOLL_EVENT_ID = "54ccea1a-16fd-469c-8018-84b375243e8a"
OUTPOLL_OUTCOME_ID = "b21f6fd8-b9d1-4b9f-bb79-ef141e3dcb76" # YES runner
OUTPOLL_WS = "wss://order-book.outpoll.com/event/ws"
# ─── Shared state ────────────────────────────────────────────────────
prices = {
"Kalshi": None,
"Polymarket": None,
"Outpoll": None,
}
def display():
"""Print the current best bids."""
print("\033[H\033[J", end="") # clear terminal
print("═══ Cross-Platform Best Bid ═══\n")
for platform, bid in prices.items():
if bid is not None:
print(f" {platform:12s} Best Bid: ${bid:.2f}")
else:
print(f" {platform:12s} Best Bid: waiting...")
print(f"\n Updated: {time.strftime('%H:%M:%S')}")
# ─── Kalshi ──────────────────────────────────────────────────────────
async def watch_kalshi():
with open(KALSHI_PRIVATE_KEY_PATH, "rb") as f:
private_key = serialization.load_pem_private_key(f.read(), password=None)
# Sign handshake
ts = str(int(time.time() * 1000))
msg_to_sign = ts + "GET" + "/trade-api/ws/v2"
sig = private_key.sign(
msg_to_sign.encode(),
padding.PSS(mgf=padding.MGF1(hashes.SHA256()), salt_length=padding.PSS.DIGEST_LENGTH),
hashes.SHA256(),
)
headers = {
"KALSHI-ACCESS-KEY": KALSHI_KEY_ID,
"KALSHI-ACCESS-SIGNATURE": base64.b64encode(sig).decode(),
"KALSHI-ACCESS-TIMESTAMP": ts,
}
async with websockets.connect(KALSHI_WS, additional_headers=headers) as ws:
# Subscribe to orderbook
await ws.send(json.dumps({
"id": 1,
"cmd": "subscribe",
"params": {"channels": ["orderbook_delta"], "market_ticker": KALSHI_MARKET_TICKER},
}))
yes_book = {} # price_str -> size
async for raw in ws:
data = json.loads(raw)
msg_type = data.get("type")
if msg_type == "orderbook_snapshot":
yes_book = {p: float(s) for p, s in data["msg"].get("yes_dollars_fp", [])}
elif msg_type == "orderbook_delta" and data["msg"]["side"] == "yes":
p = data["msg"]["price_dollars"]
delta = float(data["msg"]["delta_fp"])
yes_book[p] = yes_book.get(p, 0) + delta
if yes_book[p] <= 0:
yes_book.pop(p, None)
if yes_book:
prices["Kalshi"] = float(max(yes_book, key=float))
display()
# ─── Polymarket ──────────────────────────────────────────────────────
async def watch_polymarket():
async with websockets.connect(POLYMARKET_WS) as ws:
await ws.send(json.dumps({
"assets_ids": [POLYMARKET_ASSET_ID],
"type": "market",
"custom_feature_enabled": True,
}))
# Heartbeat
async def heartbeat():
while True:
await asyncio.sleep(10)
try:
await ws.send("PING")
except Exception:
break
hb = asyncio.create_task(heartbeat())
try:
async for raw in ws:
if raw == "PONG":
continue
events = json.loads(raw)
if not isinstance(events, list):
events = [events]
for event in events:
if event.get("event_type") == "book":
bids = event.get("bids", [])
if bids:
prices["Polymarket"] = float(max(bids, key=lambda x: float(x["price"]))["price"])
display()
elif event.get("event_type") == "price_change":
for pc in event.get("price_changes", []):
if pc.get("best_bid"):
prices["Polymarket"] = float(pc["best_bid"])
display()
finally:
hb.cancel()
# ─── Outpoll ─────────────────────────────────────────────────────────
async def watch_outpoll():
async with websockets.connect(OUTPOLL_WS) as ws:
await ws.send(json.dumps({"t": "SUBSCRIBE", "e": OUTPOLL_EVENT_ID}))
async for raw in ws:
data = json.loads(raw)
if data["t"] not in ("PRICES_DATA", "PRICES_DATA_DELTA"):
continue
for p in data["d"]:
if p["o"] == OUTPOLL_OUTCOME_ID:
prices["Outpoll"] = p["b"]
display()
# ─── Main ────────────────────────────────────────────────────────────
async def main():
display()
await asyncio.gather(
watch_kalshi(),
watch_polymarket(),
watch_outpoll(),
)
asyncio.run(main())Example output
Related pages
Last updated
Was this helpful?