Skip to content

indmoney

INDmoney's two websockets: the INDstocks price feed and the account's order updates.

The login. IndmoneySession holds the INDMoneyAPI both sockets read their credentials from, afresh on every connect, so a login made by any process is picked up. The token travels in the handshake as Authorization, with the client id as x-api-key when there is one, and INDstocks refuses a dead token at the handshake itself with HTTP 401, 403 or 513. 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. IndmoneyQuotesSocket connects to wss://ws-prices.indstocks.com/api/v1/ws/prices and subscribes its batch in full mode, the richest the feed serves: ltp carries only the last price, quote carries the day's open, high, low, close and volume without a last price, and depth, market_depth, depth5 and snapquote send nothing. Every update arrives as a JSON string whose content is itself JSON, followed by a newline, so each line of a frame is decoded twice. An update names the instrument by its bare security id, a millisecond timestamp, and a data object carrying only what changed, so each instrument's fields are merged across updates. Every socket carries instruments of one segment only, because only the socket's segment places a bare security id. A message without data is logged rather than silently dropped, and no tick is handed on until an instrument has a last price. close is sent as the last price, not the previous close, and the best bid and offer are the only depth, without quantities.

Order updates. IndmoneyOrderUpdatesSocket connects to wss://ws-order-updates.indstocks.com/api/v1/ws/trades and asks for order_update messages. Subscription acknowledgements and heartbeats share the connection; an order arrives as {"type": "order", "data": {...}}, or with the order's fields beside type when there is no data object.

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

Classes:

Name Description
IndmoneySession

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

One INDstocks price feed websocket carrying the full quotes of a batch of instruments of one segment.
IndmoneyOrderUpdatesSocket

INDstocks' order update websocket, authenticated by its handshake.

IndmoneySession 

IndmoneySession(logger)

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

Constructs INDMoneyAPI, 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 INDMoneyAPI raises when it cannot log in.

Example programs

examples/stock_brokers/websockets/indmoney/IndmoneySession/example_1_handshake_headers.py
"""Builds the handshake headers an INDmoney socket presents and logs in again only when the failed token is still current.

`IndmoneySession` holds one `INDMoneyAPI` and reads the login afresh whenever a socket connects. `credentials` returns the client id and the token in force, and `handshake_headers` turns them into the headers INDstocks expects in the websocket handshake: the token as `Authorization` and the client id as `x-api-key`. When a quotes socket is refused, `log_in_again` logs in only if the token it used is still the one in force.

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

Notice that the headers change after the login, and that the second request with the stale token makes no login.

Run it from the project root:

    python examples/stock_brokers/websockets/indmoney/IndmoneySession/example_1_handshake_headers.py
"""

import logging
import sys
import types

from stock_brokers.websockets.indmoney import (
    IndmoneySession,
)


class LoginRecord:
    """The INDmoney 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 StandInINDMoneyAPI:
    """A stand-in for `INDMoneyAPI` 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'indmoney-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'client_id': 'ind-client-7',
        }

    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 HandshakeHeadersExample:
    """Prints an INDmoney session's credentials and handshake headers around a login.

    Attributes:
        session (IndmoneySession): 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.indmoney')
        api_module.INDMoneyAPI = StandInINDMoneyAPI
        sys.modules['stock_brokers.api.indmoney'] = api_module
        self.session = IndmoneySession(logging.getLogger('indmoney.session'))

    def run(self):
        """Prints the headers, logs in again twice with the same stale token, and prints them again.

        Returns:
            None: This method returns nothing.
        """
        client_id, failed_token = self.session.credentials()
        print(f'Credentials: client_id={client_id} access_token={failed_token}')
        headers, token = self.session.handshake_headers()
        print(f'Handshake headers: {headers} for token {token}')
        self.session.log_in_again(failed_token)
        self.session.log_in_again(failed_token)
        headers, token = self.session.handshake_headers()
        print(f'Handshake headers: {headers} for token {token}')
        print(f'Logins made: {LoginRecord.login_count}')


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

It prints:

Stand-in login number 1
Credentials: client_id=ind-client-7 access_token=indmoney-token-1
Handshake headers: {'Authorization': 'indmoney-token-1', 'x-api-key': 'ind-client-7'} for token indmoney-token-1
WARNING Logging in to INDmoney again.
Stand-in login number 2
Handshake headers: {'Authorization': 'indmoney-token-2', 'x-api-key': 'ind-client-7'} for token indmoney-token-2
Logins made: 2
examples/stock_brokers/websockets/indmoney/IndmoneySession/example_2_no_client_id_and_unchecked_login.py
"""Shows the handshake headers without a client id, and a login made without checking the token.

An INDmoney account whose settings have no client id presents only the `Authorization` header, because `handshake_headers` adds `x-api-key` only when there is a client id. The order update socket logs in again with `log_in_again_without_checking`, which logs in even when another process already replaced the token that failed.

Constructing the real `INDMoneyAPI` logs in to INDmoney, so the program registers a stand-in module under `stock_brokers.api.indmoney` whose settings hold no client id and whose constructions count as logins. The program replaces the token as another process would before logging in.

Notice that the headers hold `Authorization` alone, and that the unconditional login replaces the other process's token.

Run it from the project root:

    python examples/stock_brokers/websockets/indmoney/IndmoneySession/example_2_no_client_id_and_unchecked_login.py
