dhan ¶

Dhan's two websockets: the DhanHQ v2 market feed and the account's order update stream.

The login. DhanSession holds the DhanAPI both sockets read their credentials from, afresh on every connect, so a login made by any process is picked up. The quote sockets log in again one at a time, and only when no other socket or process has already replaced the token that failed. The order socket logs in again whenever it is refused, without that check, as it always has.

Quotes. DhanQuotesSocket connects to wss://api-feed.dhan.co with the token and client id in the URL, and subscribes its batch in full mode (request code 21), a hundred instruments a message. Market data arrives as binary messages, and one message may carry several packets back to back. Every value is little endian, and every packet opens with an eight byte header: the response code, the packet's length, Dhan's numeric exchange segment and the security id.

Code	Packet	Length
2	Ticker: last price and last trade time	16
4	Quote: the trade fields and the day's open, close, high and low	50
5	Open interest	12
6	Previous close	16
8	Full: the trade fields, open interest and its range, open, close, high, low and five levels of depth	162
50	Disconnect, with a reason code	10

The previous close and open interest arrive in packets of their own. Written as ticks they would replace the instrument's full tick with one lacking a price, so they are remembered per instrument and folded into the ticks that follow. The close in Dhan's quote and full packets is the previous close during the session but the day's own close after it, so the previous close packet's value, when there is one, is what ohlc.close holds. Dhan's trade times are India wall clock seconds presented as an epoch, so 19,800 seconds are taken off. A disconnect packet with code 807 (token expired), 808 (authentication failed) or 809 (token invalid) is a refused login: the socket closes and logs in again. Other disconnect codes are logged. The quote socket gives up only after six more failed connects, not on the first refusal after logging in again, because Dhan also disconnects for reasons that are not about the login.

Order updates. DhanOrderUpdatesSocket connects to wss://api-order-update.dhan.co and sends Dhan's login message, {"LoginReq": {"MsgCode": 42, "ClientId": ..., "Token": ...}, "UserType": "SELF"}. Dhan does not answer a successful login: a good token holds the socket open, and a bad one is answered with a binary rejection frame and a disconnect, so any binary frame is taken as a refused login. Every change to an order in the account then arrives as a JSON message whose Type is order_alert and whose Data is the order.

Neither socket writes Redis. The quotes socket hands each message's ticks to on_ticks, and the order socket hands each message's order alerts to on_updates; the scripts in bin/dhan/ do the writing.

Classes:

Name	Description
`DhanSession`	The Dhan login every socket authenticates with, logged in again by one socket at a time.
`DhanQuotesSocket`	One DhanHQ market feed websocket carrying the full quotes of a batch of instruments.
`DhanOrderUpdatesSocket`	Dhan's order update websocket, which logs in with a message after connecting.

DhanSession ¶

DhanSession(logger)

The Dhan login every socket authenticates with, logged in again by one socket at a time.

Constructs DhanAPI, which logs in when the stored session is dead.

Parameters:

Name	Type	Description	Default
`logger`	`Logger`	Where logins are reported.	required

Returns:

Type	Description
`None`	This method returns nothing.

Raises:

Type	Description
`Exception`	Whatever `DhanAPI` raises when it cannot log in.

Example programs

Logging in again onceLogging in without checking

examples/stock_brokers/websockets/dhan/DhanSession/example_1_logging_in_again_once.py

"""Reads the Dhan credentials a socket connects with and logs in again only when the failed token is still current.

`DhanSession` holds one `DhanAPI` and asks it for the current login on every connect, so a token another process obtained is picked up at once. When a quotes socket is refused, it calls `log_in_again` with the token it used, and the session logs in only if that token is still the one in force, so several refused sockets cause one login between them.

Constructing the real `DhanAPI` logs in to Dhan, so the program registers a stand-in module under `stock_brokers.api.dhan` before building the session. Each construction of the stand-in counts as one login and issues the next numbered token; `_current_login` answers from a shared record standing in for the Redis `last_login` hash.

Notice that the first request logs in, that the second request with the same stale token is skipped, and that a token set by another process is what `credentials` returns next.

Run it from the project root:

    python examples/stock_brokers/websockets/dhan/DhanSession/example_1_logging_in_again_once.py
"""

import logging
import sys
import types

from stock_brokers.websockets.dhan import (
    DhanSession,
)


class LoginRecord:
    """The Dhan login state every stand-in API object reads, standing in for the Redis `last_login` hash.

    Attributes:
        access_token (str | None): The token in force now.
        login_count (int): How many logins the stand-in API has made.
    """

    access_token = None
    login_count = 0


class StandInDhanAPI:
    """A stand-in for `DhanAPI` whose construction is a pretend login."""

    def __init__(self):
        """Logs in, which here only issues the next numbered token.

        Returns:
            None: This method returns nothing.
        """
        LoginRecord.login_count = LoginRecord.login_count + 1
        LoginRecord.access_token = f'dhan-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'client_id': '1100012345',
        }

    def _current_login(self):
        """The login in force now, as the real class reads it from Redis.

        Returns:
            dict | None: The access token, or None before any login.
        """
        if LoginRecord.access_token is None:
            return None
        return {
            'access_token': LoginRecord.access_token,
        }


class LoggingInAgainOnceExample:
    """Asks a Dhan session to log in again twice with the same failed token.

    Attributes:
        session (DhanSession): The session being shown.
    """

    def __init__(self):
        """Registers the stand-in API module and builds the session.

        Returns:
            None: This method returns nothing.
        """
        logging.basicConfig(stream=sys.stdout, format='%(levelname)s %(message)s', level=logging.INFO)
        api_module = types.ModuleType('stock_brokers.api.dhan')
        api_module.DhanAPI = StandInDhanAPI
        sys.modules['stock_brokers.api.dhan'] = api_module
        self.session = DhanSession(logging.getLogger('dhan.session'))

    def run(self):
        """Refuses the same token twice, then shows a token another process obtained.

        Returns:
            None: This method returns nothing.
        """
        client_id, failed_token = self.session.credentials()
        print(f'Client {client_id} was refused with {failed_token}')
        self.session.log_in_again(failed_token)
        print(f'Token in force: {self.session.credentials()[1]}')
        self.session.log_in_again(failed_token)
        print(f'Token in force after the second request: {self.session.credentials()[1]}')
        LoginRecord.access_token = 'dhan-token-from-another-process'
        print(f'Token in force after another process logged in: {self.session.credentials()[1]}')
        print(f'Logins made: {LoginRecord.login_count}')


if __name__ == '__main__':
    LoggingInAgainOnceExample().run()

It prints:

Stand-in login number 1
Client 1100012345 was refused with dhan-token-1
WARNING Logging in to Dhan again.
Stand-in login number 2
Token in force: dhan-token-2
Token in force after the second request: dhan-token-2
Token in force after another process logged in: dhan-token-from-another-process
Logins made: 2

examples/stock_brokers/websockets/dhan/DhanSession/example_2_logging_in_without_checking.py

"""Logs in to Dhan again unconditionally, as the order update socket does after a refusal.

The order update socket cannot tell which token Dhan rejected, because the rejection is a binary frame sent after its own login message, so it calls `DhanSession.log_in_again_without_checking`. That logs in even when another process has already replaced the token, unlike `log_in_again`, which would skip the login in that case.

Constructing the real `DhanAPI` logs in to Dhan, so the program registers a stand-in module under `stock_brokers.api.dhan` whose constructions count as logins. It first sets the token as another process would, shows that `log_in_again` with the old token does nothing, and then calls `log_in_again_without_checking`.

Notice that only the unconditional call makes a login, and that it replaces the other process's token with a new one.

Run it from the project root:

    python examples/stock_brokers/websockets/dhan/DhanSession/example_2_logging_in_without_checking.py
"""

import logging
import sys
import types

from stock_brokers.websockets.dhan import (
    DhanSession,
)


class LoginRecord:
    """The Dhan login state every stand-in API object reads, standing in for the Redis `last_login` hash.

    Attributes:
        access_token (str | None): The token in force now.
        login_count (int): How many logins the stand-in API has made.
    """

    access_token = None
    login_count = 0


class StandInDhanAPI:
    """A stand-in for `DhanAPI` whose construction is a pretend login."""

    def __init__(self):
        """Logs in, which here only issues the next numbered token.

        Returns:
            None: This method returns nothing.
        """
        LoginRecord.login_count = LoginRecord.login_count + 1
        LoginRecord.access_token = f'dhan-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'client_id': '1100012345',
        }

    def _current_login(self):
        """The login in force now, as the real class reads it from Redis.

        Returns:
            dict | None: The access token, or None before any login.
        """
        if LoginRecord.access_token is None:
            return None
        return {
            'access_token': LoginRecord.access_token,
        }


class LoggingInWithoutCheckingExample:
    """Compares the checked and unchecked logins of a Dhan session.

    Attributes:
        session (DhanSession): The session being shown.
    """

    def __init__(self):
        """Registers the stand-in API module and builds the session.

        Returns:
            None: This method returns nothing.
        """
        logging.basicConfig(stream=sys.stdout, format='%(levelname)s %(message)s', level=logging.INFO)
        api_module = types.ModuleType('stock_brokers.api.dhan')
        api_module.DhanAPI = StandInDhanAPI
        sys.modules['stock_brokers.api.dhan'] = api_module
        self.session = DhanSession(logging.getLogger('dhan.session'))

    def run(self):
        """Replaces the token elsewhere, then tries both kinds of login.

        Returns:
            None: This method returns nothing.
        """
        first_token = self.session.credentials()[1]
        LoginRecord.access_token = 'dhan-token-from-another-process'
        print(f'Another process replaced {first_token}.')
        self.session.log_in_again(first_token)
        print(f'After log_in_again: {self.session.credentials()[1]}')
        self.session.log_in_again_without_checking()
        print(f'After log_in_again_without_checking: {self.session.credentials()[1]}')
        print(f'Logins made: {LoginRecord.login_count}')


if __name__ == '__main__':
    LoggingInWithoutCheckingExample().run()

It prints:

Stand-in login number 1
Another process replaced dhan-token-1.
After log_in_again: dhan-token-from-another-process
WARNING Logging in to Dhan again.
Stand-in login number 2
After log_in_again_without_checking: dhan-token-2
Logins made: 2

Methods:

Name	Description
`credentials`	The client id and the access token in force now, which may be one another process has just obtained.
`log_in_again`	Logs in again, unless another socket or process already replaced the token that failed.
`log_in_again_without_checking`	Logs in again whether or not the token was already replaced.

Source code in stock_brokers/websockets/dhan.py

def __init__(self, logger):
    """
    Constructs `DhanAPI`, which logs in when the stored session is dead.

    Args:
        logger (logging.Logger): Where logins are reported.

    Returns:
        None: This method returns nothing.

    Raises:
        Exception: Whatever `DhanAPI` raises when it cannot log in.
    """
    from stock_brokers.api.dhan import DhanAPI

    self._api_class = DhanAPI
    self._logger = logger
    self._lock = threading.Lock()
    self._dhan = DhanAPI()

credentials ¶

credentials()

The client id and the access token in force now, which may be one another process has just obtained.

Returns:

Type	Description
`tuple`	The client id and the access token, either of which may be None.

Example usage

From example_1_logging_in_again_once.py

client_id, failed_token = self.session.credentials()

