flattrade

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

Both connect to wss://piconnect.flattrade.in/PiConnectWSTp/, a separate connection each. Flattrade permits only one websocket per session, so in production the order updates socket is not run and orders come from the poller alone.

The login. FlattradeSession holds the FlattradeAPI both sockets read their credentials from, afresh on every connect, so a login made by any process is picked up. Noren refuses a dead session inside an HTTP 200, so a constructor that returned is no proof of a working one: every login is confirmed by calling UserDetails, and a session it refuses raises. 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 user id. The frame is a with the token named accesstoken: Noren also accepts a c frame naming it susertoken, and answers that NOT_OK for a token REST accepts, which reads like an expired session and is not one. Only after the ak acknowledgement says OK (documented as "Ok" at one Noren broker and "OK" at another, 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 on the default interval is not enough and the server drops the connection.

Quotes. FlattradeQuotesSocket 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, and a name that differs from the one held is handed to on_names before the frame's ticks. 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. No tick is handed on until an instrument has a price.

Order updates. FlattradeOrderUpdatesSocket subscribes to the whole account with {"t": "o", "actid": <user id>}, and every change to an order then arrives as an om message carrying the order book's fields.

Neither socket writes Redis. The quotes socket hands each frame's new names to on_names and its ticks to on_ticks, and the order socket hands each frame's om updates to on_updates; the scripts in bin/flattrade/ do the writing.

Classes:

Name	Description
FlattradeSession	The Flattrade login every socket authenticates with, confirmed against UserDetails and logged in again by one socket at a time.
FlattradeQuotesSocket	One Noren market feed websocket carrying the touchline and depth of a batch of instruments.
FlattradeOrderUpdatesSocket	Flattrade's Noren order update websocket, subscribed to the whole account once authenticated.

FlattradeSession

FlattradeSession(logger)

The Flattrade login every socket authenticates with, confirmed against UserDetails and logged in again by one socket at a time.

Constructs FlattradeAPI, which logs in when the stored session is dead, and confirms the session.

Parameters:

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

Returns:

Type	Description
None	This method returns nothing.

Raises:

Type	Description
RuntimeError	When UserDetails refuses the session.
Exception	Whatever FlattradeAPI raises when it cannot log in.

Example programs

Confirming each loginUserdetails refuses the new session

examples/stock_brokers/websockets/flattrade/FlattradeSession/example_1_confirming_each_login.py

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

Noren refuses a dead session inside an HTTP 200 answer, so a `FlattradeAPI` that constructed without error is no proof of a working session. `FlattradeSession` therefore confirms every login by calling UserDetails, and `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.

Constructing the real `FlattradeAPI` logs in to Flattrade, so the program registers a stand-in module under `stock_brokers.api.flattrade`. 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, and `post` answers the UserDetails call with `Ok`.

Notice that each login is followed by a UserDetails check, and that the second request with the same stale token makes no login.

Run it from the project root:

    python examples/stock_brokers/websockets/flattrade/FlattradeSession/example_1_confirming_each_login.py
"""

import logging
import sys
import types

from stock_brokers.websockets.flattrade import (
    FlattradeSession,
)


class LoginRecord:
    """The Flattrade 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.
        refused_logins (set): The login numbers whose session UserDetails refuses.
    """

    access_token = None
    login_count = 0
    refused_logins = set()


class StandInFlattradeAPI:
    """A stand-in for `FlattradeAPI` 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'flattrade-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'username': 'FT012345',
        }

    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,
        }

    def post(self, url, timeout):
        """Answers the UserDetails call the session confirms every login with.

        Args:
            url (str): The endpoint the session calls.
            timeout (int): The request timeout in seconds.

        Returns:
            dict: What the real class returns, with Noren's answer under `data`.
        """
        status = 'Ok'
        if LoginRecord.login_count in LoginRecord.refused_logins:
            status = 'Not_Ok'
        print(f'UserDetails called for {LoginRecord.access_token}: {status}')
        return {
            'data': {
                'stat': status,
                'uname': 'EXAMPLE USER',
            },
        }


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

    Attributes:
        session (FlattradeSession): 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.flattrade')
        api_module.FlattradeAPI = StandInFlattradeAPI
        sys.modules['stock_brokers.api.flattrade'] = api_module
        self.session = FlattradeSession(logging.getLogger('flattrade.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__':
    ConfirmingEachLoginExample().run()

It prints:

Stand-in login number 1
UserDetails called for flattrade-token-1: Ok
User FT012345 was refused with flattrade-token-1
WARNING Logging in to Flattrade again.
Stand-in login number 2
UserDetails called for flattrade-token-2: Ok
Credentials now: ('FT012345', 'flattrade-token-2')
Credentials after the second request: ('FT012345', 'flattrade-token-2')
Logins made: 2

examples/stock_brokers/websockets/flattrade/FlattradeSession/example_2_userdetails_refuses_the_new_session.py

"""Logs in to Flattrade again without checking and shows the error raised when UserDetails refuses the new session.

`FlattradeSession.log_in_again_without_checking` is what the order update socket calls after a refusal: it logs in whether or not the token was already replaced, then confirms the new session against UserDetails. If Noren refuses that session, the session raises `RuntimeError` with Noren's answer, which the socket's reconnect loop treats as a login that failed, so the socket gives up and its script exits for systemd to restart.

Constructing the real `FlattradeAPI` logs in to Flattrade, so the program registers a stand-in module under `stock_brokers.api.flattrade` whose second login is scripted to be refused by the pretend UserDetails call.

Notice that the first login is confirmed, that the second one is made even though nothing was refused, and that the `RuntimeError` carries Noren's `Not_Ok` answer.

Run it from the project root:

    python examples/stock_brokers/websockets/flattrade/FlattradeSession/example_2_userdetails_refuses_the_new_session.py
"""

import logging
import sys
import types

from stock_brokers.websockets.flattrade import (
    FlattradeSession,
)


class LoginRecord:
    """The Flattrade 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.
        refused_logins (set): The login numbers whose session UserDetails refuses.
    """

    access_token = None
    login_count = 0
    refused_logins = set()


class StandInFlattradeAPI:
    """A stand-in for `FlattradeAPI` 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'flattrade-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'username': 'FT012345',
        }

    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,
        }

    def post(self, url, timeout):
        """Answers the UserDetails call the session confirms every login with.

        Args:
            url (str): The endpoint the session calls.
            timeout (int): The request timeout in seconds.

        Returns:
            dict: What the real class returns, with Noren's answer under `data`.
        """
        status = 'Ok'
        if LoginRecord.login_count in LoginRecord.refused_logins:
            status = 'Not_Ok'
        print(f'UserDetails called for {LoginRecord.access_token}: {status}')
        return {
            'data': {
                'stat': status,
                'uname': 'EXAMPLE USER',
            },
        }


class UserdetailsRefusesTheNewSessionExample:
    """Logs in again unconditionally and catches the refusal of the new session.

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

    def __init__(self):
        """Registers the stand-in API module, scripts the second login to be refused, 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.flattrade')
        api_module.FlattradeAPI = StandInFlattradeAPI
        sys.modules['stock_brokers.api.flattrade'] = api_module
        LoginRecord.refused_logins = {
            2,
        }
        self.session = FlattradeSession(logging.getLogger('flattrade.session'))

    def run(self):
        """Logs in again without checking and prints the error.

        Returns:
            None: This method returns nothing.
        """
        try:
            self.session.log_in_again_without_checking()
        except RuntimeError as error:
            print(f'RuntimeError: {error}')
        print(f'Logins made: {LoginRecord.login_count}')


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

It prints:

Stand-in login number 1
UserDetails called for flattrade-token-1: Ok
WARNING Logging in to Flattrade again.
Stand-in login number 2
UserDetails called for flattrade-token-2: Not_Ok
RuntimeError: UserDetails refused the session: {'stat': 'Not_Ok', 'uname': 'EXAMPLE USER'}
Logins made: 2

Methods:

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

Source code in stock_brokers/websockets/flattrade.py

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

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

    Returns:
        None: This method returns nothing.

    Raises:
        RuntimeError: When UserDetails refuses the session.
        Exception: Whatever `FlattradeAPI` raises when it cannot log in.
    """
    from stock_brokers.api.flattrade import FlattradeAPI

    self._api_class = FlattradeAPI
    self._logger = logger
    self._lock = threading.Lock()
    self._flattrade = FlattradeAPI()
    self._confirm()

credentials

credentials()

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

Returns:

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

Example usage

From example_1_confirming_each_login.py

user_id, failed_token = self.session.credentials()

From example_1_confirming_each_login.py

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

From example_1_confirming_each_login.py

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

Source code in stock_brokers/websockets/flattrade.py

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

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

log_in_again

log_in_again(stale_token)

Logs in again and confirms the session, 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
RuntimeError	When UserDetails refuses the new session.
Exception	Whatever FlattradeAPI raises when it cannot log in.

Example usage

From example_1_confirming_each_login.py

self.session.log_in_again(failed_token)

Source code in stock_brokers/websockets/flattrade.py

def log_in_again(self, stale_token):
    """
    Logs in again and confirms the session, 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:
        RuntimeError: When UserDetails refuses the new session.
        Exception: Whatever `FlattradeAPI` 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 and confirms the session, whether or not the token was already replaced.

Returns:

Type	Description
None	This method returns nothing.

Raises:

Type	Description
RuntimeError	When UserDetails refuses the new session.
Exception	Whatever FlattradeAPI raises when it cannot log in.

Example usage

From example_2_userdetails_refuses_the_new_session.py

self.session.log_in_again_without_checking()

Source code in stock_brokers/websockets/flattrade.py

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

    Returns:
        None: This method returns nothing.

    Raises:
        RuntimeError: When UserDetails refuses the new session.
        Exception: Whatever `FlattradeAPI` raises when it cannot log in.
    """
    with self._lock:
        self._log_in()

_log_in

_log_in()

Logs in by constructing FlattradeAPI again and confirms the session; the caller holds the lock.

Returns:

Type	Description
None	This method returns nothing.

Raises:

Type	Description
RuntimeError	When UserDetails refuses the new session.
Exception	Whatever FlattradeAPI raises when it cannot log in.

Source code in stock_brokers/websockets/flattrade.py

def _log_in(self):
    """
    Logs in by constructing `FlattradeAPI` again and confirms the session; the caller holds the lock.

    Returns:
        None: This method returns nothing.

    Raises:
        RuntimeError: When UserDetails refuses the new session.
        Exception: Whatever `FlattradeAPI` raises when it cannot log in.
    """
    self._logger.warning("Logging in to Flattrade again.")
    self._flattrade = self._api_class()
    self._confirm()

_confirm

_confirm()

Checks that UserDetails serves the session, since Noren refuses a dead session inside an HTTP 200.

Returns:

Type	Description
None	This method returns nothing.

Raises:

Type	Description
RuntimeError	When UserDetails refuses the session.

Source code in stock_brokers/websockets/flattrade.py

def _confirm(self):
    """
    Checks that UserDetails serves the session, since Noren refuses a dead session inside an HTTP 200.

    Returns:
        None: This method returns nothing.

    Raises:
        RuntimeError: When UserDetails refuses the session.
    """
    data = (self._flattrade.post(url=USER_DETAILS_URL, timeout=10) or {}).get("data")
    if not isinstance(data, dict) or str(data.get("stat", "")).lower() != "ok":
        raise RuntimeError(f"UserDetails refused the session: {str(data)[:200]}")

FlattradeQuotesSocket

FlattradeQuotesSocket(name, tokens, names, session, on_names, on_ticks, 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 updated as Noren names instruments.	required
session	FlattradeSession	The shared login.	required
on_names	callable	Called with a dictionary of tokens to their new names when a frame names instruments, on this socket's thread.	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

Merging incremental updatesRefused acknowledgement

examples/stock_brokers/websockets/flattrade/FlattradeQuotesSocket/example_1_merging_incremental_updates.py

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

A `FlattradeQuotesSocket` 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.flattrade` module lets the real `FlattradeSession` log in and confirm its session against a pretend UserDetails call.

Notice that the new name is handed to `on_names` before the first 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/flattrade/FlattradeQuotesSocket/example_1_merging_incremental_updates.py
"""

import json
import logging
import sys
import types

from stock_brokers.websockets.flattrade import (
    FlattradeQuotesSocket,
    FlattradeSession,
)


class LoginRecord:
    """The Flattrade 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.
        refused_logins (set): The login numbers whose session UserDetails refuses.
    """

    access_token = None
    login_count = 0
    refused_logins = set()


class StandInFlattradeAPI:
    """A stand-in for `FlattradeAPI` 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'flattrade-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'username': 'FT012345',
        }

    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,
        }

    def post(self, url, timeout):
        """Answers the UserDetails call the session confirms every login with.

        Args:
            url (str): The endpoint the session calls.
            timeout (int): The request timeout in seconds.

        Returns:
            dict: What the real class returns, with Noren's answer under `data`.
        """
        status = 'Ok'
        if LoginRecord.login_count in LoginRecord.refused_logins:
            status = 'Not_Ok'
        print(f'UserDetails called for {LoginRecord.access_token}: {status}')
        return {
            'data': {
                'stat': status,
                'uname': 'EXAMPLE USER',
            },
        }


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 Flattrade quotes socket against an acknowledgement and one update.

    Attributes:
        socket (FlattradeQuotesSocket): 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.flattrade')
        api_module.FlattradeAPI = StandInFlattradeAPI
        sys.modules['stock_brokers.api.flattrade'] = api_module
        connect_acknowledgement = {
            't': 'ak',
            's': 'OK',
            'uid': 'FT012345',
        }
        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('flattrade.quotes')
        tokens = [
            'NSE|2885',
        ]
        names = {}
        session = FlattradeSession(logger)
        self.socket = FlattradeQuotesSocket('socket_0', tokens, names, session, self.print_names, self.print_ticks, logger)
        websocket_module.when_finished = self.socket.close

    def print_names(self, named):
        """Prints the instruments a frame newly named.

        Args:
            named (dict): Tokens to their new names.

        Returns:
            None: This method returns nothing.
        """
        print(f'Names learned: {named}')

    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:
            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