"""

import logging
import sys
import types

from stock_brokers.websockets.indmoney import (
    IndmoneySession,
)


class LoginRecord:
    """The INDmoney 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 StandInINDMoneyAPI:
    """A stand-in for `INDMoneyAPI` 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'indmoney-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'client_id': None,
        }

    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 NoClientIdAndUncheckedLoginExample:
    """Prints headers without a client id and logs in again unconditionally.

    Attributes:
        session (IndmoneySession): 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.indmoney')
        api_module.INDMoneyAPI = StandInINDMoneyAPI
        sys.modules['stock_brokers.api.indmoney'] = api_module
        self.session = IndmoneySession(logging.getLogger('indmoney.session'))

    def run(self):
        """Prints the headers, replaces the token elsewhere, and logs in without checking.

        Returns:
            None: This method returns nothing.
        """
        headers, token = self.session.handshake_headers()
        print(f'Handshake headers: {headers}')
        LoginRecord.access_token = 'indmoney-token-from-another-process'
        print(f'Another process replaced {token}.')
        self.session.log_in_again_without_checking()
        print(f'Credentials now: {self.session.credentials()}')
        print(f'Logins made: {LoginRecord.login_count}')


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

It prints:

Stand-in login number 1
Handshake headers: {'Authorization': 'indmoney-token-1'}
Another process replaced indmoney-token-1.
WARNING Logging in to INDmoney again.
Stand-in login number 2
Credentials now: (None, 'indmoney-token-2')
Logins made: 2

Methods:

Name Description
credentials

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

The headers that present the credentials in force now in the websocket handshake.
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/indmoney.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def __init__(self, logger):
    """
    Constructs `INDMoneyAPI`, 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 `INDMoneyAPI` raises when it cannot log in.
    """
    from stock_brokers.api.indmoney import INDMoneyAPI

    self._api_class = INDMoneyAPI
    self._logger = logger
    self._lock = threading.Lock()
    self._indmoney = INDMoneyAPI()

credentials 

credentials()

The api key (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_handshake_headers.py
client_id, failed_token = self.session.credentials()
From example_2_no_client_id_and_unchecked_login.py
print(f'Credentials now: {self.session.credentials()}')
Source code in stock_brokers/websockets/indmoney.py 
74
75
76
77
78
79
80
81
82
def credentials(self):
    """
    The api key (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._indmoney._current_login() or {}
    return self._indmoney._settings.get("client_id"), login.get("access_token")

handshake_headers 

handshake_headers()

The headers that present the credentials in force now in the websocket handshake.

Returns:

Type Description
tuple

The headers as a dictionary, and the access token they carry.

Example usage

From example_1_handshake_headers.py
headers, token = self.session.handshake_headers()
From example_2_no_client_id_and_unchecked_login.py
headers, token = self.session.handshake_headers()
Source code in stock_brokers/websockets/indmoney.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
def handshake_headers(self):
    """
    The headers that present the credentials in force now in the websocket handshake.

    Returns:
        tuple: The headers as a dictionary, and the access token they carry.
    """
    client_id, token = self.credentials()
    headers = {
        "Authorization": str(token),
    }
    if client_id:
        headers["x-api-key"] = str(client_id)
    return headers, 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 INDMoneyAPI raises when it cannot log in.

Example usage

From example_1_handshake_headers.py
self.session.log_in_again(failed_token)
Source code in stock_brokers/websockets/indmoney.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
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 `INDMoneyAPI` 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 INDMoneyAPI raises when it cannot log in.

Example usage

From example_2_no_client_id_and_unchecked_login.py
self.session.log_in_again_without_checking()
Source code in stock_brokers/websockets/indmoney.py 
117
118
119
120
121
122
123
124
125
126
127
128
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 `INDMoneyAPI` raises when it cannot log in.
    """
    with self._lock:
        self._log_in()

_log_in 

_log_in()

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

Returns:

Type Description
None

This method returns nothing.

Raises:

Type Description
Exception

Whatever INDMoneyAPI raises when it cannot log in.
Source code in stock_brokers/websockets/indmoney.py 
130
131
132
133
134
135
136
137
138
139
140
141
def _log_in(self):
    """
    Logs in by constructing `INDMoneyAPI` again; the caller holds the lock.

    Returns:
        None: This method returns nothing.

    Raises:
        Exception: Whatever `INDMoneyAPI` raises when it cannot log in.
    """
    self._logger.warning("Logging in to INDmoney again.")
    self._indmoney = self._api_class()

IndmoneyQuotesSocket 

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

Bases: BrokerWebsocket

One INDstocks price feed websocket carrying the full quotes of a batch of instruments of one segment.

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:TOKEN instrument tokens, all of one segment.

 required
names dict[str, str]

Each token to the instrument's name, which becomes a tick's id.

 required
session IndmoneySession

The shared login.

 required
on_ticks callable

Called with each frame'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

examples/stock_brokers/websockets/indmoney/IndmoneyQuotesSocket/example_1_merging_double_encoded_updates.py
"""Decodes INDstocks price feed updates, which are JSON inside JSON, and merges them into ticks.

`IndmoneyQuotesSocket` presents the token in the websocket handshake and subscribes its batch in `full` mode. Every update arrives as a JSON string whose content is itself JSON, one per line, so each line is decoded twice. An update names the instrument by its bare security id and carries only the fields that changed, so the socket maps the id back to the subscribed `SEGMENT:TOKEN` and merges each instrument's fields across updates. No tick is handed on until an instrument has a last price.

This program scripts one connection whose single frame holds two lines for NSE security 2885: the first carries the day's open, high, low, close and volume without a last price, and the second carries the last price, the best bid and offer and a millisecond timestamp. No real socket is opened: a stand-in `websocket` package plays the frame and prints the handshake headers and what the socket sends, and a stand-in `stock_brokers.api.indmoney` module lets the real `IndmoneySession` supply credentials without logging in.

Notice that only one tick comes out, that it holds the fields of both lines, that the millisecond timestamp becomes epoch seconds, and that the depth is the best bid and offer only, without quantities.

Run it from the project root:

    python examples/stock_brokers/websockets/indmoney/IndmoneyQuotesSocket/example_1_merging_double_encoded_updates.py
"""

import json
import logging
import sys
import types

from stock_brokers.websockets.indmoney import (
    IndmoneyQuotesSocket,
    IndmoneySession,
)


class LoginRecord:
    """The INDmoney 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 StandInINDMoneyAPI:
    """A stand-in for `INDMoneyAPI` 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'indmoney-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'client_id': 'ind-client-7',
        }

    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.
        """
        print(f"Handshake headers: {handlers.get('header')}")
        plan = self.plans.pop(0)
        if not self.plans:
            self.when_finished()
        return ScriptedConnection(url, plan, handlers)


class MergingDoubleEncodedUpdatesExample:
    """Runs an INDmoney quotes socket against one scripted frame of two lines.

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

    def __init__(self):
        """Builds the frame, 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.indmoney')
        api_module.INDMoneyAPI = StandInINDMoneyAPI
        sys.modules['stock_brokers.api.indmoney'] = api_module
        day_fields = {
            'instrument': '2885',
            'data': {
                'open': 2930.0,
                'high': 2960.0,
                'low': 2925.0,
                'close': 2950.0,
                'volume': 1204500.0,
            },
        }
        price_fields = {
            'instrument': '2885',
            'timestamp': 1790311529000,
            'data': {
                'ltp': 2950.5,
                'ltt': 1790311528,
                'bid_price': 2950.0,
                'ask_price': 2950.5,
                'total_buy_qty': 54000,
                'total_sell_qty': 61000,
            },
        }
        frame = self.feed_line(day_fields) + self.feed_line(price_fields)
        plans = [
            {
                'frames': [
                    frame,
                ],
            },
        ]
        websocket_module = ScriptedWebsocketModule(plans)
        sys.modules['websocket'] = websocket_module
        logger = logging.getLogger('indmoney.quotes')
        tokens = [
            'NSE:2885',
        ]
        names = {
            'NSE:2885': 'NSE:RELIANCE',
        }
        self.socket = IndmoneyQuotesSocket('socket_0', tokens, names, IndmoneySession(logger), self.print_ticks, logger)
        websocket_module.when_finished = self.socket.close

    def feed_line(self, update):
        """One line as INDstocks sends it: a JSON string whose content is itself JSON, then a newline.

        Args:
            update (dict): The update.

        Returns:
            str: The line.
        """
        return json.dumps(json.dumps(update)) + '\n'

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

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

        Returns:
            None: This method returns nothing.
        """
        for tick in ticks:
            print(f"Tick {tick['id']} ({tick['instrument_token']}) on {tick['exchange']}")
            print(f"  last_price={tick['last_price']} volume={tick['volume']} buy_quantity={tick['buy_quantity']} sell_quantity={tick['sell_quantity']}")
            print(f"  ohlc={tick['ohlc']}")
            print(f"  last_trade_time={tick['last_trade_time']} exchange_timestamp={tick['exchange_timestamp']}")
            print(f"  depth={tick['depth']}")

    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__':
    MergingDoubleEncodedUpdatesExample().run()

