shoonya

Shoonya's two websockets on the Noren platform: the market feed and the account's order updates.

Both connect to wss://api.shoonya.com/NorenWSTP/, a separate connection each, so a market data reconnect can never drop an order event.

The login. ShoonyaSession holds the ShoonyaAPI both sockets read their credentials from, afresh on every connect, so a login made by any process is picked up. Shoonya's login is a headless Chrome session, so 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.

The connect frame. Noren streams JSON text, and a connection is unusable until it is authenticated. On open a socket sends {"t": "a", "uid", "actid", "source": "API", "accesstoken"} with the account's UCC code as the user id. The frame is a with the token named accesstoken: Noren also parses a c frame naming it susertoken, and refuses a perfectly good token that way, which reads exactly like an expired session. Only after the ak acknowledgement says OK (documented as "Ok" and seen as "OK", so compared without case) does the socket subscribe; any other answer is a refused login. Noren's own heartbeat, {"t":"h"}, is sent every three seconds as the ping payload, because a plain websocket ping is not enough to keep the connection.

Quotes. ShoonyaQuotesSocket subscribes its batch with one {"t": "d", "k": "NSE|2885#MCX|565899"} frame, for depth and touchline together. The feed is incremental: an acknowledgement (tk, dk) carries every field and each update after it (tf, df) only what changed, so the last state of each instrument is kept and every update merged into it before a tick is built. Every value arrives as a string and is converted, with blanks and NA as None. An acknowledgement names the instrument with its trading symbol; a name the socket did not already have is handed to on_name before the tick. Noren spells ltt as an epoch, a date and time, or a bare India clock time; all three become an epoch, a bare clock time dated by the tick's feed time and moved back a day when it would otherwise be after it, which happens for a trade just before midnight reported just after. No tick is handed on until an instrument has a price.

Order updates. ShoonyaOrderUpdatesSocket subscribes to the whole account with {"t": "o", "actid": <ucc code>}, and every change to an order then arrives as an om message carrying the order book's fields, every value a string.

Neither socket writes Redis. The quotes socket hands names to on_name and each tick to on_tick, and the order socket hands each message's om updates to on_updates; the scripts in bin/shoonya/ do the writing.

Classes:

Name	Description
ShoonyaSession	The Shoonya login every socket authenticates with, logged in again by one socket at a time.
ShoonyaQuotesSocket	One Noren market feed websocket carrying the touchline and depth of a batch of instruments.
ShoonyaOrderUpdatesSocket	Shoonya's order update websocket, subscribed to the whole account once authenticated.

ShoonyaSession

ShoonyaSession(logger)

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

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

Example programs

Logging in again onceLogging in without checking

examples/stock_brokers/websockets/shoonya/ShoonyaSession/example_1_logging_in_again_once.py

"""Builds a Shoonya session, reads its credentials and logs in again only when the failed token is still current.

Shoonya's login drives a headless Chrome session, so `ShoonyaSession` makes sure refused sockets log in one at a time: `log_in_again` logs in only when the token a refused socket used is still the one in force, so several refused sockets cause one login between them. `credentials` returns the account's UCC code, which Noren takes as the user id, and the token in force now.

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

Notice that the second request with the same stale token makes no login.

Run it from the project root:

    python examples/stock_brokers/websockets/shoonya/ShoonyaSession/example_1_logging_in_again_once.py
"""

import logging
import sys
import types

from stock_brokers.websockets.shoonya import (
    ShoonyaSession,
)


class LoginRecord:
    """The Shoonya 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 StandInShoonyaAPI:
    """A stand-in for `ShoonyaAPI` 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'shoonya-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'ucc_code': 'FA12345',
        }

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

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


class LoggingInAgainOnceExample:
    """Builds a Shoonya session and asks it to log in again twice with the same token.

    Attributes:
        session (ShoonyaSession): 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.shoonya')
        api_module.ShoonyaAPI = StandInShoonyaAPI
        sys.modules['stock_brokers.api.shoonya'] = api_module
        self.session = ShoonyaSession(logging.getLogger('shoonya.session'))

    def run(self):
        """Refuses the same token twice and prints the credentials after each request.

        Returns:
            None: This method returns nothing.
        """
        user_id, failed_token = self.session.credentials()
        print(f'User {user_id} was refused with {failed_token}')
        self.session.log_in_again(failed_token)
        print(f'Credentials now: {self.session.credentials()}')
        self.session.log_in_again(failed_token)
        print(f'Credentials after the second request: {self.session.credentials()}')
        print(f'Logins made: {LoginRecord.login_count}')


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

It prints:

Stand-in login number 1
User FA12345 was refused with shoonya-token-1
WARNING Logging in to Shoonya again.
Stand-in login number 2
Credentials now: ('FA12345', 'shoonya-token-2')
Credentials after the second request: ('FA12345', 'shoonya-token-2')
Logins made: 2

examples/stock_brokers/websockets/shoonya/ShoonyaSession/example_2_logging_in_without_checking.py

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

The order update socket calls `ShoonyaSession.log_in_again_without_checking` when Noren refuses its connect frame. That logs in even when another process has already replaced the token, unlike `log_in_again`, which skips the login in that case.

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

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

Run it from the project root:

    python examples/stock_brokers/websockets/shoonya/ShoonyaSession/example_2_logging_in_without_checking.py
"""

import logging
import sys
import types

from stock_brokers.websockets.shoonya import (
    ShoonyaSession,
)