From example_1_logging_in_again_once.py

print(f'Token in force: {self.session.credentials()[1]}')

From example_1_logging_in_again_once.py

print(f'Token in force after the second request: {self.session.credentials()[1]}')

Source code in stock_brokers/websockets/dhan.py

def credentials(self):
    """
    The client id and the access token in force now, which may be one another process has just obtained.

    Returns:
        tuple: The client id and the access token, either of which may be None.
    """
    login = self._dhan._current_login() or {}
    return self._dhan._settings.get("client_id"), login.get("access_token")

log_in_again ¶

log_in_again(stale_token)

Logs in again, unless another socket or process already replaced the token that failed.

Parameters:

Name	Type	Description	Default
`stale_token`	`str \| None`	The access token the failing connection used.	required

Returns:

Type	Description
`None`	This method returns nothing.

Raises:

Type	Description
`Exception`	Whatever `DhanAPI` raises when it cannot log in.

Example usage

From example_1_logging_in_again_once.py

self.session.log_in_again(failed_token)

From example_2_logging_in_without_checking.py

self.session.log_in_again(first_token)

Source code in stock_brokers/websockets/dhan.py

def log_in_again(self, stale_token):
    """
    Logs in again, unless another socket or process already replaced the token that failed.

    Args:
        stale_token (str | None): The access token the failing connection used.

    Returns:
        None: This method returns nothing.

    Raises:
        Exception: Whatever `DhanAPI` raises when it cannot log in.
    """
    with self._lock:
        if self.credentials()[1] != stale_token:
            return
        self._log_in()

log_in_again_without_checking ¶

log_in_again_without_checking()

Logs in again whether or not the token was already replaced.

Returns:

Type	Description
`None`	This method returns nothing.

Raises:

Type	Description
`Exception`	Whatever `DhanAPI` raises when it cannot log in.

Example usage

From example_2_logging_in_without_checking.py

self.session.log_in_again_without_checking()

Source code in stock_brokers/websockets/dhan.py

def log_in_again_without_checking(self):
    """
    Logs in again whether or not the token was already replaced.

    Returns:
        None: This method returns nothing.

    Raises:
        Exception: Whatever `DhanAPI` raises when it cannot log in.
    """
    with self._lock:
        self._log_in()

_log_in ¶

_log_in()

Logs in by constructing DhanAPI again; the caller holds the lock.

Returns:

Type	Description
`None`	This method returns nothing.

Raises:

Type	Description
`Exception`	Whatever `DhanAPI` raises when it cannot log in.

Source code in stock_brokers/websockets/dhan.py

def _log_in(self):
    """
    Logs in by constructing `DhanAPI` again; the caller holds the lock.

    Returns:
        None: This method returns nothing.

    Raises:
        Exception: Whatever `DhanAPI` raises when it cannot log in.
    """
    self._logger.warning("Logging in to Dhan again.")
    self._dhan = self._api_class()

DhanQuotesSocket ¶

DhanQuotesSocket(name, tokens, names, session, on_ticks, logger)

Bases: BrokerWebsocket

One DhanHQ market feed websocket carrying the full quotes of a batch of instruments.

Sets up a socket for one batch of instruments.

Parameters:

Name	Type	Description	Default
`name`	`str`	The connection's name, for the log.	required
`tokens`	`list[str]`	The batch of `SEGMENT:SECURITY_ID` instrument tokens.	required
`names`	`dict[str, str]`	Each token to the instrument's name, which becomes a tick's `id`.	required
`session`	`DhanSession`	The shared login.	required
`on_ticks`	`callable`	Called with each message's list of ticks, on this socket's thread.	required
`logger`	`Logger`	Where the connection reports.	required

Returns:

Type	Description
`None`	This method returns nothing.

Example programs

Folding in the previous closeDisconnected for an expired token

examples/stock_brokers/websockets/dhan/DhanQuotesSocket/example_1_folding_in_the_previous_close.py

"""Decodes one DhanHQ market feed message holding four packets and prints the ticks the socket hands on.

A `DhanQuotesSocket` subscribes its batch in full mode and turns each binary message into normalized ticks. One message can carry several packets back to back. The previous close and open interest arrive in packets of their own, and the socket remembers them per instrument and folds them into the ticks that follow instead of writing ticks with no price.

This program builds one message the way Dhan does: a previous close packet and an open interest packet for NSE_EQ security 2885 (RELIANCE), then a 162 byte full packet for it, then a 16 byte ticker packet for NSE_EQ security 1594 (INFY). No real socket is opened. A stand-in `websocket` package delivers the message and prints what the socket sends, and a stand-in `stock_brokers.api.dhan` module lets the real `DhanSession` run without logging in.

Notice that only two ticks come out of four packets, that the full tick's `ohlc.close` is the remembered previous close of 2910.5 rather than the close in the full packet, and that Dhan's India wall clock trade time has 19,800 seconds taken off to make a true epoch.

Run it from the project root:

    python examples/stock_brokers/websockets/dhan/DhanQuotesSocket/example_1_folding_in_the_previous_close.py
"""

import logging
import struct
import sys
import types

from stock_brokers.websockets.dhan import (
    DhanQuotesSocket,
    DhanSession,
)


class LoginRecord:
    """The Dhan login state every stand-in API object reads, standing in for the Redis `last_login` hash.

    Attributes:
        access_token (str | None): The token in force now.
        login_count (int): How many logins the stand-in API has made.
    """

    access_token = None
    login_count = 0


class StandInDhanAPI:
    """A stand-in for `DhanAPI` whose construction is a pretend login."""

    def __init__(self):
        """Logs in, which here only issues the next numbered token.

        Returns:
            None: This method returns nothing.
        """
        LoginRecord.login_count = LoginRecord.login_count + 1
        LoginRecord.access_token = f'dhan-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'client_id': '1100012345',
        }

    def _current_login(self):
        """The login in force now, as the real class reads it from Redis.

        Returns:
            dict | None: The access token, or None before any login.
        """
        if LoginRecord.access_token is None:
            return None
        return {
            'access_token': LoginRecord.access_token,
        }


class HandshakeRefused(Exception):
    """A refused websocket handshake, carrying the HTTP status the way `websocket-client` does.

    Attributes:
        status_code (int): The HTTP status the broker answered with.
    """

    def __init__(self, status_code):
        """Keeps the status.

        Args:
            status_code (int): The HTTP status the broker answered with.

        Returns:
            None: This method returns nothing.
        """
        super().__init__(f'Handshake status {status_code}')
        self.status_code = status_code


class ScriptedConnection:
    """A stand-in for `websocket.WebSocketApp` that either refuses the handshake or plays frames to the socket's handlers.

    Attributes:
        url (str): The URL the socket asked for.
        plan (dict): Either a `refuse_with` status or a list of `frames`.
        handlers (dict): The socket's callbacks, by name.
        closed (bool): Whether the socket asked to close, after which no more frames are delivered.
    """

    def __init__(self, url, plan, handlers):
        """Keeps what the socket passed in.

        Args:
            url (str): The URL the socket asked for.
            plan (dict): Either a `refuse_with` status or a list of `frames`.
            handlers (dict): The socket's callbacks, by name.

        Returns:
            None: This method returns nothing.
        """
        self.url = url
        self.plan = plan
        self.handlers = handlers
        self.closed = False

    def run_forever(self, **options):
        """Plays the plan and returns when the pretend connection closes.

        Args:
            **options (dict): The ping and TLS settings the socket asks for.

        Returns:
            None: This method returns nothing.
        """
        print(f'Connecting to {self.url}')
        if 'refuse_with' in self.plan:
            self.handlers['on_error'](self, HandshakeRefused(self.plan['refuse_with']))
            self.handlers['on_close'](self, None, None)
            return
        self.handlers['on_open'](self)
        for frame in self.plan['frames']:
            if self.closed:
                break
            self.handlers['on_message'](self, frame)
        self.handlers['on_close'](self, 1000, 'normal closure')

    def send(self, data, opcode=None):
        """Prints a message the socket sends to the broker.

        Args:
            data (str | bytes): The message.
            opcode (int | None): The frame type, when the socket names one.

        Returns:
            None: This method returns nothing.
        """
        print(f'Sent: {data}')

    def close(self, **options):
        """Notes that the socket asked to close.

        Args:
            **options (dict): Any close status the socket passes.

        Returns:
            None: This method returns nothing.
        """
        self.closed = True


class ScriptedWebsocketModule:
    """A stand-in for the `websocket` package that hands out scripted connections in turn.

    Attributes:
        plans (list): What each connection does, in the order the socket connects.
        when_finished (callable | None): Called when the last connection is handed out, so the socket stops after it.
    """

    def __init__(self, plans):
        """Keeps the connection plans.

        Args:
            plans (list): What each connection does, in the order the socket connects.

        Returns:
            None: This method returns nothing.
        """
        self.plans = plans
        self.when_finished = None

    def WebSocketApp(self, url, **handlers):
        """Builds the next scripted connection, closing the socket after the last one.

        Args:
            url (str): The feed URL.
            **handlers (dict): The socket's `on_open`, `on_message`, `on_error` and `on_close` callbacks.

        Returns:
            ScriptedConnection: The connection.
        """
        plan = self.plans.pop(0)
        if not self.plans:
            self.when_finished()
        return ScriptedConnection(url, plan, handlers)


