WebSocket Integration

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

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

Public (no auth)
Authenticated (HMAC)

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.

Sign the connection handshake with your scoped API token for authenticated features:

All public events
positions (position balance updates)
authenticated (auth confirmation)

Required for subscribe_positions. Pass three HMAC headers (lmts-api-key, lmts-timestamp, lmts-signature) during the handshake — see Authentication. Generate a scoped API token at limitless.exchange → profile → Api keys.

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

Setting	Value
URL	`wss://ws.limitless.exchange`
Namespace	`/markets`
Transport	WebSocket 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())

import asyncio
import os
import socketio

from ws_auth import ws_auth_headers

token_id = os.environ.get("LMTS_TOKEN_ID")
secret = os.environ.get("LMTS_TOKEN_SECRET")
if not token_id or not secret:
    raise ValueError("LMTS_TOKEN_ID and LMTS_TOKEN_SECRET environment variables required")

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"],
        headers=ws_auth_headers(token_id, secret),
    )
    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",
)

# Use human-readable slugs for CLOB orderbook markets
await sio.emit(
    "subscribe_market_prices",
    {"marketSlugs": ["btc-100k-weekly", "eth-daily"]},
    namespace="/markets",
)

# Include both to avoid overwriting subscriptions
await sio.emit(
    "subscribe_market_prices",
    {
        "marketAddresses": ["0x1234567890abcdef1234567890abcdef12345678"],
        "marketSlugs": ["btc-100k-weekly"],
    },
    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

Event	Auth Required	Description
`newPriceData`	No	AMM market price update
`orderbookUpdate`	No	CLOB orderbook update
`positions`	Yes	Position balance update
`system`	No	System notifications
`authenticated`	Yes	Authentication confirmation
`exception`	No	Error 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())

import asyncio
import os
from limitless_ws_client import LimitlessWebSocketClient

async def main():
    token_id = os.environ["LMTS_TOKEN_ID"]
    secret = os.environ["LMTS_TOKEN_SECRET"]
    client = LimitlessWebSocketClient(token_id=token_id, secret=secret)
    await client.connect()

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

    await client.sio.wait()

asyncio.run(main())

Environment Variables

Variable	Description
`LMTS_TOKEN_ID`	Scoped API token id for authenticated features (positions). Omit for public-only.
`LMTS_TOKEN_SECRET`	Token secret used to sign the HMAC handshake. Omit for public-only.
`DEBUG`	Set 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.

Getting Started

Quick Start

TypeScript SDK

Python SDK

Go SDK

Rust SDK

WebSocket

Overview

Prerequisites

Authentication Modes

Connection Details

Subscribing to Market Prices

Subscribing to Positions

Handling Events

Event Payloads

Complete Async Python Client

Usage Examples

Environment Variables

Auto-Reconnection

​Overview

​Prerequisites

​Authentication Modes

​Connection Details

​Subscribing to Market Prices

​Subscribing to Positions

​Handling Events

​Event Payloads

​Complete Async Python Client

​Usage Examples

​Environment Variables

​Auto-Reconnection

Overview

Prerequisites

Authentication Modes

Connection Details

Subscribing to Market Prices

Subscribing to Positions

Handling Events

Event Payloads

Complete Async Python Client

Usage Examples

Environment Variables

Auto-Reconnection