It prints:

Stand-in login number 1
Handshake headers: {'Authorization': 'indmoney-token-1', 'x-api-key': 'ind-client-7'}
Connecting to wss://ws-prices.indstocks.com/api/v1/ws/prices
Sent: {"action": "subscribe", "mode": "full", "instruments": ["NSE:2885"]}
INFO socket_0 opened and subscribed to 1 instrument(s) in full mode.
Tick NSE:RELIANCE (NSE:2885) on NSE
  last_price=2950.5 volume=1204500 buy_quantity=54000 sell_quantity=61000
  ohlc={'open': 2930.0, 'high': 2960.0, 'low': 2925.0, 'close': 2950.0}
  last_trade_time=1790311528 exchange_timestamp=1790311529
  depth={'buy': [{'quantity': None, 'price': 2950.0, 'orders': None}], 'sell': [{'quantity': None, 'price': 2950.5, 'orders': None}]}
INFO socket_0 closed with status 1000: normal closure
INFO socket_0 stopped.
Gave up: False
examples/stock_brokers/websockets/indmoney/IndmoneyQuotesSocket/example_2_refused_handshake.py
"""Shows an INDmoney quotes socket logging in again after INDstocks refuses its handshake with HTTP 513.

INDstocks refuses a dead token at the websocket handshake with HTTP 401, 403 or 513. `IndmoneyQuotesSocket` takes any of those as a dead session, and its reconnect loop asks the shared `IndmoneySession` to log in again, skipping the login if another socket or process already replaced the token, before connecting with new handshake headers.

This program scripts two connections. The first is refused with 513. The second delivers a message without instrument data, which the socket logs rather than drops, and then one BSE update carrying a last price. No real socket is opened: a stand-in `websocket` package plays the connections and prints each handshake's headers, and a stand-in `stock_brokers.api.indmoney` module counts each `INDMoneyAPI` construction as one login. The backoff is set to zero seconds.

Notice that the second handshake carries the new token in `Authorization`, and that the logged message is the subscription acknowledgement INDstocks sends without `data`.

Run it from the project root:

    python examples/stock_brokers/websockets/indmoney/IndmoneyQuotesSocket/example_2_refused_handshake.py
"""

