SDK · Python

Python SDK

heisenberg — a typed, read-only client. Python 3.9+. The REST client uses only the standard library; streaming uses an optional extra.

Install

PyPI

pip install heisenberg_ai
pip install "heisenberg_ai[stream]"   # optional, for streaming

Authenticate

Pass a token or set HEISENBERG_TOKEN. The token is only sent as a Bearer header and never logged.

Python

from heisenberg import Heisenberg

hb = Heisenberg(token="…")  # or HEISENBERG_TOKEN

Venues

Python

markets = hb.polymarket.markets({"min_volume": 1000, "closed": False}).page()
book    = hb.polymarket.orderbook({"token_id": "…"}).page()
kalshi  = hb.kalshi.markets({"status": "open"}).page()
hl      = hb.hyperliquid.wallet_trades({"address": "0x…"}).page()

hb.polymarket.supports("orderbook")  # True
hb.kalshi.supports("orderbook")      # False

Intelligence

Python

profile = hb.wallets.profile("0xabc…", {"window_days": 15}).page()  # Wallet 360
pnl     = hb.wallets.pnl("0xabc…", {"granularity": "1w"}).page()
elite   = hb.leaderboards.heisenberg({"sort_by": "h_score", "limit": 10}).page()
pulse   = hb.smart_money.pulse("0xcondition…").page()    # gated (Sport)
clv     = hb.sports.closing_line("0xcondition…").page()  # gated (Sport)

Anything not wrapped is reachable via hb.query(agent_id, params). Discover everything with hb.catalog(). Every method maps to an endpoint — see the method → endpoint reference for links into the API reference (params and response fields for each agent).

Pagination

Call .page() for one page, or iterate a Query to stream every row across all pages.

Python

# one page
page = hb.polymarket.trades({"condition_id": "0x…"}).page()
print(page.results, page.pagination["has_more"])

# every row, auto-paginated
for trade in hb.polymarket.trades({"condition_id": "0x…"}):
    ...

# collect all (use with care on large sets)
rows = hb.polymarket.trades({"condition_id": "0x…"}).all()

Streaming

Python (requires heisenberg_ai[stream])

for trade in hb.stream.trades({"condition_id": "0x…"}):
    print(trade.get("price"), trade.get("side"))
# reconnects with backoff and re-subscribes automatically on disconnect

Errors & gated endpoints

Sport/audience endpoints need a subscription and raise a typed PlanError that preserves the API's message and names the plan. Other failures map to AuthError, RateLimitError, ValidationError, ApiError — all carry a request_id.

Python

from heisenberg import PlanError

try:
    hb.sports.closing_line("0x…").page()
except PlanError as e:
    print(str(e), "→ unlock:", e.required_plan)

# check ahead of time
hb.requires(602)  # "sport"

Options

Python

Heisenberg(
    token=None,        # or HEISENBERG_TOKEN
    timeout=30.0,      # seconds
    max_retries=2,     # 429/5xx/network
    base_url=None,     # override the API base
    transport=None,    # inject a transport (tests)
)

→ Same, in TypeScript