class LoginRecord:
    """The Shoonya 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 StandInShoonyaAPI:
    """A stand-in for `ShoonyaAPI` 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'shoonya-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'ucc_code': 'FA12345',
        }

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

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


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

    Attributes:
        session (ShoonyaSession): 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.shoonya')
        api_module.ShoonyaAPI = StandInShoonyaAPI
        sys.modules['stock_brokers.api.shoonya'] = api_module
        self.session = ShoonyaSession(logging.getLogger('shoonya.session'))

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

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


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

It prints:

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

Methods:

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

Source code in stock_brokers/websockets/shoonya.py

def __init__(self, logger):
    """
    Constructs `ShoonyaAPI`, 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 `ShoonyaAPI` raises when it cannot log in.
    """
    from stock_brokers.api.shoonya import ShoonyaAPI

    self._api_class = ShoonyaAPI
    self._logger = logger
    self._lock = threading.Lock()
    self._shoonya = ShoonyaAPI()

credentials

credentials()

The Noren user id and the session token in force now, which may be one another process has just obtained.

Returns:

Type	Description
tuple	The UCC code and the session token, either of which may be None.

Example usage

From example_1_logging_in_again_once.py

user_id, failed_token = self.session.credentials()

From example_1_logging_in_again_once.py

print(f'Credentials now: {self.session.credentials()}')

From example_1_logging_in_again_once.py

print(f'Credentials after the second request: {self.session.credentials()}')

Source code in stock_brokers/websockets/shoonya.py

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

    Returns:
        tuple: The UCC code and the session token, either of which may be None.
    """
    login = self._shoonya._current_login() or {}
    return self._shoonya._settings.get("ucc_code"), login.get("access_token")

log_in_again

log_in_again(stale_token)

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

Parameters:

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

Returns:

Type	Description
None	This method returns nothing.

Raises:

Type	Description
Exception	Whatever ShoonyaAPI raises when it cannot log in.

Example usage

From example_1_logging_in_again_once.py

self.session.log_in_again(failed_token)

From example_2_logging_in_without_checking.py

self.session.log_in_again(first_token)

Source code in stock_brokers/websockets/shoonya.py

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

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

    Returns:
        None: This method returns nothing.

    Raises:
        Exception: Whatever `ShoonyaAPI` 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 ShoonyaAPI raises when it cannot log in.

Example usage

From example_2_logging_in_without_checking.py

self.session.log_in_again_without_checking()

Source code in stock_brokers/websockets/shoonya.py

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

    Returns:
        None: This method returns nothing.

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

_log_in

_log_in()

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

Returns:

Type	Description
None	This method returns nothing.

Raises:

Type	Description
Exception	Whatever ShoonyaAPI raises when it cannot log in.

Source code in stock_brokers/websockets/shoonya.py

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

    Returns:
        None: This method returns nothing.

    Raises:
        Exception: Whatever `ShoonyaAPI` raises when it cannot log in.
    """
    self._logger.warning("Logging in to Shoonya again.")
    self._shoonya = self._api_class()

ShoonyaQuotesSocket

ShoonyaQuotesSocket(name, tokens, names, session, on_name, on_tick, logger)

Bases: BrokerWebsocket

One Noren market feed websocket carrying the touchline and depth of a batch of instruments.

Sets up a socket for one batch of instruments.

Parameters:

Name	Type	Description	Default
name	str	The connection's name, for the log.	required
tokens	list[str]	The batch of EXCHANGE|TOKEN instrument tokens.	required
names	dict[str, str]	Each token to the instrument's name, shared between sockets and added to as Noren names instruments.	required
session	ShoonyaSession	The shared login.	required
on_name	callable	Called with a token and its new name when an acknowledgement names an instrument, on this socket's thread.	required
on_tick	callable	Called with each tick, on this socket's thread.	required
logger	Logger	Where the connection reports.	required

Returns:

Type	Description
None	This method returns nothing.

Example programs

Merging incremental updatesRefused acknowledgement

examples/stock_brokers/websockets/shoonya/ShoonyaQuotesSocket/example_1_merging_incremental_updates.py

"""Authenticates a Shoonya quotes socket and merges Noren's incremental feed into full ticks.

A `ShoonyaQuotesSocket` sends Noren's connect frame when the connection opens and subscribes its batch only after the `ak` acknowledgement says `OK`. The feed is incremental: a touchline or depth acknowledgement (`tk`, `dk`) carries every field and names the instrument, and each update after it (`tf`, `df`) carries only what changed. The socket keeps the last state of each instrument and merges every update into it before building a tick, with every string converted to a number.

This program scripts one connection that answers the connect frame with `OK`, sends a depth acknowledgement for NSE token 2885 naming it RELIANCE-EQ, and then a depth update that changes only the last price and the best bid. No real socket is opened: a stand-in `websocket` package plays the frames and prints what the socket sends, and a stand-in `stock_brokers.api.shoonya` module lets the real `ShoonyaSession` log in without reaching Shoonya.

Notice that the new name is handed to `on_name` before the first tick is handed to `on_tick`, and that the second tick still has the open, high, low, volume and trade time from the acknowledgement although the update did not repeat them. The trade time `25-09-2026 10:15:28` becomes an epoch.

Run it from the project root:

    python examples/stock_brokers/websockets/shoonya/ShoonyaQuotesSocket/example_1_merging_incremental_updates.py
"""

import json
import logging
import sys
import types

from stock_brokers.websockets.shoonya import (
    ShoonyaQuotesSocket,
    ShoonyaSession,
)


class LoginRecord:
    """The Shoonya 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 StandInShoonyaAPI:
    """A stand-in for `ShoonyaAPI` 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'shoonya-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'ucc_code': 'FA12345',
        }

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

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


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

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

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

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

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


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

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

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

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

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

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

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

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

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

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

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

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

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

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


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

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

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

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

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

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

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

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


