WebSocket API Documentation

{
  "event": "connection_status",
  "status": "connecting",
  "message": "Connecting..."
}

{
  "event": "subscription_status",
  "status": "active",
  "operation": "initial",
  "subscription_count": 2
}

{
  "api_key": "<your_websocket_api_key>",
  "subscriptions": [
    {"code": "NASDAQ:AAPL", "type": "quote"}
  ]
}

{
  "api_key": "<your_websocket_api_key>",
  "subscriptions": [
    {"code": "NASDAQ:AAPL", "type": "series", "bar_type": "minute", "bar_interval": 1, "max_dp": 100},
    {"code": "NASDAQ:TSLA", "type": "quote"}
  ]
}

{
  "api_key": "<your_websocket_api_key>",
  "subscriptions": [
    {"code": "NASDAQ:AAPL", "type": "both", "bar_type": "minute", "bar_interval": 1, "max_dp": 100}
  ]
}

{
  "api_key": "<your_websocket_api_key>",
  "source": "Reuters",
  "related_symbols": "NASDAQ:AAPL, MSFT",
  "content": "earnings acquisition"
}

{
  "event": "subscription_status",
  "status": "active",
  "operation": "initial",
  "subscription_count": 1
}

{
  "event": "delivery_status",
  "mode": "conflated",
  "dropped_updates": 42,
  "max_rate_hz": 1
}

{
  "event": "delivery_status",
  "mode": "realtime",
  "dropped_updates": 0
}

{
  "event": "series_update",
  "code": "NASDAQ:AAPL",
  "bar_end": 1733432399.0,
  "last_update": 1733432399820,
  "bar_type": "1m",
  "series": [
    {
      "time": 1733432340.0,
      "open": 242.89,
      "high": 243.09,
      "low": 242.82,
      "close": 243.08,
      "volume": 533779.0
    }
  ]
}

{
  "event": "quote_update",
  "code": "NASDAQ:AAPL",
  "status": "PRE",
  "lp_time": 1757061117.0,
  "volume": 47549429.0,
  "last_price": 239.42,
  "change_percent": -0.15,
  "change": -0.36,
  "ask": 239.47,
  "bid": 239.42,
  "ask_size": 2.0,
  "bid_size": 1.0,
  "prev_close_price": 238.47,
  "open_price": 238.45,
  "low_price": 236.74,
  "high_price": 239.8999,
  "market_cap": 3558428648118.0,
  "currency_code": "USD",
  "delay_seconds": 0
}

{
  "link": "https://example.com/news/article",
  "source": "Reuters",
  "title": "Example company reports quarterly results",
  "content": "The company reported its latest quarterly results...",
  "published_at": 1757061265,
  "related_symbols": ["NASDAQ:AAPL"]
}

{
  "api_key": "<your_websocket_api_key>",
  "subscriptions": [
    {"code": "NASDAQ:AAPL", "type": "series", "bar_type": "minute", "bar_interval": 1, "extended": false}
  ]
}

{
  "api_key": "<your_websocket_api_key>",
  "subscriptions": [
    {"code": "NASDAQ:AAPL", "type": "both", "24h": true}
  ]
}

{
  "api_key": "<your_websocket_api_key>",
  "subscriptions": [
    {"code": "NASDAQ:AAPL", "type": "series", "bar_type": "day", "bar_interval": 1, "split": false}
  ]
}

{
  "api_key": "<your_websocket_api_key>",
  "subscriptions": [
    {"code": "NASDAQ:AAPL", "type": "series", "bar_type": "day", "bar_interval": 1, "dadj": true}
  ]
}

{
  "api_key": "<your_websocket_api_key>",
  "subscriptions": [
    {"code": "CME_MINI:ES1!", "type": "series", "bar_type": "day", "bar_interval": 1, "badj": false}
  ]
}

{
  "api_key": "<your_websocket_api_key>",
  "subscriptions": [
    {"code": "CME_MINI:ES1!", "type": "series", "bar_type": "day", "bar_interval": 1, "settlement": true}
  ]
}

{
  "api_key": "<your_websocket_api_key>",
  "subscriptions": [
    {"code": "NASDAQ:AAPL", "type": "series", "bar_type": "minute", "bar_interval": 1, "max_dp": 500}
  ]
}

{"server_time": 1741397070281}

