Fyers' two websockets: the HSM market feed and the account's order and position update stream.
The login.
FyersSession holds the FyersAPI both sockets read their credentials from, afresh on every connect, so a login made by any process is picked up.
FyersAPI logs in only when its probe raises, and Fyers can refuse a dead session inside an HTTP 200, so every login is confirmed against the profile endpoint and a refused profile raises.
The access token is a JWT: a token whose exp has passed is treated as a refused login without sending anything, because Fyers' edge bans addresses that keep sending refused requests.
The quote sockets log in again one at a time, and only when no other socket or process has already replaced the token that failed; the order socket logs in again whenever it is refused, without that check, as it always has.
Quotes.
FyersQuotesSocket speaks the binary protocol Fyers' SDK calls HSM, at wss://socket.fyers.in/hsm/v1-5/prod.
The feed is not authenticated with the access token but with the hsm_key claim inside it, sent in a binary frame after connecting.
Subscription is by HSM topic name, so each connect first resolves the symbols through POST https://api-t1.fyers.in/data/symbol-token, whose validSymbol maps each symbol to its fytoken.
The fytoken's first four digits name the exchange segment and the digits from the eleventh on the exchange token, giving sf|nse_cm|3045 for NSE:SBIN-EQ; an index is if|nse_cm|Nifty 50, by the index names Fyers' SDK carries.
Once the authentication frame is acknowledged the socket selects full mode and subscribes the topic names, 1,500 a frame, and it acknowledges every ack_count data messages, the count the authentication response gives, as the SDK does.
The feed is incremental: a snapshot packet (type 83) names a topic and carries every field, followed by the price multiplier and precision and three strings; update packets (85, and 76 for lite) carry the topic id and positional values, where -2147483648 means "no value", read against the fixed field tables below.
Prices are scaled integers, divided by 10 to the precision times the multiplier.
A Cloudflare ban on the symbol lookup pauses the socket thirty minutes and a rate limit five, rather than logging in, because every refused request extends a ban; that pause is why the quote socket keeps a reconnect loop of its own.
Symbol lookup codes -8, -15, -16 and -17, or HTTP 401, mean the session is dead.
Order updates.
FyersOrderUpdatesSocket connects to wss://socket.fyers.in/trade/v3 with an authorization header of the app id and access token joined by a colon, so a bad credential fails the HTTP upgrade itself.
Once open it sends {"T": "SUB_ORD", "SLIST": ["orders", "positions"], "SUB_T": 1}, Fyers answers with code 1605, and every change then arrives as JSON: {"s": "ok", "orders": {...}} for an order and {"s": "ok", "positions": {...}} for a position.
An error message carrying a session code closes the socket, after the message's orders and positions have been handed on.
Fyers answers its keepalive with a bare ping or pong rather than a frame.
Neither socket writes Redis.
The quotes socket hands each message's ticks to on_ticks, and the order socket hands each message's orders and positions to on_updates; the scripts in bin/fyers/ do the writing.
Classes:
| Name | Description |
|---|---|
FyersRefusal |
Fyers refused a request, with its code and message. |
FyersSession |
The Fyers login every socket authenticates with, confirmed against the profile and logged in again by one socket at a time. |
FyersHsmFrames |
Builds and reads the HSM binary frames of Fyers' market feed. |
FyersQuotesSocket |
One Fyers HSM market feed websocket for a batch of symbols. |
FyersOrderUpdatesSocket |
Fyers' order and position update websocket, authenticated in its handshake. |
Bases: Exception
Fyers refused a request, with its code and message.
Attributes:
| Name | Type | Description |
|---|---|---|
code |
int | str | None
|
Fyers' code, or the HTTP status. |
message |
str
|
What Fyers said. |
Keeps the code and the message.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
code
|
int | str | None
|
Fyers' code, or the HTTP status. |
required |
message
|
str
|
What Fyers said. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example programs
examples/stock_brokers/websockets/fyers/FyersRefusal/example_1_raising_and_catching.py
"""Raises a `FyersRefusal` the way the Fyers socket code does and reads its code and message.
`FyersRefusal` is the exception the Fyers socket code raises when Fyers refuses a request: the symbol-token lookup answering with an error, or the profile check refusing a session. It keeps Fyers' own code, or the HTTP status when there is no code, in `code`, and what Fyers said in `message`, so the reconnect loop can tell a dead session (codes -8, -15, -16, -17 and HTTP 401) from a rate limit (HTTP 429).
This program raises three refusals with the shapes the real code uses and catches each one. It needs no stand-ins, because the exception is plain data.
Notice that `code` keeps its original type, an integer for Fyers codes and HTTP statuses and None when Fyers gave none, and that the exception prints as the pair of its arguments.
Run it from the project root:
python examples/stock_brokers/websockets/fyers/FyersRefusal/example_1_raising_and_catching.py
"""
from stock_brokers.websockets.fyers import (
FyersRefusal,
)
class RaisingAndCatchingExample:
"""Raises and catches several Fyers refusals.
Attributes:
refusals (list): Pairs of a code and a message to raise.
"""
def __init__(self):
"""Lists the refusals to raise.
Returns:
None: This method returns nothing.
"""
self.refusals = [
(
-16,
'Could not authenticate the user',
),
(
429,
'Too many requests',
),
(
None,
'no symbol resolved to a topic',
),
]
def raise_refusal(self, code, message):
"""Raises one refusal.
Args:
code (int | None): Fyers' code, or the HTTP status.
message (str): What Fyers said.
Returns:
None: This method returns nothing.
Raises:
FyersRefusal: Always.
"""
raise FyersRefusal(code, message)
def run(self):
"""Raises each refusal and prints what the caught exception carries.
Returns:
None: This method returns nothing.
"""
for code, message in self.refusals:
try:
self.raise_refusal(code, message)
except FyersRefusal as refusal:
print(f'Caught {refusal!r}: code={refusal.code!r} message={refusal.message!r}')
if __name__ == '__main__':
RaisingAndCatchingExample().run()
It prints:
Caught FyersRefusal(-16, 'Could not authenticate the user'): code=-16 message='Could not authenticate the user'
Caught FyersRefusal(429, 'Too many requests'): code=429 message='Too many requests'
Caught FyersRefusal(None, 'no symbol resolved to a topic'): code=None message='no symbol resolved to a topic'
examples/stock_brokers/websockets/fyers/FyersRefusal/example_2_refused_profile_at_startup.py
"""Catches the `FyersRefusal` a Fyers session raises when the profile refuses its very first login.
`FyersSession` confirms its first login against the profile endpoint while it is being built, so a session Fyers refuses never reaches a socket: the constructor raises `FyersRefusal`. A broker script builds its session before its sockets, so this is the error it sees and exits on when the stored login is dead and a fresh one is refused.
Constructing the real `FyersAPI` logs in to Fyers, so the program registers a stand-in module under `stock_brokers.api.fyers` whose first login is scripted to have its profile refused with code -16.
Notice that the refusal carries the profile's code and a message quoting Fyers' answer.
Run it from the project root:
python examples/stock_brokers/websockets/fyers/FyersRefusal/example_2_refused_profile_at_startup.py
"""
import base64
import json
import logging
import sys
import types
from stock_brokers.websockets.fyers import (
FyersRefusal,
FyersSession,
)
class LoginRecord:
"""The Fyers 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.
expired_logins (set): The login numbers whose token has already expired.
refused_profiles (set): The login numbers whose session the profile endpoint refuses.
"""
access_token = None
login_count = 0
expired_logins = set()
refused_profiles = set()
class StandInFyersAPI:
"""A stand-in for `FyersAPI` whose construction is a pretend login that issues a JWT access token."""
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
expiry = 4102444800
if LoginRecord.login_count in LoginRecord.expired_logins:
expiry = 1
claims = {
'exp': expiry,
'hsm_key': f'hsm-key-{LoginRecord.login_count}',
}
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
LoginRecord.access_token = f'header.{payload}.signature'
print(f'Stand-in login number {LoginRecord.login_count}')
self._settings = {
'app_id': 'XY1234-100',
}
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 get(self, url, timeout):
"""Answers the profile 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 Fyers' answer under `data`.
"""
if LoginRecord.login_count in LoginRecord.refused_profiles:
print(f'Profile refused for login {LoginRecord.login_count}')
return {
'data': {
's': 'error',
'code': -16,
'message': 'Could not authenticate the user',
},
}
print(f'Profile served for login {LoginRecord.login_count}')
return {
'data': {
's': 'ok',
'data': {
'fy_id': 'XY01234',
},
},
}
class RefusedProfileAtStartupExample:
"""Builds a Fyers session whose first profile check is refused.
Attributes:
logger (logging.Logger): Where the session would report.
"""
def __init__(self):
"""Registers the stand-in API module and scripts the first profile to be refused.
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.fyers')
api_module.FyersAPI = StandInFyersAPI
sys.modules['stock_brokers.api.fyers'] = api_module
LoginRecord.refused_profiles = {
1,
}
self.logger = logging.getLogger('fyers.session')
def run(self):
"""Builds the session and prints the refusal it raises.
Returns:
None: This method returns nothing.
"""
try:
FyersSession(self.logger)
except FyersRefusal as refusal:
print(f'The session could not start: code={refusal.code}')
print(f'Message: {refusal.message}')
if __name__ == '__main__':
RefusedProfileAtStartupExample().run()
It prints:
Source code in stock_brokers/websockets/fyers.py
The Fyers login every socket authenticates with, confirmed against the profile and logged in again by one socket at a time.
Constructs FyersAPI, which logs in when its probe raises, and confirms the session against the profile.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
logger
|
Logger
|
Where logins are reported. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Raises:
| Type | Description |
|---|---|
FyersRefusal
|
When the profile is refused. |
Exception
|
Whatever |
Example programs
examples/stock_brokers/websockets/fyers/FyersSession/example_1_reading_the_jwt.py
"""Reads a Fyers session's credentials and the claims inside its JWT access token, and logs in again once.
`FyersSession` holds one `FyersAPI`, confirms every login against the profile endpoint because Fyers can refuse a dead session inside an HTTP 200, and reads the login afresh on every connect. The access token is a JWT, and `token_claims` decodes its claims: the quotes socket reads `exp` to avoid sending an expired token and `hsm_key` to authenticate the market feed. `log_in_again` logs in only when the token a refused socket used is still the one in force.
Constructing the real `FyersAPI` logs in to Fyers, so the program registers a stand-in module under `stock_brokers.api.fyers`. Each construction counts as one login and issues a JWT with the next numbered hsm key, `get` serves the profile, and `_current_login` answers from a shared record standing in for the Redis `last_login` hash.
Notice the profile check after each login, the claims of the old and new tokens, and that the second request with the stale token makes no login.
Run it from the project root:
python examples/stock_brokers/websockets/fyers/FyersSession/example_1_reading_the_jwt.py
"""
import base64
import json
import logging
import sys
import types
from stock_brokers.websockets.fyers import (
FyersSession,
)
class LoginRecord:
"""The Fyers 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.
expired_logins (set): The login numbers whose token has already expired.
refused_profiles (set): The login numbers whose session the profile endpoint refuses.
"""
access_token = None
login_count = 0
expired_logins = set()
refused_profiles = set()
class StandInFyersAPI:
"""A stand-in for `FyersAPI` whose construction is a pretend login that issues a JWT access token."""
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
expiry = 4102444800
if LoginRecord.login_count in LoginRecord.expired_logins:
expiry = 1
claims = {
'exp': expiry,
'hsm_key': f'hsm-key-{LoginRecord.login_count}',
}
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
LoginRecord.access_token = f'header.{payload}.signature'
print(f'Stand-in login number {LoginRecord.login_count}')
self._settings = {
'app_id': 'XY1234-100',
}
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 get(self, url, timeout):
"""Answers the profile 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 Fyers' answer under `data`.
"""
if LoginRecord.login_count in LoginRecord.refused_profiles:
print(f'Profile refused for login {LoginRecord.login_count}')
return {
'data': {
's': 'error',
'code': -16,
'message': 'Could not authenticate the user',
},
}
print(f'Profile served for login {LoginRecord.login_count}')
return {
'data': {
's': 'ok',
'data': {
'fy_id': 'XY01234',
},
},
}
class ReadingTheJwtExample:
"""Reads a Fyers session's token claims around a login.
Attributes:
session (FyersSession): 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.fyers')
api_module.FyersAPI = StandInFyersAPI
sys.modules['stock_brokers.api.fyers'] = api_module
self.session = FyersSession(logging.getLogger('fyers.session'))
def run(self):
"""Prints the claims, logs in again twice with the same stale token, and prints the new claims.
Returns:
None: This method returns nothing.
"""
app_id, failed_token = self.session.credentials()
print(f'App id: {app_id}')
print(f'Claims: {self.session.token_claims(failed_token)}')
self.session.log_in_again(failed_token)
self.session.log_in_again(failed_token)
new_token = self.session.credentials()[1]
print(f'Claims: {self.session.token_claims(new_token)}')
print(f'Logins made: {LoginRecord.login_count}')
if __name__ == '__main__':
ReadingTheJwtExample().run()
It prints:
examples/stock_brokers/websockets/fyers/FyersSession/example_2_profile_refuses_the_new_login.py
"""Logs in to Fyers again without checking, and shows `FyersRefusal` when the profile refuses the new session.
`FyersSession.log_in_again_without_checking` is what the order update socket calls after a refusal: it logs in whether or not another process already replaced the token, then confirms the new session against the profile. When the profile is refused the session raises `FyersRefusal` carrying Fyers' code, and a socket's reconnect loop takes that as a failed login and gives up. Separately, `token_claims` answers an empty dictionary for a token that is not a JWT, which the sockets treat as an expired token.
Constructing the real `FyersAPI` logs in to Fyers, so the program registers a stand-in module under `stock_brokers.api.fyers` whose second login is scripted to have its profile refused with code -16.
Notice that the refusal's code and message come from the profile answer, and that a malformed token yields no claims instead of an error.
Run it from the project root:
python examples/stock_brokers/websockets/fyers/FyersSession/example_2_profile_refuses_the_new_login.py
"""
import base64
import json
import logging
import sys
import types
from stock_brokers.websockets.fyers import (
FyersRefusal,
FyersSession,
)
class LoginRecord:
"""The Fyers 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.
expired_logins (set): The login numbers whose token has already expired.
refused_profiles (set): The login numbers whose session the profile endpoint refuses.
"""
access_token = None
login_count = 0
expired_logins = set()
refused_profiles = set()
class StandInFyersAPI:
"""A stand-in for `FyersAPI` whose construction is a pretend login that issues a JWT access token."""
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
expiry = 4102444800
if LoginRecord.login_count in LoginRecord.expired_logins:
expiry = 1
claims = {
'exp': expiry,
'hsm_key': f'hsm-key-{LoginRecord.login_count}',
}
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
LoginRecord.access_token = f'header.{payload}.signature'
print(f'Stand-in login number {LoginRecord.login_count}')
self._settings = {
'app_id': 'XY1234-100',
}
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 get(self, url, timeout):
"""Answers the profile 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 Fyers' answer under `data`.
"""
if LoginRecord.login_count in LoginRecord.refused_profiles:
print(f'Profile refused for login {LoginRecord.login_count}')
return {
'data': {
's': 'error',
'code': -16,
'message': 'Could not authenticate the user',
},
}
print(f'Profile served for login {LoginRecord.login_count}')
return {
'data': {
's': 'ok',
'data': {
'fy_id': 'XY01234',
},
},
}
class ProfileRefusesTheNewLoginExample:
"""Logs in again unconditionally and catches the profile's refusal.
Attributes:
session (FyersSession): The session being shown.
"""
def __init__(self):
"""Registers the stand-in API module, scripts the second profile 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.fyers')
api_module.FyersAPI = StandInFyersAPI
sys.modules['stock_brokers.api.fyers'] = api_module
LoginRecord.refused_profiles = {
2,
}
self.session = FyersSession(logging.getLogger('fyers.session'))
def run(self):
"""Logs in again without checking, prints the refusal, and decodes a malformed token.
Returns:
None: This method returns nothing.
"""
try:
self.session.log_in_again_without_checking()
except FyersRefusal as refusal:
print(f'FyersRefusal: code={refusal.code} message={refusal.message}')
print(f'Claims of a malformed token: {self.session.token_claims("not-a-jwt")}')
print(f'Logins made: {LoginRecord.login_count}')
if __name__ == '__main__':
ProfileRefusesTheNewLoginExample().run()
It prints:
Stand-in login number 1
Profile served for login 1
WARNING Logging in to Fyers again.
Stand-in login number 2
Profile refused for login 2
FyersRefusal: code=-16 message=the profile was refused: {'s': 'error', 'code': -16, 'message': 'Could not authenticate the user'}
Claims of a malformed token: {}
Logins made: 2
Methods:
| Name | Description |
|---|---|
credentials |
The app id and the access token in force now, which may be one another process has just obtained. |
token_claims |
The claims inside a Fyers access token, a JWT. |
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/fyers.py
The app id and the access token in force now, which may be one another process has just obtained.
Returns:
| Type | Description |
|---|---|
tuple
|
The app id and the access token, either of which may be None. |
Example usage
Source code in stock_brokers/websockets/fyers.py
The claims inside a Fyers access token, a JWT.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
access_token
|
str | None
|
The token. |
required |
Returns:
| Type | Description |
|---|---|
dict
|
The claims, or an empty dictionary when the token cannot be decoded. |
Example usage
From example_2_profile_refuses_the_new_login.py
print(f'Claims of a malformed token: {self.session.token_claims("not-a-jwt")}')
Source code in stock_brokers/websockets/fyers.py
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 |
|---|---|
FyersRefusal
|
When the new session's profile is refused. |
Exception
|
Whatever |
Example usage
Source code in stock_brokers/websockets/fyers.py
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 |
|---|---|
FyersRefusal
|
When the new session's profile is refused. |
Exception
|
Whatever |
Example usage
Source code in stock_brokers/websockets/fyers.py
Logs in by constructing FyersAPI again and confirms it; the caller holds the lock.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Raises:
| Type | Description |
|---|---|
FyersRefusal
|
When the new session's profile is refused. |
Exception
|
Whatever |
Source code in stock_brokers/websockets/fyers.py
Checks that the profile is served, since a constructor that returned is no proof of a session.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
fyers
|
FyersAPI
|
The API object to check. |
required |
Returns:
| Type | Description |
|---|---|
FyersAPI
|
The same object. |
Raises:
| Type | Description |
|---|---|
FyersRefusal
|
When the profile is refused. |
Source code in stock_brokers/websockets/fyers.py
Builds and reads the HSM binary frames of Fyers' market feed.
Example programs
examples/stock_brokers/websockets/fyers/FyersHsmFrames/example_1_building_frames.py
"""Builds the binary frames a Fyers quotes socket sends and the HSM topic names it subscribes.
`FyersHsmFrames` holds the byte layouts of Fyers' HSM market feed protocol, the one Fyers' own SDK speaks. The quotes socket uses it to build the authentication frame carrying the hsm key, the frame that selects full mode, the subscription frame listing topic names, and the frame acknowledging data messages. `topic_name` turns a symbol and its fytoken into a topic name: the fytoken's first four digits name the exchange segment and the digits from the eleventh on the exchange token, and an index uses the index name Fyers' SDK carries.
This program builds each frame and prints it as hex, and asks for the topic names of an equity, an index, and a symbol whose segment the feed does not carry. It needs no stand-ins, because the class only packs and unpacks bytes.
Notice that `NSE:SBIN-EQ` becomes `sf|nse_cm|3045`, that `NSE:NIFTY50-INDEX` becomes `if|nse_cm|Nifty 50`, and that an unknown segment gives None.
Run it from the project root:
python examples/stock_brokers/websockets/fyers/FyersHsmFrames/example_1_building_frames.py
"""
from stock_brokers.websockets.fyers import (
FyersHsmFrames,
)
class BuildingFramesExample:
"""Builds HSM frames and topic names and prints them.
Attributes:
frames (FyersHsmFrames): The frame builder being shown.
"""
def __init__(self):
"""Builds the frame builder.
Returns:
None: This method returns nothing.
"""
self.frames = FyersHsmFrames()
def run(self):
"""Prints each frame as hex and each topic name.
Returns:
None: This method returns nothing.
"""
print(f"Authentication frame: {self.frames.auth_frame('hsm-key-1').hex()}")
print(f'Full mode frame: {self.frames.full_mode_frame().hex()}')
topics = [
self.frames.topic_name('NSE:SBIN-EQ', '10100000003045'),
self.frames.topic_name('NSE:NIFTY50-INDEX', '101000000026000'),
self.frames.topic_name('NSE:SOMETHING-EQ', '99990000001234'),
]
print(f'Topic names: {topics}')
print(f'Subscription frame: {self.frames.subscribe_frame(topics[:2]).hex()}')
print(f'Acknowledgement of message 42: {self.frames.acknowledge_frame(42).hex()}')
if __name__ == '__main__':
BuildingFramesExample().run()
It prints:
Authentication frame: 0028010401000968736d2d6b65792d31020001500300010104000f507974686f6e53444b2d332e302e39
Full mode frame: 00000c02010008000000000000080002000146
Topic names: ['sf|nse_cm|3045', 'if|nse_cm|Nifty 50', None]
Subscription frame: 0036040201002400020e73667c6e73655f636d7c333034351269667c6e73655f636d7c4e696674792035300200010b
Acknowledgement of message 42: 000903010100040000002a
examples/stock_brokers/websockets/fyers/FyersHsmFrames/example_2_reading_responses.py
"""Reads Fyers HSM authentication responses and picks the price scale and field table a data packet is read with.
When the feed answers the authentication frame, `auth_response` reads whether it accepted the hsm key (a `K` in the first field) and how many data messages the socket must acknowledge at a time. Data packets carry scaled integer prices: `price_divisor` gives 10 to the precision times the multiplier from the instrument's snapshot, or 100 before any snapshot has said. A packet's values are positional, and `fields_for` picks the table they are read against from the topic name and the number of values: depth topics start `dp`, index topics start `if`, three values or fewer are a lite packet, and anything else is a full data packet.
This program builds an accepted and a refused authentication response byte by byte, reads a truncated one, and asks for divisors and field tables. It needs no stand-ins, because the class only packs and unpacks bytes.
Notice that a truncated response reads as refused with a count of zero rather than raising.
Run it from the project root:
python examples/stock_brokers/websockets/fyers/FyersHsmFrames/example_2_reading_responses.py
"""
import struct
from stock_brokers.websockets.fyers import (
FyersHsmFrames,
)
class ReadingResponsesExample:
"""Reads authentication responses and picks price scales and field tables.
Attributes:
frames (FyersHsmFrames): The frame reader being shown.
"""
def __init__(self):
"""Builds the frame reader.
Returns:
None: This method returns nothing.
"""
self.frames = FyersHsmFrames()
def auth_response(self, status, ack_count):
"""An authentication response as the feed sends it.
Args:
status (bytes): `K` for accepted, anything else for refused.
ack_count (int): How many data messages to acknowledge at a time.
Returns:
bytes: The frame.
"""
return struct.pack('!HBBBH', 0, 1, 2, 1, len(status)) + status + bytes([2]) + struct.pack('!H', 4) + struct.pack('>I', ack_count)
def run(self):
"""Prints what each response, state and topic reads as.
Returns:
None: This method returns nothing.
"""
print(f"Accepted response: {self.frames.auth_response(self.auth_response(b'K', 50))}")
print(f"Refused response: {self.frames.auth_response(self.auth_response(b'N', 0))}")
print(f"Truncated response: {self.frames.auth_response(bytes([0, 0, 1, 2, 1, 0]))}")
snapshot_state = {
'precision': 4,
'multiplier': 1,
}
print(f'Divisor before any snapshot: {self.frames.price_divisor({})}')
print(f'Divisor for a currency snapshot: {self.frames.price_divisor(snapshot_state)}')
topics = [
(
'dp|nse_cm|3045',
32,
),
(
'if|nse_cm|Nifty 50',
8,
),
(
'sf|nse_cm|3045',
3,
),
(
'sf|nse_cm|3045',
23,
),
]
for topic, field_count in topics:
fields = self.frames.fields_for(topic, field_count)
print(f'{topic} with {field_count} values: {len(fields)} fields starting {fields[:3]}')
if __name__ == '__main__':
ReadingResponsesExample().run()
It prints:
Accepted response: (True, 50)
Refused response: (False, 0)
Truncated response: (False, 0)
Divisor before any snapshot: 100
Divisor for a currency snapshot: 10000
dp|nse_cm|3045 with 32 values: 32 fields starting ['bid_price1', 'bid_price2', 'bid_price3']
if|nse_cm|Nifty 50 with 8 values: 8 fields starting ['ltp', 'prev_close_price', 'exch_feed_time']
sf|nse_cm|3045 with 3 values: 3 fields starting ['ltp', 'symbol', 'type']
sf|nse_cm|3045 with 23 values: 23 fields starting ['ltp', 'vol_traded_today', 'last_traded_time']
Methods:
| Name | Description |
|---|---|
auth_frame |
The frame that authenticates the feed with the hsm key. |
auth_response |
Reads an authentication response: a |
full_mode_frame |
The frame that puts the channel in full mode, so data packets carry every field. |
subscribe_frame |
The frame that subscribes HSM topic names on the channel. |
acknowledge_frame |
The frame that acknowledges data messages up to a message number. |
topic_name |
The HSM topic name of a Fyers symbol. |
price_divisor |
What an instrument's scaled integer prices are divided by. |
fields_for |
The field table a packet's positional values are read against. |
The frame that authenticates the feed with the hsm key.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
hsm_key
|
str
|
The |
required |
Returns:
| Type | Description |
|---|---|
bytes
|
The frame. |
Example usage
From example_1_building_frames.py
print(f"Authentication frame: {self.frames.auth_frame('hsm-key-1').hex()}")
Source code in stock_brokers/websockets/fyers.py
Reads an authentication response: a K in its first field on success, then the acknowledgement count.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
bytes
|
The response. |
required |
Returns:
| Type | Description |
|---|---|
tuple
|
Whether the feed was accepted, and how many data messages to acknowledge at a time. |
Example usage
From example_2_reading_responses.py
print(f"Accepted response: {self.frames.auth_response(self.auth_response(b'K', 50))}")
From example_2_reading_responses.py
print(f"Refused response: {self.frames.auth_response(self.auth_response(b'N', 0))}")
From example_2_reading_responses.py
print(f"Truncated response: {self.frames.auth_response(bytes([0, 0, 1, 2, 1, 0]))}")
Source code in stock_brokers/websockets/fyers.py
The frame that puts the channel in full mode, so data packets carry every field.
Returns:
| Type | Description |
|---|---|
bytes
|
The frame. |
Example usage
Source code in stock_brokers/websockets/fyers.py
The frame that subscribes HSM topic names on the channel.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
topics
|
list[str]
|
The topic names, such as |
required |
Returns:
| Type | Description |
|---|---|
bytes
|
The frame. |
Example usage
From example_1_building_frames.py
print(f'Subscription frame: {self.frames.subscribe_frame(topics[:2]).hex()}')
Source code in stock_brokers/websockets/fyers.py
The frame that acknowledges data messages up to a message number.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message_number
|
int
|
The number in a data message's header. |
required |
Returns:
| Type | Description |
|---|---|
bytes
|
The frame. |
Example usage
From example_1_building_frames.py
print(f'Acknowledgement of message 42: {self.frames.acknowledge_frame(42).hex()}')
Source code in stock_brokers/websockets/fyers.py
The HSM topic name of a Fyers symbol.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
symbol
|
str
|
The Fyers symbol, such as |
required |
fytoken
|
str
|
The symbol's fytoken from the symbol-token endpoint. |
required |
Returns:
| Type | Description |
|---|---|
str | None
|
The topic name, or None when its segment is not one the feed carries. |
Example usage
From example_1_building_frames.py
topics = [
self.frames.topic_name('NSE:SBIN-EQ', '10100000003045'),
self.frames.topic_name('NSE:NIFTY50-INDEX', '101000000026000'),
self.frames.topic_name('NSE:SOMETHING-EQ', '99990000001234'),
]
Source code in stock_brokers/websockets/fyers.py
What an instrument's scaled integer prices are divided by.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
state
|
dict
|
The instrument's merged field state. |
required |
Returns:
| Type | Description |
|---|---|
int
|
10 to its precision times its multiplier, from its snapshot, or 100 when no snapshot has said. |
Example usage
From example_2_reading_responses.py
print(f'Divisor before any snapshot: {self.frames.price_divisor({})}')
From example_2_reading_responses.py
print(f'Divisor for a currency snapshot: {self.frames.price_divisor(snapshot_state)}')
Source code in stock_brokers/websockets/fyers.py
The field table a packet's positional values are read against.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
topic_name
|
str
|
The topic name its snapshot gave, which encodes the feed kind. |
required |
field_count
|
int
|
The number of values in the packet. |
required |
Returns:
| Type | Description |
|---|---|
list[str]
|
The field names. |
Example usage
Source code in stock_brokers/websockets/fyers.py
Bases: BrokerWebsocket
One Fyers HSM market feed websocket for a batch of symbols.
Sets up a socket for one batch of symbols.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The connection's name, for the log. |
required |
symbols
|
list[str]
|
The batch of Fyers symbols, which are also the ticks' |
required |
session
|
FyersSession
|
The shared login. |
required |
on_ticks
|
callable
|
Called with each data message's list of ticks that carry a price or a book, on this socket's thread. |
required |
logger
|
Logger
|
Where the connection reports. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example programs
examples/stock_brokers/websockets/fyers/FyersQuotesSocket/example_1_snapshots_and_updates.py
"""Runs a Fyers HSM quotes socket through symbol lookup, authentication, snapshots and an update.
`FyersQuotesSocket` first resolves its symbols to HSM topic names through Fyers' symbol-token endpoint, then opens `wss://socket.fyers.in/hsm/v1-5/prod` and authenticates with the `hsm_key` claim inside the access token. Once the feed accepts, the socket selects full mode and subscribes the topics. Data arrives incrementally: a snapshot packet (type 83) names a topic and carries every field, and update packets (type 85) carry only positional values for a topic id, with -2147483648 meaning no value. Prices are scaled integers divided by 10 to the precision times the multiplier. The socket also acknowledges every `ack_count` data messages, as the authentication response asks.
This program resolves `NSE:SBIN-EQ` and `NSE:NIFTY50-INDEX` (a third symbol does not resolve), accepts the feed with an acknowledgement count of 2, sends one data message with a snapshot of each topic, and then an update that changes only SBIN's last price and volume. No real socket or HTTP request is made: a stand-in `websocket` package plays the frames and prints what the socket sends as hex, a stand-in `requests` package answers the symbol lookup, and a stand-in `stock_brokers.api.fyers` module issues JWT tokens and serves the profile so the real `FyersSession` runs unchanged.
Notice the warning about the unresolved symbol, the three binary frames sent after authentication, the acknowledgement frame sent after the second data message, and that the update's tick keeps the open, high, low and close from the snapshot.
Run it from the project root:
python examples/stock_brokers/websockets/fyers/FyersQuotesSocket/example_1_snapshots_and_updates.py
"""
import base64
import json
import logging
import struct
import sys
import types
from stock_brokers.websockets.fyers import (
FyersQuotesSocket,
FyersSession,
)
class LoginRecord:
"""The Fyers 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.
expired_logins (set): The login numbers whose token has already expired.
refused_profiles (set): The login numbers whose session the profile endpoint refuses.
"""
access_token = None
login_count = 0
expired_logins = set()
refused_profiles = set()
class StandInFyersAPI:
"""A stand-in for `FyersAPI` whose construction is a pretend login that issues a JWT access token."""
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
expiry = 4102444800
if LoginRecord.login_count in LoginRecord.expired_logins:
expiry = 1
claims = {
'exp': expiry,
'hsm_key': f'hsm-key-{LoginRecord.login_count}',
}
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
LoginRecord.access_token = f'header.{payload}.signature'
print(f'Stand-in login number {LoginRecord.login_count}')
self._settings = {
'app_id': 'XY1234-100',
}
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 get(self, url, timeout):
"""Answers the profile 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 Fyers' answer under `data`.
"""
if LoginRecord.login_count in LoginRecord.refused_profiles:
print(f'Profile refused for login {LoginRecord.login_count}')
return {
'data': {
's': 'error',
'code': -16,
'message': 'Could not authenticate the user',
},
}
print(f'Profile served for login {LoginRecord.login_count}')
return {
'data': {
's': 'ok',
'data': {
'fy_id': 'XY01234',
},
},
}
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.
"""
if isinstance(data, bytes):
print(f'Sent binary frame: {data.hex()}')
else:
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.
"""
if 'header' in handlers:
print(f"Handshake header names: {list(handlers['header'])}")
plan = self.plans.pop(0)
if not self.plans:
self.when_finished()
return ScriptedConnection(url, plan, handlers)
class SymbolLookupResponse:
"""A stand-in for the `requests` response of Fyers' symbol-token endpoint.
Attributes:
status_code (int): The HTTP status.
body (dict): The JSON body.
text (str): The body as text.
"""
def __init__(self, status_code, body):
"""Keeps the answer.
Args:
status_code (int): The HTTP status.
body (dict): The JSON body.
Returns:
None: This method returns nothing.
"""
self.status_code = status_code
self.body = body
self.text = json.dumps(body)
def json(self):
"""The JSON body.
Returns:
dict: The body.
"""
return self.body
class StandInRequests:
"""A stand-in for the `requests` package that answers the symbol lookup from a fixed table.
Attributes:
fytokens (dict): Each Fyers symbol to its fytoken.
"""
def __init__(self, fytokens):
"""Keeps the table.
Args:
fytokens (dict): Each Fyers symbol to its fytoken.
Returns:
None: This method returns nothing.
"""
self.fytokens = fytokens
def post(self, url, headers, json, timeout):
"""Answers one symbol lookup.
Args:
url (str): The endpoint.
headers (dict): The request headers.
json (dict): The request body with its `symbols`.
timeout (int): The request timeout in seconds.
Returns:
SymbolLookupResponse: The answer.
"""
print(f"Symbol lookup for {json['symbols']}")
valid = {}
invalid = []
for symbol in json['symbols']:
if symbol in self.fytokens:
valid[symbol] = self.fytokens[symbol]
else:
invalid.append(symbol)
body = {
's': 'ok',
'validSymbol': valid,
'invalidSymbol': invalid,
}
return SymbolLookupResponse(200, body)
class HsmResponses:
"""Builds the binary frames Fyers' HSM feed sends, byte by byte."""
def auth_response(self, accepted, ack_count):
"""An authentication response.
Args:
accepted (bool): Whether the feed accepted the hsm key.
ack_count (int): How many data messages to acknowledge at a time.
Returns:
bytes: The frame.
"""
status = b'K'
if not accepted:
status = b'N'
return struct.pack('!HBBBH', 0, 1, 2, 1, 1) + status + bytes([2]) + struct.pack('!H', 4) + struct.pack('>I', ack_count)
def values(self, values):
"""Positional field values, with None as Fyers' no-value marker.
Args:
values (list): The values.
Returns:
bytes: The packed values.
"""
packed = b''
for value in values:
if value is None:
value = -2147483648
packed = packed + struct.pack('>i', value)
return packed
def snapshot(self, topic_id, topic_name, values):
"""A snapshot packet naming a topic and carrying every field, then a precision of 2 and three strings.
Args:
topic_id (int): The topic id later updates refer to.
topic_name (str): The topic name.
values (list): The positional values.
Returns:
bytes: The packet.
"""
name = topic_name.encode('ascii')
packet = bytes([83]) + struct.pack('H', topic_id) + bytes([len(name)]) + name
packet = packet + bytes([len(values)]) + self.values(values)
packet = packet + b'\x00\x00' + struct.pack('>H', 1) + bytes([2])
return packet + bytes([3]) + b'NSE' + bytes([1]) + b'-' + bytes([1]) + b'-'
def update(self, topic_id, values):
"""An update packet carrying positional values for a topic id.
Args:
topic_id (int): The topic id.
values (list): The positional values.
Returns:
bytes: The packet.
"""
return bytes([85]) + struct.pack('H', topic_id) + bytes([len(values)]) + self.values(values)
def datafeed(self, message_number, packets):
"""A data feed response holding packets.
Args:
message_number (int): The message number an acknowledgement refers to.
packets (list): The packets.
Returns:
bytes: The frame.
"""
frame = struct.pack('!H', 0) + bytes([6]) + struct.pack('>I', message_number) + struct.pack('!H', len(packets))
for packet in packets:
frame = frame + packet
return frame
class SnapshotsAndUpdatesExample:
"""Runs a Fyers quotes socket against a scripted lookup and feed.
Attributes:
socket (FyersQuotesSocket): 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.fyers')
api_module.FyersAPI = StandInFyersAPI
sys.modules['stock_brokers.api.fyers'] = api_module
fytokens = {
'NSE:SBIN-EQ': '10100000003045',
'NSE:NIFTY50-INDEX': '101000000026000',
}
sys.modules['requests'] = StandInRequests(fytokens)
responses = HsmResponses()
stock_values = [
82050,
1204500,
1790311528,
1790311529,
500,
700,
82045,
82055,
25,
54000,
61000,
81990,
0,
81500,
82500,
91200,
60010,
73500,
89800,
81700,
81000,
None,
None,
]
index_values = [
2515065,
2490000,
1790311529,
2521000,
2498000,
2500000,
None,
None,
]
update_values = [
82100,
1205000,
]
for _ in range(21):
update_values.append(None)
first_message = responses.datafeed(1, [
responses.snapshot(1, 'sf|nse_cm|3045', stock_values),
responses.snapshot(2, 'if|nse_cm|Nifty 50', index_values),
])
second_message = responses.datafeed(2, [
responses.update(1, update_values),
])
plans = [
{
'frames': [
responses.auth_response(True, 2),
first_message,
second_message,
],
},
]
websocket_module = ScriptedWebsocketModule(plans)
sys.modules['websocket'] = websocket_module
logger = logging.getLogger('fyers.quotes')
symbols = [
'NSE:SBIN-EQ',
'NSE:NIFTY50-INDEX',
'NSE:NOSUCHSTOCK-EQ',
]
self.socket = FyersQuotesSocket('socket_0', symbols, FyersSession(logger), self.print_ticks, logger)
websocket_module.when_finished = self.socket.close
def print_ticks(self, ticks):
"""Prints the main fields of each tick the socket built.
Args:
ticks (list): The normalized ticks from one data message.
Returns:
None: This method returns nothing.
"""
for tick in ticks:
change = tick['change']
if change is not None:
change = round(change, 4)
print(f"Tick {tick['id']} in {tick['mode']} mode")
print(f" last_price={tick['last_price']} volume={tick['volume']} change={change}")
print(f" ohlc={tick['ohlc']} 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 own 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__':
SnapshotsAndUpdatesExample().run()
It prints:
Stand-in login number 1
Profile served for login 1
Symbol lookup for ['NSE:SBIN-EQ', 'NSE:NIFTY50-INDEX', 'NSE:NOSUCHSTOCK-EQ']
WARNING socket_0: 1 symbol(s) did not resolve and are not subscribed: NSE:NOSUCHSTOCK-EQ
Connecting to wss://socket.fyers.in/hsm/v1-5/prod
Sent binary frame: 0028010401000968736d2d6b65792d31020001500300010104000f507974686f6e53444b2d332e302e39
INFO socket_0 opened. Authenticating.
Sent binary frame: 00000c02010008000000000000080002000146
Sent binary frame: 0036040201002400020e73667c6e73655f636d7c333034351269667c6e73655f636d7c4e696674792035300200010b
INFO socket_0 authenticated and subscribed to 2 instrument(s).
Tick NSE:SBIN-EQ in full mode
last_price=820.5 volume=1204500 change=1.2963
ohlc={'open': 817.0, 'high': 825.0, 'low': 815.0, 'close': 810.0} exchange_timestamp=1790311529
best bid={'quantity': 500, 'price': 820.45, 'orders': None} best offer={'quantity': 700, 'price': 820.55, 'orders': None}
Tick NSE:NIFTY50-INDEX in quote mode
last_price=25150.65 volume=None change=1.0066
ohlc={'open': 25000.0, 'high': 25210.0, 'low': 24980.0, 'close': 24900.0} exchange_timestamp=1790311529
Sent binary frame: 0009030101000400000002
Tick NSE:SBIN-EQ in full mode
last_price=821.0 volume=1205000 change=1.358
ohlc={'open': 817.0, 'high': 825.0, 'low': 815.0, 'close': 810.0} exchange_timestamp=1790311529
best bid={'quantity': 500, 'price': 820.45, 'orders': None} best offer={'quantity': 700, 'price': 820.55, 'orders': None}
INFO socket_0 closed with status 1000: normal closure
INFO socket_0 stopped.
Gave up: False
examples/stock_brokers/websockets/fyers/FyersQuotesSocket/example_2_expired_token_then_refused_feed.py
"""Shows a Fyers quotes socket logging in once for an expired token and giving up when the feed still refuses it.
`FyersQuotesSocket` has a reconnect loop of its own, `run_forever`, because a Cloudflare ban or rate limit on the symbol lookup must pause the socket for minutes rather than log in again. Otherwise it follows the shared loop: a refusal logs in again once, and a refusal straight after that login makes the socket give up so its script can exit and let systemd restart it. The access token is a JWT, and a token whose `exp` has passed is treated as refused without sending anything, because Fyers' edge bans addresses that keep sending refused requests.
This program's first login issues an already expired token, so the first connect sends nothing and logs in again. The second connect resolves the symbol and opens, but the feed refuses the hsm key with an `N` in its authentication response, so the socket gives up. No real socket or HTTP request is made: stand-in `websocket` and `requests` packages play the connection and the lookup, and a stand-in `stock_brokers.api.fyers` module issues the tokens and serves the profile. The backoff is set to zero seconds; the ban and rate limit pauses are not shown because they wait for minutes.
Notice that the expired token never reaches the network, that each login is confirmed against the profile, and that `gave_up` ends True.
Run it from the project root:
python examples/stock_brokers/websockets/fyers/FyersQuotesSocket/example_2_expired_token_then_refused_feed.py
"""
import base64
import json
import logging
import struct
import sys
import types
from stock_brokers.websockets.fyers import (
FyersQuotesSocket,
FyersSession,
)
class LoginRecord:
"""The Fyers 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.
expired_logins (set): The login numbers whose token has already expired.
refused_profiles (set): The login numbers whose session the profile endpoint refuses.
"""
access_token = None
login_count = 0
expired_logins = set()
refused_profiles = set()
class StandInFyersAPI:
"""A stand-in for `FyersAPI` whose construction is a pretend login that issues a JWT access token."""
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
expiry = 4102444800
if LoginRecord.login_count in LoginRecord.expired_logins:
expiry = 1
claims = {
'exp': expiry,
'hsm_key': f'hsm-key-{LoginRecord.login_count}',
}
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
LoginRecord.access_token = f'header.{payload}.signature'
print(f'Stand-in login number {LoginRecord.login_count}')
self._settings = {
'app_id': 'XY1234-100',
}
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 get(self, url, timeout):
"""Answers the profile 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 Fyers' answer under `data`.
"""
if LoginRecord.login_count in LoginRecord.refused_profiles:
print(f'Profile refused for login {LoginRecord.login_count}')
return {
'data': {
's': 'error',
'code': -16,
'message': 'Could not authenticate the user',
},
}
print(f'Profile served for login {LoginRecord.login_count}')
return {
'data': {
's': 'ok',
'data': {
'fy_id': 'XY01234',
},
},
}
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.
"""
if isinstance(data, bytes):
print(f'Sent binary frame: {data.hex()}')
else:
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.
"""
if 'header' in handlers:
print(f"Handshake header names: {list(handlers['header'])}")
plan = self.plans.pop(0)
if not self.plans:
self.when_finished()
return ScriptedConnection(url, plan, handlers)
class SymbolLookupResponse:
"""A stand-in for the `requests` response of Fyers' symbol-token endpoint.
Attributes:
status_code (int): The HTTP status.
body (dict): The JSON body.
text (str): The body as text.
"""
def __init__(self, status_code, body):
"""Keeps the answer.
Args:
status_code (int): The HTTP status.
body (dict): The JSON body.
Returns:
None: This method returns nothing.
"""
self.status_code = status_code
self.body = body
self.text = json.dumps(body)
def json(self):
"""The JSON body.
Returns:
dict: The body.
"""
return self.body
class StandInRequests:
"""A stand-in for the `requests` package that answers the symbol lookup from a fixed table.
Attributes:
fytokens (dict): Each Fyers symbol to its fytoken.
"""
def __init__(self, fytokens):
"""Keeps the table.
Args:
fytokens (dict): Each Fyers symbol to its fytoken.
Returns:
None: This method returns nothing.
"""
self.fytokens = fytokens
def post(self, url, headers, json, timeout):
"""Answers one symbol lookup.
Args:
url (str): The endpoint.
headers (dict): The request headers.
json (dict): The request body with its `symbols`.
timeout (int): The request timeout in seconds.
Returns:
SymbolLookupResponse: The answer.
"""
print(f"Symbol lookup for {json['symbols']}")
valid = {}
invalid = []
for symbol in json['symbols']:
if symbol in self.fytokens:
valid[symbol] = self.fytokens[symbol]
else:
invalid.append(symbol)
body = {
's': 'ok',
'validSymbol': valid,
'invalidSymbol': invalid,
}
return SymbolLookupResponse(200, body)
class HsmResponses:
"""Builds the binary frames Fyers' HSM feed sends, byte by byte."""
def auth_response(self, accepted, ack_count):
"""An authentication response.
Args:
accepted (bool): Whether the feed accepted the hsm key.
ack_count (int): How many data messages to acknowledge at a time.
Returns:
bytes: The frame.
"""
status = b'K'
if not accepted:
status = b'N'
return struct.pack('!HBBBH', 0, 1, 2, 1, 1) + status + bytes([2]) + struct.pack('!H', 4) + struct.pack('>I', ack_count)
def values(self, values):
"""Positional field values, with None as Fyers' no-value marker.
Args:
values (list): The values.
Returns:
bytes: The packed values.
"""
packed = b''
for value in values:
if value is None:
value = -2147483648
packed = packed + struct.pack('>i', value)
return packed
def snapshot(self, topic_id, topic_name, values):
"""A snapshot packet naming a topic and carrying every field, then a precision of 2 and three strings.
Args:
topic_id (int): The topic id later updates refer to.
topic_name (str): The topic name.
values (list): The positional values.
Returns:
bytes: The packet.
"""
name = topic_name.encode('ascii')
packet = bytes([83]) + struct.pack('H', topic_id) + bytes([len(name)]) + name
packet = packet + bytes([len(values)]) + self.values(values)
packet = packet + b'\x00\x00' + struct.pack('>H', 1) + bytes([2])
return packet + bytes([3]) + b'NSE' + bytes([1]) + b'-' + bytes([1]) + b'-'
def update(self, topic_id, values):
"""An update packet carrying positional values for a topic id.
Args:
topic_id (int): The topic id.
values (list): The positional values.
Returns:
bytes: The packet.
"""
return bytes([85]) + struct.pack('H', topic_id) + bytes([len(values)]) + self.values(values)
def datafeed(self, message_number, packets):
"""A data feed response holding packets.
Args:
message_number (int): The message number an acknowledgement refers to.
packets (list): The packets.
Returns:
bytes: The frame.
"""
frame = struct.pack('!H', 0) + bytes([6]) + struct.pack('>I', message_number) + struct.pack('!H', len(packets))
for packet in packets:
frame = frame + packet
return frame
class ExpiredTokenThenRefusedFeedExample:
"""Runs a Fyers quotes socket through an expired token and a refused feed.
Attributes:
socket (FyersQuotesSocket): 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.fyers')
api_module.FyersAPI = StandInFyersAPI
sys.modules['stock_brokers.api.fyers'] = api_module
LoginRecord.expired_logins = {
1,
}
fytokens = {
'NSE:SBIN-EQ': '10100000003045',
}
sys.modules['requests'] = StandInRequests(fytokens)
responses = HsmResponses()
plans = [
{
'frames': [
responses.auth_response(False, 0),
],
},
]
websocket_module = ScriptedWebsocketModule(plans)
websocket_module.when_finished = self.do_nothing
sys.modules['websocket'] = websocket_module
logger = logging.getLogger('fyers.quotes')
symbols = [
'NSE:SBIN-EQ',
]
self.socket = FyersQuotesSocket('socket_0', symbols, FyersSession(logger), self.print_ticks, logger)
self.socket.MIN_BACKOFF_SECONDS = 0
def do_nothing(self):
"""Stands in for closing the socket, which this program leaves to the socket giving up.
Returns:
None: This method returns nothing.
"""
def print_ticks(self, ticks):
"""Prints the main fields of each tick the socket built.
Args:
ticks (list): The normalized ticks from one data message.
Returns:
None: This method returns nothing.
"""
for tick in ticks:
change = tick['change']
if change is not None:
change = round(change, 4)
print(f"Tick {tick['id']} in {tick['mode']} mode")
print(f" last_price={tick['last_price']} volume={tick['volume']} change={change}")
print(f" ohlc={tick['ohlc']} 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 own reconnect loop until it gives up.
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__':
ExpiredTokenThenRefusedFeedExample().run()
It prints:
Stand-in login number 1
Profile served for login 1
WARNING socket_0: the Fyers access token has expired or carries no hsm_key.
WARNING Logging in to Fyers again.
Stand-in login number 2
Profile served for login 2
WARNING socket_0 disconnected. Reconnecting in 0 seconds.
Symbol lookup for ['NSE:SBIN-EQ']
Connecting to wss://socket.fyers.in/hsm/v1-5/prod
Sent binary frame: 0028010401000968736d2d6b65792d32020001500300010104000f507974686f6e53444b2d332e302e39
INFO socket_0 opened. Authenticating.
WARNING socket_0: Fyers refused the feed authentication.
INFO socket_0 closed with status 1000: normal closure
ERROR socket_0 still cannot connect after logging in again. Giving up.
INFO socket_0 stopped.
Gave up: True
Logins made: 2
Methods:
| Name | Description |
|---|---|
run_forever |
Connects, and reconnects with backoff, until closed or unable to recover, pausing instead of logging in after a ban or a rate limit. |
close |
Closes the connection and stops reconnecting. |
Source code in stock_brokers/websockets/fyers.py
Connects, and reconnects with backoff, until closed or unable to recover, pausing instead of logging in after a ban or a rate limit.
This is BrokerWebsocket.run_forever with one step added: a connect that ended in a Cloudflare ban or a rate limit waits the pause out and connects again, without counting a failure or logging in, since a login would only extend a ban.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example usage
Source code in stock_brokers/websockets/fyers.py
Checks the token, resolves the symbols to topics, then opens the websocket and blocks until it closes.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Raises:
| Type | Description |
|---|---|
FyersRefusal
|
When no symbol resolves to a topic, or the symbol lookup fails for a reason other than the session, a ban or a limit. |
Source code in stock_brokers/websockets/fyers.py
The Fyers symbol of each HSM topic name, from the symbol-token endpoint, five hundred symbols a request.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
app_id
|
str
|
The app id the request is authorized with. |
required |
token
|
str
|
The access token the request is authorized with. |
required |
Returns:
| Type | Description |
|---|---|
dict[str, str]
|
Each topic name to its symbol; a symbol listed as invalid, or whose segment the feed does not carry, is left out. |
Raises:
| Type | Description |
|---|---|
FyersRefusal
|
When the endpoint answers with an error or something that is not JSON. |
Source code in stock_brokers/websockets/fyers.py
The numeric code a refusal carries.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
exception
|
Exception
|
The refusal. |
required |
Returns:
| Type | Description |
|---|---|
int | None
|
The code, or None. |
Source code in stock_brokers/websockets/fyers.py
Whether a refusal is Cloudflare's address ban rather than anything Fyers itself said.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
exception
|
Exception
|
The refusal. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True for a ban. |
Source code in stock_brokers/websockets/fyers.py
Whether a refusal is a rate limit.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
exception
|
Exception
|
The refusal. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True for a rate limit. |
Source code in stock_brokers/websockets/fyers.py
Whether Fyers refused the request because the session is dead, and not for a ban or a limit.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
exception
|
Exception
|
The refusal. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True for a dead session. |
Source code in stock_brokers/websockets/fyers.py
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. |
Authenticates with the hsm key; the subscription waits for the acknowledgement.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The connection that opened. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/fyers.py
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/fyers.py
Handles the acknowledgements, and hands on the ticks of each instrument a data message updated.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The connection the message arrived on. |
required |
message
|
bytes | str
|
The message. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/fyers.py
Merges a data feed response into the per-instrument state.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
bytes
|
The data feed response. |
required |
Returns:
| Type | Description |
|---|---|
dict[str, dict]
|
The state of each instrument the response touched, by symbol. |
Source code in stock_brokers/websockets/fyers.py
980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 | |
Builds a normalized tick from an instrument's merged field state.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
symbol
|
str
|
The Fyers symbol. |
required |
state
|
dict
|
The merged field state. |
required |
divisor
|
int
|
What turns Fyers' scaled integers into rupees. |
required |
Returns:
| Type | Description |
|---|---|
dict
|
The tick. |
Source code in stock_brokers/websockets/fyers.py
1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 | |
A scaled integer price from the state in rupees.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
state
|
dict
|
The merged field state. |
required |
key
|
str
|
The field. |
required |
divisor
|
int
|
The scale. |
required |
Returns:
| Type | Description |
|---|---|
float | None
|
The price, or None when the field is missing. |
Source code in stock_brokers/websockets/fyers.py
Closes the connection and stops reconnecting.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example usage
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
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
Bases: BrokerWebsocket
Fyers' order and position update websocket, authenticated in its handshake.
Sets up the order updates socket.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
session
|
FyersSession
|
The login. |
required |
on_updates
|
callable
|
Called on this socket's thread with a message's orders and positions, each a list of Fyers' objects exactly as sent, and the |
required |
logger
|
Logger
|
Where the connection reports. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example programs
examples/stock_brokers/websockets/fyers/FyersOrderUpdatesSocket/example_1_orders_and_positions.py
"""Receives orders and positions on Fyers' order update socket and prints what it hands on.
`FyersOrderUpdatesSocket` connects to `wss://socket.fyers.in/trade/v3` with an `authorization` header of the app id and the access token, so a bad credential fails the handshake itself. Once open it subscribes to orders and positions, Fyers confirms with code 1605, and every change then arrives as `{"s": "ok", "orders": {...}}` or `{"s": "ok", "positions": {...}}`. Fyers answers the keepalive with a bare `ping` or `pong`, which the socket ignores.
This program scripts one connection delivering the subscription confirmation, a `pong`, a pending order, a position and the same order filled. No real socket is opened: a stand-in `websocket` package plays the frames and prints the handshake header names and what the socket sends, and a stand-in `stock_brokers.api.fyers` module issues a JWT token and serves the profile so the real `FyersSession` runs unchanged.
Notice that the callback receives orders and positions separately, and that the confirmation is logged rather than handed on.
Run it from the project root:
python examples/stock_brokers/websockets/fyers/FyersOrderUpdatesSocket/example_1_orders_and_positions.py
"""
import base64
import json
import logging
import sys
import types
from stock_brokers.websockets.fyers import (
FyersOrderUpdatesSocket,
FyersSession,
)
class LoginRecord:
"""The Fyers 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.
expired_logins (set): The login numbers whose token has already expired.
refused_profiles (set): The login numbers whose session the profile endpoint refuses.
"""
access_token = None
login_count = 0
expired_logins = set()
refused_profiles = set()
class StandInFyersAPI:
"""A stand-in for `FyersAPI` whose construction is a pretend login that issues a JWT access token."""
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
expiry = 4102444800
if LoginRecord.login_count in LoginRecord.expired_logins:
expiry = 1
claims = {
'exp': expiry,
'hsm_key': f'hsm-key-{LoginRecord.login_count}',
}
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
LoginRecord.access_token = f'header.{payload}.signature'
print(f'Stand-in login number {LoginRecord.login_count}')
self._settings = {
'app_id': 'XY1234-100',
}
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 get(self, url, timeout):
"""Answers the profile 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 Fyers' answer under `data`.
"""
if LoginRecord.login_count in LoginRecord.refused_profiles:
print(f'Profile refused for login {LoginRecord.login_count}')
return {
'data': {
's': 'error',
'code': -16,
'message': 'Could not authenticate the user',
},
}
print(f'Profile served for login {LoginRecord.login_count}')
return {
'data': {
's': 'ok',
'data': {
'fy_id': 'XY01234',
},
},
}
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.
"""
if isinstance(data, bytes):
print(f'Sent binary frame: {data.hex()}')
else:
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.
"""
if 'header' in handlers:
print(f"Handshake header names: {list(handlers['header'])}")
plan = self.plans.pop(0)
if not self.plans:
self.when_finished()
return ScriptedConnection(url, plan, handlers)
class OrdersAndPositionsExample:
"""Runs the Fyers order update socket against one scripted connection.
Attributes:
socket (FyersOrderUpdatesSocket): The socket being shown.
"""
def __init__(self):
"""Builds the messages, the stand-ins, the session and the socket.
Returns:
None: This method returns nothing.
"""
logging.basicConfig(stream=sys.stdout, format='%(levelname)s %(message)s', level=logging.INFO)
api_module = types.ModuleType('stock_brokers.api.fyers')
api_module.FyersAPI = StandInFyersAPI
sys.modules['stock_brokers.api.fyers'] = api_module
subscribed = {
's': 'ok',
'code': 1605,
'message': 'Successfully subscribed',
}
position = {
's': 'ok',
'positions': {
'symbol': 'NSE:SBIN-EQ',
'netQty': 10,
'productType': 'INTRADAY',
},
}
plans = [
{
'frames': [
json.dumps(subscribed),
'pong',
self.order_message(6, 0),
json.dumps(position),
self.order_message(2, 10),
],
},
]
websocket_module = ScriptedWebsocketModule(plans)
sys.modules['websocket'] = websocket_module
logger = logging.getLogger('fyers.orders')
self.socket = FyersOrderUpdatesSocket(FyersSession(logger), self.print_updates, logger)
websocket_module.when_finished = self.socket.close
def order_message(self, status, filled_quantity):
"""One of Fyers' order messages for an SBIN buy.
Args:
status (int): Fyers' numeric order status, 6 for pending and 2 for filled.
filled_quantity (int): How much has filled.
Returns:
str: The message as JSON text.
"""
message = {
's': 'ok',
'orders': {
'id': '26092500012345',
'symbol': 'NSE:SBIN-EQ',
'status': status,
'side': 1,
'qty': 10,
'filledQty': filled_quantity,
'limitPrice': 820.5,
},
}
return json.dumps(message)
def print_updates(self, orders, positions, received_at):
"""Prints each order and position the socket handed on.
Args:
orders (list): The message's orders, exactly as Fyers sent them.
positions (list): The message's positions, exactly as Fyers sent them.
received_at (datetime.datetime): When the message arrived, which is not printed.
Returns:
None: This method returns nothing.
"""
for order in orders:
print(f"Order {order['id']} status {order['status']}: {order['filledQty']} of {order['qty']} {order['symbol']}")
for position in positions:
print(f"Position {position['symbol']}: net quantity {position['netQty']}")
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__':
OrdersAndPositionsExample().run()
It prints:
Stand-in login number 1
Profile served for login 1
Handshake header names: ['authorization']
Connecting to wss://socket.fyers.in/trade/v3
Sent: {"T": "SUB_ORD", "SLIST": ["orders", "positions"], "SUB_T": 1}
INFO Order updates socket opened and subscribed to orders, positions.
INFO Fyers: Successfully subscribed.
Order 26092500012345 status 6: 0 of 10 NSE:SBIN-EQ
Position NSE:SBIN-EQ: net quantity 10
Order 26092500012345 status 2: 10 of 10 NSE:SBIN-EQ
INFO Order updates socket closed with status 1000: normal closure
INFO Order updates socket stopped.
Gave up: False
examples/stock_brokers/websockets/fyers/FyersOrderUpdatesSocket/example_2_session_error_closes_the_socket.py
"""Shows Fyers' order update socket closing on a session error and logging in again without checking.
A Fyers error message whose code is -8, -15, -16 or -17 means the session is dead. `FyersOrderUpdatesSocket` first hands on any orders and positions in the same message, then closes the connection, and its reconnect loop calls `FyersSession.log_in_again_without_checking`, which logs in and confirms the new session against the profile whether or not another process already replaced the token.
This program scripts two connections. The first delivers one frame holding a JSON list of an order and a session error with code -16, followed by an order that is never delivered because the socket has closed. The second delivers the order filled. No real socket is opened: a stand-in `websocket` package plays the connections, and a stand-in `stock_brokers.api.fyers` module counts each `FyersAPI` construction as a login and serves the profile. The backoff is set to zero seconds.
Notice that the order in the same frame as the error still reaches the callback, and that the login is confirmed against the profile before the socket reconnects.
Run it from the project root:
python examples/stock_brokers/websockets/fyers/FyersOrderUpdatesSocket/example_2_session_error_closes_the_socket.py
"""
import base64
import json
import logging
import sys
import types
from stock_brokers.websockets.fyers import (
FyersOrderUpdatesSocket,
FyersSession,
)
class LoginRecord:
"""The Fyers 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.
expired_logins (set): The login numbers whose token has already expired.
refused_profiles (set): The login numbers whose session the profile endpoint refuses.
"""
access_token = None
login_count = 0
expired_logins = set()
refused_profiles = set()
class StandInFyersAPI:
"""A stand-in for `FyersAPI` whose construction is a pretend login that issues a JWT access token."""
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
expiry = 4102444800
if LoginRecord.login_count in LoginRecord.expired_logins:
expiry = 1
claims = {
'exp': expiry,
'hsm_key': f'hsm-key-{LoginRecord.login_count}',
}
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
LoginRecord.access_token = f'header.{payload}.signature'
print(f'Stand-in login number {LoginRecord.login_count}')
self._settings = {
'app_id': 'XY1234-100',
}
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 get(self, url, timeout):
"""Answers the profile 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 Fyers' answer under `data`.
"""
if LoginRecord.login_count in LoginRecord.refused_profiles:
print(f'Profile refused for login {LoginRecord.login_count}')
return {
'data': {
's': 'error',
'code': -16,
'message': 'Could not authenticate the user',
},
}
print(f'Profile served for login {LoginRecord.login_count}')
return {
'data': {
's': 'ok',
'data': {
'fy_id': 'XY01234',
},
},
}
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.
"""
if isinstance(data, bytes):
print(f'Sent binary frame: {data.hex()}')
else:
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.
"""
if 'header' in handlers:
print(f"Handshake header names: {list(handlers['header'])}")
plan = self.plans.pop(0)
if not self.plans:
self.when_finished()
return ScriptedConnection(url, plan, handlers)
class SessionErrorClosesTheSocketExample:
"""Runs the Fyers order update socket through a session error and a good connection.
Attributes:
socket (FyersOrderUpdatesSocket): The socket being shown.
"""
def __init__(self):
"""Builds the messages, the stand-ins, the session and the socket.
Returns:
None: This method returns nothing.
"""
logging.basicConfig(stream=sys.stdout, format='%(levelname)s %(message)s', level=logging.INFO)
api_module = types.ModuleType('stock_brokers.api.fyers')
api_module.FyersAPI = StandInFyersAPI
sys.modules['stock_brokers.api.fyers'] = api_module
session_error = {
's': 'error',
'code': -16,
'message': 'Token is invalid or expired',
}
batch = [
json.loads(self.order_message(6, 0)),
session_error,
]
plans = [
{
'frames': [
json.dumps(batch),
self.order_message(6, 5),
],
},
{
'frames': [
self.order_message(2, 10),
],
},
]
websocket_module = ScriptedWebsocketModule(plans)
sys.modules['websocket'] = websocket_module
logger = logging.getLogger('fyers.orders')
self.socket = FyersOrderUpdatesSocket(FyersSession(logger), self.print_updates, logger)
self.socket.MIN_BACKOFF_SECONDS = 0
websocket_module.when_finished = self.socket.close
def order_message(self, status, filled_quantity):
"""One of Fyers' order messages for an SBIN buy.
Args:
status (int): Fyers' numeric order status, 6 for pending and 2 for filled.
filled_quantity (int): How much has filled.
Returns:
str: The message as JSON text.
"""
message = {
's': 'ok',
'orders': {
'id': '26092500012345',
'symbol': 'NSE:SBIN-EQ',
'status': status,
'side': 1,
'qty': 10,
'filledQty': filled_quantity,
'limitPrice': 820.5,
},
}
return json.dumps(message)
def print_updates(self, orders, positions, received_at):
"""Prints each order and position the socket handed on.
Args:
orders (list): The message's orders, exactly as Fyers sent them.
positions (list): The message's positions, exactly as Fyers sent them.
received_at (datetime.datetime): When the message arrived, which is not printed.
Returns:
None: This method returns nothing.
"""
for order in orders:
print(f"Order {order['id']} status {order['status']}: {order['filledQty']} of {order['qty']} {order['symbol']}")
for position in positions:
print(f"Position {position['symbol']}: net quantity {position['netQty']}")
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__':
SessionErrorClosesTheSocketExample().run()
It prints:
Stand-in login number 1
Profile served for login 1
Handshake header names: ['authorization']
Connecting to wss://socket.fyers.in/trade/v3
Sent: {"T": "SUB_ORD", "SLIST": ["orders", "positions"], "SUB_T": 1}
INFO Order updates socket opened and subscribed to orders, positions.
ERROR Fyers reported an error: Token is invalid or expired
Order 26092500012345 status 6: 0 of 10 NSE:SBIN-EQ
INFO Order updates socket closed with status 1000: normal closure
WARNING Logging in to Fyers again.
Stand-in login number 2
Profile served for login 2
WARNING Order updates socket disconnected. Reconnecting in 0 seconds.
Handshake header names: ['authorization']
Connecting to wss://socket.fyers.in/trade/v3
Sent: {"T": "SUB_ORD", "SLIST": ["orders", "positions"], "SUB_T": 1}
INFO Order updates socket opened and subscribed to orders, positions.
Order 26092500012345 status 2: 10 of 10 NSE:SBIN-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/fyers.py
Opens the websocket with the token in force now, authenticated in the handshake, and blocks until it closes.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/fyers.py
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/fyers.py
Subscribes to orders and positions; the session was authenticated by the handshake.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The connection that opened. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/fyers.py
Notes a refused handshake and reports the error.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The connection that failed. |
required |
error
|
Exception
|
The error, which carries |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/fyers.py
Hands a message's orders and positions on, then closes the socket if the message said the session is dead.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The connection the message arrived on. |
required |
message
|
bytes | str
|
The message. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/fyers.py
The orders and positions a message carries, acting on error and subscription messages.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
bytes | str
|
The message. |
required |
Returns:
| Type | Description |
|---|---|
tuple
|
The orders and the positions, each a list of Fyers' objects in arrival order. |
Source code in stock_brokers/websockets/fyers.py
Whether a field holds something other than a blank or an NA placeholder.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
value
|
object
|
The field. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True when it holds a value. |
Source code in stock_brokers/websockets/fyers.py
Closes the connection and stops reconnecting.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example usage
Connects, and reconnects with backoff, until closed or unable to recover.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example usage
Source code in stock_brokers/websockets/base.py
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
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. |