class MergingIncrementalUpdatesExample:
    """Runs a Shoonya quotes socket against an acknowledgement and one update.

    Attributes:
        socket (ShoonyaQuotesSocket): 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.shoonya')
        api_module.ShoonyaAPI = StandInShoonyaAPI
        sys.modules['stock_brokers.api.shoonya'] = api_module
        connect_acknowledgement = {
            't': 'ak',
            's': 'OK',
            'uid': 'FA12345',
        }
        depth_acknowledgement = {
            't': 'dk',
            'e': 'NSE',
            'tk': '2885',
            'ts': 'RELIANCE-EQ',
            'lp': '2950.50',
            'pc': '1.39',
            'c': '2910.00',
            'o': '2930.00',
            'h': '2960.00',
            'l': '2925.00',
            'v': '1204500',
            'ap': '2949.10',
            'ltt': '25-09-2026 10:15:28',
            'ft': '1790311529',
            'bp1': '2950.00',
            'bq1': '100',
            'bo1': '3',
            'sp1': '2950.50',
            'sq1': '120',
            'so1': '4',
        }
        depth_update = {
            't': 'df',
            'e': 'NSE',
            'tk': '2885',
            'lp': '2951.00',
            'pc': '1.41',
            'bp1': '2950.50',
            'bq1': '80',
            'ft': '1790311530',
        }
        plans = [
            {
                'frames': [
                    json.dumps(connect_acknowledgement),
                    json.dumps(depth_acknowledgement),
                    json.dumps(depth_update),
                ],
            },
        ]
        websocket_module = ScriptedWebsocketModule(plans)
        sys.modules['websocket'] = websocket_module
        logger = logging.getLogger('shoonya.quotes')
        tokens = [
            'NSE|2885',
        ]
        names = {}
        session = ShoonyaSession(logger)
        self.socket = ShoonyaQuotesSocket('socket_0', tokens, names, session, self.print_name, self.print_tick, logger)
        websocket_module.when_finished = self.socket.close

    def print_name(self, token, name):
        """Prints an instrument an acknowledgement named.

        Args:
            token (str): The instrument's `EXCHANGE|TOKEN`.
            name (str): Its new name.

        Returns:
            None: This method returns nothing.
        """
        print(f'Name learned: {token} is {name}')

    def print_tick(self, tick):
        """Prints the main fields of one tick the socket built.

        Args:
            tick (dict): The normalized tick.

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

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

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


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

It prints:

Stand-in login number 1
Connecting to wss://api.shoonya.com/NorenWSTP/
Sent: {"t": "a", "uid": "FA12345", "actid": "FA12345", "source": "API", "accesstoken": "shoonya-token-1"}
INFO socket_0 opened. Authenticating.
Sent: {"t": "d", "k": "NSE|2885"}
INFO socket_0 authenticated and subscribed to 1 instrument(s).
Name learned: NSE|2885 is NSE:RELIANCE-EQ
Tick NSE:RELIANCE-EQ (NSE|2885) in full mode
  last_price=2950.5 volume=1204500 change=1.39
  ohlc={'open': 2930.0, 'high': 2960.0, 'low': 2925.0, 'close': 2910.0}
  last_trade_time=1790311528 exchange_timestamp=1790311529
  best bid={'quantity': 100, 'price': 2950.0, 'orders': 3} best offer={'quantity': 120, 'price': 2950.5, 'orders': 4}
Tick NSE:RELIANCE-EQ (NSE|2885) in full mode
  last_price=2951.0 volume=1204500 change=1.41
  ohlc={'open': 2930.0, 'high': 2960.0, 'low': 2925.0, 'close': 2910.0}
  last_trade_time=1790311528 exchange_timestamp=1790311530
  best bid={'quantity': 80, 'price': 2950.5, 'orders': 3} best offer={'quantity': 120, 'price': 2950.5, 'orders': 4}
INFO socket_0 closed with status 1000: normal closure
INFO socket_0 stopped.
Gave up: False

examples/stock_brokers/websockets/shoonya/ShoonyaQuotesSocket/example_2_refused_acknowledgement.py

"""Shows a Shoonya quotes socket logging in again when Noren refuses its connect frame.

Noren accepts every websocket handshake and refuses a dead session in its `ak` acknowledgement instead, with a status other than `OK`. `ShoonyaQuotesSocket` then closes the connection and its reconnect loop asks the shared `ShoonyaSession` to log in again, before the socket connects with the new token.

This program scripts two connections. The first answers the connect frame with `NOT_OK`. The second accepts it and sends a touchline acknowledgement for MCX token 565899 whose trade time is a bare India clock time, `10:15:28`, and a touchline acknowledgement for a token with no price yet. No real socket is opened: a stand-in `websocket` package plays the connections, and a stand-in `stock_brokers.api.shoonya` module counts each `ShoonyaAPI` construction as a login. The backoff is set to zero seconds.

Notice the second login, the new token in the second connect frame, and that the bare clock time is dated by the frame's feed time. The instrument without a price is named but produces no tick.

Run it from the project root:

    python examples/stock_brokers/websockets/shoonya/ShoonyaQuotesSocket/example_2_refused_acknowledgement.py
"""

import json
import logging
import sys
import types

from stock_brokers.websockets.shoonya import (
    ShoonyaQuotesSocket,
    ShoonyaSession,
)