import json
import logging
import sys
import types

from stock_brokers.websockets.indmoney import (
    IndmoneyQuotesSocket,
    IndmoneySession,
)


class LoginRecord:
    """The INDmoney 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 StandInINDMoneyAPI:
    """A stand-in for `INDMoneyAPI` 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'indmoney-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'client_id': 'ind-client-7',
        }

    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.
        """
        print(f"Handshake headers: {handlers.get('header')}")
        plan = self.plans.pop(0)
        if not self.plans:
            self.when_finished()
        return ScriptedConnection(url, plan, handlers)


class RefusedHandshakeExample:
    """Runs an INDmoney quotes socket through a refused handshake and a good connection.

    Attributes:
        socket (IndmoneyQuotesSocket): 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.indmoney')
        api_module.INDMoneyAPI = StandInINDMoneyAPI
        sys.modules['stock_brokers.api.indmoney'] = api_module
        acknowledgement = {
            'status': 'subscribed',
            'mode': 'full',
        }
        update = {
            'instrument': '500325',
            'timestamp': 1790311529000,
            'data': {
                'ltp': 2951.0,
                'volume': 88000,
            },
        }
        plans = [
            {
                'refuse_with': 513,
            },
            {
                'frames': [
                    self.feed_line(acknowledgement),
                    self.feed_line(update),
                ],
            },
        ]
        websocket_module = ScriptedWebsocketModule(plans)
        sys.modules['websocket'] = websocket_module
        logger = logging.getLogger('indmoney.quotes')
        tokens = [
            'BSE:500325',
        ]
        names = {
            'BSE:500325': 'BSE:RELIANCE',
        }
        self.socket = IndmoneyQuotesSocket('socket_0', tokens, names, IndmoneySession(logger), self.print_ticks, logger)
        self.socket.MIN_BACKOFF_SECONDS = 0
        websocket_module.when_finished = self.socket.close

    def feed_line(self, update):
        """One line as INDstocks sends it: a JSON string whose content is itself JSON, then a newline.

        Args:
            update (dict): The update.

        Returns:
            str: The line.
        """
        return json.dumps(json.dumps(update)) + '\n'

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

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

        Returns:
            None: This method returns nothing.
        """
        for tick in ticks:
            print(f"Tick {tick['id']} ({tick['instrument_token']}) on {tick['exchange']}")
            print(f"  last_price={tick['last_price']} volume={tick['volume']} buy_quantity={tick['buy_quantity']} sell_quantity={tick['sell_quantity']}")
            print(f"  ohlc={tick['ohlc']}")
            print(f"  last_trade_time={tick['last_trade_time']} exchange_timestamp={tick['exchange_timestamp']}")
            print(f"  depth={tick['depth']}")

    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__':
    RefusedHandshakeExample().run()

It prints:

Stand-in login number 1
Handshake headers: {'Authorization': 'indmoney-token-1', 'x-api-key': 'ind-client-7'}
Connecting to wss://ws-prices.indstocks.com/api/v1/ws/prices
WARNING socket_0 handshake refused with HTTP 513, taken as a dead session.
INFO socket_0 closed with status None: None
WARNING Logging in to INDmoney again.
Stand-in login number 2
WARNING socket_0 disconnected. Reconnecting in 0 seconds.
Handshake headers: {'Authorization': 'indmoney-token-2', 'x-api-key': 'ind-client-7'}
Connecting to wss://ws-prices.indstocks.com/api/v1/ws/prices
Sent: {"action": "subscribe", "mode": "full", "instruments": ["BSE:500325"]}
INFO socket_0 opened and subscribed to 1 instrument(s) in full mode.
INFO socket_0 message without instrument data: {"status": "subscribed", "mode": "full"}
Tick BSE:RELIANCE (BSE:500325) on BSE
  last_price=2951.0 volume=88000 buy_quantity=None sell_quantity=None
  ohlc={'open': None, 'high': None, 'low': None, 'close': None}
  last_trade_time=None exchange_timestamp=1790311529
  depth={'buy': [], 'sell': []}
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/indmoney.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
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:TOKEN` instrument tokens, all of one segment.
        names (dict[str, str]): Each token to the instrument's name, which becomes a tick's `id`.
        session (IndmoneySession): The shared login.
        on_ticks (callable): Called with each frame'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._state = {}

_connect 

_connect()

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

Returns:

Type Description
None

This method returns nothing.
Source code in stock_brokers/websockets/indmoney.py 
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def _connect(self):
    """
    Opens the websocket with the token in force now, presented in the handshake, and blocks until it closes.

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

    headers, self._token = self._session.handshake_headers()
    self._websocket_application = websocket.WebSocketApp(
        FEED_URL,
        header=headers,
        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/indmoney.py 
195
196
197
198
199
200
201
202
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.

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/indmoney.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
def _on_open(self, websocket_connection):
    """
    Subscribes the batch in full mode.

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

    Returns:
        None: This method returns nothing.
    """
    self._opened = True
    self._state = {}
    websocket_connection.send(json.dumps({
        "action": "subscribe",
        "mode": SUBSCRIPTION_MODE,
        "instruments": list(self._tokens),
    }))
    self._logger.info(f"{self.name} opened and subscribed to {len(self._tokens)} instrument(s) in {SUBSCRIPTION_MODE} mode.")

_on_error 

_on_error(websocket_connection, error)

Takes a refused handshake as a dead session, and reports any other error.

Parameters:

Name Type Description Default
websocket_connection WebSocketApp

The connection that failed.

 required
error Exception

The error, which carries status_code when the handshake was refused.

 required

Returns:

Type Description
None

This method returns nothing.
Source code in stock_brokers/websockets/indmoney.py 
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
def _on_error(self, websocket_connection, error):
    """
    Takes a refused handshake as a dead session, and reports any other error.

    Args:
        websocket_connection (websocket.WebSocketApp): The connection that failed.
        error (Exception): The error, which carries `status_code` when the handshake was refused.

    Returns:
        None: This method returns nothing.
    """
    status = getattr(error, "status_code", None)
    if status in AUTHENTICATION_STATUS_CODES:
        self._authentication_rejected = True
        self._logger.warning(f"{self.name} handshake refused with HTTP {status}, taken as a dead session.")
        return
    self._logger.error(f"{self.name} error: {error}")

_on_message 

_on_message(websocket_connection, message)

Decodes a frame into ticks and hands them on.

Parameters:

Name Type Description Default
websocket_connection WebSocketApp

The connection the frame arrived on.

 required
message bytes | str

The frame.

 required

Returns:

Type Description
None

This method returns nothing.
Source code in stock_brokers/websockets/indmoney.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
def _on_message(self, websocket_connection, message):
    """
    Decodes a frame into ticks and hands them on.

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

    Returns:
        None: This method returns nothing.
    """
    ticks = self._parse_ticks(message)
    if ticks:
        self._on_ticks(ticks)

_decode_entries 

_decode_entries(message)

Decodes one frame into its update dictionaries, decoding a line a second time when it holds a JSON string.

Parameters:

Name Type Description Default
message bytes | str

The frame.

 required

Returns:

Type Description
list[dict]

The decoded updates, in order.
Source code in stock_brokers/websockets/indmoney.py 
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def _decode_entries(self, message):
    """
    Decodes one frame into its update dictionaries, decoding a line a second time when it holds a JSON string.

    Args:
        message (bytes | str): The frame.

    Returns:
        list[dict]: The decoded updates, in order.
    """
    if isinstance(message, (bytes, bytearray)):
        message = message.decode("utf-8", errors="replace")
    if not isinstance(message, str):
        return []

    entries = []
    for line in message.splitlines():
        if not line.strip():
            continue
        try:
            payload = json.loads(line)
            if isinstance(payload, str):
                payload = json.loads(payload)
        except json.JSONDecodeError:
            self._logger.debug(f"{self.name} ignored a line that is not JSON: {line[:120]}")
            continue
        if isinstance(payload, list):
            for item in payload:
                if isinstance(item, dict):
                    entries.append(item)
        elif isinstance(payload, dict):
            entries.append(payload)
    return entries

_parse_ticks 

_parse_ticks(message)

Builds the ticks in one frame, merging each instrument's fields across updates.

Parameters:

Name Type Description Default
message bytes | str

The frame.

 required

Returns:

Type Description
list[dict]

One tick per update that left its instrument with a last price.
Source code in stock_brokers/websockets/indmoney.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
def _parse_ticks(self, message):
    """
    Builds the ticks in one frame, merging each instrument's fields across updates.

    Args:
        message (bytes | str): The frame.

    Returns:
        list[dict]: One tick per update that left its instrument with a last price.
    """
    ticks = []
    for entry in self._decode_entries(message):
        data = entry.get("data")
        instrument = entry.get("instrument")
        if not isinstance(data, dict) or instrument is None:
            self._logger.info(f"{self.name} message without instrument data: {json.dumps(entry)[:200]}")
            continue

        instrument = str(instrument)
        if ":" in instrument:
            token = instrument
        else:
            token = self._subscribed_token(instrument)
        state = self._state.setdefault(token, {})
        state.update(data)
        if entry.get("timestamp") is not None:
            state["timestamp"] = entry["timestamp"]

        tick = self._build_tick(token, state)
        if tick["last_price"] is None:
            continue
        ticks.append(tick)
    return ticks

_subscribed_token 

_subscribed_token(bare)

The subscribed SEGMENT:TOKEN for a bare security id; a socket carries one segment, so at most one matches.

Parameters:

Name Type Description Default
bare str

The security id as the feed sent it.

 required

Returns:

Type Description
str

The subscribed token, or the bare id when this socket did not subscribe it.
Source code in stock_brokers/websockets/indmoney.py 
324
325
326
327
328
329
330
331
332
333
334
335
336
337
def _subscribed_token(self, bare):
    """
    The subscribed `SEGMENT:TOKEN` for a bare security id; a socket carries one segment, so at most one matches.

    Args:
        bare (str): The security id as the feed sent it.

    Returns:
        str: The subscribed token, or the bare id when this socket did not subscribe it.
    """
    for token in self._tokens:
        if token.endswith(f":{bare}"):
            return token
    return bare

_epoch_seconds 

_epoch_seconds(value)

A feed time as whole epoch seconds.

Parameters:

Name Type Description Default
value int | float | str | None

The time as sent, in milliseconds, or in seconds for a small value.

 required

Returns:

Type Description
int | None

Epoch seconds, or None for a missing or zero time.
Source code in stock_brokers/websockets/indmoney.py 
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
def _epoch_seconds(self, value):
    """
    A feed time as whole epoch seconds.

    Args:
        value (int | float | str | None): The time as sent, in milliseconds, or in seconds for a small value.

    Returns:
        int | None: Epoch seconds, or None for a missing or zero time.
    """
    moment = self._number(value)
    if not moment:
        return None
    if moment > 1e11:
        return int(moment / 1000)
    return int(moment)

_build_tick 

_build_tick(token, state)

Builds a normalized tick from an instrument's merged full mode fields.

Parameters:

Name Type Description Default
token str

The instrument's SEGMENT:TOKEN.

 required
state dict

Its fields merged across updates, with the latest update's timestamp.

 required

Returns:

Type Description
dict

The tick.
Source code in stock_brokers/websockets/indmoney.py 
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
def _build_tick(self, token, state):
    """
    Builds a normalized tick from an instrument's merged full mode fields.

    Args:
        token (str): The instrument's `SEGMENT:TOKEN`.
        state (dict): Its fields merged across updates, with the latest update's `timestamp`.

    Returns:
        dict: The tick.
    """
    buy_levels = []
    bid_price = self._number(state.get("bid_price"))
    if bid_price is not None:
        buy_levels.append({
            "quantity": None,
            "price": bid_price,
            "orders": None,
        })
    sell_levels = []
    ask_price = self._number(state.get("ask_price"))
    if ask_price is not None:
        sell_levels.append({
            "quantity": None,
            "price": ask_price,
            "orders": None,
        })

    exchange = None
    if ":" in token:
        exchange = token.split(":", 1)[0]
    return {
        "id": self._names.get(token) or token,
        "broker": "indmoney",
        "instrument_token": token,
        "exchange": exchange,
        "mode": "quote",
        "last_price": self._number(state.get("ltp")),
        "last_quantity": None,
        "average_price": None,
        "volume": self._whole(state.get("volume")),
        "buy_quantity": self._whole(state.get("total_buy_qty")),
        "sell_quantity": self._whole(state.get("total_sell_qty")),
        "ohlc": {
            "open": self._number(state.get("open")),
            "high": self._number(state.get("high")),
            "low": self._number(state.get("low")),
            "close": self._number(state.get("close")),
        },
        "change": None,
        "oi": self._whole(state.get("oi")),
        "oi_day_high": None,
        "oi_day_low": None,
        "last_trade_time": self._epoch_seconds(state.get("ltt")),
        "exchange_timestamp": self._epoch_seconds(state.get("timestamp")),
        "depth": {
            "buy": buy_levels,
            "sell": sell_levels,
        },
        "received_at": time.time(),
    }

_number 

_number(value)

A feed value as a float.

Parameters:

Name Type Description Default
value object

The raw value.

 required

Returns:

Type Description
float | None

The number, or None for a blank, null or unreadable value.
Source code in stock_brokers/websockets/indmoney.py 
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
def _number(self, value):
    """
    A feed value as a float.

    Args:
        value (object): The raw value.

    Returns:
        float | None: The number, or None for a blank, null or unreadable value.
    """
    if value is None or value == "":
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None

_whole 

_whole(value)

A feed quantity as an integer; quantities are counts, so a float such as 12.0 becomes 12.

Parameters:

Name Type Description Default
value object

The raw value.

 required

Returns:

Type Description
int | None

The count, or None.
Source code in stock_brokers/websockets/indmoney.py 
435
436
437
438
439
440
441
442
443
444
445
446
447
448
def _whole(self, value):
    """
    A feed quantity as an integer; quantities are counts, so a float such as 12.0 becomes 12.

    Args:
        value (object): The raw value.

    Returns:
        int | None: The count, or None.
    """
    value = self._number(value)
    if value is None:
        return None
    return int(round(value))

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 
53
54
55
56
57
58
59
60
61
62
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 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
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 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
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 
152
153
154
155
156
157
158
159
160
161
162
163
164
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}")