class DhanPackets:
    """Builds DhanHQ v2 market feed packets byte by byte, little endian, as `wss://api-feed.dhan.co` sends them."""

    def header(self, feed_code, length, segment_code, security_id):
        """The eight byte header every packet opens with.

        Args:
            feed_code (int): The response code, such as 8 for a full packet.
            length (int): The packet's length in bytes.
            segment_code (int): Dhan's numeric exchange segment, such as 1 for NSE_EQ.
            security_id (int): Dhan's security id.

        Returns:
            bytes: The header.
        """
        return struct.pack('<BhBi', feed_code, length, segment_code, security_id)

    def previous_close(self, segment_code, security_id, close):
        """A 16 byte previous close packet.

        Args:
            segment_code (int): Dhan's numeric exchange segment.
            security_id (int): Dhan's security id.
            close (float): The previous session's close.

        Returns:
            bytes: The packet.
        """
        return self.header(6, 16, segment_code, security_id) + struct.pack('<fi', close, 0)

    def open_interest(self, segment_code, security_id, open_interest):
        """A 12 byte open interest packet.

        Args:
            segment_code (int): Dhan's numeric exchange segment.
            security_id (int): Dhan's security id.
            open_interest (int): The open interest.

        Returns:
            bytes: The packet.
        """
        return self.header(5, 12, segment_code, security_id) + struct.pack('<i', open_interest)

    def trade_fields(self, last_price, wall_clock_epoch):
        """The trade fields quote and full packets share, from byte 8 to byte 34.

        Args:
            last_price (float): The last traded price.
            wall_clock_epoch (int): The last trade time as Dhan sends it, India wall clock seconds presented as an epoch.

        Returns:
            bytes: The fields.
        """
        return struct.pack('<fhifiii', last_price, 25, wall_clock_epoch, last_price - 1.5, 1204500, 54000, 61000)

    def quote(self, segment_code, security_id, last_price, wall_clock_epoch):
        """A 50 byte quote packet.

        Args:
            segment_code (int): Dhan's numeric exchange segment.
            security_id (int): Dhan's security id.
            last_price (float): The last traded price.
            wall_clock_epoch (int): The last trade time as Dhan sends it.

        Returns:
            bytes: The packet.
        """
        packet = self.header(4, 50, segment_code, security_id)
        packet = packet + self.trade_fields(last_price, wall_clock_epoch)
        return packet + struct.pack('<ffff', last_price - 20.0, last_price - 40.0, last_price + 10.0, last_price - 25.0)

    def full(self, segment_code, security_id, last_price, wall_clock_epoch):
        """A 162 byte full packet with open interest and five levels of depth.

        Args:
            segment_code (int): Dhan's numeric exchange segment.
            security_id (int): Dhan's security id.
            last_price (float): The last traded price.
            wall_clock_epoch (int): The last trade time as Dhan sends it.

        Returns:
            bytes: The packet.
        """
        packet = self.header(8, 162, segment_code, security_id)
        packet = packet + self.trade_fields(last_price, wall_clock_epoch)
        packet = packet + struct.pack('<iii', 0, 0, 0)
        packet = packet + struct.pack('<ffff', last_price - 20.0, last_price - 40.0, last_price + 10.0, last_price - 25.0)
        for level in range(5):
            packet = packet + struct.pack('<iihhff', 100 + level, 200 + level, 3 + level, 4 + level, last_price - 0.5 - level, last_price + 0.5 + level)
        return packet

    def ticker(self, segment_code, security_id, last_price, wall_clock_epoch):
        """A 16 byte ticker packet with the last price and the last trade time.

        Args:
            segment_code (int): Dhan's numeric exchange segment.
            security_id (int): Dhan's security id.
            last_price (float): The last traded price.
            wall_clock_epoch (int): The last trade time as Dhan sends it.

        Returns:
            bytes: The packet.
        """
        return self.header(2, 16, segment_code, security_id) + struct.pack('<fi', last_price, wall_clock_epoch)

    def disconnect(self, reason_code):
        """A 10 byte disconnect packet with Dhan's reason code.

        Args:
            reason_code (int): Why Dhan is disconnecting, such as 808 for a failed authentication.

        Returns:
            bytes: The packet.
        """
        return self.header(50, 10, 0, 0) + struct.pack('<h', reason_code)


class FoldingInThePreviousCloseExample:
    """Runs a Dhan quotes socket against one scripted message.

    Attributes:
        socket (DhanQuotesSocket): The socket being shown.
    """

    def __init__(self):
        """Builds the message, the stand-ins, the session and the socket.

        Returns:
            None: This method returns nothing.
        """
        logging.basicConfig(stream=sys.stdout, format='%(levelname)s %(message)s', level=logging.INFO)
        api_module = types.ModuleType('stock_brokers.api.dhan')
        api_module.DhanAPI = StandInDhanAPI
        sys.modules['stock_brokers.api.dhan'] = api_module
        packets = DhanPackets()
        wall_clock_epoch = 1790311528 + 19800
        message = b''
        message = message + packets.previous_close(1, 2885, 2910.5)
        message = message + packets.open_interest(1, 2885, 0)
        message = message + packets.full(1, 2885, 2950.5, wall_clock_epoch)
        message = message + packets.ticker(1, 1594, 1500.25, wall_clock_epoch)
        plans = [
            {
                'frames': [
                    message,
                ],
            },
        ]
        websocket_module = ScriptedWebsocketModule(plans)
        sys.modules['websocket'] = websocket_module
        logger = logging.getLogger('dhan.quotes')
        tokens = [
            'NSE_EQ:2885',
            'NSE_EQ:1594',
        ]
        names = {
            'NSE_EQ:2885': 'NSE:RELIANCE',
            'NSE_EQ:1594': 'NSE:INFY',
        }
        self.socket = DhanQuotesSocket('socket_0', tokens, names, DhanSession(logger), self.print_ticks, logger)
        websocket_module.when_finished = self.socket.close

    def print_ticks(self, ticks):
        """Prints the main fields of each tick the socket decoded.

        Args:
            ticks (list): The normalized ticks from one message.

        Returns:
            None: This method returns nothing.
        """
        for tick in ticks:
            change = tick['change']
            if change is not None:
                change = round(change, 4)
            print(f"Tick {tick['id']} ({tick['instrument_token']}) on {tick['exchange']} in {tick['mode']} mode")
            print(f"  last_price={tick['last_price']} last_trade_time={tick['last_trade_time']} volume={tick['volume']}")
            print(f"  ohlc={tick['ohlc']} change={change}")
            print(f"  oi={tick['oi']}")
            if tick['depth']['buy']:
                print(f"  best bid={tick['depth']['buy'][0]} best offer={tick['depth']['sell'][0]}")

    def run(self):
        """Runs the socket's reconnect loop until the scripted connection ends.

        Returns:
            None: This method returns nothing.
        """
        self.socket.run_forever()
        print(f'Gave up: {self.socket.gave_up}')


if __name__ == '__main__':
    FoldingInThePreviousCloseExample().run()

It prints:

Stand-in login number 1
Connecting to wss://api-feed.dhan.co?version=2&token=dhan-token-1&clientId=1100012345&authType=2
Sent: {"RequestCode": 21, "InstrumentCount": 2, "InstrumentList": [{"ExchangeSegment": "NSE_EQ", "SecurityId": "2885"}, {"ExchangeSegment": "NSE_EQ", "SecurityId": "1594"}]}
INFO socket_0 opened and subscribed to 2 instrument(s).
Tick NSE:RELIANCE (NSE_EQ:2885) on NSE_EQ in full mode
  last_price=2950.5 last_trade_time=1790311528 volume=1204500
  ohlc={'open': 2930.5, 'close': 2910.5, 'high': 2960.5, 'low': 2925.5} change=1.3743
  oi=0
  best bid={'quantity': 100, 'price': 2950.0, 'orders': 3} best offer={'quantity': 200, 'price': 2951.0, 'orders': 4}
Tick NSE:INFY (NSE_EQ:1594) on NSE_EQ in ltp mode
  last_price=1500.25 last_trade_time=1790311528 volume=None
  ohlc={'open': None, 'high': None, 'low': None, 'close': None} change=None
  oi=None
INFO socket_0 closed with status 1000: normal closure
INFO socket_0 stopped.
Gave up: False

examples/stock_brokers/websockets/dhan/DhanQuotesSocket/example_2_disconnected_for_an_expired_token.py

"""Shows a Dhan quotes socket logging in again after Dhan disconnects it for an expired token.

Dhan does not refuse a dead token at the handshake. It accepts the connection and then sends a disconnect packet whose reason code says why: 807 (token expired), 808 (authentication failed) or 809 (token invalid). `DhanQuotesSocket` treats those codes as a refused login, closes the connection, and its reconnect loop asks the shared `DhanSession` to log in again before connecting with the new token.

This program scripts two connections. The first opens and receives a disconnect packet with code 807, followed by a quote packet that is never delivered because the socket has already closed. The second opens with the new token and delivers a 50 byte quote packet for NSE_FNO security 35001. No real socket is opened: a stand-in `websocket` package plays the connections, and a stand-in `stock_brokers.api.dhan` module counts each `DhanAPI` construction as one login so the real `DhanSession` runs unchanged. The backoff is set to zero seconds.

Notice the warning about the disconnect packet, the second login, the new token in the second URL, and that a quote tick with no remembered previous close keeps the close from its own packet.

Run it from the project root:

    python examples/stock_brokers/websockets/dhan/DhanQuotesSocket/example_2_disconnected_for_an_expired_token.py
"""

import logging
import struct
import sys
import types

from stock_brokers.websockets.dhan import (
    DhanQuotesSocket,
    DhanSession,
)


class LoginRecord:
    """The Dhan login state every stand-in API object reads, standing in for the Redis `last_login` hash.

    Attributes:
        access_token (str | None): The token in force now.
        login_count (int): How many logins the stand-in API has made.
    """

    access_token = None
    login_count = 0


class StandInDhanAPI:
    """A stand-in for `DhanAPI` whose construction is a pretend login."""

    def __init__(self):
        """Logs in, which here only issues the next numbered token.

        Returns:
            None: This method returns nothing.
        """
        LoginRecord.login_count = LoginRecord.login_count + 1
        LoginRecord.access_token = f'dhan-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'client_id': '1100012345',
        }

    def _current_login(self):
        """The login in force now, as the real class reads it from Redis.

        Returns:
            dict | None: The access token, or None before any login.
        """
        if LoginRecord.access_token is None:
            return None
        return {
            'access_token': LoginRecord.access_token,
        }


class HandshakeRefused(Exception):
    """A refused websocket handshake, carrying the HTTP status the way `websocket-client` does.

    Attributes:
        status_code (int): The HTTP status the broker answered with.
    """

    def __init__(self, status_code):
        """Keeps the status.

        Args:
            status_code (int): The HTTP status the broker answered with.

        Returns:
            None: This method returns nothing.
        """
        super().__init__(f'Handshake status {status_code}')
        self.status_code = status_code


class ScriptedConnection:
    """A stand-in for `websocket.WebSocketApp` that either refuses the handshake or plays frames to the socket's handlers.

    Attributes:
        url (str): The URL the socket asked for.
        plan (dict): Either a `refuse_with` status or a list of `frames`.
        handlers (dict): The socket's callbacks, by name.
        closed (bool): Whether the socket asked to close, after which no more frames are delivered.
    """

    def __init__(self, url, plan, handlers):
        """Keeps what the socket passed in.

        Args:
            url (str): The URL the socket asked for.
            plan (dict): Either a `refuse_with` status or a list of `frames`.
            handlers (dict): The socket's callbacks, by name.

        Returns:
            None: This method returns nothing.
        """
        self.url = url
        self.plan = plan
        self.handlers = handlers
        self.closed = False

    def run_forever(self, **options):
        """Plays the plan and returns when the pretend connection closes.

        Args:
            **options (dict): The ping and TLS settings the socket asks for.

        Returns:
            None: This method returns nothing.
        """
        print(f'Connecting to {self.url}')
        if 'refuse_with' in self.plan:
            self.handlers['on_error'](self, HandshakeRefused(self.plan['refuse_with']))
            self.handlers['on_close'](self, None, None)
            return
        self.handlers['on_open'](self)
        for frame in self.plan['frames']:
            if self.closed:
                break
            self.handlers['on_message'](self, frame)
        self.handlers['on_close'](self, 1000, 'normal closure')

    def send(self, data, opcode=None):
        """Prints a message the socket sends to the broker.

        Args:
            data (str | bytes): The message.
            opcode (int | None): The frame type, when the socket names one.

        Returns:
            None: This method returns nothing.
        """
        print(f'Sent: {data}')

    def close(self, **options):
        """Notes that the socket asked to close.

        Args:
            **options (dict): Any close status the socket passes.

        Returns:
            None: This method returns nothing.
        """
        self.closed = True