class LoginRecord:
    """The Shoonya 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 StandInShoonyaAPI:
    """A stand-in for `ShoonyaAPI` 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'shoonya-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'ucc_code': 'FA12345',
        }

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

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


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

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

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

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

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


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

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

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

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

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

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

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

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

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

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

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

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

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

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


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

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

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

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

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

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

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

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


class RefusedAcknowledgementExample:
    """Runs a Shoonya quotes socket through a refused connect frame and a good connection.

    Attributes:
        socket (ShoonyaQuotesSocket): 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.shoonya')
        api_module.ShoonyaAPI = StandInShoonyaAPI
        sys.modules['stock_brokers.api.shoonya'] = api_module
        refused = {
            't': 'ak',
            's': 'NOT_OK',
            'emsg': 'Session Expired :  Invalid Session Key',
        }
        accepted = {
            't': 'ak',
            's': 'Ok',
        }
        crude_touchline = {
            't': 'tk',
            'e': 'MCX',
            'tk': '565899',
            'ts': 'CRUDEOIL19OCT26',
            'lp': '6120.00',
            'c': '6100.00',
            'v': '8450',
            'oi': '12500',
            'ltt': '10:15:28',
            'ft': '1790311529',
        }
        unpriced_touchline = {
            't': 'tk',
            'e': 'MCX',
            'tk': '565900',
            'ts': 'CRUDEOIL19NOV26',
            'lp': 'NA',
        }
        plans = [
            {
                'frames': [
                    json.dumps(refused),
                ],
            },
            {
                'frames': [
                    json.dumps(accepted),
                    json.dumps(crude_touchline),
                    json.dumps(unpriced_touchline),
                ],
            },
        ]
        websocket_module = ScriptedWebsocketModule(plans)
        sys.modules['websocket'] = websocket_module
        logger = logging.getLogger('shoonya.quotes')
        tokens = [
            'MCX|565899',
            'MCX|565900',
        ]
        names = {}
        session = ShoonyaSession(logger)
        self.socket = ShoonyaQuotesSocket('socket_0', tokens, names, session, self.print_name, self.print_tick, logger)
        self.socket.MIN_BACKOFF_SECONDS = 0
        websocket_module.when_finished = self.socket.close

    def print_name(self, token, name):
        """Prints an instrument an acknowledgement named.

        Args:
            token (str): The instrument's `EXCHANGE|TOKEN`.
            name (str): Its new name.

        Returns:
            None: This method returns nothing.
        """
        print(f'Name learned: {token} is {name}')

    def print_tick(self, tick):
        """Prints the main fields of one tick the socket built.

        Args:
            tick (dict): The normalized tick.

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

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

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


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

It prints:

Stand-in login number 1
Connecting to wss://api.shoonya.com/NorenWSTP/
Sent: {"t": "a", "uid": "FA12345", "actid": "FA12345", "source": "API", "accesstoken": "shoonya-token-1"}
INFO socket_0 opened. Authenticating.
ERROR socket_0 authentication refused: {'t': 'ak', 's': 'NOT_OK', 'emsg': 'Session Expired :  Invalid Session Key'}
INFO socket_0 closed with status 1000: normal closure
WARNING Logging in to Shoonya again.
Stand-in login number 2
WARNING socket_0 disconnected. Reconnecting in 0 seconds.
Connecting to wss://api.shoonya.com/NorenWSTP/
Sent: {"t": "a", "uid": "FA12345", "actid": "FA12345", "source": "API", "accesstoken": "shoonya-token-2"}
INFO socket_0 opened. Authenticating.
Sent: {"t": "d", "k": "MCX|565899#MCX|565900"}
INFO socket_0 authenticated and subscribed to 2 instrument(s).
Name learned: MCX|565899 is MCX:CRUDEOIL19OCT26
Tick MCX:CRUDEOIL19OCT26 (MCX|565899) in quote mode
  last_price=6120.0 volume=8450 change=0.3279
  ohlc={'open': None, 'high': None, 'low': None, 'close': 6100.0}
  last_trade_time=1790311528 exchange_timestamp=1790311529
Name learned: MCX|565900 is MCX:CRUDEOIL19NOV26
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/shoonya.py