UserDetails called for flattrade-token-1: Ok
Connecting to wss://piconnect.flattrade.in/PiConnectWSTp/
Sent: {"t": "a", "uid": "FT012345", "actid": "FT012345", "source": "API", "accesstoken": "flattrade-token-1"}
INFO socket_0 opened. Authenticating.
Sent: {"t": "d", "k": "NSE|2885"}
INFO socket_0 authenticated and subscribed to 1 instrument(s).
Names learned: {'NSE|2885': '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/flattrade/FlattradeQuotesSocket/example_2_refused_acknowledgement.py

"""Shows a Flattrade 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`. `FlattradeQuotesSocket` then closes the connection and its reconnect loop asks the shared `FlattradeSession` to log in again, which the session confirms against UserDetails 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.flattrade` module counts each `FlattradeAPI` construction as a login. The backoff is set to zero seconds.

Notice the second login and its UserDetails check, 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/flattrade/FlattradeQuotesSocket/example_2_refused_acknowledgement.py
"""

import json
import logging
import sys
import types

from stock_brokers.websockets.flattrade import (
    FlattradeQuotesSocket,
    FlattradeSession,
)


class LoginRecord:
    """The Flattrade 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.
        refused_logins (set): The login numbers whose session UserDetails refuses.
    """

    access_token = None
    login_count = 0
    refused_logins = set()


class StandInFlattradeAPI:
    """A stand-in for `FlattradeAPI` 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'flattrade-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'username': 'FT012345',
        }

    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,
        }

    def post(self, url, timeout):
        """Answers the UserDetails call the session confirms every login with.

        Args:
            url (str): The endpoint the session calls.
            timeout (int): The request timeout in seconds.

        Returns:
            dict: What the real class returns, with Noren's answer under `data`.
        """
        status = 'Ok'
        if LoginRecord.login_count in LoginRecord.refused_logins:
            status = 'Not_Ok'
        print(f'UserDetails called for {LoginRecord.access_token}: {status}')
        return {
            'data': {
                'stat': status,
                'uname': 'EXAMPLE USER',
            },
        }


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 Flattrade quotes socket through a refused connect frame and a good connection.

    Attributes:
        socket (FlattradeQuotesSocket): 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.flattrade')
        api_module.FlattradeAPI = StandInFlattradeAPI
        sys.modules['stock_brokers.api.flattrade'] = 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('flattrade.quotes')
        tokens = [
            'MCX|565899',
            'MCX|565900',
        ]
        names = {}
        session = FlattradeSession(logger)
        self.socket = FlattradeQuotesSocket('socket_0', tokens, names, session, self.print_names, self.print_ticks, logger)
        self.socket.MIN_BACKOFF_SECONDS = 0
        websocket_module.when_finished = self.socket.close

    def print_names(self, named):
        """Prints the instruments a frame newly named.

        Args:
            named (dict): Tokens to their new names.

        Returns:
            None: This method returns nothing.
        """
        print(f'Names learned: {named}')

    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:
            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
UserDetails called for flattrade-token-1: Ok
Connecting to wss://piconnect.flattrade.in/PiConnectWSTp/
Sent: {"t": "a", "uid": "FT012345", "actid": "FT012345", "source": "API", "accesstoken": "flattrade-token-1"}
INFO socket_0 opened. Authenticating.
ERROR socket_0 authentication rejected: {'t': 'ak', 's': 'NOT_OK', 'emsg': 'Session Expired :  Invalid Session Key'}
INFO socket_0 closed with status 1000: normal closure
WARNING Logging in to Flattrade again.
Stand-in login number 2
UserDetails called for flattrade-token-2: Ok
WARNING socket_0 disconnected. Reconnecting in 0 seconds.
Connecting to wss://piconnect.flattrade.in/PiConnectWSTp/
Sent: {"t": "a", "uid": "FT012345", "actid": "FT012345", "source": "API", "accesstoken": "flattrade-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).
Names learned: {'MCX|565899': '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
Names learned: {'MCX|565900': '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/flattrade.py

def __init__(self, name, tokens, names, session, on_names, 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 `EXCHANGE|TOKEN` instrument tokens.
        names (dict[str, str]): Each token to the instrument's name, shared between sockets and updated as Noren names instruments.
        session (FlattradeSession): The shared login.
        on_names (callable): Called with a dictionary of tokens to their new names when a frame names instruments, on this socket's thread.
        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_names = on_names
    self._on_ticks = on_ticks
    self._token = None
    self._user_id = None
    self._state = {}

_connect

_connect()

Opens the websocket with the credentials in force now and blocks until it closes, sending Noren's heartbeat meanwhile.

Returns:

Type	Description
None	This method returns nothing.

Source code in stock_brokers/websockets/flattrade.py

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

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

    self._user_id, self._token = self._session.credentials()
    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_PAYLOAD,
    )

_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/flattrade.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)

Sends the connect frame; the subscription waits for its 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/flattrade.py

def _on_open(self, websocket_connection):
    """
    Sends the connect frame; the subscription waits for its acknowledgement.

    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({
        "t": "a",
        "uid": self._user_id,
        "actid": self._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/flattrade.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 one Noren frame, handing its new names on first and then its ticks.

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

def _on_message(self, websocket_connection, message):
    """
    Handles one Noren frame, handing its new names on first and then its ticks.

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

    Returns:
        None: This method returns nothing.
    """
    ticks, named = self._parse(message)
    if named:
        self._on_names(named)
    if ticks:
        self._on_ticks(ticks)

_parse

_parse(message)

Reads one Noren frame, subscribing on the connect acknowledgement and learning names on the way.

Parameters:

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

Returns:

Type	Description
tuple	The frame's ticks as a list, and the instruments it newly names as a dictionary of token to name.

Source code in stock_brokers/websockets/flattrade.py

def _parse(self, message):
    """
    Reads one Noren frame, subscribing on the connect acknowledgement and learning names on the way.

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

    Returns:
        tuple: The frame's ticks as a list, and the instruments it newly names as a dictionary of token to name.
    """
    if isinstance(message, (bytes, bytearray)):
        message = message.decode("utf-8", errors="replace")
    try:
        payload = json.loads(message)
    except (TypeError, json.JSONDecodeError):
        self._logger.warning(f"{self.name} could not decode a frame: {str(message)[:200]}")
        return [], {}
    if not isinstance(payload, dict):
        return [], {}

    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 rejected: {payload}")
            self._authentication_rejected = True
            self._websocket_application.close()
        return [], {}

    if message_type not in (TOUCHLINE_ACK, TOUCHLINE_FEED, DEPTH_ACK, DEPTH_FEED):
        return [], {}
    exchange = payload.get("e")
    token = payload.get("tk")
    if not exchange or not token:
        return [], {}
    instrument_token = f"{exchange}|{token}"

    named = {}
    if message_type in (TOUCHLINE_ACK, DEPTH_ACK) and payload.get("ts"):
        name = f"{exchange}:{payload['ts']}"
        if self._names.get(instrument_token) != name:
            self._names[instrument_token] = name
            named[instrument_token] = name

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

_build_tick

_build_tick(instrument_token, state, name)

Builds a normalized tick from the merged state of an instrument.

Parameters:

Name	Type	Description	Default
instrument_token	str	The EXCHANGE|TOKEN key.	required
state	dict	The instrument's merged fields.	required
name	str | None	The instrument's name, or None to key the tick by its token.	required

Returns:

Type	Description
dict	The tick.

Source code in stock_brokers/websockets/flattrade.py

def _build_tick(self, instrument_token, state, name):
    """
    Builds a normalized tick from the merged state of an instrument.

    Args:
        instrument_token (str): The `EXCHANGE|TOKEN` key.
        state (dict): The instrument's merged fields.
        name (str | None): The instrument's name, or None to key the tick by its token.

    Returns:
        dict: The tick.
    """
    exchange = instrument_token.partition("|")[0]
    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):
        if state.get(f"bp{level}") is not None:
            depth["buy"].append({
                "quantity": self._number(state.get(f"bq{level}"), int),
                "price": self._number(state.get(f"bp{level}")),
                "orders": self._number(state.get(f"bo{level}"), int),
            })
        if state.get(f"sp{level}") is not None:
            depth["sell"].append({
                "quantity": self._number(state.get(f"sq{level}"), int),
                "price": self._number(state.get(f"sp{level}")),
                "orders": self._number(state.get(f"so{level}"), int),
            })

    mode = "quote"
    if depth["buy"] or depth["sell"]:
        mode = "full"
    feed_time = self._number(state.get("ft"), int)
    return {
        "id": name or instrument_token,
        "broker": "flattrade",
        "instrument_token": instrument_token,
        "exchange": exchange,
        "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, treating blanks and placeholders as missing.

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

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

    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, whichever way it is spelled.

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

def _trade_time(self, value, feed_time):
    """
    Noren's last trade time as an epoch, whichever way it is spelled.

    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}")

FlattradeOrderUpdatesSocket

FlattradeOrderUpdatesSocket(session, on_updates, logger)

Bases: BrokerWebsocket

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

Sets up the order updates socket.

Parameters:

Name	Type	Description	Default
session	FlattradeSession	The login.	required
on_updates	callable	Called on this socket's thread with each frame's om updates that carry an order number, as dictionaries exactly as Noren sent them, and the datetime the frame 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/flattrade/FlattradeOrderUpdatesSocket/example_1_subscribing_to_the_account.py

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

`FlattradeOrderUpdatesSocket` 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. In production Flattrade permits one websocket per session, so this socket is not run there; the program shows what it does when it is.

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.flattrade` module lets the real `FlattradeSession` log in and confirm its session without reaching Flattrade.

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/flattrade/FlattradeOrderUpdatesSocket/example_1_subscribing_to_the_account.py
"""

import json
import logging
import sys
import types

from stock_brokers.websockets.flattrade import (
    FlattradeOrderUpdatesSocket,
    FlattradeSession,
)


class LoginRecord:
    """The Flattrade 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.
        refused_logins (set): The login numbers whose session UserDetails refuses.
    """

    access_token = None
    login_count = 0
    refused_logins = set()


class StandInFlattradeAPI:
    """A stand-in for `FlattradeAPI` 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'flattrade-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'username': 'FT012345',
        }

    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,
        }

    def post(self, url, timeout):
        """Answers the UserDetails call the session confirms every login with.

        Args:
            url (str): The endpoint the session calls.
            timeout (int): The request timeout in seconds.

        Returns:
            dict: What the real class returns, with Noren's answer under `data`.
        """
        status = 'Ok'
        if LoginRecord.login_count in LoginRecord.refused_logins:
            status = 'Not_Ok'
        print(f'UserDetails called for {LoginRecord.access_token}: {status}')
        return {
            'data': {
                'stat': status,
                'uname': 'EXAMPLE USER',
            },
        }


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 Flattrade order update socket against one scripted connection.

    Attributes:
        socket (FlattradeOrderUpdatesSocket): 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.flattrade')
        api_module.FlattradeAPI = StandInFlattradeAPI
        sys.modules['stock_brokers.api.flattrade'] = 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('flattrade.orders')
        self.socket = FlattradeOrderUpdatesSocket(FlattradeSession(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': 'FT012345',
            'actid': 'FT012345',
            '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
UserDetails called for flattrade-token-1: Ok
Connecting to wss://piconnect.flattrade.in/PiConnectWSTp/
Sent: {"t": "a", "uid": "FT012345", "actid": "FT012345", "source": "API", "accesstoken": "flattrade-token-1"}
INFO Order updates socket opened. Authenticating.
Sent: {"t": "o", "actid": "FT012345"}
INFO Authenticated and subscribed to order updates. 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/flattrade/FlattradeOrderUpdatesSocket/example_2_rejected_then_logged_in_again.py

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

When the `ak` acknowledgement is not `OK`, `FlattradeOrderUpdatesSocket` closes the connection, and its reconnect loop calls `FlattradeSession.log_in_again_without_checking`, which logs in and confirms the new session against UserDetails 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.flattrade` module counts each `FlattradeAPI` construction as a login. The backoff is set to zero seconds.

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

Run it from the project root:

    python examples/stock_brokers/websockets/flattrade/FlattradeOrderUpdatesSocket/example_2_rejected_then_logged_in_again.py
"""

import json
import logging
import sys
import types

from stock_brokers.websockets.flattrade import (
    FlattradeOrderUpdatesSocket,
    FlattradeSession,
)


class LoginRecord:
    """The Flattrade 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.
        refused_logins (set): The login numbers whose session UserDetails refuses.
    """

    access_token = None
    login_count = 0
    refused_logins = set()


class StandInFlattradeAPI:
    """A stand-in for `FlattradeAPI` 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'flattrade-token-{LoginRecord.login_count}'
        print(f'Stand-in login number {LoginRecord.login_count}')
        self._settings = {
            'username': 'FT012345',
        }

    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,
        }

    def post(self, url, timeout):
        """Answers the UserDetails call the session confirms every login with.

        Args:
            url (str): The endpoint the session calls.
            timeout (int): The request timeout in seconds.

        Returns:
            dict: What the real class returns, with Noren's answer under `data`.
        """
        status = 'Ok'
        if LoginRecord.login_count in LoginRecord.refused_logins:
            status = 'Not_Ok'
        print(f'UserDetails called for {LoginRecord.access_token}: {status}')
        return {
            'data': {
                'stat': status,
                'uname': 'EXAMPLE USER',
            },
        }


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 Flattrade order update socket through a rejection and a good connection.

    Attributes:
        socket (FlattradeOrderUpdatesSocket): 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.flattrade')
        api_module.FlattradeAPI = StandInFlattradeAPI
        sys.modules['stock_brokers.api.flattrade'] = 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('flattrade.orders')
        self.socket = FlattradeOrderUpdatesSocket(FlattradeSession(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': 'FT012345',
            'actid': 'FT012345',
            '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
UserDetails called for flattrade-token-1: Ok
Connecting to wss://piconnect.flattrade.in/PiConnectWSTp/
Sent: {"t": "a", "uid": "FT012345", "actid": "FT012345", "source": "API", "accesstoken": "flattrade-token-1"}
INFO Order updates socket opened. Authenticating.
ERROR Authentication rejected: {'t': 'ak', 's': 'NOT_OK'}
INFO Order updates socket closed with status 1000: normal closure
WARNING Logging in to Flattrade again.
Stand-in login number 2
UserDetails called for flattrade-token-2: Ok
WARNING Order updates socket disconnected. Reconnecting in 0 seconds.
Connecting to wss://piconnect.flattrade.in/PiConnectWSTp/
Sent: {"t": "a", "uid": "FT012345", "actid": "FT012345", "source": "API", "accesstoken": "flattrade-token-2"}
INFO Order updates socket opened. Authenticating.
Sent: {"t": "o", "actid": "FT012345"}
INFO Authenticated and subscribed to order updates. 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/flattrade.py

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

    Args:
        session (FlattradeSession): The login.
        on_updates (callable): Called on this socket's thread with each frame's `om` updates that carry an order number, as dictionaries exactly as Noren sent them, and the `datetime` the frame 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/flattrade.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_PAYLOAD,
    )

_log_in_again

_log_in_again()

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

Returns:

Type	Description
None	This method returns nothing.

Source code in stock_brokers/websockets/flattrade.py

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

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

_on_open

_on_open(websocket_connection)

Sends the connect frame with the token in force now; the subscription waits for its 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/flattrade.py

def _on_open(self, websocket_connection):
    """
    Sends the connect frame with the token in force now; the subscription waits for its 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/flattrade.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 once authenticated, and hands each frame's order updates on.

Parameters:

Name	Type	Description	Default
websocket_connection	WebSocketApp	The connection the frame 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/flattrade.py

def _on_message(self, websocket_connection, message):
    """
    Subscribes once authenticated, and hands each frame's order updates on.

    Args:
        websocket_connection (websocket.WebSocketApp): The connection the frame 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 to order updates. Waiting for order updates.")
            else:
                self._logger.error(f"Authentication rejected: {entry}")
                self._authentication_rejected = True
                websocket_connection.close()
        elif message_type == ORDER_UPDATE and entry.get("norenordno"):
            orders.append(entry)
        else:
            self._logger.debug(f"Ignoring a message of type {message_type!r}: {json.dumps(entry)[:200]}")
    if orders:
        self._on_updates(orders, received_at)

close

close()

Closes the connection and stops reconnecting.

Returns:

Type	Description
None	This method returns nothing.

Example usage

From example_2_giving_up_after_a_second_refusal.py

self.closed_socket.close()

Source code in stock_brokers/websockets/base.py

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

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

run_forever

run_forever()

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

Returns:

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

Example usage

From example_1_reconnecting_after_a_drop.py

self.socket.run_forever()

From example_2_giving_up_after_a_second_refusal.py

self.refused_socket.run_forever()

From example_2_giving_up_after_a_second_refusal.py

self.closed_socket.run_forever()

Source code in stock_brokers/websockets/base.py

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

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

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

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

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

_gives_up_after_logging_in_again

_gives_up_after_logging_in_again(failed_connects)

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

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

Parameters:

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

Returns:

Type	Description
bool	True to give up now.

Source code in stock_brokers/websockets/base.py

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

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

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

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

_on_close

_on_close(websocket_connection, status_code, message)

Reports that the connection closed.

Parameters:

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

Returns:

Type	Description
None	This method returns nothing.

Source code in stock_brokers/websockets/base.py

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

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

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