class ScriptedWebsocketModule:
    """A stand-in for the `websocket` package that hands out scripted connections in turn.

    Attributes:
        plans (list): What each connection does, in the order the socket connects.
        when_finished (callable | None): Called when the last connection is handed out, so the socket stops after it.
    """

    def __init__(self, plans):
        """Keeps the connection plans.

        Args:
            plans (list): What each connection does, in the order the socket connects.

        Returns:
            None: This method returns nothing.
        """
        self.plans = plans
        self.when_finished = None

    def WebSocketApp(self, url, **handlers):
        """Builds the next scripted connection, closing the socket after the last one.

        Args:
            url (str): The feed URL.
            **handlers (dict): The socket's `on_open`, `on_message`, `on_error` and `on_close` callbacks.

        Returns:
            ScriptedConnection: The connection.
        """
        plan = self.plans.pop(0)
        if not self.plans:
            self.when_finished()
        return ScriptedConnection(url, plan, handlers)


class DhanPackets:
    """Builds DhanHQ v2 market feed packets byte by byte, little endian, as `wss://api-feed.dhan.co` sends them."""

    def header(self, feed_code, length, segment_code, security_id):
        """The eight byte header every packet opens with.

        Args:
            feed_code (int): The response code, such as 8 for a full packet.
            length (int): The packet's length in bytes.
            segment_code (int): Dhan's numeric exchange segment, such as 1 for NSE_EQ.
            security_id (int): Dhan's security id.

        Returns:
            bytes: The header.
        """
        return struct.pack('<BhBi', feed_code, length, segment_code, security_id)

    def previous_close(self, segment_code, security_id, close):
        """A 16 byte previous close packet.

        Args:
            segment_code (int): Dhan's numeric exchange segment.
            security_id (int): Dhan's security id.
            close (float): The previous session's close.

        Returns:
            bytes: The packet.
        """
        return self.header(6, 16, segment_code, security_id) + struct.pack('<fi', close, 0)

    def open_interest(self, segment_code, security_id, open_interest):
        """A 12 byte open interest packet.

        Args:
            segment_code (int): Dhan's numeric exchange segment.
            security_id (int): Dhan's security id.
            open_interest (int): The open interest.

        Returns:
            bytes: The packet.
        """
        return self.header(5, 12, segment_code, security_id) + struct.pack('<i', open_interest)

    def trade_fields(self, last_price, wall_clock_epoch):
        """The trade fields quote and full packets share, from byte 8 to byte 34.

        Args:
            last_price (float): The last traded price.
            wall_clock_epoch (int): The last trade time as Dhan sends it, India wall clock seconds presented as an epoch.

        Returns:
            bytes: The fields.
        """
        return struct.pack('<fhifiii', last_price, 25, wall_clock_epoch, last_price - 1.5, 1204500, 54000, 61000)

    def quote(self, segment_code, security_id, last_price, wall_clock_epoch):
        """A 50 byte quote packet.

        Args:
            segment_code (int): Dhan's numeric exchange segment.
            security_id (int): Dhan's security id.
            last_price (float): The last traded price.
            wall_clock_epoch (int): The last trade time as Dhan sends it.

        Returns:
            bytes: The packet.
        """
        packet = self.header(4, 50, segment_code, security_id)
        packet = packet + self.trade_fields(last_price, wall_clock_epoch)
        return packet + struct.pack('<ffff', last_price - 20.0, last_price - 40.0, last_price + 10.0, last_price - 25.0)

    def full(self, segment_code, security_id, last_price, wall_clock_epoch):
        """A 162 byte full packet with open interest and five levels of depth.

        Args:
            segment_code (int): Dhan's numeric exchange segment.
            security_id (int): Dhan's security id.
            last_price (float): The last traded price.
            wall_clock_epoch (int): The last trade time as Dhan sends it.

        Returns:
            bytes: The packet.
        """
        packet = self.header(8, 162, segment_code, security_id)
        packet = packet + self.trade_fields(last_price, wall_clock_epoch)
        packet = packet + struct.pack('<iii', 0, 0, 0)
        packet = packet + struct.pack('<ffff', last_price - 20.0, last_price - 40.0, last_price + 10.0, last_price - 25.0)
        for level in range(5):
            packet = packet + struct.pack('<iihhff', 100 + level, 200 + level, 3 + level, 4 + level, last_price - 0.5 - level, last_price + 0.5 + level)
        return packet

    def ticker(self, segment_code, security_id, last_price, wall_clock_epoch):
        """A 16 byte ticker packet with the last price and the last trade time.

        Args:
            segment_code (int): Dhan's numeric exchange segment.
            security_id (int): Dhan's security id.
            last_price (float): The last traded price.
            wall_clock_epoch (int): The last trade time as Dhan sends it.

        Returns:
            bytes: The packet.
        """
        return self.header(2, 16, segment_code, security_id) + struct.pack('<fi', last_price, wall_clock_epoch)

    def disconnect(self, reason_code):
        """A 10 byte disconnect packet with Dhan's reason code.

        Args:
            reason_code (int): Why Dhan is disconnecting, such as 808 for a failed authentication.

        Returns:
            bytes: The packet.
        """
        return self.header(50, 10, 0, 0) + struct.pack('<h', reason_code)


class DisconnectedForAnExpiredTokenExample:
    """Runs a Dhan quotes socket through an expired token disconnect and a good connection.

    Attributes:
        socket (DhanQuotesSocket): The socket being shown.
    """

    def __init__(self):
        """Builds the messages, the stand-ins, the session and the socket.

        Returns:
            None: This method returns nothing.
        """
        logging.basicConfig(stream=sys.stdout, format='%(levelname)s %(message)s', level=logging.INFO)
        api_module = types.ModuleType('stock_brokers.api.dhan')
        api_module.DhanAPI = StandInDhanAPI
        sys.modules['stock_brokers.api.dhan'] = api_module
        packets = DhanPackets()
        wall_clock_epoch = 1790311528 + 19800
        plans = [
            {
                'frames': [
                    packets.disconnect(807),
                    packets.quote(2, 35001, 25150.5, wall_clock_epoch),
                ],
            },
            {
                'frames': [
                    packets.quote(2, 35001, 25151.25, wall_clock_epoch),
                ],
            },
        ]
        websocket_module = ScriptedWebsocketModule(plans)
        sys.modules['websocket'] = websocket_module
        logger = logging.getLogger('dhan.quotes')
        tokens = [
            'NSE_FNO:35001',
        ]
        names = {
            'NSE_FNO:35001': 'NFO:NIFTY26SEPFUT',
        }
        self.socket = DhanQuotesSocket('socket_0', tokens, names, DhanSession(logger), self.print_ticks, logger)
        self.socket.MIN_BACKOFF_SECONDS = 0
        websocket_module.when_finished = self.socket.close

    def print_ticks(self, ticks):
        """Prints the main fields of each tick the socket decoded.

        Args:
            ticks (list): The normalized ticks from one message.

        Returns:
            None: This method returns nothing.
        """
        for tick in ticks:
            change = tick['change']
            if change is not None:
                change = round(change, 4)
            print(f"Tick {tick['id']} ({tick['instrument_token']}) on {tick['exchange']} in {tick['mode']} mode")
            print(f"  last_price={tick['last_price']} last_trade_time={tick['last_trade_time']} volume={tick['volume']}")
            print(f"  ohlc={tick['ohlc']} change={change}")
            print(f"  oi={tick['oi']}")
            if tick['depth']['buy']:
                print(f"  best bid={tick['depth']['buy'][0]} best offer={tick['depth']['sell'][0]}")

    def run(self):
        """Runs the socket's reconnect loop until the scripted connections run out.

        Returns:
            None: This method returns nothing.
        """
        self.socket.run_forever()
        print(f'Gave up: {self.socket.gave_up}')
        print(f'Logins made: {LoginRecord.login_count}')


if __name__ == '__main__':
    DisconnectedForAnExpiredTokenExample().run()

It prints:

Stand-in login number 1
Connecting to wss://api-feed.dhan.co?version=2&token=dhan-token-1&clientId=1100012345&authType=2
Sent: {"RequestCode": 21, "InstrumentCount": 1, "InstrumentList": [{"ExchangeSegment": "NSE_FNO", "SecurityId": "35001"}]}
INFO socket_0 opened and subscribed to 1 instrument(s).
WARNING socket_0 received a disconnect packet, code 807.
INFO socket_0 closed with status 1000: normal closure
WARNING Logging in to Dhan again.
Stand-in login number 2
WARNING socket_0 disconnected. Reconnecting in 0 seconds.
Connecting to wss://api-feed.dhan.co?version=2&token=dhan-token-2&clientId=1100012345&authType=2
Sent: {"RequestCode": 21, "InstrumentCount": 1, "InstrumentList": [{"ExchangeSegment": "NSE_FNO", "SecurityId": "35001"}]}
INFO socket_0 opened and subscribed to 1 instrument(s).
Tick NFO:NIFTY26SEPFUT (NSE_FNO:35001) on NSE_FNO in quote mode
  last_price=25151.25 last_trade_time=1790311528 volume=1204500
  ohlc={'open': 25131.25, 'close': 25111.25, 'high': 25161.25, 'low': 25126.25} change=0.1593
  oi=None
INFO socket_0 closed with status 1000: normal closure
INFO socket_0 stopped.
Gave up: False
Logins made: 2

Methods:

Name	Description
`close`	Closes the connection and stops reconnecting.
`run_forever`	Connects, and reconnects with backoff, until closed or unable to recover.

Source code in stock_brokers/websockets/dhan.py

def __init__(self, name, tokens, names, session, on_ticks, logger):
    """
    Sets up a socket for one batch of instruments.

    Args:
        name (str): The connection's name, for the log.
        tokens (list[str]): The batch of `SEGMENT:SECURITY_ID` instrument tokens.
        names (dict[str, str]): Each token to the instrument's name, which becomes a tick's `id`.
        session (DhanSession): The shared login.
        on_ticks (callable): Called with each message's list of ticks, on this socket's thread.
        logger (logging.Logger): Where the connection reports.

    Returns:
        None: This method returns nothing.
    """
    super().__init__(name, logger)
    self._tokens = tokens
    self._names = names
    self._session = session
    self._on_ticks = on_ticks
    self._token = None
    self._remembered = {}

_gives_up_after_logging_in_again ¶

_gives_up_after_logging_in_again(failed_connects)

Gives up only once six connects in a row have failed, since Dhan also disconnects for reasons other than the login.

Parameters:

Name	Type	Description	Default
`failed_connects`	`int`	How many connects in a row have failed since the last login.	required

Returns:

Type	Description
`bool`	True to give up now.

Source code in stock_brokers/websockets/dhan.py

def _gives_up_after_logging_in_again(self, failed_connects):
    """
    Gives up only once six connects in a row have failed, since Dhan also disconnects for reasons other than the login.

    Args:
        failed_connects (int): How many connects in a row have failed since the last login.

    Returns:
        bool: True to give up now.
    """
    return failed_connects >= self.MAX_FAILED_CONNECTS

_connect ¶

_connect()

Opens the websocket with the token in force now and blocks until it closes.

Returns:

Type	Description
`None`	This method returns nothing.

Source code in stock_brokers/websockets/dhan.py

def _connect(self):
    """
    Opens the websocket with the token in force now and blocks until it closes.

    Returns:
        None: This method returns nothing.
    """
    import websocket

    client_id, self._token = self._session.credentials()
    url = f"{FEED_URL}?version=2&token={self._token}&clientId={client_id}&authType=2"
    self._websocket_application = websocket.WebSocketApp(
        url,
        on_open=self._on_open,
        on_message=self._on_message,
        on_error=self._on_error,
        on_close=self._on_close,
    )
    self._websocket_application.run_forever(
        ping_interval=PING_INTERVAL_SECONDS,
        ping_timeout=PING_TIMEOUT_SECONDS,
    )