def __init__(self, name, tokens, names, session, on_name, on_tick, 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 `EXCHANGE|TOKEN` instrument tokens.
        names (dict[str, str]): Each token to the instrument's name, shared between sockets and added to as Noren names instruments.
        session (ShoonyaSession): The shared login.
        on_name (callable): Called with a token and its new name when an acknowledgement names an instrument, on this socket's thread.
        on_tick (callable): Called with each tick, 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_name = on_name
    self._on_tick = on_tick
    self._token = None
    self._state = {}

_connect

_connect()

Opens the websocket and blocks until it closes, sending Noren's heartbeat meanwhile.

Returns:

Type	Description
None	This method returns nothing.

Source code in stock_brokers/websockets/shoonya.py

def _connect(self):
    """
    Opens the websocket and blocks until it closes, sending Noren's heartbeat meanwhile.

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

    self._websocket_application = websocket.WebSocketApp(
        FEED_URL,
        on_open=self._on_open,
        on_message=self._on_message,
        on_error=self._on_error,
        on_close=self._on_close,
    )
    self._websocket_application.run_forever(
        ping_interval=HEARTBEAT_INTERVAL_SECONDS,
        ping_timeout=HEARTBEAT_TIMEOUT_SECONDS,
        ping_payload=HEARTBEAT,
    )

_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/shoonya.py

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

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

_on_open

_on_open(websocket_connection)

Authenticates with the token in force now; the subscription waits for the acknowledgement.

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/shoonya.py

def _on_open(self, websocket_connection):
    """
    Authenticates with the token in force now; the subscription waits for the acknowledgement.

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

    Returns:
        None: This method returns nothing.
    """
    self._opened = True
    self._state = {}
    user_id, self._token = self._session.credentials()
    websocket_connection.send(json.dumps({
        "t": "a",
        "uid": user_id,
        "actid": user_id,
        "source": "API",
        "accesstoken": self._token,
    }))
    self._logger.info(f"{self.name} opened. Authenticating.")

_on_error

_on_error(websocket_connection, error)

Reports an error; Noren refuses a login in its acknowledgement rather than at the handshake.

Parameters:

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

Returns:

Type	Description
None	This method returns nothing.

Source code in stock_brokers/websockets/shoonya.py

def _on_error(self, websocket_connection, error):
    """
    Reports an error; Noren refuses a login in its acknowledgement rather than at the handshake.

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

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

_on_message

_on_message(websocket_connection, message)

Handles the connect acknowledgement, and hands each tick on.

Parameters:

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

Returns:

Type	Description
None	This method returns nothing.

Source code in stock_brokers/websockets/shoonya.py

def _on_message(self, websocket_connection, message):
    """
    Handles the connect acknowledgement, and hands each tick on.

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

    Returns:
        None: This method returns nothing.
    """
    tick = self._parse(message)
    if tick:
        self._on_tick(tick)

_parse

_parse(message)

Builds a tick from one Noren message, subscribing on the connect acknowledgement and learning names on the way.

Parameters:

Name	Type	Description	Default
message	bytes | str	The message.	required

Returns:

Type	Description
dict | None	The tick, or None for a message that carries none.

Source code in stock_brokers/websockets/shoonya.py

def _parse(self, message):
    """
    Builds a tick from one Noren message, subscribing on the connect acknowledgement and learning names on the way.

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

    Returns:
        dict | None: The tick, or None for a message that carries none.
    """
    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"{self.name} ignored a frame that is not JSON: {str(message)[:160]}")
        return None
    if not isinstance(payload, dict):
        return None

    message_type = payload.get("t")
    if message_type == CONNECT_ACK:
        if str(payload.get("s", "")).upper() == "OK":
            self._websocket_application.send(json.dumps({"t": "d", "k": "#".join(self._tokens)}))
            self._logger.info(f"{self.name} authenticated and subscribed to {len(self._tokens)} instrument(s).")
        else:
            self._logger.error(f"{self.name} authentication refused: {payload}")
            self._authentication_rejected = True
            self._websocket_application.close()
        return None

    if message_type not in TICK_MESSAGES or not payload.get("e") or not payload.get("tk"):
        return None
    token = f"{payload['e']}|{payload['tk']}"

    if message_type in ACKNOWLEDGEMENTS and payload.get("ts") and not self._names.get(token):
        self._names[token] = f"{payload['e']}:{payload['ts']}"
        self._on_name(token, self._names[token])

    state = self._state.setdefault(token, {})
    for key, value in payload.items():
        if key not in ("t", "e", "tk"):
            state[key] = value
    tick = self._build_tick(token, state, self._names.get(token) or token)
    if tick["last_price"] is None:
        return None
    return tick

_build_tick

_build_tick(token, state, identifier)

Builds a normalized tick from an instrument's merged state.

Parameters:

Name	Type	Description	Default
token	str	The instrument's EXCHANGE|TOKEN.	required
state	dict	Its merged field state.	required
identifier	str	The name the tick is keyed by.	required

Returns:

Type	Description
dict	The tick.

Source code in stock_brokers/websockets/shoonya.py

def _build_tick(self, token, state, identifier):
    """
    Builds a normalized tick from an instrument's merged state.

    Args:
        token (str): The instrument's `EXCHANGE|TOKEN`.
        state (dict): Its merged field state.
        identifier (str): The name the tick is keyed by.

    Returns:
        dict: The tick.
    """
    last_price = self._number(state.get("lp"))
    close = self._number(state.get("c"))
    change = self._number(state.get("pc"))
    if change is None and last_price is not None and close:
        change = (last_price - close) * 100 / close

    depth = {
        "buy": [],
        "sell": [],
    }
    for level in range(1, 6):
        for side, prefix in (("buy", "b"), ("sell", "s")):
            if state.get(f"{prefix}p{level}") is not None:
                depth[side].append({
                    "quantity": self._number(state.get(f"{prefix}q{level}"), int),
                    "price": self._number(state.get(f"{prefix}p{level}")),
                    "orders": self._number(state.get(f"{prefix}o{level}"), int),
                })

    mode = "quote"
    if depth["buy"] or depth["sell"]:
        mode = "full"
    feed_time = self._number(state.get("ft"), int)
    return {
        "id": identifier,
        "broker": "shoonya",
        "instrument_token": token,
        "exchange": token.partition("|")[0],
        "mode": mode,
        "last_price": last_price,
        "last_quantity": self._number(state.get("ltq"), int),
        "average_price": self._number(state.get("ap")),
        "volume": self._number(state.get("v"), int),
        "buy_quantity": self._number(state.get("tbq"), int),
        "sell_quantity": self._number(state.get("tsq"), int),
        "ohlc": {
            "open": self._number(state.get("o")),
            "high": self._number(state.get("h")),
            "low": self._number(state.get("l")),
            "close": close,
        },
        "change": change,
        "oi": self._number(state.get("oi"), int),
        "oi_day_high": None,
        "oi_day_low": None,
        "last_trade_time": self._trade_time(state.get("ltt"), feed_time),
        "exchange_timestamp": feed_time,
        "depth": depth,
        "received_at": time.time(),
    }

_number