{
  "error": "invalid_message",
  "message": "The message must be a valid JSON subscription request.",
  "reference_id": "550e8400-e29b-41d4-a716-446655440000",
  "retryable": false,
  "action": "fix_request"
}

{
  "error": "invalid_apikey",
  "message": "The API key is invalid or no longer current. Check the current WebSocket API key in your dashboard.",
  "reference_id": "550e8400-e29b-41d4-a716-446655440000",
  "retryable": false,
  "action": "update_credentials"
}

{
  "error": "subscription_unavailable",
  "message": "One or more requested subscriptions are temporarily unavailable. Please reconnect. If this persists, contact support with the reference ID.",
  "reference_id": "550e8400-e29b-41d4-a716-446655440000",
  "retryable": true,
  "action": "reconnect"
}

{
  "error": "invalid_bar_interval",
  "message": "Invalid bar_interval for second interval. Allowed values: 1, 5, 10, 15, 30, 45",
  "symbol": "NASDAQ:AAPL",
  "reference_id": "550e8400-e29b-41d4-a716-446655440000",
  "retryable": false,
  "action": "fix_request"
}

{
  "error": "rate_limited",
  "message": "Too many connection attempts were received from this network. Wait 2 seconds before trying again and review your reconnect logic.",
  "reference_id": "550e8400-e29b-41d4-a716-446655440000",
  "retryable": true,
  "action": "wait_and_retry",
  "retry_after_seconds": 2
}

# pip install "websockets>=14"
import asyncio
import json
import logging
import os

import websockets

URI = "wss://realtime.insightsentry.com/live"
ACTIVATION_TIMEOUT_SECONDS = 5 * 60
REQUEST = {
    "api_key": os.environ["INSIGHTSENTRY_WEBSOCKET_API_KEY"],
    "subscriptions": [
        {
            "code": "NASDAQ:AAPL",
            "type": "both",
            "bar_type": "minute",
            "bar_interval": 1,
            "max_dp": 100,
        },
    ],
}


def message_summary(message):
    if message.get("event") == "series_update":
        return f"series update for {message.get('code', 'unknown symbol')}"
    if "published_at" in message:
        return "news item"
    if message.get("event") == "quote_update":
        return f"quote update for {message.get('code', 'unknown symbol')}"
    return None


async def main():
    activated = False
    reported_first_data = False
    activation_deadline = asyncio.get_running_loop().time() + ACTIVATION_TIMEOUT_SECONDS

    async with websockets.connect(
        URI,
        open_timeout=10,
        close_timeout=5,
    ) as websocket:
        await websocket.send(json.dumps(REQUEST))

        while True:
            timeout = None
            if not activated:
                timeout = max(
                    0.0,
                    activation_deadline - asyncio.get_running_loop().time(),
                )
            try:
                raw_message = await asyncio.wait_for(websocket.recv(), timeout=timeout)
            except asyncio.TimeoutError as error:
                raise RuntimeError("Subscription did not become active in time") from error

            if raw_message == "pong":
                continue

            try:
                message = json.loads(raw_message)
            except json.JSONDecodeError:
                logging.warning("Ignored an unexpected non-JSON WebSocket message")
                continue

            if error_code := message.get("error"):
                logging.warning(
                    "WebSocket error=%s action=%s reference_id=%s",
                    error_code,
                    message.get("action"),
                    message.get("reference_id"),
                )
                if message.get("action") == "wait_and_retry":
                    continue
                return

            if (
                message.get("event") == "subscription_status"
                and message.get("status") == "active"
            ):
                activated = True
                logging.info("Subscription active")
                continue

            summary = message_summary(message)
            if summary is not None and not reported_first_data:
                logging.info("Received %s", summary)
                reported_first_data = True


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

# pip install "websockets>=14"
import asyncio
import json
import logging
import os
import random
import time

import websockets
from websockets.exceptions import ConnectionClosed, InvalidStatus

URI = "wss://realtime.insightsentry.com/live"
API_KEY = os.environ["INSIGHTSENTRY_WEBSOCKET_API_KEY"]
REQUEST = {
    "api_key": API_KEY,
    "subscriptions": [
        {
            "code": "NASDAQ:AAPL",
            "type": "both",
            "bar_type": "minute",
            "bar_interval": 1,
            "max_dp": 100,
        },
    ],
}