_log_in_again ¶

_log_in_again()

Logs in again through the shared session, which skips the login when the token was already replaced.

Returns:

Type	Description
`None`	This method returns nothing.

Source code in stock_brokers/websockets/dhan.py

def _log_in_again(self):
    """
    Logs in again through the shared session, which skips the login when the token was already replaced.

    Returns:
        None: This method returns nothing.
    """
    self._session.log_in_again(self._token)

_on_open ¶

_on_open(websocket_connection)

Subscribes the batch in full mode, a hundred instruments a message.

Parameters:

Name	Type	Description	Default
`websocket_connection`	`WebSocketApp`	The connection that opened.	required

Returns:

Type	Description
`None`	This method returns nothing.

Source code in stock_brokers/websockets/dhan.py

def _on_open(self, websocket_connection):
    """
    Subscribes the batch in full mode, a hundred instruments a message.

    Args:
        websocket_connection (websocket.WebSocketApp): The connection that opened.

    Returns:
        None: This method returns nothing.
    """
    self._opened = True
    for start in range(0, len(self._tokens), MAX_INSTRUMENTS_PER_MESSAGE):
        batch = self._tokens[start:start + MAX_INSTRUMENTS_PER_MESSAGE]
        instruments = []
        for token in batch:
            instruments.append({
                "ExchangeSegment": token.partition(":")[0],
                "SecurityId": token.partition(":")[2],
            })
        websocket_connection.send(json.dumps({
            "RequestCode": SUBSCRIBE_FULL,
            "InstrumentCount": len(batch),
            "InstrumentList": instruments,
        }))
    self._logger.info(f"{self.name} opened and subscribed to {len(self._tokens)} instrument(s).")

_on_error ¶

_on_error(websocket_connection, error)

Reports an error; Dhan refuses a login with a disconnect packet rather than at the handshake.

Parameters:

Name	Type	Description	Default
`websocket_connection`	`WebSocketApp`	The connection that failed.	required
`error`	`Exception`	The error.	required

Returns:

Type	Description
`None`	This method returns nothing.

Source code in stock_brokers/websockets/dhan.py

def _on_error(self, websocket_connection, error):
    """
    Reports an error; Dhan refuses a login with a disconnect packet rather than at the handshake.

    Args:
        websocket_connection (websocket.WebSocketApp): The connection that failed.
        error (Exception): The error.

    Returns:
        None: This method returns nothing.
    """
    self._logger.error(f"{self.name} error: {error}")

_on_message ¶

_on_message(websocket_connection, message)

Decodes a binary message into ticks and hands them on; text frames are ignored.

Parameters:

Name	Type	Description	Default
`websocket_connection`	`WebSocketApp`	The connection the message arrived on.	required
`message`	`bytes \| str`	The message.	required

Returns:

Type	Description
`None`	This method returns nothing.

Source code in stock_brokers/websockets/dhan.py

def _on_message(self, websocket_connection, message):
    """
    Decodes a binary message into ticks and hands them on; text frames are ignored.

    Args:
        websocket_connection (websocket.WebSocketApp): The connection the message arrived on.
        message (bytes | str): The message.

    Returns:
        None: This method returns nothing.
    """
    if not isinstance(message, bytes):
        return
    ticks = []
    for packet in self._split_packets(message):
        tick = self._parse_packet(packet)
        if tick:
            ticks.append(tick)
    if ticks:
        self._on_ticks(ticks)

_split_packets ¶

_split_packets(message)

Splits a message into its packets, each of which declares its own length.

Parameters:

Name	Type	Description	Default
`message`	`bytes`	The binary message.	required

Returns:

Type	Description
`list[bytes]`	The packets, stopping at a packet that claims no length or more than is left.

Source code in stock_brokers/websockets/dhan.py

def _split_packets(self, message):
    """
    Splits a message into its packets, each of which declares its own length.

    Args:
        message (bytes): The binary message.

    Returns:
        list[bytes]: The packets, stopping at a packet that claims no length or more than is left.
    """
    packets = []
    offset = 0
    while offset + 8 <= len(message):
        length = self._unpack(message, offset + 1, offset + 3, "h")
        if length <= 0 or offset + length > len(message):
            break
        packets.append(message[offset:offset + length])
        offset = offset + length
    return packets

_parse_packet ¶

_parse_packet(packet)

Builds a normalized tick from one packet, remembering the single-value packets.

Parameters:

Name	Type	Description	Default
`packet`	`bytes`	The binary packet.	required

Returns:

Type	Description
`dict \| None`	The tick, or None for a packet that carries no tick.

Source code in stock_brokers/websockets/dhan.py

def _parse_packet(self, packet):
    """
    Builds a normalized tick from one packet, remembering the single-value packets.

    Args:
        packet (bytes): The binary packet.

    Returns:
        dict | None: The tick, or None for a packet that carries no tick.
    """
    if len(packet) < 8:
        return None
    feed_code = self._unpack(packet, 0, 1, "B")
    segment_code = self._unpack(packet, 3, 4, "B")
    token = f"{SEGMENT_NAMES.get(segment_code, segment_code)}:{self._unpack(packet, 4, 8, 'i')}"

    if feed_code == DISCONNECT:
        code = None
        if len(packet) >= 10:
            code = self._unpack(packet, 8, 10, "h")
        self._logger.warning(f"{self.name} received a disconnect packet, code {code}.")
        if code in AUTHENTICATION_DISCONNECT_CODES:
            self._authentication_rejected = True
            self._websocket_application.close()
        return None

    remembered = self._remembered.setdefault(token, {})
    if feed_code == PREVIOUS_CLOSE and len(packet) >= 16:
        remembered["previous_close"] = self._unpack(packet, 8, 12, "f")
        return None
    if feed_code == OPEN_INTEREST and len(packet) >= 12:
        remembered["oi"] = self._unpack(packet, 8, 12, "i")
        return None

    if feed_code == TICKER and len(packet) >= 16:
        tick = self._empty_tick(token, segment_code, "ltp")
        tick["last_price"] = self._unpack(packet, 8, 12, "f")
        tick["last_trade_time"] = self._true_epoch(self._unpack(packet, 12, 16, "i"))
        return tick

    if feed_code == QUOTE and len(packet) >= 50:
        tick = self._empty_tick(token, segment_code, "quote")
        self._fill_trade(tick, packet)
        tick["ohlc"] = {
            "open": self._unpack(packet, 34, 38, "f"),
            "close": self._unpack(packet, 38, 42, "f"),
            "high": self._unpack(packet, 42, 46, "f"),
            "low": self._unpack(packet, 46, 50, "f"),
        }
        tick["oi"] = remembered.get("oi")
        self._apply_previous_close(tick, remembered)
        return tick

    if feed_code == FULL and len(packet) >= 162:
        tick = self._empty_tick(token, segment_code, "full")
        self._fill_trade(tick, packet)
        tick["oi"] = self._unpack(packet, 34, 38, "i")
        tick["oi_day_high"] = self._unpack(packet, 38, 42, "i")
        tick["oi_day_low"] = self._unpack(packet, 42, 46, "i")
        tick["ohlc"] = {
            "open": self._unpack(packet, 46, 50, "f"),
            "close": self._unpack(packet, 50, 54, "f"),
            "high": self._unpack(packet, 54, 58, "f"),
            "low": self._unpack(packet, 58, 62, "f"),
        }
        self._apply_previous_close(tick, remembered)
        for offset in range(62, 162, 20):
            tick["depth"]["buy"].append({
                "quantity": self._unpack(packet, offset, offset + 4, "i"),
                "price": self._unpack(packet, offset + 12, offset + 16, "f"),
                "orders": self._unpack(packet, offset + 8, offset + 10, "h"),
            })
            tick["depth"]["sell"].append({
                "quantity": self._unpack(packet, offset + 4, offset + 8, "i"),
                "price": self._unpack(packet, offset + 16, offset + 20, "f"),
                "orders": self._unpack(packet, offset + 10, offset + 12, "h"),
            })
        return tick

    self._logger.debug(f"{self.name} skipped a packet with feed code {feed_code} and length {len(packet)}.")
    return None

_fill_trade ¶

_fill_trade(tick, packet)

Fills the trade fields the quote and full packets share, at the same offsets.

Parameters:

Name	Type	Description	Default
`tick`	`dict`	The tick to fill.	required
`packet`	`bytes`	The binary packet.	required

Returns:

Type	Description
`None`	This method returns nothing.

Source code in stock_brokers/websockets/dhan.py

def _fill_trade(self, tick, packet):
    """
    Fills the trade fields the quote and full packets share, at the same offsets.

    Args:
        tick (dict): The tick to fill.
        packet (bytes): The binary packet.

    Returns:
        None: This method returns nothing.
    """
    tick["last_price"] = self._unpack(packet, 8, 12, "f")
    tick["last_quantity"] = self._unpack(packet, 12, 14, "h")
    tick["last_trade_time"] = self._true_epoch(self._unpack(packet, 14, 18, "i"))
    tick["average_price"] = self._unpack(packet, 18, 22, "f")
    tick["volume"] = self._unpack(packet, 22, 26, "i")
    tick["sell_quantity"] = self._unpack(packet, 26, 30, "i")
    tick["buy_quantity"] = self._unpack(packet, 30, 34, "i")

_empty_tick ¶

_empty_tick(token, segment_code, mode)

A tick with every field unset.

Parameters:

Name	Type	Description	Default
`token`	`str`	The instrument's `SEGMENT:SECURITY_ID`.	required
`segment_code`	`int`	Dhan's numeric segment.	required
`mode`	`str`	`ltp`, `quote` or `full`.	required

Returns:

Type	Description
`dict`	The tick, named from the batch's names or by its token.

Source code in stock_brokers/websockets/dhan.py

def _empty_tick(self, token, segment_code, mode):
    """
    A tick with every field unset.

    Args:
        token (str): The instrument's `SEGMENT:SECURITY_ID`.
        segment_code (int): Dhan's numeric segment.
        mode (str): `ltp`, `quote` or `full`.

    Returns:
        dict: The tick, named from the batch's names or by its token.
    """
    return {
        "id": self._names.get(token) or token,
        "broker": "dhan",
        "instrument_token": token,
        "exchange": SEGMENT_NAMES.get(segment_code),
        "mode": mode,
        "last_price": None,
        "last_quantity": None,
        "average_price": None,
        "volume": None,
        "buy_quantity": None,
        "sell_quantity": None,
        "ohlc": {
            "open": None,
            "high": None,
            "low": None,
            "close": None,
        },
        "change": None,
        "oi": None,
        "oi_day_high": None,
        "oi_day_low": None,
        "last_trade_time": None,
        "exchange_timestamp": None,
        "depth": {
            "buy": [],
            "sell": [],
        },
        "received_at": time.time(),
    }

_unpack ¶

_unpack(data, start, end, byte_format)

A little endian value from a slice of a packet.

Parameters:

Name	Type	Description	Default
`data`	`bytes`	The packet or message.	required
`start`	`int`	Where the slice starts.	required
`end`	`int`	Where the slice ends.	required
`byte_format`	`str`	The `struct` format character.	required