_number(value, cast=float)

A Noren string field as a number, with blanks and placeholders as None.

Parameters:

Name	Type	Description	Default
value	str | None	The raw field.	required
cast	type	float or int.	float

Returns:

Type	Description
float | int | None	The number, or None.

Source code in stock_brokers/websockets/shoonya.py

def _number(self, value, cast=float):
    """
    A Noren string field as a number, with blanks and placeholders as None.

    Args:
        value (str | None): The raw field.
        cast (type): `float` or `int`.

    Returns:
        float | int | None: The number, or None.
    """
    if value is None or value == "" or value == "NA":
        return None
    try:
        if cast is int:
            return cast(float(value))
        return cast(value)
    except (TypeError, ValueError):
        return None

_trade_time

_trade_time(value, feed_time)

Noren's last trade time as an epoch, whether sent as an epoch, a date and time, or a bare India clock time.

Parameters:

Name	Type	Description	Default
value	str | None	The raw ltt field.	required
feed_time	int | None	The tick's ft epoch, or None to date a bare clock time by the wall clock.	required

Returns:

Type	Description
int | None	The epoch, or None when the field cannot be read.

Source code in stock_brokers/websockets/shoonya.py

def _trade_time(self, value, feed_time):
    """
    Noren's last trade time as an epoch, whether sent as an epoch, a date and time, or a bare India clock time.

    Args:
        value (str | None): The raw `ltt` field.
        feed_time (int | None): The tick's `ft` epoch, or None to date a bare clock time by the wall clock.

    Returns:
        int | None: The epoch, or None when the field cannot be read.
    """
    if value is None or value == "" or value == "NA":
        return None
    text = str(value).strip()
    if text.isdigit():
        return int(text)
    for time_format in TRADE_TIME_FORMATS:
        try:
            return int(datetime.strptime(text, time_format).replace(tzinfo=INDIA).timestamp())
        except ValueError:
            continue
    try:
        clock = datetime.strptime(text, "%H:%M:%S").time()
    except ValueError:
        return None
    reference_epoch = feed_time
    if not reference_epoch:
        reference_epoch = time.time()
    reference = datetime.fromtimestamp(reference_epoch, tz=INDIA)
    moment = datetime.combine(reference.date(), clock, tzinfo=INDIA)
    if moment > reference + timedelta(minutes=5):
        moment = moment - timedelta(days=1)
    return int(moment.timestamp())

close

close()

Closes the connection and stops reconnecting.

Returns:

Type	Description
None	This method returns nothing.

Example usage

From example_2_giving_up_after_a_second_refusal.py

self.closed_socket.close()

Source code in stock_brokers/websockets/base.py

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

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

run_forever

run_forever()

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

Returns:

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

Example usage

From example_1_reconnecting_after_a_drop.py

self.socket.run_forever()

From example_2_giving_up_after_a_second_refusal.py

self.refused_socket.run_forever()

From example_2_giving_up_after_a_second_refusal.py

self.closed_socket.run_forever()

Source code in stock_brokers/websockets/base.py

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

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

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

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

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

_gives_up_after_logging_in_again

_gives_up_after_logging_in_again(failed_connects)

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

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

Parameters:

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

Returns:

Type	Description
bool	True to give up now.

Source code in stock_brokers/websockets/base.py

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

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

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

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

_on_close

_on_close(websocket_connection, status_code, message)

Reports that the connection closed.

Parameters:

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

Returns:

Type	Description
None	This method returns nothing.

Source code in stock_brokers/websockets/base.py

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

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

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

ShoonyaOrderUpdatesSocket

ShoonyaOrderUpdatesSocket(session, on_updates, logger)

Bases: BrokerWebsocket

Shoonya's order update websocket, subscribed to the whole account once authenticated.

Sets up the order updates socket.

Parameters:

Name	Type	Description	Default
session	ShoonyaSession	The login.	required
on_updates	callable	Called on this socket's thread with each message's om updates that carry an order number, as dictionaries exactly as Noren 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

Subscribing to the accountRejected then logged in again

examples/stock_brokers/websockets/shoonya/ShoonyaOrderUpdatesSocket/example_1_subscribing_to_the_account.py

"""Authenticates Shoonya's order update socket, subscribes to the account and prints the order updates it hands on.

`ShoonyaOrderUpdatesSocket` sends Noren's connect frame when the connection opens, and once the `ak` acknowledgement says `OK` it subscribes to the whole account with `{"t": "o", "actid": ...}`. Every change to an order then arrives as an `om` message, and the socket hands each frame's `om` updates that carry an order number on, exactly as Noren sent them. Shoonya runs it as a connection of its own, so a market data reconnect can never drop an order event.

This program scripts one connection that sends the acknowledgement and then one frame holding a list of two `om` updates and a message of another type. No real socket is opened: a stand-in `websocket` package plays the frames and prints what the socket sends, and a stand-in `stock_brokers.api.shoonya` module lets the real `ShoonyaSession` log in without reaching Shoonya.

Notice the connect frame and the account subscription in the output, and that both updates of the list reach the callback in one call.

Run it from the project root:

    python examples/stock_brokers/websockets/shoonya/ShoonyaOrderUpdatesSocket/example_1_subscribing_to_the_account.py
"""

import json
import logging
import sys
import types

from stock_brokers.websockets.shoonya import (
    ShoonyaOrderUpdatesSocket,
    ShoonyaSession,
)


class LoginRecord:
    """The Shoonya 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 StandInShoonyaAPI:
    """A stand-in for `ShoonyaAPI` 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'shoonya-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'ucc_code': 'FA12345',
        }

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

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


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

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

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

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

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


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

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

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

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

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

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

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

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

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

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

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

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

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

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


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

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

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

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

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

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

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

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