MAX_ATTEMPTS = 10
BASE_DELAY_SECONDS = 0.5
MAX_DELAY_SECONDS = 30.0
STABLE_AFTER_SECONDS = 60.0
ACTIVATION_TIMEOUT_SECONDS = 5 * 60
DATA_QUEUE_CAPACITY = 1024
PARTIAL_VALIDATION_ERRORS = {
    "invalid_symbol",
    "invalid_bar_interval",
    "invalid_currency_code",
    "invalid_unit_code",
}


class StopRetrying(Exception):
    pass


class ReconnectRequested(Exception):
    def __init__(self, delay_seconds: float, stable: bool):
        self.delay_seconds = delay_seconds
        self.stable = stable


def number(value, default=0.0):
    try:
        return max(0.0, float(value))
    except (TypeError, ValueError):
        return default


def is_stable(activated_at):
    return activated_at is not None and time.monotonic() - activated_at >= STABLE_AFTER_SECONDS


async def resend_after(websocket, delay_seconds):
    await asyncio.sleep(delay_seconds)
    await websocket.send(json.dumps(REQUEST))


def enqueue_data(data_queue, message):
    if (
        message.get("event") not in ("quote_update", "series_update")
        and "published_at" not in message
    ):
        return

    try:
        data_queue.put_nowait(message)
    except asyncio.QueueFull as error:
        raise StopRetrying(
            "local data queue is full; fix the consumer before reconnecting"
        ) from error


async def consume_data(data_queue):
    processed_messages = 0
    while True:
        message = await data_queue.get()
        try:
            # Replace this counter with your storage or processing call.
            # Await async work here, or use asyncio.to_thread for blocking work.
            processed_messages += 1
            if processed_messages == 1 or processed_messages % 1000 == 0:
                logging.info("Processed %d data messages", processed_messages)
        finally:
            data_queue.task_done()


async def connect_once(data_queue):
    activated_at = None
    retry_task = None
    request_retry_attempt = 0
    request_needs_fix = False
    activation_deadline = time.monotonic() + ACTIVATION_TIMEOUT_SECONDS

    async with websockets.connect(
        URI,
        open_timeout=10,
        close_timeout=5,
    ) as websocket:
        await websocket.send(json.dumps(REQUEST))

        try:
            while True:
                timeout = None
                if activated_at is None:
                    timeout = max(0.0, activation_deadline - time.monotonic())
                try:
                    raw_message = await asyncio.wait_for(websocket.recv(), timeout=timeout)
                except asyncio.TimeoutError:
                    if request_needs_fix:
                        raise StopRetrying("activation failed because the request must be fixed")
                    raise ReconnectRequested(0.0, False)

                if raw_message == "pong":
                    continue

                try:
                    message = json.loads(raw_message)
                except json.JSONDecodeError:
                    logging.warning("Ignored an unexpected non-JSON WebSocket message")
                    continue

                if error_code := message.get("error"):
                    action = message.get("action")
                    reference_id = message.get("reference_id")
                    logging.warning(
                        "WebSocket error=%s action=%s reference_id=%s",
                        error_code,
                        action,
                        reference_id,
                    )

                    if action == "wait_and_retry":
                        # The current upgraded request resumes automatically.
                        continue

                    if action == "retry_request":
                        request_retry_attempt += 1
                        if request_retry_attempt >= MAX_ATTEMPTS:
                            raise StopRetrying(
                                f"request retry limit reached; reference_id={reference_id}"
                            )
                        if retry_task is None or retry_task.done():
                            if retry_task is not None:
                                await asyncio.gather(retry_task, return_exceptions=True)
                            cap = min(
                                BASE_DELAY_SECONDS * (2 ** (request_retry_attempt - 1)),
                                MAX_DELAY_SECONDS,
                            )
                            delay = max(
                                number(message.get("delay_seconds")),
                                random.uniform(cap / 2, cap),
                            )
                            retry_task = asyncio.create_task(resend_after(websocket, delay))
                        continue

                    if action == "reconnect":
                        delay = number(message.get("retry_after_seconds"))
                        raise ReconnectRequested(delay, is_stable(activated_at))

                    if action == "reconnect_after_fixing_consumer":
                        raise StopRetrying(
                            "consumer must be fixed before reconnecting; "
                            f"reference_id={reference_id}"
                        )

                    if action == "fix_request" and error_code in PARTIAL_VALIDATION_ERRORS:
                        # Other valid legs may still produce an active status.
                        request_needs_fix = True
                        continue

                    if not message.get("retryable", False):
                        raise StopRetrying(
                            f"{error_code}; reference_id={reference_id}; action={action}"
                        )
                    raise StopRetrying(
                        f"unsupported retry action={action}; reference_id={reference_id}"
                    )

                if (
                    message.get("event") == "subscription_status"
                    and message.get("status") == "active"
                ):
                    if activated_at is None:
                        activated_at = time.monotonic()
                    if retry_task is not None:
                        retry_task.cancel()
                        await asyncio.gather(retry_task, return_exceptions=True)
                        retry_task = None
                    request_retry_attempt = 0
                    request_needs_fix = False
                    logging.info(
                        "Subscription active (%s)",
                        message.get("operation", "unknown"),
                    )
                    continue

                enqueue_data(data_queue, message)
        except ConnectionClosed as closed:
            received_close = closed.rcvd
            logging.info(
                "WebSocket closed code=%s reason=%s",
                received_close.code if received_close is not None else None,
                received_close.reason if received_close is not None else None,
            )
        finally:
            if retry_task is not None:
                retry_task.cancel()
                await asyncio.gather(retry_task, return_exceptions=True)

    return is_stable(activated_at)