Returns:

Type	Description
`int \| float`	The value.

Source code in stock_brokers/websockets/dhan.py

def _unpack(self, data, start, end, byte_format):
    """
    A little endian value from a slice of a packet.

    Args:
        data (bytes): The packet or message.
        start (int): Where the slice starts.
        end (int): Where the slice ends.
        byte_format (str): The `struct` format character.

    Returns:
        int | float: The value.
    """
    return struct.unpack("<" + byte_format, data[start:end])[0]

_true_epoch ¶

_true_epoch(timestamp)

A Dhan India-wall-clock-as-epoch timestamp on the true epoch.

Parameters:

Name	Type	Description	Default
`timestamp`	`int`	The timestamp Dhan sent.	required

Returns:

Type	Description
`int \| None`	The true epoch, or None for zero.

Source code in stock_brokers/websockets/dhan.py

def _true_epoch(self, timestamp):
    """
    A Dhan India-wall-clock-as-epoch timestamp on the true epoch.

    Args:
        timestamp (int): The timestamp Dhan sent.

    Returns:
        int | None: The true epoch, or None for zero.
    """
    if timestamp:
        return timestamp - IST_OFFSET_SECONDS
    return None

_apply_previous_close ¶

_apply_previous_close(tick, remembered)

Puts the previous session's close in ohlc.close once Dhan has sent it, and the change against it.

Parameters:

Name	Type	Description	Default
`tick`	`dict`	The tick to adjust.	required
`remembered`	`dict`	What the instrument's single-value packets left.	required

Returns:

Type	Description
`None`	This method returns nothing.

Source code in stock_brokers/websockets/dhan.py

def _apply_previous_close(self, tick, remembered):
    """
    Puts the previous session's close in `ohlc.close` once Dhan has sent it, and the change against it.

    Args:
        tick (dict): The tick to adjust.
        remembered (dict): What the instrument's single-value packets left.

    Returns:
        None: This method returns nothing.
    """
    if remembered.get("previous_close"):
        tick["ohlc"]["close"] = remembered["previous_close"]
    close = tick["ohlc"]["close"]
    if close:
        tick["change"] = (tick["last_price"] - close) * 100 / close
    else:
        tick["change"] = None

close ¶

close()

Closes the connection and stops reconnecting.

Returns:

Type	Description
`None`	This method returns nothing.

Example usage

From example_2_giving_up_after_a_second_refusal.py

self.closed_socket.close()

Source code in stock_brokers/websockets/base.py

def close(self):
    """
    Closes the connection and stops reconnecting.

    Returns:
        None: This method returns nothing.
    """
    self._stop.set()
    if self._websocket_application is not None:
        self._websocket_application.close()

run_forever ¶

run_forever()

Connects, and reconnects with backoff, until closed or unable to recover.

Returns:

Type	Description
`None`	This method returns nothing. `gave_up` says whether the socket stopped because it could not recover.

Example usage

From example_1_reconnecting_after_a_drop.py

self.socket.run_forever()

From example_2_giving_up_after_a_second_refusal.py

self.refused_socket.run_forever()

From example_2_giving_up_after_a_second_refusal.py

self.closed_socket.run_forever()

Source code in stock_brokers/websockets/base.py

def run_forever(self):
    """
    Connects, and reconnects with backoff, until closed or unable to recover.

    Returns:
        None: This method returns nothing. `gave_up` says whether the socket stopped because it could not recover.
    """
    backoff = self.MIN_BACKOFF_SECONDS
    failed_connects = 0
    logged_in_again = False
    while not self._stop.is_set():
        self._opened = False
        self._authentication_rejected = False
        try:
            self._connect()
        except Exception as exception:
            self._logger.error(f"{self.name} connection failed: {type(exception).__name__}: {exception}")
        if self._stop.is_set():
            break

        if self._opened and not self._authentication_rejected:
            failed_connects = 0
            logged_in_again = False
            backoff = self.MIN_BACKOFF_SECONDS
        else:
            failed_connects = failed_connects + 1

        if self._authentication_rejected or failed_connects >= self.MAX_FAILED_CONNECTS:
            if logged_in_again and self._gives_up_after_logging_in_again(failed_connects):
                self.gave_up = True
                self._logger.error(f"{self.name} still cannot connect after logging in again. Giving up.")
                break
            try:
                self._log_in_again()
            except Exception as exception:
                self.gave_up = True
                self._logger.error(f"{self.name} could not log in again: {type(exception).__name__}: {exception}")
                break
            failed_connects = 0
            logged_in_again = True

        self._logger.warning(f"{self.name} disconnected. Reconnecting in {backoff} seconds.")
        self._stop.wait(backoff)
        backoff = min(backoff * 2, self.MAX_BACKOFF_SECONDS)
    self._logger.info(f"{self.name} stopped.")

_on_close ¶

_on_close(websocket_connection, status_code, message)

Reports that the connection closed.

Parameters:

Name	Type	Description	Default
`websocket_connection`	`WebSocketApp`	The connection that closed.	required
`status_code`	`int \| None`	The close status the broker sent.	required
`message`	`str \| None`	The close reason the broker sent.	required

Returns:

Type	Description
`None`	This method returns nothing.

Source code in stock_brokers/websockets/base.py

def _on_close(self, websocket_connection, status_code, message):
    """
    Reports that the connection closed.

    Args:
        websocket_connection (websocket.WebSocketApp): The connection that closed.
        status_code (int | None): The close status the broker sent.
        message (str | None): The close reason the broker sent.

    Returns:
        None: This method returns nothing.
    """
    self._logger.info(f"{self.name} closed with status {status_code}: {message}")

DhanOrderUpdatesSocket ¶

DhanOrderUpdatesSocket(session, on_updates, logger)

Bases: BrokerWebsocket

Dhan's order update websocket, which logs in with a message after connecting.

Sets up the order updates socket.

Parameters:

Name	Type	Description	Default
`session`	`DhanSession`	The login.	required
`on_updates`	`callable`	Called on this socket's thread with each message's order alerts, as a list of `Data` dictionaries exactly as Dhan sent them, and the `datetime` the message was received.	required
`logger`	`Logger`	Where the connection reports.	required

Returns:

Type	Description
`None`	This method returns nothing.

Example programs

Logging in and receiving alertsBinary rejection logs in again

examples/stock_brokers/websockets/dhan/DhanOrderUpdatesSocket/example_1_logging_in_and_receiving_alerts.py

"""Connects Dhan's order update socket, sends its login message and prints the order alerts it hands on.

`DhanOrderUpdatesSocket` connects to `wss://api-order-update.dhan.co` and, once the connection opens, sends Dhan's login message with the client id and the token in force. Dhan does not answer a good login; it simply keeps the socket open and sends a JSON message whose `Type` is `order_alert` for every change to an order. The socket hands each message's `Data` dictionaries on, exactly as Dhan sent them, and ignores other message types.

This program scripts one connection that delivers a pending alert, a message of another type, a frame that is not JSON, and a traded alert. No real socket is opened: a stand-in `websocket` package plays the frames and prints what the socket sends, and a stand-in `stock_brokers.api.dhan` module lets the real `DhanSession` supply credentials without logging in. The arrival time is not printed because it is the wall clock.

Notice the login message in the output, and that the callback runs once for each alert while the other message is logged as ignored. The frame that is not JSON leaves no line, because the socket notes it only at debug level.

Run it from the project root:

    python examples/stock_brokers/websockets/dhan/DhanOrderUpdatesSocket/example_1_logging_in_and_receiving_alerts.py
"""

import json
import logging
import sys
import types

from stock_brokers.websockets.dhan import (
    DhanOrderUpdatesSocket,
    DhanSession,
)


class LoginRecord:
    """The Dhan login state every stand-in API object reads, standing in for the Redis `last_login` hash.

    Attributes:
        access_token (str | None): The token in force now.
        login_count (int): How many logins the stand-in API has made.
    """

    access_token = None
    login_count = 0


class StandInDhanAPI:
    """A stand-in for `DhanAPI` whose construction is a pretend login."""

    def __init__(self):
        """Logs in, which here only issues the next numbered token.

        Returns:
            None: This method returns nothing.
        """
        LoginRecord.login_count = LoginRecord.login_count + 1
        LoginRecord.access_token = f'dhan-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'client_id': '1100012345',
        }

    def _current_login(self):
        """The login in force now, as the real class reads it from Redis.

        Returns:
            dict | None: The access token, or None before any login.
        """
        if LoginRecord.access_token is None:
            return None
        return {
            'access_token': LoginRecord.access_token,
        }


class HandshakeRefused(Exception):
    """A refused websocket handshake, carrying the HTTP status the way `websocket-client` does.

    Attributes:
        status_code (int): The HTTP status the broker answered with.
    """

    def __init__(self, status_code):
        """Keeps the status.

        Args:
            status_code (int): The HTTP status the broker answered with.

        Returns:
            None: This method returns nothing.
        """
        super().__init__(f'Handshake status {status_code}')
        self.status_code = status_code


class ScriptedConnection:
    """A stand-in for `websocket.WebSocketApp` that either refuses the handshake or plays frames to the socket's handlers.

    Attributes:
        url (str): The URL the socket asked for.
        plan (dict): Either a `refuse_with` status or a list of `frames`.
        handlers (dict): The socket's callbacks, by name.
        closed (bool): Whether the socket asked to close, after which no more frames are delivered.
    """

    def __init__(self, url, plan, handlers):
        """Keeps what the socket passed in.

        Args:
            url (str): The URL the socket asked for.
            plan (dict): Either a `refuse_with` status or a list of `frames`.
            handlers (dict): The socket's callbacks, by name.

        Returns:
            None: This method returns nothing.
        """
        self.url = url
        self.plan = plan
        self.handlers = handlers
        self.closed = False

    def run_forever(self, **options):
        """Plays the plan and returns when the pretend connection closes.

        Args:
            **options (dict): The ping and TLS settings the socket asks for.

        Returns:
            None: This method returns nothing.
        """
        print(f'Connecting to {self.url}')
        if 'refuse_with' in self.plan:
            self.handlers['on_error'](self, HandshakeRefused(self.plan['refuse_with']))
            self.handlers['on_close'](self, None, None)
            return
        self.handlers['on_open'](self)
        for frame in self.plan['frames']:
            if self.closed:
                break
            self.handlers['on_message'](self, frame)
        self.handlers['on_close'](self, 1000, 'normal closure')

    def send(self, data, opcode=None):
        """Prints a message the socket sends to the broker.

        Args:
            data (str | bytes): The message.
            opcode (int | None): The frame type, when the socket names one.

        Returns:
            None: This method returns nothing.
        """
        print(f'Sent: {data}')

    def close(self, **options):
        """Notes that the socket asked to close.

        Args:
            **options (dict): Any close status the socket passes.

        Returns:
            None: This method returns nothing.
        """
        self.closed = True


