Skip to main content

​
Overview

The Limitless WebSocket API provides real-time market prices, orderbook updates, and position notifications. This guide covers connecting with Python using python-socketio with asyncio.

​
Prerequisites

1

Install dependencies

Install the required packages:
pip install "python-socketio[asyncio]" eth-account==0.10.0 requests
Use a virtual environment to isolate dependencies. Run python -m venv venv then source venv/bin/activate (Unix) or venv\Scripts\activate (Windows) before installing.

​
Authentication Modes

Connect without authentication to receive public market data only:
  • newPriceData (AMM market prices)
  • orderbookUpdate (CLOB orderbook)
  • system (system notifications)
No API key required. Use this for read-only market monitoring.
Never commit your API token id or secret to version control or expose them in client-side code. Use environment variables (e.g. LMTS_TOKEN_ID / LMTS_TOKEN_SECRET) and load them at runtime.
Cookie-based session authentication is deprecated. Authenticate the handshake with HMAC-signed headers (lmts-api-key, lmts-timestamp, lmts-signature) — see Authentication.

​
Connection Details

SettingValue
URLwss://ws.limitless.exchange
Namespace/markets
TransportWebSocket only (no polling)
Authenticated connections sign the socket.io handshake with HMAC. The signed canonical message uses a fixed path and empty body, and the HMAC key is the base64-decoded secret:
ws_auth.py
import base64
import hashlib
import hmac
from datetime import datetime, timezone


def ws_auth_headers(token_id: str, secret: str) -> dict:
    timestamp = datetime.now(timezone.utc).isoformat()
    message = f"{timestamp}\nGET\n/socket.io/?EIO=4&transport=websocket\n"
    signature = base64.b64encode(
        hmac.new(
            base64.b64decode(secret), message.encode("utf-8"), hashlib.sha256
        ).digest()
    ).decode("utf-8")
    return {
        "lmts-api-key": token_id,
        "lmts-timestamp": timestamp,
        "lmts-signature": signature,
    }
See Authentication for how to generate the token id and secret. 
import asyncio
import os
import socketio

sio = socketio.AsyncClient(logger=os.environ.get("DEBUG") == "1")

@sio.event(namespace="/markets")
async def connect():
    print("Connected to /markets")

async def main():
    await sio.connect(
        "wss://ws.limitless.exchange",
        transports=["websocket"],
        namespaces=["/markets"],
    )
    await sio.wait()

asyncio.run(main())

​
Subscribing to Market Prices

Emit subscribe_market_prices with market identifiers. Subscriptions replace previous ones, so include all markets you want in a single call. 
# Market addresses are Ethereum contract addresses in hex format
await sio.emit(
    "subscribe_market_prices",
    {"marketAddresses": ["0x1234567890abcdef1234567890abcdef12345678"]},
    namespace="/markets",
)
Subscriptions replace previous ones. If you want both AMM prices and CLOB orderbook, send both marketAddresses and marketSlugs together in a single subscribe_market_prices call.

​
Subscribing to Positions

Position updates require authentication. Emit subscribe_positions after connecting: 
# Requires HMAC auth headers during the handshake (see ws_auth_headers above)
await sio.emit("subscribe_positions", {}, namespace="/markets")

​
Handling Events

Register handlers for each event type. All handlers use namespace="/markets".
EventAuth RequiredDescription
newPriceDataNoAMM market price update
orderbookUpdateNoCLOB orderbook update
positionsYesPosition balance update
systemNoSystem notifications
authenticatedYesAuthentication confirmation
exceptionNoError notifications

​
Event Payloads

newPriceData — AMM market price update: 
{
  "marketAddress": "0x1234...",
  "updatedPrices": { "yes": "0.65", "no": "0.35" },
  "blockNumber": 12345678,
  "timestamp": "2024-01-01T00:00:00.000Z"
}
positions — Position balance update (auth required): 
{
  "account": "0xabcd...",
  "marketAddress": "0x1234...",
  "positions": [
    { "tokenId": "123456", "balance": "1000000", "outcomeIndex": 0 }
  ],
  "type": "AMM"
}
system — System notification: 
{
  "message": "Notification text",
  "markets": ["0x1234..."]
}

​
Complete Async Python Client

import asyncio
import base64
import hashlib
import hmac
import logging
import os
from datetime import datetime, timezone
from typing import Optional

import socketio

logging.basicConfig(
    level=logging.DEBUG if os.environ.get("DEBUG") == "1" else logging.INFO
)
logger = logging.getLogger(__name__)

WS_URL = "wss://ws.limitless.exchange"
NAMESPACE = "/markets"