IndmoneyOrderUpdatesSocket 

IndmoneyOrderUpdatesSocket(session, on_updates, logger)

Bases: BrokerWebsocket

INDstocks' order update websocket, authenticated by its handshake.

Sets up the order updates socket.

Parameters:

Name Type Description Default
session IndmoneySession

The login.

 required
on_updates callable

Called on this socket's thread with each message's orders, as dictionaries exactly as INDstocks 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

examples/stock_brokers/websockets/indmoney/IndmoneyOrderUpdatesSocket/example_1_orders_in_both_shapes.py
"""Receives INDstocks order updates in both of the shapes the feed uses and prints the orders handed on.

`IndmoneyOrderUpdatesSocket` is authenticated by its handshake headers and asks for `order_update` messages once it opens. The subscription acknowledgement and heartbeats share the connection with the orders. An order arrives either as `{"type": "order", "data": {...}}` or with the order's fields beside `type` when there is no `data` object, and the socket hands on the order in both cases.

This program scripts one connection delivering the acknowledgement, a heartbeat, an order wrapped in `data`, and an order with its fields at the top level. No real socket is opened: a stand-in `websocket` package plays the frames and prints the handshake headers and what the socket sends, and a stand-in `stock_brokers.api.indmoney` module lets the real `IndmoneySession` supply credentials without logging in.

Notice the acknowledgement line, the ignored heartbeat, and that the flat order reaches the callback with its `type` field still in it, because it is handed on exactly as sent.

Run it from the project root:

    python examples/stock_brokers/websockets/indmoney/IndmoneyOrderUpdatesSocket/example_1_orders_in_both_shapes.py
"""