def http_status(error):
    response = getattr(error, "response", None)
    return getattr(response, "status_code", None)


def http_error_payload(error):
    response = getattr(error, "response", None)
    body = getattr(response, "body", None)
    if not body:
        return {}

    try:
        payload = json.loads(body)
    except (json.JSONDecodeError, TypeError, UnicodeDecodeError):
        return {}
    return payload if isinstance(payload, dict) else {}


def retry_after(error):
    response = getattr(error, "response", None)
    headers = getattr(response, "headers", {}) or {}
    return number(headers.get("Retry-After"))


async def run_connections(data_queue):
    attempt = 0

    while attempt < MAX_ATTEMPTS:
        stable = False
        server_delay = 0.0

        try:
            stable = await connect_once(data_queue)
        except StopRetrying:
            raise
        except ReconnectRequested as requested:
            stable = requested.stable
            server_delay = requested.delay_seconds
        except InvalidStatus as error:
            status = http_status(error)
            payload = http_error_payload(error)
            reference_id = payload.get("reference_id")
            logging.warning(
                "WebSocket upgrade rejected HTTP %s error=%s reference_id=%s",
                status,
                payload.get("error"),
                reference_id,
            )
            if status != 429:
                raise StopRetrying(
                    f"WebSocket upgrade failed with HTTP {status}; "
                    f"reference_id={reference_id}"
                )
            server_delay = max(
                retry_after(error),
                number(payload.get("retry_after_seconds")),
            )
        except Exception:
            logging.exception("WebSocket connection failed")

        if stable:
            attempt = 0
        attempt += 1
        if attempt >= MAX_ATTEMPTS:
            break

        cap = min(BASE_DELAY_SECONDS * (2 ** (attempt - 1)), MAX_DELAY_SECONDS)
        client_delay = random.uniform(cap / 2, cap)
        delay = max(server_delay, client_delay)
        logging.info("Reconnecting in %.2f seconds (attempt %d)", delay, attempt + 1)
        await asyncio.sleep(delay)

    raise RuntimeError("WebSocket reconnect limit reached")


async def run():
    data_queue = asyncio.Queue(maxsize=DATA_QUEUE_CAPACITY)
    consumer_task = asyncio.create_task(consume_data(data_queue))
    connection_task = asyncio.create_task(run_connections(data_queue))
    try:
        done, _ = await asyncio.wait(
            {consumer_task, connection_task},
            return_when=asyncio.FIRST_COMPLETED,
        )
        if connection_task in done:
            await connection_task
            raise RuntimeError("Connection loop stopped unexpectedly")
        await consumer_task
        raise RuntimeError("Data consumer stopped unexpectedly")
    finally:
        for task in (consumer_task, connection_task):
            task.cancel()
        await asyncio.gather(consumer_task, connection_task, return_exceptions=True)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(run())

URI = "wss://realtime.insightsentry.com/newsfeed"
REQUEST = {
    "api_key": API_KEY,
    "related_symbols": "NASDAQ:AAPL, MSFT",
    "content": "earnings acquisition",
}