class SubscribingToTheAccountExample:
    """Runs the Shoonya order update socket against one scripted connection.

    Attributes:
        socket (ShoonyaOrderUpdatesSocket): 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.shoonya')
        api_module.ShoonyaAPI = StandInShoonyaAPI
        sys.modules['stock_brokers.api.shoonya'] = api_module
        acknowledgement = {
            't': 'ak',
            's': 'OK',
        }
        batch = [
            self.order_update('OPEN', '0'),
            self.order_update('COMPLETE', '5'),
            {
                't': 'dk',
                'e': 'NSE',
            },
        ]
        plans = [
            {
                'frames': [
                    json.dumps(acknowledgement),
                    json.dumps(batch),
                ],
            },
        ]
        websocket_module = ScriptedWebsocketModule(plans)
        sys.modules['websocket'] = websocket_module
        logger = logging.getLogger('shoonya.orders')
        self.socket = ShoonyaOrderUpdatesSocket(ShoonyaSession(logger), self.print_orders, logger)
        websocket_module.when_finished = self.socket.close

    def order_update(self, status, filled_quantity):
        """One Noren `om` order update for a TCS sell, every value a string.

        Args:
            status (str): Noren's order status.
            filled_quantity (str): How much has filled.

        Returns:
            dict: The update.
        """
        return {
            't': 'om',
            'norenordno': '26092500012345',
            'uid': 'FA12345',
            'actid': 'FA12345',
            'exch': 'NSE',
            'tsym': 'TCS-EQ',
            'trantype': 'S',
            'qty': '5',
            'fillshares': filled_quantity,
            'prc': '4120.00',
            'prctyp': 'LMT',
            'status': status,
        }

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

        Args:
            orders (list): The `om` updates of one frame, exactly as Noren sent them.
            received_at (datetime.datetime): When the frame arrived, which is not printed.

        Returns:
            None: This method returns nothing.
        """
        print(f'Callback received {len(orders)} update(s)')
        for order in orders:
            print(f"  Order {order['norenordno']} {order['status']}: {order['trantype']} {order['fillshares']} of {order['qty']} {order['tsym']}")

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

It prints:

Stand-in login number 1
Connecting to wss://api.shoonya.com/NorenWSTP/
Sent: {"t": "a", "uid": "FA12345", "actid": "FA12345", "source": "API", "accesstoken": "shoonya-token-1"}
INFO Order updates socket opened. Authenticating.
Sent: {"t": "o", "actid": "FA12345"}
INFO Authenticated and subscribed. Waiting for order updates.
Callback received 2 update(s)
  Order 26092500012345 OPEN: S 0 of 5 TCS-EQ
  Order 26092500012345 COMPLETE: S 5 of 5 TCS-EQ
INFO Order updates socket closed with status 1000: normal closure
INFO Order updates socket stopped.
Gave up: False

examples/stock_brokers/websockets/shoonya/ShoonyaOrderUpdatesSocket/example_2_rejected_then_logged_in_again.py

"""Shows Shoonya's order update socket logging in again, without checking, after Noren rejects its connect frame.

When the `ak` acknowledgement is not `OK`, `ShoonyaOrderUpdatesSocket` closes the connection, and its reconnect loop calls `ShoonyaSession.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 rejected, the second is accepted and delivers one completed order update. No real socket is opened: a stand-in `websocket` package plays the connections, and a stand-in `stock_brokers.api.shoonya` module counts each `ShoonyaAPI` construction as a login. The backoff is set to zero seconds.

Notice the error line for the rejection, the second login, and the new token in the second connect frame.

Run it from the project root:

    python examples/stock_brokers/websockets/shoonya/ShoonyaOrderUpdatesSocket/example_2_rejected_then_logged_in_again.py
"""

import json
import logging
import sys
import types

from stock_brokers.websockets.shoonya import (
    ShoonyaOrderUpdatesSocket,
    ShoonyaSession,
)


class LoginRecord:
    """The Shoonya 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 StandInShoonyaAPI:
    """A stand-in for `ShoonyaAPI` 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'shoonya-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'ucc_code': 'FA12345',
        }

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

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


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

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

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

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

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


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

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

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

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

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

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

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

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

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

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

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

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

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

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


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

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

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

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

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

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

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

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


class RejectedThenLoggedInAgainExample:
    """Runs the Shoonya order update socket through a rejection and a good connection.

    Attributes:
        socket (ShoonyaOrderUpdatesSocket): 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.shoonya')
        api_module.ShoonyaAPI = StandInShoonyaAPI
        sys.modules['stock_brokers.api.shoonya'] = api_module
        rejected = {
            't': 'ak',
            's': 'NOT_OK',
        }
        accepted = {
            't': 'ak',
            's': 'OK',
        }
        plans = [
            {
                'frames': [
                    json.dumps(rejected),
                ],
            },
            {
                'frames': [
                    json.dumps(accepted),
                    json.dumps(self.order_update('COMPLETE', '5')),
                ],
            },
        ]
        websocket_module = ScriptedWebsocketModule(plans)
        sys.modules['websocket'] = websocket_module
        logger = logging.getLogger('shoonya.orders')
        self.socket = ShoonyaOrderUpdatesSocket(ShoonyaSession(logger), self.print_orders, logger)
        self.socket.MIN_BACKOFF_SECONDS = 0
        websocket_module.when_finished = self.socket.close

    def order_update(self, status, filled_quantity):
        """One Noren `om` order update for a TCS sell, every value a string.

        Args:
            status (str): Noren's order status.
            filled_quantity (str): How much has filled.

        Returns:
            dict: The update.
        """
        return {
            't': 'om',
            'norenordno': '26092500012345',
            'uid': 'FA12345',
            'actid': 'FA12345',
            'exch': 'NSE',
            'tsym': 'TCS-EQ',
            'trantype': 'S',
            'qty': '5',
            'fillshares': filled_quantity,
            'prc': '4120.00',
            'prctyp': 'LMT',
            'status': status,
        }

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

        Args:
            orders (list): The `om` updates of one frame, exactly as Noren sent them.
            received_at (datetime.datetime): When the frame arrived, which is not printed.

        Returns:
            None: This method returns nothing.
        """
        print(f'Callback received {len(orders)} update(s)')
        for order in orders:
            print(f"  Order {order['norenordno']} {order['status']}: {order['trantype']} {order['fillshares']} of {order['qty']} {order['tsym']}")

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