class ScriptedWebsocketModule:
    """A stand-in for the `websocket` package that hands out scripted connections in turn.

    Attributes:
        plans (list): What each connection does, in the order the socket connects.
        when_finished (callable | None): Called when the last connection is handed out, so the socket stops after it.
    """

    def __init__(self, plans):
        """Keeps the connection plans.

        Args:
            plans (list): What each connection does, in the order the socket connects.

        Returns:
            None: This method returns nothing.
        """
        self.plans = plans
        self.when_finished = None

    def WebSocketApp(self, url, **handlers):
        """Builds the next scripted connection, closing the socket after the last one.

        Args:
            url (str): The feed URL.
            **handlers (dict): The socket's `on_open`, `on_message`, `on_error` and `on_close` callbacks.

        Returns:
            ScriptedConnection: The connection.
        """
        plan = self.plans.pop(0)
        if not self.plans:
            self.when_finished()
        return ScriptedConnection(url, plan, handlers)


class LoggingInAndReceivingAlertsExample:
    """Runs the Dhan order update socket against one scripted connection.

    Attributes:
        socket (DhanOrderUpdatesSocket): The socket being shown.
    """

    def __init__(self):
        """Builds the messages, the stand-ins, the session and the socket.

        Returns:
            None: This method returns nothing.
        """
        logging.basicConfig(stream=sys.stdout, format='%(levelname)s %(message)s', level=logging.INFO)
        api_module = types.ModuleType('stock_brokers.api.dhan')
        api_module.DhanAPI = StandInDhanAPI
        sys.modules['stock_brokers.api.dhan'] = api_module
        heartbeat = {
            'Type': 'heartbeat',
            'Data': {},
        }
        plans = [
            {
                'frames': [
                    self.order_alert('Pending', 0),
                    json.dumps(heartbeat),
                    'not json',
                    self.order_alert('Traded', 10),
                ],
            },
        ]
        websocket_module = ScriptedWebsocketModule(plans)
        sys.modules['websocket'] = websocket_module
        logger = logging.getLogger('dhan.orders')
        self.socket = DhanOrderUpdatesSocket(DhanSession(logger), self.print_orders, logger)
        websocket_module.when_finished = self.socket.close

    def order_alert(self, status, traded_quantity):
        """One of Dhan's order alert messages for an INFY buy.

        Args:
            status (str): Dhan's order status.
            traded_quantity (int): How much has traded.

        Returns:
            str: The message as JSON text.
        """
        alert = {
            'Type': 'order_alert',
            'Data': {
                'OrderNo': '1122092500001',
                'Status': status,
                'Symbol': 'INFY',
                'Exchange': 'NSE',
                'Segment': 'E',
                'TxnType': 'B',
                'Quantity': 10,
                'TradedQty': traded_quantity,
                'TradedPrice': 1499.5,
                'Product': 'C',
            },
        }
        return json.dumps(alert)

    def print_orders(self, orders, received_at):
        """Prints each order alert the socket handed on.

        Args:
            orders (list): The `Data` dictionaries of one message, exactly as Dhan sent them.
            received_at (datetime.datetime): When the message arrived, which is not printed.

        Returns:
            None: This method returns nothing.
        """
        for order in orders:
            print(f"Order {order['OrderNo']} {order['Status']}: {order['TxnType']} {order['TradedQty']} of {order['Quantity']} {order['Symbol']} at {order['TradedPrice']}")

    def run(self):
        """Runs the socket's reconnect loop until the scripted connection ends.

        Returns:
            None: This method returns nothing.
        """
        self.socket.run_forever()
        print(f'Gave up: {self.socket.gave_up}')


if __name__ == '__main__':
    LoggingInAndReceivingAlertsExample().run()

It prints:

Stand-in login number 1
Connecting to wss://api-order-update.dhan.co
Sent: {"LoginReq": {"MsgCode": 42, "ClientId": "1100012345", "Token": "dhan-token-1"}, "UserType": "SELF"}
INFO Order updates socket opened and logged in. Waiting for order updates.
Order 1122092500001 Pending: B 0 of 10 INFY at 1499.5
INFO Ignoring a message of type 'heartbeat': {"Type": "heartbeat", "Data": {}}
Order 1122092500001 Traded: B 10 of 10 INFY at 1499.5
INFO Order updates socket closed with status 1000: normal closure
INFO Order updates socket stopped.
Gave up: False

examples/stock_brokers/websockets/dhan/DhanOrderUpdatesSocket/example_2_binary_rejection_logs_in_again.py

"""Shows Dhan's order update socket taking a binary frame as a refused login and logging in again.

Dhan answers a bad token on the order update socket with a binary rejection frame and a disconnect, so `DhanOrderUpdatesSocket` treats any binary frame as a refused login. It closes the connection, and its reconnect loop calls `DhanSession.log_in_again_without_checking`, which logs in whether or not another process already replaced the token, before connecting again and sending the login message with the new token.

This program scripts two connections. The first receives a binary frame; the second receives one traded order alert. No real socket is opened: a stand-in `websocket` package plays the connections, and a stand-in `stock_brokers.api.dhan` module counts each `DhanAPI` construction as one login so the real `DhanSession` runs unchanged. The backoff is set to zero seconds.

Notice that the second login message carries `dhan-token-2`, the token from the login made after the rejection.

Run it from the project root:

    python examples/stock_brokers/websockets/dhan/DhanOrderUpdatesSocket/example_2_binary_rejection_logs_in_again.py
"""

import json
import logging
import sys
import types

from stock_brokers.websockets.dhan import (
    DhanOrderUpdatesSocket,
    DhanSession,
)


class LoginRecord:
    """The Dhan login state every stand-in API object reads, standing in for the Redis `last_login` hash.

    Attributes:
        access_token (str | None): The token in force now.
        login_count (int): How many logins the stand-in API has made.
    """

    access_token = None
    login_count = 0


class StandInDhanAPI:
    """A stand-in for `DhanAPI` whose construction is a pretend login."""

    def __init__(self):
        """Logs in, which here only issues the next numbered token.

        Returns:
            None: This method returns nothing.
        """
        LoginRecord.login_count = LoginRecord.login_count + 1
        LoginRecord.access_token = f'dhan-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'client_id': '1100012345',
        }

    def _current_login(self):
        """The login in force now, as the real class reads it from Redis.

        Returns:
            dict | None: The access token, or None before any login.
        """
        if LoginRecord.access_token is None:
            return None
        return {
            'access_token': LoginRecord.access_token,
        }


class HandshakeRefused(Exception):
    """A refused websocket handshake, carrying the HTTP status the way `websocket-client` does.

    Attributes:
        status_code (int): The HTTP status the broker answered with.
    """

    def __init__(self, status_code):
        """Keeps the status.

        Args:
            status_code (int): The HTTP status the broker answered with.

        Returns:
            None: This method returns nothing.
        """
        super().__init__(f'Handshake status {status_code}')
        self.status_code = status_code


class ScriptedConnection:
    """A stand-in for `websocket.WebSocketApp` that either refuses the handshake or plays frames to the socket's handlers.

    Attributes:
        url (str): The URL the socket asked for.
        plan (dict): Either a `refuse_with` status or a list of `frames`.
        handlers (dict): The socket's callbacks, by name.
        closed (bool): Whether the socket asked to close, after which no more frames are delivered.
    """

    def __init__(self, url, plan, handlers):
        """Keeps what the socket passed in.

        Args:
            url (str): The URL the socket asked for.
            plan (dict): Either a `refuse_with` status or a list of `frames`.
            handlers (dict): The socket's callbacks, by name.

        Returns:
            None: This method returns nothing.
        """
        self.url = url
        self.plan = plan
        self.handlers = handlers
        self.closed = False

    def run_forever(self, **options):
        """Plays the plan and returns when the pretend connection closes.

        Args:
            **options (dict): The ping and TLS settings the socket asks for.

        Returns:
            None: This method returns nothing.
        """
        print(f'Connecting to {self.url}')
        if 'refuse_with' in self.plan:
            self.handlers['on_error'](self, HandshakeRefused(self.plan['refuse_with']))
            self.handlers['on_close'](self, None, None)
            return
        self.handlers['on_open'](self)
        for frame in self.plan['frames']:
            if self.closed:
                break
            self.handlers['on_message'](self, frame)
        self.handlers['on_close'](self, 1000, 'normal closure')

    def send(self, data, opcode=None):
        """Prints a message the socket sends to the broker.

        Args:
            data (str | bytes): The message.
            opcode (int | None): The frame type, when the socket names one.

        Returns:
            None: This method returns nothing.
        """
        print(f'Sent: {data}')

    def close(self, **options):
        """Notes that the socket asked to close.

        Args:
            **options (dict): Any close status the socket passes.

        Returns:
            None: This method returns nothing.
        """
        self.closed = True


class ScriptedWebsocketModule:
    """A stand-in for the `websocket` package that hands out scripted connections in turn.

    Attributes:
        plans (list): What each connection does, in the order the socket connects.
        when_finished (callable | None): Called when the last connection is handed out, so the socket stops after it.
    """

    def __init__(self, plans):
        """Keeps the connection plans.

        Args:
            plans (list): What each connection does, in the order the socket connects.

        Returns:
            None: This method returns nothing.
        """
        self.plans = plans
        self.when_finished = None

    def WebSocketApp(self, url, **handlers):
        """Builds the next scripted connection, closing the socket after the last one.

        Args:
            url (str): The feed URL.
            **handlers (dict): The socket's `on_open`, `on_message`, `on_error` and `on_close` callbacks.

        Returns:
            ScriptedConnection: The connection.
        """
        plan = self.plans.pop(0)
        if not self.plans:
            self.when_finished()
        return ScriptedConnection(url, plan, handlers)


class BinaryRejectionLogsInAgainExample:
    """Runs the Dhan order update socket through a rejected login and a good connection.

    Attributes:
        socket (DhanOrderUpdatesSocket): The socket being shown.
    """

    def __init__(self):
        """Builds the frames, the stand-ins, the session and the socket.

        Returns:
            None: This method returns nothing.
        """
        logging.basicConfig(stream=sys.stdout, format='%(levelname)s %(message)s', level=logging.INFO)
        api_module = types.ModuleType('stock_brokers.api.dhan')
        api_module.DhanAPI = StandInDhanAPI
        sys.modules['stock_brokers.api.dhan'] = api_module
        plans = [
            {
                'frames': [
                    b'\x00\x00\x32\x03invalid token',
                ],
            },
            {
                'frames': [
                    self.order_alert('Traded', 10),
                ],
            },
        ]
        websocket_module = ScriptedWebsocketModule(plans)
        sys.modules['websocket'] = websocket_module
        logger = logging.getLogger('dhan.orders')
        self.socket = DhanOrderUpdatesSocket(DhanSession(logger), self.print_orders, logger)
        self.socket.MIN_BACKOFF_SECONDS = 0
        websocket_module.when_finished = self.socket.close

    def order_alert(self, status, traded_quantity):
        """One of Dhan's order alert messages for an INFY buy.

        Args:
            status (str): Dhan's order status.
            traded_quantity (int): How much has traded.

        Returns:
            str: The message as JSON text.
        """
        alert = {
            'Type': 'order_alert',
            'Data': {
                'OrderNo': '1122092500001',
                'Status': status,
                'Symbol': 'INFY',
                'Exchange': 'NSE',
                'Segment': 'E',
                'TxnType': 'B',
                'Quantity': 10,
                'TradedQty': traded_quantity,
                'TradedPrice': 1499.5,
                'Product': 'C',
            },
        }
        return json.dumps(alert)

    def print_orders(self, orders, received_at):
        """Prints each order alert the socket handed on.

        Args:
            orders (list): The `Data` dictionaries of one message, exactly as Dhan sent them.
            received_at (datetime.datetime): When the message arrived, which is not printed.

        Returns:
            None: This method returns nothing.
        """
        for order in orders:
            print(f"Order {order['OrderNo']} {order['Status']}: {order['TxnType']} {order['TradedQty']} of {order['Quantity']} {order['Symbol']} at {order['TradedPrice']}")

    def run(self):
        """Runs the socket's reconnect loop until the scripted connections run out.

        Returns:
            None: This method returns nothing.
        """
        self.socket.run_forever()
        print(f'Gave up: {self.socket.gave_up}')
        print(f'Logins made: {LoginRecord.login_count}')