import json
import logging
import sys
import types

from stock_brokers.websockets.indmoney import (
    IndmoneyOrderUpdatesSocket,
    IndmoneySession,
)


class LoginRecord:
    """The INDmoney 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 StandInINDMoneyAPI:
    """A stand-in for `INDMoneyAPI` 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'indmoney-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'client_id': 'ind-client-7',
        }

    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.
        """
        print(f"Handshake headers: {handlers.get('header')}")
        plan = self.plans.pop(0)
        if not self.plans:
            self.when_finished()
        return ScriptedConnection(url, plan, handlers)


class OrdersInBothShapesExample:
    """Runs the INDmoney order update socket against one scripted connection.

    Attributes:
        socket (IndmoneyOrderUpdatesSocket): 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.indmoney')
        api_module.INDMoneyAPI = StandInINDMoneyAPI
        sys.modules['stock_brokers.api.indmoney'] = api_module
        acknowledgement = {
            'action': 'subscribe',
            'status': 'success',
        }
        heartbeat = {
            'type': 'heartbeat',
        }
        wrapped = {
            'type': 'order',
            'data': self.order('OPEN', 0),
        }
        flat = self.order('COMPLETE', 5)
        flat['type'] = 'order'
        plans = [
            {
                'frames': [
                    json.dumps(acknowledgement),
                    json.dumps(heartbeat),
                    json.dumps(wrapped),
                    json.dumps(flat),
                ],
            },
        ]
        websocket_module = ScriptedWebsocketModule(plans)
        sys.modules['websocket'] = websocket_module
        logger = logging.getLogger('indmoney.orders')
        self.socket = IndmoneyOrderUpdatesSocket(IndmoneySession(logger), self.print_orders, logger)
        websocket_module.when_finished = self.socket.close

    def order(self, status, filled_quantity):
        """An INDstocks order for a TCS sell.

        Args:
            status (str): The order status.
            filled_quantity (int): How much has filled.

        Returns:
            dict: The order.
        """
        return {
            'order_id': 'IND2609250001',
            'order_status': status,
            'trading_symbol': 'TCS',
            'exchange': 'NSE',
            'txn_type': 'SELL',
            'qty': 5,
            'filled_qty': filled_quantity,
            'price': 4120.0,
        }

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

        Args:
            orders (list): The orders of one message, exactly as INDstocks 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 type={order.get('type')} {order['order_id']} {order['order_status']}: {order['txn_type']} {order['filled_qty']} of {order['qty']} {order['trading_symbol']}")

    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__':
    OrdersInBothShapesExample().run()

