Propheseer Python SDK

Official Python SDK for the Propheseer prediction markets API. Access normalized data from Polymarket, Kalshi, and Gemini through a single, type-safe interface.

Installation

pip install propheseer

For WebSocket support (real-time market updates):

pip install propheseer[websocket]

Quick Start

from propheseer import Propheseer

client = Propheseer(
    api_key="pk_test_...",  # or set PROPHESEER_API_KEY env var
)

# List prediction markets
page = client.markets.list(source="polymarket", limit=10)
for market in page.data:
    print(f"{market.question}: {market.outcomes[0].probability:.0%}")

# Check your API key usage
result = client.keys.me()
print(f"Plan: {result.data.plan}, Daily usage: {result.data.usage.daily}")

Async Support

import asyncio
from propheseer import AsyncPropheseer

async def main():
    async with AsyncPropheseer(api_key="pk_test_...") as client:
        page = await client.markets.list(source="polymarket")
        for market in page.data:
            print(market.question)

asyncio.run(main())

Configuration

client = Propheseer(
    api_key="pk_test_...",                      # Required (or PROPHESEER_API_KEY env var)
    base_url="https://api.propheseer.com",      # Default
    timeout=30.0,                               # Request timeout in seconds (default: 30)
    max_retries=2,                              # Retries on 429/5xx errors (default: 2)
)

Resources

Markets

# List markets with filters
page = client.markets.list(
    source="polymarket",   # "polymarket" | "kalshi" | "gemini"
    category="politics",   # "politics" | "sports" | "finance" | ...
    status="open",
    q="election",          # search query
    limit=50,
    offset=0,
)

# Get a single market
result = client.markets.get("pm_12345")
market = result.data

# Auto-paginate through all markets
for market in client.markets.list_auto_paginate(source="kalshi"):
    print(market.question)

Categories

result = client.categories.list()
# [Category(id='politics', name='Politics', subcategories=['elections', ...]), ...]

Arbitrage (Pro+)

result = client.arbitrage.find(min_spread=0.05, category="politics")

for opp in result.data:
    print(f"{opp.question}: spread={opp.spread}, return={opp.potential_return}")

Unusual Trades (Pro+)

page = client.unusual_trades.list(
    reason="high_amount",
    since="2025-01-01T00:00:00Z",
    limit=20,
)

for trade in page.data:
    print(f"${trade.trade.usdc_value} {trade.trade.side} on \"{trade.market.question}\"")

# Auto-paginate
for trade in client.unusual_trades.list_auto_paginate():
    print(trade.detection.reason, trade.detection.anomaly_score)

History (Business+)

# Market price history
result = client.history.list(market_id="pm_12345", days=7)

# Available snapshot dates
result = client.history.dates()

Ticker (Public)

# No auth required
result = client.ticker.list(limit=10)

Pagination

Paginated endpoints return a SyncPage[T] (or AsyncPage[T]) object:

page = client.markets.list(limit=50)
print(page.data)          # list[Market]
print(page.meta.total)    # total matching markets
print(page.has_more())    # bool
print(page.next_offset()) # int | None

For automatic pagination:

for market in client.markets.list_auto_paginate(limit=50, max_items=200):
    # yields individual items, fetching new pages as needed
    pass

Error Handling

All API errors are raised as typed exceptions:

from propheseer import PermissionDeniedError, RateLimitError, AuthenticationError

try:
    client.arbitrage.find()
except PermissionDeniedError as err:
    print(f"Upgrade to {err.required_plan} plan")
except RateLimitError as err:
    print(f"Rate limited, retry after {err.retry_after}s")
except AuthenticationError:
    print("Invalid API key")

WebSocket (Real-Time Updates)

from propheseer import PropheseerWebSocket

ws = PropheseerWebSocket(
    api_key="pk_test_...",
    reconnect=True,              # Auto-reconnect (default: True)
    max_reconnect_attempts=5,    # Default: 5
    ping_interval=25.0,          # Keepalive interval (default: 25s)
)

@ws.on("connected")
def on_connected(msg):
    print(f"Connected: {msg['sessionId']} ({msg['plan']})")
    ws.subscribe(["pm_12345", "ks_67890"])

@ws.on("market_update")
def on_update(msg):
    print("Market update:", msg["market"])

@ws.on("error")
def on_error(err):
    print("Error:", err)

ws.connect()

# Later:
ws.unsubscribe(["pm_12345"])
ws.close()

Async WebSocket

import asyncio
from propheseer import AsyncPropheseerWebSocket

async def main():
    ws = AsyncPropheseerWebSocket(api_key="pk_test_...")

    ws.on("connected", lambda msg: print(f"Connected: {msg['sessionId']}"))
    ws.on("market_update", lambda msg: print("Update:", msg))

    await ws.connect()
    ws.subscribe(["pm_12345"])

    await asyncio.sleep(60)
    await ws.close()

asyncio.run(main())

Rate Limit Information

Every API response includes rate limit info:

page = client.markets.list()
print(page.rate_limit)
# RateLimitInfo(
#   plan='pro',
#   billing_type='subscription',
#   limit_day=10000,
#   remaining_day=9950,
#   limit_minute=100,
#   remaining_minute=98,
# )

Requirements

• Python >= 3.9
• httpx >= 0.24.0
• websockets >= 11.0 (optional, for WebSocket support)

License

MIT