if __name__ == '__main__':
    BinaryRejectionLogsInAgainExample().run()

It prints:

Stand-in login number 1
Connecting to wss://api-order-update.dhan.co
Sent: {"LoginReq": {"MsgCode": 42, "ClientId": "1100012345", "Token": "dhan-token-1"}, "UserType": "SELF"}
INFO Order updates socket opened and logged in. Waiting for order updates.
WARNING Dhan sent a binary frame, taken as a refused login: b'\x00\x002\x03invalid token'
INFO Order updates socket closed with status 1000: normal closure
WARNING Logging in to Dhan again.
Stand-in login number 2
WARNING Order updates socket disconnected. Reconnecting in 0 seconds.
Connecting to wss://api-order-update.dhan.co
Sent: {"LoginReq": {"MsgCode": 42, "ClientId": "1100012345", "Token": "dhan-token-2"}, "UserType": "SELF"}
INFO Order updates socket opened and logged in. Waiting for order updates.
Order 1122092500001 Traded: B 10 of 10 INFY at 1499.5
INFO Order updates socket closed with status 1000: normal closure
INFO Order updates socket stopped.
Gave up: False
Logins made: 2

Methods:

Name	Description
`close`	Closes the connection and stops reconnecting.
`run_forever`	Connects, and reconnects with backoff, until closed or unable to recover.

Source code in stock_brokers/websockets/dhan.py

def __init__(self, session, on_updates, logger):
    """
    Sets up the order updates socket.

    Args:
        session (DhanSession): The login.
        on_updates (callable): Called on this socket's thread with each message's order alerts, as a list of `Data` dictionaries exactly as Dhan sent them, and the `datetime` the message was received.
        logger (logging.Logger): Where the connection reports.

    Returns:
        None: This method returns nothing.
    """
    super().__init__("Order updates socket", logger)
    self._session = session
    self._on_updates = on_updates

_connect ¶

_connect()

Opens the websocket and blocks until it closes; the login message is sent once it opens.

Returns:

Type	Description
`None`	This method returns nothing.

Source code in stock_brokers/websockets/dhan.py

def _connect(self):
    """
    Opens the websocket and blocks until it closes; the login message is sent once it opens.

    Returns:
        None: This method returns nothing.
    """
    import websocket

    self._websocket_application = websocket.WebSocketApp(
        ORDER_UPDATES_URL,
        on_open=self._on_open,
        on_message=self._on_message,
        on_error=self._on_error,
        on_close=self._on_close,
    )
    self._websocket_application.run_forever(
        ping_interval=PING_INTERVAL_SECONDS,
        ping_timeout=PING_TIMEOUT_SECONDS,
    )

_log_in_again ¶

_log_in_again()

Logs in again, without checking whether another process already replaced the token.

Returns:

Type	Description
`None`	This method returns nothing.

Source code in stock_brokers/websockets/dhan.py

def _log_in_again(self):
    """
    Logs in again, without checking whether another process already replaced the token.

    Returns:
        None: This method returns nothing.
    """
    self._session.log_in_again_without_checking()

_on_open ¶

_on_open(websocket_connection)

Sends the login message with the token in force now.

Parameters:

Name	Type	Description	Default
`websocket_connection`	`WebSocketApp`	The connection that opened.	required

Returns:

Type	Description
`None`	This method returns nothing.

Source code in stock_brokers/websockets/dhan.py

def _on_open(self, websocket_connection):
    """
    Sends the login message with the token in force now.

    Args:
        websocket_connection (websocket.WebSocketApp): The connection that opened.

    Returns:
        None: This method returns nothing.
    """
    self._opened = True
    client_id, token = self._session.credentials()
    websocket_connection.send(json.dumps({
        "LoginReq": {
            "MsgCode": LOGIN_MESSAGE_CODE,
            "ClientId": str(client_id),
            "Token": token,
        },
        "UserType": "SELF",
    }))
    self._logger.info(f"{self.name} opened and logged in. Waiting for order updates.")

_on_error ¶

_on_error(websocket_connection, error)

Reports an error.

Parameters:

Name	Type	Description	Default
`websocket_connection`	`WebSocketApp`	The connection that failed.	required
`error`	`Exception`	The error.	required

Returns:

Type	Description
`None`	This method returns nothing.

Source code in stock_brokers/websockets/dhan.py

def _on_error(self, websocket_connection, error):
    """
    Reports an error.

    Args:
        websocket_connection (websocket.WebSocketApp): The connection that failed.
        error (Exception): The error.

    Returns:
        None: This method returns nothing.
    """
    self._logger.error(f"Order updates error: {error}")

_on_message ¶

_on_message(websocket_connection, message)

Hands a message's order alerts on, and treats a binary frame as a refused login.

Parameters:

Name	Type	Description	Default
`websocket_connection`	`WebSocketApp`	The connection the message arrived on.	required
`message`	`bytes \| str`	The frame: one JSON message, a JSON list of them, or Dhan's binary rejection.	required

Returns:

Type	Description
`None`	This method returns nothing.

Source code in stock_brokers/websockets/dhan.py

def _on_message(self, websocket_connection, message):
    """
    Hands a message's order alerts on, and treats a binary frame as a refused login.

    Args:
        websocket_connection (websocket.WebSocketApp): The connection the message arrived on.
        message (bytes | str): The frame: one JSON message, a JSON list of them, or Dhan's binary rejection.

    Returns:
        None: This method returns nothing.
    """
    if isinstance(message, (bytes, bytearray)):
        self._logger.warning(f"Dhan sent a binary frame, taken as a refused login: {bytes(message)[:80]!r}")
        self._authentication_rejected = True
        websocket_connection.close()
        return

    try:
        payload = json.loads(message)
    except (TypeError, json.JSONDecodeError):
        self._logger.debug(f"Ignoring a frame that is not JSON: {str(message)[:160]}")
        return

    received_at = datetime.now()
    entries = payload
    if not isinstance(payload, list):
        entries = [payload]
    orders = []
    for entry in entries:
        if not isinstance(entry, dict):
            continue
        if entry.get("Type") == ORDER_ALERT:
            orders.append(entry.get("Data"))
        else:
            self._logger.info(f"Ignoring a message of type {entry.get('Type')!r}: {json.dumps(entry)[:200]}")
    if orders:
        self._on_updates(orders, received_at)

close ¶

close()

Closes the connection and stops reconnecting.

Returns:

Type	Description
`None`	This method returns nothing.

Example usage

From example_2_giving_up_after_a_second_refusal.py

self.closed_socket.close()

Source code in stock_brokers/websockets/base.py

def close(self):
    """
    Closes the connection and stops reconnecting.

    Returns:
        None: This method returns nothing.
    """
    self._stop.set()
    if self._websocket_application is not None:
        self._websocket_application.close()

run_forever ¶

run_forever()

Connects, and reconnects with backoff, until closed or unable to recover.

Returns:

Type	Description
`None`	This method returns nothing. `gave_up` says whether the socket stopped because it could not recover.

Example usage

From example_1_reconnecting_after_a_drop.py

self.socket.run_forever()

From example_2_giving_up_after_a_second_refusal.py

self.refused_socket.run_forever()

From example_2_giving_up_after_a_second_refusal.py

self.closed_socket.run_forever()

Source code in stock_brokers/websockets/base.py

def run_forever(self):
    """
    Connects, and reconnects with backoff, until closed or unable to recover.

    Returns:
        None: This method returns nothing. `gave_up` says whether the socket stopped because it could not recover.
    """
    backoff = self.MIN_BACKOFF_SECONDS
    failed_connects = 0
    logged_in_again = False
    while not self._stop.is_set():
        self._opened = False
        self._authentication_rejected = False
        try:
            self._connect()
        except Exception as exception:
            self._logger.error(f"{self.name} connection failed: {type(exception).__name__}: {exception}")
        if self._stop.is_set():
            break

        if self._opened and not self._authentication_rejected:
            failed_connects = 0
            logged_in_again = False
            backoff = self.MIN_BACKOFF_SECONDS
        else:
            failed_connects = failed_connects + 1

        if self._authentication_rejected or failed_connects >= self.MAX_FAILED_CONNECTS:
            if logged_in_again and self._gives_up_after_logging_in_again(failed_connects):
                self.gave_up = True
                self._logger.error(f"{self.name} still cannot connect after logging in again. Giving up.")
                break
            try:
                self._log_in_again()
            except Exception as exception:
                self.gave_up = True
                self._logger.error(f"{self.name} could not log in again: {type(exception).__name__}: {exception}")
                break
            failed_connects = 0
            logged_in_again = True

        self._logger.warning(f"{self.name} disconnected. Reconnecting in {backoff} seconds.")
        self._stop.wait(backoff)
        backoff = min(backoff * 2, self.MAX_BACKOFF_SECONDS)
    self._logger.info(f"{self.name} stopped.")

_gives_up_after_logging_in_again ¶

_gives_up_after_logging_in_again(failed_connects)

Whether a socket that has already logged in again gives up on this failure.

The loop asks only when the connection was refused or has failed six times in a row. By default the answer is yes, so a refusal straight after logging in again ends the socket. A broker whose refusals are not always about the login overrides this to give up only after six more failed connects.

Parameters:

Name	Type	Description	Default
`failed_connects`	`int`	How many connects in a row have failed since the last login.	required

Returns:

Type	Description
`bool`	True to give up now.

Source code in stock_brokers/websockets/base.py

def _gives_up_after_logging_in_again(self, failed_connects):
    """
    Whether a socket that has already logged in again gives up on this failure.

    The loop asks only when the connection was refused or has failed six times in a row.
    By default the answer is yes, so a refusal straight after logging in again ends the socket.
    A broker whose refusals are not always about the login overrides this to give up only after six more failed connects.

    Args:
        failed_connects (int): How many connects in a row have failed since the last login.

    Returns:
        bool: True to give up now.
    """
    return True

_on_close ¶

_on_close(websocket_connection, status_code, message)

Reports that the connection closed.

Parameters:

Name	Type	Description	Default
`websocket_connection`	`WebSocketApp`	The connection that closed.	required
`status_code`	`int \| None`	The close status the broker sent.	required
`message`	`str \| None`	The close reason the broker sent.	required

Returns:

Type	Description
`None`	This method returns nothing.

Source code in stock_brokers/websockets/base.py

def _on_close(self, websocket_connection, status_code, message):
    """
    Reports that the connection closed.

    Args:
        websocket_connection (websocket.WebSocketApp): The connection that closed.
        status_code (int | None): The close status the broker sent.
        message (str | None): The close reason the broker sent.

    Returns:
        None: This method returns nothing.
    """
    self._logger.info(f"{self.name} closed with status {status_code}: {message}")