It prints:

Stand-in login number 1
Handshake headers: {'Authorization': 'indmoney-token-1', 'x-api-key': 'ind-client-7'}
Connecting to wss://ws-order-updates.indstocks.com/api/v1/ws/trades
Sent: {"action": "subscribe", "mode": "order_update"}
INFO Order updates socket opened and subscribed. Waiting for order updates.
INFO Order subscription acknowledged.
INFO Ignoring a message of type 'heartbeat': {"type": "heartbeat"}
Order type=None IND2609250001 OPEN: SELL 0 of 5 TCS
Order type=order IND2609250001 COMPLETE: SELL 5 of 5 TCS
INFO Order updates socket closed with status 1000: normal closure
INFO Order updates socket stopped.
Gave up: False
examples/stock_brokers/websockets/indmoney/IndmoneyOrderUpdatesSocket/example_2_refused_then_logged_in_again.py
"""Shows INDmoney's order update socket logging in again, without checking, after a refused handshake.

When INDstocks refuses the order socket's handshake with HTTP 401, 403 or 513, `IndmoneyOrderUpdatesSocket` treats it as a dead session and its reconnect loop calls `IndmoneySession.log_in_again_without_checking`, which logs in whether or not another process already replaced the token.

This program scripts two connections: the first is refused with HTTP 401, and the second opens and delivers one completed order. No real socket is opened: a stand-in `websocket` package plays the connections and prints each handshake's headers, and a stand-in `stock_brokers.api.indmoney` module counts each `INDMoneyAPI` construction as one login. The backoff is set to zero seconds.

Notice the warning about the refused handshake, the second login, and the new token in the second handshake.

Run it from the project root:

    python examples/stock_brokers/websockets/indmoney/IndmoneyOrderUpdatesSocket/example_2_refused_then_logged_in_again.py
"""

import json
import logging
import sys
import types

from stock_brokers.websockets.indmoney import (
    IndmoneyOrderUpdatesSocket,
    IndmoneySession,
)


class LoginRecord:
    """The INDmoney 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 StandInINDMoneyAPI:
    """A stand-in for `INDMoneyAPI` 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'indmoney-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'client_id': 'ind-client-7',
        }

    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.
        """
        print(f"Handshake headers: {handlers.get('header')}")
        plan = self.plans.pop(0)
        if not self.plans:
            self.when_finished()
        return ScriptedConnection(url, plan, handlers)


class RefusedThenLoggedInAgainExample:
    """Runs the INDmoney order update socket through a refused handshake and a good connection.

    Attributes:
        socket (IndmoneyOrderUpdatesSocket): 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.indmoney')
        api_module.INDMoneyAPI = StandInINDMoneyAPI
        sys.modules['stock_brokers.api.indmoney'] = api_module
        update = {
            'type': 'order',
            'data': self.order('COMPLETE', 5),
        }
        plans = [
            {
                'refuse_with': 401,
            },
            {
                'frames': [
                    json.dumps(update),
                ],
            },
        ]
        websocket_module = ScriptedWebsocketModule(plans)
        sys.modules['websocket'] = websocket_module
        logger = logging.getLogger('indmoney.orders')
        self.socket = IndmoneyOrderUpdatesSocket(IndmoneySession(logger), self.print_orders, logger)
        self.socket.MIN_BACKOFF_SECONDS = 0
        websocket_module.when_finished = self.socket.close

    def order(self, status, filled_quantity):
        """An INDstocks order for a TCS sell.

        Args:
            status (str): The order status.
            filled_quantity (int): How much has filled.

        Returns:
            dict: The order.
        """
        return {
            'order_id': 'IND2609250001',
            'order_status': status,
            'trading_symbol': 'TCS',
            'exchange': 'NSE',
            'txn_type': 'SELL',
            'qty': 5,
            'filled_qty': filled_quantity,
            'price': 4120.0,
        }

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

        Args:
            orders (list): The orders of one message, exactly as INDstocks 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['order_id']} {order['order_status']}: {order['txn_type']} {order['filled_qty']} of {order['qty']} {order['trading_symbol']}")

    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__':
    RefusedThenLoggedInAgainExample().run()