It prints:

Stand-in login number 1
Connecting to wss://api.shoonya.com/NorenWSTP/
Sent: {"t": "a", "uid": "FA12345", "actid": "FA12345", "source": "API", "accesstoken": "shoonya-token-1"}
INFO Order updates socket opened. Authenticating.
ERROR Authentication refused: {'t': 'ak', 's': 'NOT_OK'}
INFO Order updates socket closed with status 1000: normal closure
WARNING Logging in to Shoonya again.
Stand-in login number 2
WARNING Order updates socket disconnected. Reconnecting in 0 seconds.
Connecting to wss://api.shoonya.com/NorenWSTP/
Sent: {"t": "a", "uid": "FA12345", "actid": "FA12345", "source": "API", "accesstoken": "shoonya-token-2"}
INFO Order updates socket opened. Authenticating.
Sent: {"t": "o", "actid": "FA12345"}
INFO Authenticated and subscribed. Waiting for order updates.
Callback received 1 update(s)
  Order 26092500012345 COMPLETE: S 5 of 5 TCS-EQ
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/shoonya.py

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

    Args:
        session (ShoonyaSession): The login.
        on_updates (callable): Called on this socket's thread with each message's `om` updates that carry an order number, as dictionaries exactly as Noren 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
    self._user_id = None

_connect

_connect()

Opens the websocket and blocks until it closes, sending Noren's heartbeat meanwhile.

Returns:

Type	Description
None	This method returns nothing.

Source code in stock_brokers/websockets/shoonya.py

def _connect(self):
    """
    Opens the websocket and blocks until it closes, sending Noren's heartbeat meanwhile.

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

    self._websocket_application = websocket.WebSocketApp(
        FEED_URL,
        on_open=self._on_open,
        on_message=self._on_message,
        on_error=self._on_error,
        on_close=self._on_close,
    )
    self._websocket_application.run_forever(
        ping_interval=HEARTBEAT_INTERVAL_SECONDS,
        ping_timeout=HEARTBEAT_TIMEOUT_SECONDS,
        ping_payload=HEARTBEAT,
    )

_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/shoonya.py

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

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

_on_open

_on_open(websocket_connection)

Authenticates with the token in force now; the subscription waits for the acknowledgement.

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/shoonya.py

def _on_open(self, websocket_connection):
    """
    Authenticates with the token in force now; the subscription waits for the acknowledgement.

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

    Returns:
        None: This method returns nothing.
    """
    self._opened = True
    self._user_id, token = self._session.credentials()
    websocket_connection.send(json.dumps({
        "t": "a",
        "uid": self._user_id,
        "actid": self._user_id,
        "source": "API",
        "accesstoken": token,
    }))
    self._logger.info(f"{self.name} opened. Authenticating.")

_on_error

_on_error(websocket_connection, error)

Reports an error.

Parameters:

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

Returns:

Type	Description
None	This method returns nothing.

Source code in stock_brokers/websockets/shoonya.py

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

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

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

_on_message

_on_message(websocket_connection, message)

Subscribes on the connect acknowledgement, and hands each message's order updates on.

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/shoonya.py

def _on_message(self, websocket_connection, message):
    """
    Subscribes on the connect acknowledgement, and hands each message's order updates on.

    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
        message_type = entry.get("t")
        if message_type == CONNECT_ACK:
            if str(entry.get("s", "")).upper() == "OK":
                websocket_connection.send(json.dumps({"t": "o", "actid": self._user_id}))
                self._logger.info("Authenticated and subscribed. Waiting for order updates.")
            else:
                self._logger.error(f"Authentication refused: {entry}")
                self._authentication_rejected = True
                websocket_connection.close()
        elif message_type == ORDER_UPDATE and entry.get("norenordno"):
            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

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

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

run_forever

run_forever()

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

Returns:

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

Example usage

From example_1_reconnecting_after_a_drop.py

self.socket.run_forever()

From example_2_giving_up_after_a_second_refusal.py

self.refused_socket.run_forever()

From example_2_giving_up_after_a_second_refusal.py

self.closed_socket.run_forever()

Source code in stock_brokers/websockets/base.py

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

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

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

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

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

_gives_up_after_logging_in_again

_gives_up_after_logging_in_again(failed_connects)

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

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

Parameters:

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

Returns:

Type	Description
bool	True to give up now.

Source code in stock_brokers/websockets/base.py

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

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

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

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

_on_close

_on_close(websocket_connection, status_code, message)

Reports that the connection closed.

Parameters:

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

Returns:

Type	Description
None	This method returns nothing.

Source code in stock_brokers/websockets/base.py

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

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

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