def ws_auth_headers(token_id: str, secret: str) -> dict:
    timestamp = datetime.now(timezone.utc).isoformat()
    message = f"{timestamp}\nGET\n/socket.io/?EIO=4&transport=websocket\n"
    signature = base64.b64encode(
        hmac.new(
            base64.b64decode(secret), message.encode("utf-8"), hashlib.sha256
        ).digest()
    ).decode("utf-8")
    return {
        "lmts-api-key": token_id,
        "lmts-timestamp": timestamp,
        "lmts-signature": signature,
    }


class LimitlessWebSocketClient:
    def __init__(
        self,
        token_id: Optional[str] = None,
        secret: Optional[str] = None,
    ):
        self.token_id = token_id or os.environ.get("LMTS_TOKEN_ID")
        self.secret = secret or os.environ.get("LMTS_TOKEN_SECRET")
        self.sio = socketio.AsyncClient(
            reconnection=True,
            reconnection_attempts=0,
            logger=os.environ.get("DEBUG") == "1",
        )
        self._market_addresses: list[str] = []
        self._market_slugs: list[str] = []
        self._subscribe_positions = False
        self._setup_handlers()

    def _setup_handlers(self) -> None:
        @self.sio.event(namespace=NAMESPACE)
        async def connect():
            logger.info("Connected to %s", NAMESPACE)
            await self._resubscribe()

        @self.sio.event(namespace=NAMESPACE)
        async def disconnect():
            logger.info("Disconnected from %s", NAMESPACE)

        @self.sio.event(namespace=NAMESPACE)
        async def newPriceData(data: dict):
            logger.info("Price update: %s", data)

        @self.sio.event(namespace=NAMESPACE)
        async def orderbookUpdate(data: dict):
            logger.info("Orderbook update: %s", data)

        @self.sio.event(namespace=NAMESPACE)
        async def positions(data: dict):
            logger.info("Positions: %s", data)

        @self.sio.event(namespace=NAMESPACE)
        async def system(data: dict):
            logger.info("System: %s", data)

        @self.sio.event(namespace=NAMESPACE)
        async def authenticated(data: dict):
            logger.info("Authenticated: %s", data)

        @self.sio.event(namespace=NAMESPACE)
        async def exception(data: dict):
            logger.error("Exception: %s", data)

    async def _resubscribe(self) -> None:
        if self._market_addresses or self._market_slugs:
            payload = {}
            if self._market_addresses:
                payload["marketAddresses"] = self._market_addresses
            if self._market_slugs:
                payload["marketSlugs"] = self._market_slugs
            await self.sio.emit(
                "subscribe_market_prices",
                payload,
                namespace=NAMESPACE,
            )
        if self._subscribe_positions:
            await self.sio.emit("subscribe_positions", {}, namespace=NAMESPACE)

    async def subscribe_market_prices(
        self,
        market_addresses: Optional[list[str]] = None,
        market_slugs: Optional[list[str]] = None,
    ) -> None:
        self._market_addresses = market_addresses or []
        self._market_slugs = market_slugs or []
        await self._resubscribe()

    async def subscribe_positions(self) -> None:
        if not self.token_id or not self.secret:
            raise ValueError("API token required for position subscription")
        self._subscribe_positions = True
        await self._resubscribe()

    async def connect(self) -> None:
        kwargs = {
            "transports": ["websocket"],
            "namespaces": [NAMESPACE],
        }
        if self.token_id and self.secret:
            kwargs["headers"] = ws_auth_headers(self.token_id, self.secret)

        await self.sio.connect(WS_URL, **kwargs)

    async def disconnect(self) -> None:
        await self.sio.disconnect()

    async def run(self) -> None:
        await self.connect()
        await self.sio.wait()

​
Usage Examples

import asyncio
from limitless_ws_client import LimitlessWebSocketClient

async def main():
    client = LimitlessWebSocketClient()
    await client.connect()

    await client.subscribe_market_prices(
        market_addresses=["0x1234567890abcdef1234567890abcdef12345678"],
        market_slugs=["btc-100k-weekly"],
    )

    await client.sio.wait()

asyncio.run(main())

​
Environment Variables

VariableDescription
LMTS_TOKEN_IDScoped API token id for authenticated features (positions). Omit for public-only.
LMTS_TOKEN_SECRETToken secret used to sign the HMAC handshake. Omit for public-only.
DEBUGSet to 1 to enable verbose socket.io logging.

​
Auto-Reconnection

The client uses reconnection=True by default. On reconnect, call _resubscribe() in the connect handler to restore market and position subscriptions, since the server does not persist them across disconnects.
Store your API token id and secret securely. Rotate the token if exposure is suspected. Do not log or print the secret.