> ## Documentation Index
> Fetch the complete documentation index at: https://docs.limitless.exchange/llms.txt
> Use this file to discover all available pages before exploring further.

# WebSocket Integration

> Real-time market data and position updates using Python WebSocket client

## 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

<Steps>
  <Step title="Install dependencies">
    Install the required packages:
  </Step>
</Steps>

```bash theme={null}
pip install "python-socketio[asyncio]" eth-account==0.10.0 requests
```

<Tip>
  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.
</Tip>

## Authentication Modes

<Tabs>
  <Tab title="Public (no auth)">
    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.
  </Tab>

  <Tab title="Authenticated (HMAC)">
    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](/developers/authentication). Generate a scoped API token at [limitless.exchange](https://limitless.exchange) → profile → Api keys.
  </Tab>
</Tabs>

<Warning>
  **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.
</Warning>

<Note>
  Cookie-based session authentication is **deprecated**. Authenticate the handshake with HMAC-signed headers (`lmts-api-key`, `lmts-timestamp`, `lmts-signature`) — see [Authentication](/developers/authentication).
</Note>

## 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`:

```python ws_auth.py theme={null}
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](/developers/authentication) for how to generate the token id and secret.

<CodeGroup>
  ```python Public (no auth) theme={null}
  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())
  ```

  ```python Authenticated (HMAC) theme={null}
  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())
  ```
</CodeGroup>

## 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.

<CodeGroup>
  ```python AMM markets (contract addresses) theme={null}
  # Market addresses are Ethereum contract addresses in hex format
  await sio.emit(
      "subscribe_market_prices",
      {"marketAddresses": ["0x1234567890abcdef1234567890abcdef12345678"]},
      namespace="/markets",
  )
  ```

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

  ```python Both AMM and CLOB theme={null}
  # Include both to avoid overwriting subscriptions
  await sio.emit(
      "subscribe_market_prices",
      {
          "marketAddresses": ["0x1234567890abcdef1234567890abcdef12345678"],
          "marketSlugs": ["btc-100k-weekly"],
      },
      namespace="/markets",
  )
  ```
</CodeGroup>

<Warning>
  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.
</Warning>

## Subscribing to Positions

Position updates require authentication. Emit `subscribe_positions` after connecting:

```python theme={null}
# 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"`.

| 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:

```json theme={null}
{
  "marketAddress": "0x1234...",
  "updatedPrices": { "yes": "0.65", "no": "0.35" },
  "blockNumber": 12345678,
  "timestamp": "2024-01-01T00:00:00.000Z"
}
```

**`positions`** — Position balance update (auth required):

```json theme={null}
{
  "account": "0xabcd...",
  "marketAddress": "0x1234...",
  "positions": [
    { "tokenId": "123456", "balance": "1000000", "outcomeIndex": 0 }
  ],
  "type": "AMM"
}
```

**`system`** — System notification:

```json theme={null}
{
  "message": "Notification text",
  "markets": ["0x1234..."]
}
```

## Complete Async Python Client

<CodeGroup>
  ```python limitless_ws_client.py theme={null}
  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()
  ```
</CodeGroup>

## Usage Examples

<CodeGroup>
  ```python Public market data only theme={null}
  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())
  ```

  ```python Authenticated with positions theme={null}
  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())
  ```
</CodeGroup>

## 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.

<Warning>
  Store your API token id and secret securely. Rotate the token if exposure is suspected. Do not log or print the secret.
</Warning>