It prints:

Stand-in login number 1
Handshake headers: {'Authorization': 'indmoney-token-1', 'x-api-key': 'ind-client-7'}
Connecting to wss://ws-order-updates.indstocks.com/api/v1/ws/trades
WARNING Order updates handshake refused with HTTP 401, taken as a dead session.
INFO Order updates socket closed with status None: None
WARNING Logging in to INDmoney again.
Stand-in login number 2
WARNING Order updates socket disconnected. Reconnecting in 0 seconds.
Handshake headers: {'Authorization': 'indmoney-token-2', 'x-api-key': 'ind-client-7'}
Connecting to wss://ws-order-updates.indstocks.com/api/v1/ws/trades
Sent: {"action": "subscribe", "mode": "order_update"}
INFO Order updates socket opened and subscribed. Waiting for order updates.
Order IND2609250001 COMPLETE: SELL 5 of 5 TCS
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/indmoney.py 
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
def __init__(self, session, on_updates, logger):
    """
    Sets up the order updates socket.

    Args:
        session (IndmoneySession): The login.
        on_updates (callable): Called on this socket's thread with each message's orders, as dictionaries exactly as INDstocks 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 with the token in force now, presented in the handshake, and blocks until it closes.

Returns:

Type Description
None

This method returns nothing.
Source code in stock_brokers/websockets/indmoney.py 
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
def _connect(self):
    """
    Opens the websocket with the token in force now, presented in the handshake, and blocks until it closes.

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

    headers, _ = self._session.handshake_headers()
    self._websocket_application = websocket.WebSocketApp(
        ORDER_UPDATES_URL,
        header=headers,
        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/indmoney.py 
495
496
497
498
499
500
501
502
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)

Asks for order updates; the session is already authenticated by the handshake.

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/indmoney.py 
504
505
506
507
508
509
510
511
512
513
514
515
516
def _on_open(self, websocket_connection):
    """
    Asks for order updates; the session is already authenticated by the handshake.

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

    Returns:
        None: This method returns nothing.
    """
    self._opened = True
    websocket_connection.send(json.dumps({"action": "subscribe", "mode": "order_update"}))
    self._logger.info(f"{self.name} opened and subscribed. Waiting for order updates.")

_on_error 

_on_error(websocket_connection, error)

Takes a refused handshake as a dead session, and reports any other error.

Parameters:

Name Type Description Default
websocket_connection WebSocketApp

The connection that failed.

 required
error Exception

The error, which carries status_code when the handshake was refused.

 required

Returns:

Type Description
None

This method returns nothing.
Source code in stock_brokers/websockets/indmoney.py 
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
def _on_error(self, websocket_connection, error):
    """
    Takes a refused handshake as a dead session, and reports any other error.

    Args:
        websocket_connection (websocket.WebSocketApp): The connection that failed.
        error (Exception): The error, which carries `status_code` when the handshake was refused.

    Returns:
        None: This method returns nothing.
    """
    status = getattr(error, "status_code", None)
    if status in AUTHENTICATION_STATUS_CODES:
        self._authentication_rejected = True
        self._logger.warning(f"Order updates handshake refused with HTTP {status}, taken as a dead session.")
        return
    self._logger.error(f"Order updates error: {error}")

_on_message 

_on_message(websocket_connection, message)

Hands a message's orders on, and logs acknowledgements and anything else.

Parameters:

Name Type Description Default
websocket_connection WebSocketApp

The connection the message arrived on.

 required
message bytes | str

The frame: one JSON message or a JSON list of them.

 required

Returns:

Type Description
None

This method returns nothing.
Source code in stock_brokers/websockets/indmoney.py 
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
def _on_message(self, websocket_connection, message):
    """
    Hands a message's orders on, and logs acknowledgements and anything else.

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

    Returns:
        None: This method returns nothing.
    """
    if isinstance(message, (bytes, bytearray)):
        message = message.decode("utf-8", errors="replace")
    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:
            if entry.get("status") in ("subscribed", "success") or entry.get("action") == "subscribe":
                self._logger.info("Order subscription acknowledged.")
            else:
                self._logger.info(f"Ignoring a message of type {entry.get('type')!r}: {json.dumps(entry)[:200]}")
            continue
        if isinstance(entry.get("data"), dict):
            orders.append(entry.get("data"))
        else:
            orders.append(entry)
    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 
53
54
55
56
57
58
59
60
61
62
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 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
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 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
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 
152
153
154
155
156
157
158
159
160
161
162
163
164
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}")