Wisdom Capital's two XTS websockets: the market data feed and the interactive (trading) namespace that pushes order and position events.
Two applications, two sessions.
XTS splits a broker into a market data application and an interactive one.
Quotes need the market data session, whose token is minted from price_api_key and price_api_secret; XTS issues exactly one per application key, and a second login invalidates the first.
WisdomCapitalMarketDataSession reads the market data token from the shared login and asks WisdomCapitalAPI to replace it when it is refused, which the API does under a Redis lock, and only when nobody has replaced it already.
Orders and positions live in the interactive application, which also keeps one session per key: WisdomCapitalInteractiveSession joins with the token WisdomCapitalAPI shares through last_login, reads the user id from that token's JWT payload, and never logs in for itself except by constructing WisdomCapitalAPI again, which keeps a stored token that still works.
The transport.
The host's certificate is issued to the platform provider's domain, so the exact certificate is pinned instead of the hostname being checked: HTTPS requests go through a urllib3 pool that asserts the fingerprint, and a websocket's certificate is compared with the pin as soon as it opens.
A connection opens an Engine.IO session over HTTPS polling, attaches a websocket with the session id, sends the 2probe probe, answers 3probe with 5 to complete the upgrade, and then pings (2) at half the shorter of the ping interval and timeout the server asked for, since the two disagree and the interactive namespace waits only twenty seconds.
A server ping is answered with 3.
XTS reports a refused token with e-session, e-token or "Invalid Token" in the answer.
Quotes.
WisdomCapitalQuotesSocket attaches with publishFormat=JSON&broadcastMode=Full and subscribes over REST, for touchline (1501) and market depth (1502); the answer carries the current snapshot, which is handed on straight away.
XTS reports instruments still subscribed from an earlier connection as e-session-0002 Instrument Already Subscribed, with the same prefix as a refused token, so that answer is met by dropping the subscription and asking again rather than by replacing a good token.
Events arrive as socket.io frames 42["<event>", <payload>], the payload a quote, a list of them, or either as a JSON string; touchline and depth for one instrument are merged, a depth message nesting its touchline fields, before a tick is built.
XTS counts seconds from 1980-01-01 in India time, so its timestamps are moved onto the true Unix epoch, and it misspells the last traded quantity as LastTradedQunatity, sending the key with a null value.
A tick with neither a price nor a book is not handed on.
Order updates.
WisdomCapitalOrderUpdatesSocket joins the interactive namespace with apiType=INTERACTIVE, which is enough for XTS to push every order, trade and position event for the account; without it the server cuts the connection after ten seconds.
When another login replaces the token, XTS sends a logout event and leaves the connection open while delivering nothing more, so the socket closes, waits up to ten seconds for the replacing login to reach Redis, and reconnects with it; this wait is why the socket keeps a reconnect loop of its own.
A trade event is not handed on, because the order event that follows carries the fill.
Neither socket writes Redis.
The quotes socket hands ticks to on_ticks, and the order socket hands each event's orders or positions to on_updates; the scripts in bin/wisdom_capital/ do the writing.
Classes:
| Name | Description |
|---|---|
WisdomCapitalTransport |
The pinned HTTPS requests, the Engine.IO handshake and the heartbeat both XTS sockets use. |
WisdomCapitalMarketDataSession |
The market data login every quotes socket authenticates with. |
WisdomCapitalInteractiveSession |
The interactive login the order updates socket joins with, shared with every Wisdom Capital process. |
WisdomCapitalQuotesSocket |
One XTS market data socket carrying the touchline and depth of a batch of instruments. |
WisdomCapitalOrderUpdatesSocket |
Wisdom Capital's XTS interactive socket, which carries the account's order and position events. |
The pinned HTTPS requests, the Engine.IO handshake and the heartbeat both XTS sockets use.
Example programs
examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalTransport/example_1_engine_io_handshake.py
"""Opens an XTS Engine.IO session through `WisdomCapitalTransport` and derives the websocket URL, heartbeat and TLS options from it.
`WisdomCapitalTransport` holds what both Wisdom Capital sockets need to reach XTS. `request` sends an HTTPS request through a urllib3 pool pinned to the platform's certificate, because the certificate is issued to the platform provider's domain rather than Wisdom Capital's. `handshake` opens an Engine.IO session over polling and decodes the answer with `first_json_object`, since the JSON is framed with a length prefix. `heartbeat_seconds` picks half the shorter of the server's ping interval and timeout, `websocket_url` attaches a websocket to the session, and `ssl_options` relaxes the hostname check because the pin is checked once the websocket opens.
This program performs one handshake and one plain request against a stand-in `urllib3` package, whose pool prints each request and gives a scripted answer, and reads a polling body with no JSON in it. No network is used.
Notice that the heartbeat is 10 seconds, half of the 20 second ping timeout, and that `check_hostname` is off while certificate verification stays required.
Run it from the project root:
python examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalTransport/example_1_engine_io_handshake.py
"""
import json
import ssl
import sys
from stock_brokers.websockets.wisdom_capital import (
WisdomCapitalTransport,
)
class PinnedResponse:
"""A stand-in for the `urllib3` response of one XTS request.
Attributes:
status (int): The HTTP status.
data (bytes): The body.
"""
def __init__(self, status, text):
"""Keeps the answer.
Args:
status (int): The HTTP status.
text (str): The body as text.
Returns:
None: This method returns nothing.
"""
self.status = status
self.data = text.encode()
class PinnedPool:
"""A stand-in for `urllib3.HTTPSConnectionPool` that answers from a shared script.
Attributes:
answers (list): The answers still to give, shared by every pool.
"""
def __init__(self, answers):
"""Keeps the shared answers.
Args:
answers (list): The answers still to give, shared by every pool.
Returns:
None: This method returns nothing.
"""
self.answers = answers
def request(self, method, path, body=None, headers=None, timeout=None):
"""Prints the request and gives the next scripted answer.
Args:
method (str): The HTTP method.
path (str): The path, with its query string.
body (str | None): The JSON body.
headers (dict | None): The request headers.
timeout (int | None): The request timeout in seconds.
Returns:
PinnedResponse: The answer.
"""
shown = path.split('?')[0]
if body is not None:
code = json.loads(body).get('xtsMessageCode')
shown = f'{shown} for message code {code}'
answer = self.answers.pop(0)
print(f'HTTPS {method} {shown}: HTTP {answer.status}')
return answer
class StandInUrllib3:
"""A stand-in for the `urllib3` package whose pools answer from one script.
Attributes:
answers (list): The answers, in the order requests are made.
"""
def __init__(self, answers):
"""Keeps the answers.
Args:
answers (list): The answers, in the order requests are made.
Returns:
None: This method returns nothing.
"""
self.answers = answers
def HTTPSConnectionPool(self, host, port, assert_fingerprint=None):
"""Builds a pool, with the same signature as the real class.
Args:
host (str): The host.
port (int): The port.
assert_fingerprint (str | None): The certificate pin.
Returns:
PinnedPool: The pool.
"""
return PinnedPool(self.answers)
class EngineIoHandshakeExample:
"""Performs a handshake and derives the connection details from it.
Attributes:
transport (WisdomCapitalTransport): The transport being shown.
"""
def __init__(self):
"""Registers the stand-in urllib3 package and builds the transport.
Returns:
None: This method returns nothing.
"""
handshake = {
'sid': 'engine-session-1',
'upgrades': [
'websocket',
],
'pingInterval': 25000,
'pingTimeout': 20000,
}
text = json.dumps(handshake)
success = {
'type': 'success',
}
answers = [
PinnedResponse(200, f'{len(text) + 1}:0{text}2:40'),
PinnedResponse(200, json.dumps(success)),
]
sys.modules['urllib3'] = StandInUrllib3(answers)
self.transport = WisdomCapitalTransport()
def run(self):
"""Prints the handshake and everything derived from it.
Returns:
None: This method returns nothing.
"""
query = 'token=market-data-token-1&userID=WCMD01&publishFormat=JSON&broadcastMode=Full'
handshake, status, body = self.transport.handshake('/apimarketdata', query)
print(f'Handshake {status}: {handshake}')
print(f'Heartbeat every {self.transport.heartbeat_seconds(handshake, 25.0)} seconds')
print(f'Heartbeat without limits from the server: {self.transport.heartbeat_seconds({}, 25.0)} seconds')
print(f"Websocket URL: {self.transport.websocket_url('/apimarketdata', handshake, query)}")
context = self.transport.ssl_options()['context']
print(f'check_hostname={context.check_hostname} verify_mode={context.verify_mode == ssl.CERT_REQUIRED}')
subscription = {
'xtsMessageCode': 1501,
}
response = self.transport.request('PUT', '/apimarketdata/instruments/subscription', token='market-data-token-1', body=subscription)
print(f'Request answered {response.status}: {response.data.decode()}')
print(f"First JSON object in a body without one: {self.transport.first_json_object('2:40')}")
if __name__ == '__main__':
EngineIoHandshakeExample().run()
It prints:
HTTPS GET /apimarketdata/socket.io/: HTTP 200
Handshake 200: {'sid': 'engine-session-1', 'upgrades': ['websocket'], 'pingInterval': 25000, 'pingTimeout': 20000}
Heartbeat every 10.0 seconds
Heartbeat without limits from the server: 25.0 seconds
Websocket URL: wss://trade.wisdomcapital.in/apimarketdata/socket.io/?EIO=3&transport=websocket&sid=engine-session-1&token=market-data-token-1&userID=WCMD01&publishFormat=JSON&broadcastMode=Full
check_hostname=False verify_mode=True
HTTPS PUT /apimarketdata/instruments/subscription for message code 1501: HTTP 200
Request answered 200: {"type": "success"}
First JSON object in a body without one: None
examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalTransport/example_2_pins_refusals_and_heartbeat.py
"""Checks certificate fingerprints, recognises refused tokens, and runs one heartbeat with `WisdomCapitalTransport`.
Once an XTS websocket opens, the sockets compare `peer_fingerprint`, the SHA-256 of the certificate the peer presented, with the pin; a connection whose certificate cannot be read counts as matching, since the HTTPS requests are pinned already. `is_authentication_refusal` recognises XTS's refused token answers by `e-session`, `e-token` or `Invalid Token`. `handshake` returns None for a refused handshake together with the status and body. `start_heartbeat` runs a thread that sends the Engine.IO ping `2` on an interval until the connection's `closed` event is set.
This program uses a stand-in connection presenting made-up certificate bytes and one presenting nothing, a stand-in `urllib3` package that refuses a handshake, and a heartbeat with a 0.05 second interval whose stand-in connection closes itself after the first ping. No network is used.
Notice that the made-up certificate does not match the pin, that `Instrument Already Subscribed` counts as a refusal to this check, which is why the quotes socket looks for it first, and that exactly one ping is sent.
Run it from the project root:
python examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalTransport/example_2_pins_refusals_and_heartbeat.py
"""
import json
import sys
import threading
from stock_brokers.websockets.wisdom_capital import (
WisdomCapitalTransport,
)
class PinnedResponse:
"""A stand-in for the `urllib3` response of one XTS request.
Attributes:
status (int): The HTTP status.
data (bytes): The body.
"""
def __init__(self, status, text):
"""Keeps the answer.
Args:
status (int): The HTTP status.
text (str): The body as text.
Returns:
None: This method returns nothing.
"""
self.status = status
self.data = text.encode()
class PinnedPool:
"""A stand-in for `urllib3.HTTPSConnectionPool` that answers from a shared script.
Attributes:
answers (list): The answers still to give, shared by every pool.
"""
def __init__(self, answers):
"""Keeps the shared answers.
Args:
answers (list): The answers still to give, shared by every pool.
Returns:
None: This method returns nothing.
"""
self.answers = answers
def request(self, method, path, body=None, headers=None, timeout=None):
"""Prints the request and gives the next scripted answer.
Args:
method (str): The HTTP method.
path (str): The path, with its query string.
body (str | None): The JSON body.
headers (dict | None): The request headers.
timeout (int | None): The request timeout in seconds.
Returns:
PinnedResponse: The answer.
"""
shown = path.split('?')[0]
if body is not None:
code = json.loads(body).get('xtsMessageCode')
shown = f'{shown} for message code {code}'
answer = self.answers.pop(0)
print(f'HTTPS {method} {shown}: HTTP {answer.status}')
return answer
class StandInUrllib3:
"""A stand-in for the `urllib3` package whose pools answer from one script.
Attributes:
answers (list): The answers, in the order requests are made.
"""
def __init__(self, answers):
"""Keeps the answers.
Args:
answers (list): The answers, in the order requests are made.
Returns:
None: This method returns nothing.
"""
self.answers = answers
def HTTPSConnectionPool(self, host, port, assert_fingerprint=None):
"""Builds a pool, with the same signature as the real class.
Args:
host (str): The host.
port (int): The port.
assert_fingerprint (str | None): The certificate pin.
Returns:
PinnedPool: The pool.
"""
return PinnedPool(self.answers)
class PeerCertificate:
"""A stand-in for the TLS socket inside a websocket, presenting fixed certificate bytes."""
def getpeercert(self, binary_form=False):
"""The peer's certificate.
Args:
binary_form (bool): Whether to return the DER bytes.
Returns:
bytes: Made-up certificate bytes.
"""
return b'not the pinned certificate'
class CertificateHolder:
"""A stand-in for the websocket-client socket object that holds the TLS socket.
Attributes:
sock (PeerCertificate): The TLS socket.
"""
def __init__(self):
"""Holds the stand-in TLS socket.
Returns:
None: This method returns nothing.
"""
self.sock = PeerCertificate()
class ConnectionWithCertificate:
"""A stand-in for an open websocket whose peer presented a certificate.
Attributes:
sock (CertificateHolder): The socket object.
"""
def __init__(self):
"""Holds the stand-in socket object.
Returns:
None: This method returns nothing.
"""
self.sock = CertificateHolder()
class ConnectionWithoutCertificate:
"""A stand-in for an open websocket whose certificate cannot be read."""
class PingedConnection:
"""A stand-in for an open websocket that records pings and closes after the first.
Attributes:
closed (threading.Event): The connection's closed event, set after the first ping.
sent (list): The messages sent.
"""
def __init__(self):
"""Starts open with nothing sent.
Returns:
None: This method returns nothing.
"""
self.closed = threading.Event()
self.sent = []
def send(self, data):
"""Records a ping and closes the connection.
Args:
data (str): The message.
Returns:
None: This method returns nothing.
"""
self.sent.append(data)
self.closed.set()
class PinsRefusalsAndHeartbeatExample:
"""Checks fingerprints, refusals and one heartbeat.
Attributes:
transport (WisdomCapitalTransport): The transport being shown.
"""
def __init__(self):
"""Registers the stand-in urllib3 package and builds the transport.
Returns:
None: This method returns nothing.
"""
refused = {
'type': 'error',
'code': 'e-session-0001',
'description': 'Invalid Token',
}
sys.modules['urllib3'] = StandInUrllib3([
PinnedResponse(400, json.dumps(refused)),
])
self.transport = WisdomCapitalTransport()
def run(self):
"""Prints each check and the heartbeat's pings.
Returns:
None: This method returns nothing.
"""
print(f'Made-up certificate: {self.transport.peer_fingerprint(ConnectionWithCertificate())[:16]}...')
print(f'Unreadable certificate: {self.transport.peer_fingerprint(ConnectionWithoutCertificate())[:16]}...')
answers = [
'{"code": "e-session-0001", "description": "Invalid Token"}',
'{"code": "e-session-0002", "description": "Instrument Already Subscribed !"}',
'{"type": "success"}',
]
for answer in answers:
print(f'Refusal: {self.transport.is_authentication_refusal(answer)} for {answer}')
handshake, status, body = self.transport.handshake('/interactive', 'token=stale&userID=WC0001&apiType=INTERACTIVE')
print(f'Refused handshake: {handshake}, {status}, {body}')
connection = PingedConnection()
stop = threading.Event()
self.transport.start_heartbeat(connection, connection.closed, 0.05, stop, 'example_heartbeat')
connection.closed.wait(5)
print(f'Heartbeat sent: {connection.sent}')
if __name__ == '__main__':
PinsRefusalsAndHeartbeatExample().run()
It prints:
Made-up certificate: 2CC853CECC8C4B03...
Unreadable certificate: 1ACF7CA747F2D678...
Refusal: True for {"code": "e-session-0001", "description": "Invalid Token"}
Refusal: True for {"code": "e-session-0002", "description": "Instrument Already Subscribed !"}
Refusal: False for {"type": "success"}
HTTPS GET /interactive/socket.io/: HTTP 400
Refused handshake: None, 400, {"type": "error", "code": "e-session-0001", "description": "Invalid Token"}
Heartbeat sent: ['2']
Methods:
| Name | Description |
|---|---|
request |
Sends an HTTPS request to the platform over a connection pinned to its certificate. |
handshake |
Opens an Engine.IO session over HTTPS polling. |
first_json_object |
The first balanced |
heartbeat_seconds |
Half the shorter of the ping interval and timeout the server asked for, so a ping sits inside both. |
websocket_url |
The websocket URL that attaches to an Engine.IO session. |
ssl_options |
The TLS options for the websocket: the chain verified and the hostname check relaxed, since the pin is checked once it opens. |
peer_fingerprint |
The SHA-256 fingerprint of the certificate an open websocket's peer presented. |
start_heartbeat |
Starts a thread that sends the Engine.IO ping on the interval the server asked for, until the connection closes. |
is_authentication_refusal |
Whether an answer says the token was refused. |
Sends an HTTPS request to the platform over a connection pinned to its certificate.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
method
|
str
|
The HTTP method. |
required |
path
|
str
|
The path to request. |
required |
token
|
str | None
|
The token to send as |
None
|
body
|
object | None
|
An object to send as JSON, if any. |
None
|
Returns:
| Type | Description |
|---|---|
HTTPResponse
|
The response. |
Raises:
| Type | Description |
|---|---|
HTTPError
|
When the request fails or the certificate does not match the pin. |
Example usage
From example_1_engine_io_handshake.py
response = self.transport.request('PUT', '/apimarketdata/instruments/subscription', token='market-data-token-1', body=subscription)
Source code in stock_brokers/websockets/wisdom_capital.py
Opens an Engine.IO session over HTTPS polling.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str
|
The application path, market data or interactive. |
required |
query
|
str
|
The query string that authenticates the session. |
required |
Returns:
| Type | Description |
|---|---|
tuple
|
The decoded handshake, or None when it was refused or unreadable, and the answer's status and body for reporting. |
Example usage
From example_1_engine_io_handshake.py
handshake, status, body = self.transport.handshake('/apimarketdata', query)
From example_2_pins_refusals_and_heartbeat.py
handshake, status, body = self.transport.handshake('/interactive', 'token=stale&userID=WC0001&apiType=INTERACTIVE')
Source code in stock_brokers/websockets/wisdom_capital.py
The first balanced {...} in an Engine.IO polling body, which frames the handshake JSON with a length prefix.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
The body. |
required |
Returns:
| Type | Description |
|---|---|
dict | None
|
The decoded object, or None when there is none. |
Example usage
From example_1_engine_io_handshake.py
print(f"First JSON object in a body without one: {self.transport.first_json_object('2:40')}")
Source code in stock_brokers/websockets/wisdom_capital.py
Half the shorter of the ping interval and timeout the server asked for, so a ping sits inside both.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
handshake
|
dict
|
The decoded Engine.IO handshake. |
required |
default_seconds
|
float
|
The interval to use when the server asked for neither. |
required |
Returns:
| Type | Description |
|---|---|
float
|
The interval in seconds, at least one. |
Example usage
From example_1_engine_io_handshake.py
print(f'Heartbeat every {self.transport.heartbeat_seconds(handshake, 25.0)} seconds')
From example_1_engine_io_handshake.py
print(f'Heartbeat without limits from the server: {self.transport.heartbeat_seconds({}, 25.0)} seconds')
Source code in stock_brokers/websockets/wisdom_capital.py
The websocket URL that attaches to an Engine.IO session.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path
|
str
|
The application path, market data or interactive. |
required |
handshake
|
dict
|
The decoded Engine.IO handshake. |
required |
query
|
str
|
The query string that authenticates the session. |
required |
Returns:
| Type | Description |
|---|---|
str
|
The URL. |
Example usage
From example_1_engine_io_handshake.py
print(f"Websocket URL: {self.transport.websocket_url('/apimarketdata', handshake, query)}")
Source code in stock_brokers/websockets/wisdom_capital.py
The TLS options for the websocket: the chain verified and the hostname check relaxed, since the pin is checked once it opens.
Returns:
| Type | Description |
|---|---|
dict
|
The |
Example usage
Source code in stock_brokers/websockets/wisdom_capital.py
The SHA-256 fingerprint of the certificate an open websocket's peer presented.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The open connection. |
required |
Returns:
| Type | Description |
|---|---|
str
|
The fingerprint in upper case hex, or the pinned one when the certificate cannot be read. |
Example usage
From example_2_pins_refusals_and_heartbeat.py
print(f'Made-up certificate: {self.transport.peer_fingerprint(ConnectionWithCertificate())[:16]}...')
From example_2_pins_refusals_and_heartbeat.py
print(f'Unreadable certificate: {self.transport.peer_fingerprint(ConnectionWithoutCertificate())[:16]}...')
Source code in stock_brokers/websockets/wisdom_capital.py
Starts a thread that sends the Engine.IO ping on the interval the server asked for, until the connection closes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The connection to ping. |
required |
closed
|
Event
|
Set when this connection closes. |
required |
interval
|
float
|
The interval in seconds. |
required |
stop
|
Event
|
Set when the socket is stopped. |
required |
thread_name
|
str
|
The thread's name. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example usage
From example_2_pins_refusals_and_heartbeat.py
self.transport.start_heartbeat(connection, connection.closed, 0.05, stop, 'example_heartbeat')
Source code in stock_brokers/websockets/wisdom_capital.py
Sends 2 every interval until the connection closes, the socket stops, or a send fails.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The connection to ping. |
required |
closed
|
Event
|
Set when this connection closes. |
required |
interval
|
float
|
The interval in seconds. |
required |
stop
|
Event
|
Set when the socket is stopped. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/wisdom_capital.py
Whether an answer says the token was refused.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
text
|
str
|
The answer's body. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True for a refused token. |
Example usage
From example_2_pins_refusals_and_heartbeat.py
print(f'Refusal: {self.transport.is_authentication_refusal(answer)} for {answer}')
Source code in stock_brokers/websockets/wisdom_capital.py
The market data login every quotes socket authenticates with.
Constructs WisdomCapitalAPI, which establishes both of Wisdom Capital's sessions when either has stopped working.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
logger
|
Logger
|
Where token replacements are reported. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Raises:
| Type | Description |
|---|---|
Exception
|
Whatever |
Example programs
examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalMarketDataSession/example_1_reading_and_replacing_the_token.py
"""Reads Wisdom Capital's market data token and replaces it, once, when a socket reports it refused.
XTS issues one market data token per application key, and a second login invalidates the first, so `WisdomCapitalMarketDataSession` never logs in on its own. `market_data` reads the token and user id in force from `WisdomCapitalAPI`, and `replace` asks the API to replace a refused token, which the API does under a Redis lock and only when nobody has replaced it already.
Constructing the real `WisdomCapitalAPI` logs in to both XTS applications, so the program registers a stand-in module under `stock_brokers.api.wisdom_capital` that issues numbered market data tokens and replaces one only when it is still the stale one.
Notice that the second `replace` with the same stale token makes no login.
Run it from the project root:
python examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalMarketDataSession/example_1_reading_and_replacing_the_token.py
"""
import base64
import json
import logging
import sys
import types
from stock_brokers.websockets.wisdom_capital import (
WisdomCapitalMarketDataSession,
)
class LoginRecord:
"""The Wisdom Capital login state every stand-in API object reads, standing in for the Redis `last_login` hash.
Attributes:
market_data_token (str | None): The market data token in force now.
interactive_user_id (str | None): The user id inside the interactive token in force now.
interactive_session (str | None): The session inside the interactive token in force now.
login_count (int): How many interactive logins the stand-in API has made.
market_data_logins (int): How many market data logins the stand-in API has made.
market_data_fails (bool): Whether the market data login fails.
"""
market_data_token = None
interactive_user_id = 'WC0001'
interactive_session = None
login_count = 0
market_data_logins = 0
market_data_fails = False
class StandInWisdomCapitalAPI:
"""A stand-in for `WisdomCapitalAPI` whose construction is a pretend login to both XTS applications.
Attributes:
market_data_session_error (Exception | None): The market data login failure, recorded rather than raised, as the real class does.
"""
def __init__(self):
"""Logs in, which here only issues numbered tokens.
Returns:
None: This method returns nothing.
"""
LoginRecord.login_count = LoginRecord.login_count + 1
LoginRecord.interactive_session = f'interactive-{LoginRecord.login_count}'
print(f'Stand-in login number {LoginRecord.login_count}')
self.market_data_session_error = None
if LoginRecord.market_data_fails:
self.market_data_session_error = RuntimeError('the market data login was refused: Invalid secret key')
elif LoginRecord.market_data_token is None:
self.log_in_to_market_data()
def log_in_to_market_data(self):
"""Issues the next numbered market data token.
Returns:
None: This method returns nothing.
"""
LoginRecord.market_data_logins = LoginRecord.market_data_logins + 1
LoginRecord.market_data_token = f'market-data-token-{LoginRecord.market_data_logins}'
def _current_login(self):
"""The interactive login in force now, its token a JWT carrying the user id as XTS signs it.
Returns:
dict: The access token.
"""
if LoginRecord.interactive_session is None:
return {
'access_token': None,
}
claims = {
'session': LoginRecord.interactive_session,
}
if LoginRecord.interactive_user_id is not None:
claims['userID'] = LoginRecord.interactive_user_id
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
return {
'access_token': f'header.{payload}.signature',
}
def market_data_session(self):
"""The market data token and user id in force now.
Returns:
dict: `access_token` and `user_id`.
"""
return {
'access_token': LoginRecord.market_data_token,
'user_id': 'WCMD01',
}
def replace_market_data_session(self, stale_access_token=None):
"""Logs in to market data again, unless the token in force is no longer the stale one.
Args:
stale_access_token (str | None): The token that was refused.
Returns:
dict: The market data session now in force.
"""
if LoginRecord.market_data_token == stale_access_token:
self.log_in_to_market_data()
print(f'Stand-in market data login issued {LoginRecord.market_data_token}')
else:
print('The market data token was already replaced; no login.')
return self.market_data_session()
class ReadingAndReplacingTheTokenExample:
"""Reads and replaces a market data token.
Attributes:
session (WisdomCapitalMarketDataSession): 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.wisdom_capital')
api_module.WisdomCapitalAPI = StandInWisdomCapitalAPI
sys.modules['stock_brokers.api.wisdom_capital'] = api_module
self.session = WisdomCapitalMarketDataSession(logging.getLogger('wisdom_capital.session'))
def run(self):
"""Prints the session, replaces its token twice with the same stale token, and prints it again.
Returns:
None: This method returns nothing.
"""
market_data = self.session.market_data()
print(f'Market data session: {market_data}')
stale_token = market_data['access_token']
self.session.replace(stale_token)
self.session.replace(stale_token)
print(f'Market data session: {self.session.market_data()}')
if __name__ == '__main__':
ReadingAndReplacingTheTokenExample().run()
It prints:
Stand-in login number 1
Market data session: {'access_token': 'market-data-token-1', 'user_id': 'WCMD01'}
WARNING Replacing the Wisdom Capital market data token.
Stand-in market data login issued market-data-token-2
WARNING Replacing the Wisdom Capital market data token.
The market data token was already replaced; no login.
Market data session: {'access_token': 'market-data-token-2', 'user_id': 'WCMD01'}
examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalMarketDataSession/example_2_market_data_login_failed.py
"""Shows `WisdomCapitalMarketDataSession` raising the market data login failure the API recorded.
`WisdomCapitalAPI` establishes both XTS sessions when it is constructed, and records a failed market data login in `market_data_session_error` rather than raising it, because the interactive session may still work. `WisdomCapitalMarketDataSession` needs the market data session, so its constructor raises that recorded error, and the quotes script exits instead of connecting without a token.
Constructing the real `WisdomCapitalAPI` logs in to both XTS applications, so the program registers a stand-in module under `stock_brokers.api.wisdom_capital` whose market data login is scripted to fail.
Notice that the stand-in's login itself succeeds, and that the error comes from building the session.
Run it from the project root:
python examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalMarketDataSession/example_2_market_data_login_failed.py
"""
import base64
import json
import logging
import sys
import types
from stock_brokers.websockets.wisdom_capital import (
WisdomCapitalMarketDataSession,
)
class LoginRecord:
"""The Wisdom Capital login state every stand-in API object reads, standing in for the Redis `last_login` hash.
Attributes:
market_data_token (str | None): The market data token in force now.
interactive_user_id (str | None): The user id inside the interactive token in force now.
interactive_session (str | None): The session inside the interactive token in force now.
login_count (int): How many interactive logins the stand-in API has made.
market_data_logins (int): How many market data logins the stand-in API has made.
market_data_fails (bool): Whether the market data login fails.
"""
market_data_token = None
interactive_user_id = 'WC0001'
interactive_session = None
login_count = 0
market_data_logins = 0
market_data_fails = False
class StandInWisdomCapitalAPI:
"""A stand-in for `WisdomCapitalAPI` whose construction is a pretend login to both XTS applications.
Attributes:
market_data_session_error (Exception | None): The market data login failure, recorded rather than raised, as the real class does.
"""
def __init__(self):
"""Logs in, which here only issues numbered tokens.
Returns:
None: This method returns nothing.
"""
LoginRecord.login_count = LoginRecord.login_count + 1
LoginRecord.interactive_session = f'interactive-{LoginRecord.login_count}'
print(f'Stand-in login number {LoginRecord.login_count}')
self.market_data_session_error = None
if LoginRecord.market_data_fails:
self.market_data_session_error = RuntimeError('the market data login was refused: Invalid secret key')
elif LoginRecord.market_data_token is None:
self.log_in_to_market_data()
def log_in_to_market_data(self):
"""Issues the next numbered market data token.
Returns:
None: This method returns nothing.
"""
LoginRecord.market_data_logins = LoginRecord.market_data_logins + 1
LoginRecord.market_data_token = f'market-data-token-{LoginRecord.market_data_logins}'
def _current_login(self):
"""The interactive login in force now, its token a JWT carrying the user id as XTS signs it.
Returns:
dict: The access token.
"""
if LoginRecord.interactive_session is None:
return {
'access_token': None,
}
claims = {
'session': LoginRecord.interactive_session,
}
if LoginRecord.interactive_user_id is not None:
claims['userID'] = LoginRecord.interactive_user_id
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
return {
'access_token': f'header.{payload}.signature',
}
def market_data_session(self):
"""The market data token and user id in force now.
Returns:
dict: `access_token` and `user_id`.
"""
return {
'access_token': LoginRecord.market_data_token,
'user_id': 'WCMD01',
}
def replace_market_data_session(self, stale_access_token=None):
"""Logs in to market data again, unless the token in force is no longer the stale one.
Args:
stale_access_token (str | None): The token that was refused.
Returns:
dict: The market data session now in force.
"""
if LoginRecord.market_data_token == stale_access_token:
self.log_in_to_market_data()
print(f'Stand-in market data login issued {LoginRecord.market_data_token}')
else:
print('The market data token was already replaced; no login.')
return self.market_data_session()
class MarketDataLoginFailedExample:
"""Builds a market data session whose login failed.
Attributes:
logger (logging.Logger): Where the session would report.
"""
def __init__(self):
"""Registers the stand-in API module and scripts the market data login to fail.
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.wisdom_capital')
api_module.WisdomCapitalAPI = StandInWisdomCapitalAPI
sys.modules['stock_brokers.api.wisdom_capital'] = api_module
LoginRecord.market_data_fails = True
self.logger = logging.getLogger('wisdom_capital.session')
def run(self):
"""Builds the session and prints the error, or the session when it builds.
Returns:
None: This method returns nothing.
"""
try:
session = WisdomCapitalMarketDataSession(self.logger)
print(f'Market data session: {session.market_data()}')
except RuntimeError as error:
print(f'The market data session could not start: {error}')
if __name__ == '__main__':
MarketDataLoginFailedExample().run()
It prints:
Methods:
| Name | Description |
|---|---|
market_data |
The market data session in force now. |
replace |
Replaces a refused market data token, unless another socket or process already has. |
Source code in stock_brokers/websockets/wisdom_capital.py
Replaces a refused market data token, unless another socket or process already has.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
stale_token
|
str | None
|
The token the failing connection used. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Raises:
| Type | Description |
|---|---|
Exception
|
Whatever |
Example usage
Source code in stock_brokers/websockets/wisdom_capital.py
The interactive login the order updates socket joins with, shared with every Wisdom Capital process.
Constructs WisdomCapitalAPI, which logs in when the stored session is dead.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
logger
|
Logger
|
Where logins are reported. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Raises:
| Type | Description |
|---|---|
Exception
|
Whatever |
Example programs
examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalInteractiveSession/example_1_shared_login_and_user_id.py
"""Reads the interactive token Wisdom Capital processes share and the user id inside it, then checks the login again.
`WisdomCapitalInteractiveSession` joins with the interactive token every Wisdom Capital process shares through `last_login`. `current_token` returns that token, `shared_login` returns it together with the `userID` claim of its JWT payload, which XTS wants in the connection query, and `log_in_again_without_checking` constructs `WisdomCapitalAPI` again, which keeps a stored token that still works and logs in only when it does not.
Constructing the real `WisdomCapitalAPI` logs in to both XTS applications, so the program registers a stand-in module under `stock_brokers.api.wisdom_capital` whose tokens are JWTs carrying the user id. Each construction counts as one login.
Notice that the user id is read from inside the token, and that the token changes after the login check.
Run it from the project root:
python examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalInteractiveSession/example_1_shared_login_and_user_id.py
"""
import base64
import json
import logging
import sys
import types
from stock_brokers.websockets.wisdom_capital import (
WisdomCapitalInteractiveSession,
)
class LoginRecord:
"""The Wisdom Capital login state every stand-in API object reads, standing in for the Redis `last_login` hash.
Attributes:
market_data_token (str | None): The market data token in force now.
interactive_user_id (str | None): The user id inside the interactive token in force now.
interactive_session (str | None): The session inside the interactive token in force now.
login_count (int): How many interactive logins the stand-in API has made.
market_data_logins (int): How many market data logins the stand-in API has made.
market_data_fails (bool): Whether the market data login fails.
"""
market_data_token = None
interactive_user_id = 'WC0001'
interactive_session = None
login_count = 0
market_data_logins = 0
market_data_fails = False
class StandInWisdomCapitalAPI:
"""A stand-in for `WisdomCapitalAPI` whose construction is a pretend login to both XTS applications.
Attributes:
market_data_session_error (Exception | None): The market data login failure, recorded rather than raised, as the real class does.
"""
def __init__(self):
"""Logs in, which here only issues numbered tokens.
Returns:
None: This method returns nothing.
"""
LoginRecord.login_count = LoginRecord.login_count + 1
LoginRecord.interactive_session = f'interactive-{LoginRecord.login_count}'
print(f'Stand-in login number {LoginRecord.login_count}')
self.market_data_session_error = None
if LoginRecord.market_data_fails:
self.market_data_session_error = RuntimeError('the market data login was refused: Invalid secret key')
elif LoginRecord.market_data_token is None:
self.log_in_to_market_data()
def log_in_to_market_data(self):
"""Issues the next numbered market data token.
Returns:
None: This method returns nothing.
"""
LoginRecord.market_data_logins = LoginRecord.market_data_logins + 1
LoginRecord.market_data_token = f'market-data-token-{LoginRecord.market_data_logins}'
def _current_login(self):
"""The interactive login in force now, its token a JWT carrying the user id as XTS signs it.
Returns:
dict: The access token.
"""
if LoginRecord.interactive_session is None:
return {
'access_token': None,
}
claims = {
'session': LoginRecord.interactive_session,
}
if LoginRecord.interactive_user_id is not None:
claims['userID'] = LoginRecord.interactive_user_id
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
return {
'access_token': f'header.{payload}.signature',
}
def market_data_session(self):
"""The market data token and user id in force now.
Returns:
dict: `access_token` and `user_id`.
"""
return {
'access_token': LoginRecord.market_data_token,
'user_id': 'WCMD01',
}
def replace_market_data_session(self, stale_access_token=None):
"""Logs in to market data again, unless the token in force is no longer the stale one.
Args:
stale_access_token (str | None): The token that was refused.
Returns:
dict: The market data session now in force.
"""
if LoginRecord.market_data_token == stale_access_token:
self.log_in_to_market_data()
print(f'Stand-in market data login issued {LoginRecord.market_data_token}')
else:
print('The market data token was already replaced; no login.')
return self.market_data_session()
class SharedLoginAndUserIdExample:
"""Reads the shared interactive login around a login check.
Attributes:
session (WisdomCapitalInteractiveSession): 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.wisdom_capital')
api_module.WisdomCapitalAPI = StandInWisdomCapitalAPI
sys.modules['stock_brokers.api.wisdom_capital'] = api_module
self.session = WisdomCapitalInteractiveSession(logging.getLogger('wisdom_capital.session'))
def claims(self, token):
"""The JWT claims of a token, decoded for display.
Args:
token (str): The token.
Returns:
dict: The claims.
"""
payload = token.split('.')[1]
return json.loads(base64.urlsafe_b64decode(payload + '=' * (-len(payload) % 4)))
def run(self):
"""Prints the shared login before and after a login check.
Returns:
None: This method returns nothing.
"""
token, user_id = self.session.shared_login()
print(f'User id {user_id}, token claims {self.claims(token)}')
self.session.log_in_again_without_checking()
print(f'Token claims after the check: {self.claims(self.session.current_token())}')
if __name__ == '__main__':
SharedLoginAndUserIdExample().run()
It prints:
examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalInteractiveSession/example_2_unusable_stored_tokens.py
"""Shows `WisdomCapitalInteractiveSession.shared_login` refusing a missing token and a token without a user id.
`shared_login` raises `RuntimeError` when no usable interactive token is stored, or when the stored token's JWT payload carries no `userID`, because the socket cannot join the interactive namespace without both. The order update socket takes either as a refused login.
Constructing the real `WisdomCapitalAPI` logs in to both XTS applications, so the program registers a stand-in module under `stock_brokers.api.wisdom_capital`. The program then empties the shared login, and afterwards stores a token without the claim, as a broken login would.
Notice that `current_token` answers None rather than raising, and that the two errors say which part is missing.
Run it from the project root:
python examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalInteractiveSession/example_2_unusable_stored_tokens.py
"""
import base64
import json
import logging
import sys
import types
from stock_brokers.websockets.wisdom_capital import (
WisdomCapitalInteractiveSession,
)
class LoginRecord:
"""The Wisdom Capital login state every stand-in API object reads, standing in for the Redis `last_login` hash.
Attributes:
market_data_token (str | None): The market data token in force now.
interactive_user_id (str | None): The user id inside the interactive token in force now.
interactive_session (str | None): The session inside the interactive token in force now.
login_count (int): How many interactive logins the stand-in API has made.
market_data_logins (int): How many market data logins the stand-in API has made.
market_data_fails (bool): Whether the market data login fails.
"""
market_data_token = None
interactive_user_id = 'WC0001'
interactive_session = None
login_count = 0
market_data_logins = 0
market_data_fails = False
class StandInWisdomCapitalAPI:
"""A stand-in for `WisdomCapitalAPI` whose construction is a pretend login to both XTS applications.
Attributes:
market_data_session_error (Exception | None): The market data login failure, recorded rather than raised, as the real class does.
"""
def __init__(self):
"""Logs in, which here only issues numbered tokens.
Returns:
None: This method returns nothing.
"""
LoginRecord.login_count = LoginRecord.login_count + 1
LoginRecord.interactive_session = f'interactive-{LoginRecord.login_count}'
print(f'Stand-in login number {LoginRecord.login_count}')
self.market_data_session_error = None
if LoginRecord.market_data_fails:
self.market_data_session_error = RuntimeError('the market data login was refused: Invalid secret key')
elif LoginRecord.market_data_token is None:
self.log_in_to_market_data()
def log_in_to_market_data(self):
"""Issues the next numbered market data token.
Returns:
None: This method returns nothing.
"""
LoginRecord.market_data_logins = LoginRecord.market_data_logins + 1
LoginRecord.market_data_token = f'market-data-token-{LoginRecord.market_data_logins}'
def _current_login(self):
"""The interactive login in force now, its token a JWT carrying the user id as XTS signs it.
Returns:
dict: The access token.
"""
if LoginRecord.interactive_session is None:
return {
'access_token': None,
}
claims = {
'session': LoginRecord.interactive_session,
}
if LoginRecord.interactive_user_id is not None:
claims['userID'] = LoginRecord.interactive_user_id
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
return {
'access_token': f'header.{payload}.signature',
}
def market_data_session(self):
"""The market data token and user id in force now.
Returns:
dict: `access_token` and `user_id`.
"""
return {
'access_token': LoginRecord.market_data_token,
'user_id': 'WCMD01',
}
def replace_market_data_session(self, stale_access_token=None):
"""Logs in to market data again, unless the token in force is no longer the stale one.
Args:
stale_access_token (str | None): The token that was refused.
Returns:
dict: The market data session now in force.
"""
if LoginRecord.market_data_token == stale_access_token:
self.log_in_to_market_data()
print(f'Stand-in market data login issued {LoginRecord.market_data_token}')
else:
print('The market data token was already replaced; no login.')
return self.market_data_session()
class UnusableStoredTokensExample:
"""Asks for the shared login when the stored token is unusable.
Attributes:
session (WisdomCapitalInteractiveSession): 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.wisdom_capital')
api_module.WisdomCapitalAPI = StandInWisdomCapitalAPI
sys.modules['stock_brokers.api.wisdom_capital'] = api_module
self.session = WisdomCapitalInteractiveSession(logging.getLogger('wisdom_capital.session'))
def try_shared_login(self):
"""Asks for the shared login and prints the outcome.
Returns:
None: This method returns nothing.
"""
try:
token, user_id = self.session.shared_login()
print(f'Shared login for {user_id}')
except RuntimeError as error:
print(f'RuntimeError: {error}')
def run(self):
"""Empties the stored login, then stores one without a user id, asking for the shared login each time.
Returns:
None: This method returns nothing.
"""
LoginRecord.interactive_session = None
print(f'Current token: {self.session.current_token()}')
self.try_shared_login()
LoginRecord.interactive_session = 'interactive-without-user'
LoginRecord.interactive_user_id = None
self.try_shared_login()
if __name__ == '__main__':
UnusableStoredTokensExample().run()
It prints:
Methods:
| Name | Description |
|---|---|
current_token |
The interactive token stored in the shared login now. |
shared_login |
The interactive token every Wisdom Capital process shares, and the user id it was issued to. |
log_in_again_without_checking |
Constructs |
Source code in stock_brokers/websockets/wisdom_capital.py
The interactive token stored in the shared login now.
Returns:
| Type | Description |
|---|---|
str | None
|
The token, or None when none is stored. |
Example usage
From example_1_shared_login_and_user_id.py
print(f'Token claims after the check: {self.claims(self.session.current_token())}')
Source code in stock_brokers/websockets/wisdom_capital.py
The interactive token every Wisdom Capital process shares, and the user id it was issued to.
Returns:
| Type | Description |
|---|---|
tuple
|
The access token and the user id. |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
No usable token is stored, or the stored token carries no user id. |
Example usage
Source code in stock_brokers/websockets/wisdom_capital.py
Constructs WisdomCapitalAPI again, which keeps a stored token that still works and logs in only when it does not.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Raises:
| Type | Description |
|---|---|
Exception
|
Whatever |
Example usage
Source code in stock_brokers/websockets/wisdom_capital.py
The user id XTS signs into an interactive token.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
The interactive access token, a JWT. |
required |
Returns:
| Type | Description |
|---|---|
str | None
|
The token's |
Source code in stock_brokers/websockets/wisdom_capital.py
Bases: BrokerWebsocket
One XTS market data socket carrying the touchline and depth of a batch of instruments.
Sets up a socket for one batch of instruments.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
The connection's name, for the log. |
required |
tokens
|
list[str]
|
The batch of |
required |
names
|
dict[str, str]
|
Each token to the instrument's name, which becomes a tick's |
required |
session
|
WisdomCapitalMarketDataSession
|
The shared market data login. |
required |
on_ticks
|
callable
|
Called with each 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/wisdom_capital/WisdomCapitalQuotesSocket/example_1_snapshot_then_depth.py
"""Runs a Wisdom Capital XTS quotes socket through the Engine.IO handshake, the REST subscription snapshot and a depth event.
`WisdomCapitalQuotesSocket` opens an Engine.IO session over pinned HTTPS polling with the market data token, attaches a websocket to it, and completes the upgrade: it sends `2probe`, and when the server answers `3probe` it sends `5`, starts pinging, and subscribes its instruments over REST for touchline (1501) and market depth (1502). Each subscription answer carries the current snapshot, handed on at once. Events then arrive as socket.io frames `42["<event>", <payload>]`; touchline and depth for one instrument are merged, and XTS timestamps counted from 1980 in India time are moved onto the Unix epoch.
This program subscribes NSE RELIANCE (`1:2885`). The touchline subscription's snapshot carries a touchline and the depth subscription's carries nothing, then the websocket delivers a server ping and a depth event nesting a new touchline. No real socket or HTTPS request is made: a stand-in `urllib3` package answers the handshake and the subscriptions, a stand-in `websocket` package plays the frames and prints what the socket sends, and a stand-in `stock_brokers.api.wisdom_capital` module lets the real `WisdomCapitalMarketDataSession` supply the market data token. The stand-in connection presents no certificate, so the pin check falls back to the pinned fingerprint.
Notice the `2probe`, `5` and `3` the socket sends, that the snapshot's tick has a one-level book from the touchline's best bid and offer, that the depth event's tick has five levels because the padded zero row is dropped, and that the misspelt `LastTradedQunatity` still fills `last_quantity`.
Run it from the project root:
python examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalQuotesSocket/example_1_snapshot_then_depth.py
"""
import base64
import json
import logging
import sys
import types
from stock_brokers.websockets.wisdom_capital import (
WisdomCapitalMarketDataSession,
WisdomCapitalQuotesSocket,
)
class LoginRecord:
"""The Wisdom Capital login state every stand-in API object reads, standing in for the Redis `last_login` hash.
Attributes:
market_data_token (str | None): The market data token in force now.
interactive_user_id (str | None): The user id inside the interactive token in force now.
interactive_session (str | None): The session inside the interactive token in force now.
login_count (int): How many interactive logins the stand-in API has made.
market_data_logins (int): How many market data logins the stand-in API has made.
market_data_fails (bool): Whether the market data login fails.
"""
market_data_token = None
interactive_user_id = 'WC0001'
interactive_session = None
login_count = 0
market_data_logins = 0
market_data_fails = False
class StandInWisdomCapitalAPI:
"""A stand-in for `WisdomCapitalAPI` whose construction is a pretend login to both XTS applications.
Attributes:
market_data_session_error (Exception | None): The market data login failure, recorded rather than raised, as the real class does.
"""
def __init__(self):
"""Logs in, which here only issues numbered tokens.
Returns:
None: This method returns nothing.
"""
LoginRecord.login_count = LoginRecord.login_count + 1
LoginRecord.interactive_session = f'interactive-{LoginRecord.login_count}'
print(f'Stand-in login number {LoginRecord.login_count}')
self.market_data_session_error = None
if LoginRecord.market_data_fails:
self.market_data_session_error = RuntimeError('the market data login was refused: Invalid secret key')
elif LoginRecord.market_data_token is None:
self.log_in_to_market_data()
def log_in_to_market_data(self):
"""Issues the next numbered market data token.
Returns:
None: This method returns nothing.
"""
LoginRecord.market_data_logins = LoginRecord.market_data_logins + 1
LoginRecord.market_data_token = f'market-data-token-{LoginRecord.market_data_logins}'
def _current_login(self):
"""The interactive login in force now, its token a JWT carrying the user id as XTS signs it.
Returns:
dict: The access token.
"""
if LoginRecord.interactive_session is None:
return {
'access_token': None,
}
claims = {
'session': LoginRecord.interactive_session,
}
if LoginRecord.interactive_user_id is not None:
claims['userID'] = LoginRecord.interactive_user_id
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
return {
'access_token': f'header.{payload}.signature',
}
def market_data_session(self):
"""The market data token and user id in force now.
Returns:
dict: `access_token` and `user_id`.
"""
return {
'access_token': LoginRecord.market_data_token,
'user_id': 'WCMD01',
}
def replace_market_data_session(self, stale_access_token=None):
"""Logs in to market data again, unless the token in force is no longer the stale one.
Args:
stale_access_token (str | None): The token that was refused.
Returns:
dict: The market data session now in force.
"""
if LoginRecord.market_data_token == stale_access_token:
self.log_in_to_market_data()
print(f'Stand-in market data login issued {LoginRecord.market_data_token}')
else:
print('The market data token was already replaced; no login.')
return self.market_data_session()
class PinnedResponse:
"""A stand-in for the `urllib3` response of one XTS request.
Attributes:
status (int): The HTTP status.
data (bytes): The body.
"""
def __init__(self, status, text):
"""Keeps the answer.
Args:
status (int): The HTTP status.
text (str): The body as text.
Returns:
None: This method returns nothing.
"""
self.status = status
self.data = text.encode()
class PinnedPool:
"""A stand-in for `urllib3.HTTPSConnectionPool` that answers from a shared script.
Attributes:
answers (list): The answers still to give, shared by every pool.
"""
def __init__(self, answers):
"""Keeps the shared answers.
Args:
answers (list): The answers still to give, shared by every pool.
Returns:
None: This method returns nothing.
"""
self.answers = answers
def request(self, method, path, body=None, headers=None, timeout=None):
"""Prints the request and gives the next scripted answer.
Args:
method (str): The HTTP method.
path (str): The path, with its query string.
body (str | None): The JSON body.
headers (dict | None): The request headers.
timeout (int | None): The request timeout in seconds.
Returns:
PinnedResponse: The answer.
"""
shown = path.split('?')[0]
if body is not None:
code = json.loads(body).get('xtsMessageCode')
shown = f'{shown} for message code {code}'
answer = self.answers.pop(0)
print(f'HTTPS {method} {shown}: HTTP {answer.status}')
return answer
class StandInUrllib3:
"""A stand-in for the `urllib3` package whose pools answer from one script.
Attributes:
answers (list): The answers, in the order requests are made.
"""
def __init__(self, answers):
"""Keeps the answers.
Args:
answers (list): The answers, in the order requests are made.
Returns:
None: This method returns nothing.
"""
self.answers = answers
def HTTPSConnectionPool(self, host, port, assert_fingerprint=None):
"""Builds a pool, with the same signature as the real class.
Args:
host (str): The host.
port (int): The port.
assert_fingerprint (str | None): The certificate pin.
Returns:
PinnedPool: The pool.
"""
return PinnedPool(self.answers)
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`, where a frame may also be a function to call at that point.
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.split('?')[0]} with Engine.IO session {self.url.split('sid=')[1].split('&')[0]}")
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
if callable(frame):
frame()
continue
self.handlers['on_message'](self, frame)
self.handlers['on_close'](self, 1000, 'normal closure')
def send(self, data, opcode=None):
"""Prints a message the socket sends to the broker.
Args:
data (str | bytes): The message.
opcode (int | None): The frame type, when the socket names one.
Returns:
None: This method returns nothing.
"""
print(f'Sent: {data}')
def close(self, **options):
"""Notes that the socket asked to close.
Args:
**options (dict): Any close status the socket passes.
Returns:
None: This method returns nothing.
"""
self.closed = True
class ScriptedWebsocketModule:
"""A stand-in for the `websocket` package that hands out scripted connections in turn.
Attributes:
plans (list): What each connection does, in the order the socket connects.
when_finished (callable | None): Called when the last connection is handed out, so the socket stops after it.
"""
def __init__(self, plans):
"""Keeps the connection plans.
Args:
plans (list): What each connection does, in the order the socket connects.
Returns:
None: This method returns nothing.
"""
self.plans = plans
self.when_finished = None
def WebSocketApp(self, url, **handlers):
"""Builds the next scripted connection, closing the socket after the last one.
Args:
url (str): The feed URL.
**handlers (dict): The socket's `on_open`, `on_message`, `on_error` and `on_close` callbacks.
Returns:
ScriptedConnection: The connection.
"""
plan = self.plans.pop(0)
if not self.plans:
self.when_finished()
return ScriptedConnection(url, plan, handlers)
class SnapshotThenDepthExample:
"""Runs a Wisdom Capital quotes socket against a scripted handshake, subscription and connection.
Attributes:
socket (WisdomCapitalQuotesSocket): The socket being shown.
"""
def __init__(self):
"""Builds the answers, 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.wisdom_capital')
api_module.WisdomCapitalAPI = StandInWisdomCapitalAPI
sys.modules['stock_brokers.api.wisdom_capital'] = api_module
answers = [
self.handshake_answer('engine-session-1'),
self.subscription_answer([
self.touchline(2950.5),
]),
self.subscription_answer([]),
]
sys.modules['urllib3'] = StandInUrllib3(answers)
plans = [
{
'frames': [
'3probe',
'2',
self.depth_event(2951.0),
],
},
]
websocket_module = ScriptedWebsocketModule(plans)
sys.modules['websocket'] = websocket_module
logger = logging.getLogger('wisdom_capital.quotes')
tokens = [
'1:2885',
]
names = {
'1:2885': 'NSE:RELIANCE',
}
session = WisdomCapitalMarketDataSession(logger)
self.socket = WisdomCapitalQuotesSocket('socket_0', tokens, names, session, self.print_ticks, logger)
websocket_module.when_finished = self.socket.close
def handshake_answer(self, session_id):
"""An Engine.IO polling handshake answer, framed with its length prefix as XTS sends it.
Args:
session_id (str): The Engine.IO session id.
Returns:
PinnedResponse: The answer.
"""
handshake = {
'sid': session_id,
'upgrades': [
'websocket',
],
'pingInterval': 25000,
'pingTimeout': 20000,
}
text = json.dumps(handshake)
return PinnedResponse(200, f'{len(text) + 1}:0{text}2:40')
def touchline(self, last_price):
"""An XTS touchline for NSE RELIANCE (segment 1, instrument 2885), with XTS's misspelt last traded quantity.
Args:
last_price (float): The last traded price.
Returns:
dict: The touchline.
"""
return {
'ExchangeSegment': 1,
'ExchangeInstrumentID': 2885,
'LastTradedPrice': last_price,
'LastTradedQunatity': 25,
'TotalTradedQuantity': 1204500,
'AverageTradedPrice': 2949.0,
'Open': 2930.0,
'High': 2960.0,
'Low': 2925.0,
'Close': 2910.0,
'LastTradedTime': 1790311528 - 315513000,
'ExchangeTimeStamp': 1790311529 - 315513000,
'BidInfo': {
'Size': 100,
'Price': 2950.0,
'TotalOrders': 3,
},
'AskInfo': {
'Size': 120,
'Price': 2950.5,
'TotalOrders': 4,
},
}
def depth_event(self, last_price):
"""A socket.io market depth event whose quote nests its touchline, with padded zero rows.
Args:
last_price (float): The last traded price.
Returns:
str: The frame.
"""
bids = []
asks = []
for level in range(5):
bids.append({
'Size': 100 + level,
'Price': 2950.0 - level * 0.05,
'TotalOrders': 3,
})
asks.append({
'Size': 200 + level,
'Price': 2951.0 + level * 0.05,
'TotalOrders': 4,
})
bids.append({
'Size': 0,
'Price': 0,
'TotalOrders': 0,
})
quote = {
'ExchangeSegment': 1,
'ExchangeInstrumentID': 2885,
'Bids': bids,
'Asks': asks,
'Touchline': self.touchline(last_price),
}
event = [
'1502-json-full',
json.dumps(quote),
]
return '42' + json.dumps(event)
def subscription_answer(self, quotes):
"""A successful subscription answer carrying the current snapshot as JSON strings.
Args:
quotes (list): The quotes in the snapshot.
Returns:
PinnedResponse: The answer.
"""
encoded = []
for quote in quotes:
encoded.append(json.dumps(quote))
answer = {
'type': 'success',
'result': {
'listQuotes': encoded,
},
}
return PinnedResponse(200, json.dumps(answer))
def print_ticks(self, ticks):
"""Prints the main fields of each tick the socket handed on.
Args:
ticks (list): The ticks.
Returns:
None: This method returns nothing.
"""
for tick in ticks:
change = tick['change']
if change is not None:
change = round(change, 4)
print(f"Tick {tick['id']} ({tick['instrument_token']}) on {tick['exchange']} in {tick['mode']} mode")
print(f" last_price={tick['last_price']} last_quantity={tick['last_quantity']} volume={tick['volume']} change={change}")
print(f" last_trade_time={tick['last_trade_time']} exchange_timestamp={tick['exchange_timestamp']}")
if tick['depth']['buy']:
print(f" best bid={tick['depth']['buy'][0]} best offer={tick['depth']['sell'][0]} levels={len(tick['depth']['buy'])}")
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__':
SnapshotThenDepthExample().run()
It prints:
Stand-in login number 1
HTTPS GET /apimarketdata/socket.io/: HTTP 200
Connecting to wss://trade.wisdomcapital.in/apimarketdata/socket.io/ with Engine.IO session engine-session-1
Sent: 2probe
Sent: 5
HTTPS POST /apimarketdata/instruments/subscription for message code 1501: HTTP 200
HTTPS POST /apimarketdata/instruments/subscription for message code 1502: HTTP 200
Tick NSE:RELIANCE (1:2885) on NSECM in full mode
last_price=2950.5 last_quantity=25 volume=1204500 change=1.3918
last_trade_time=1790311528 exchange_timestamp=1790311529
best bid={'quantity': 100, 'price': 2950.0, 'orders': 3} best offer={'quantity': 120, 'price': 2950.5, 'orders': 4} levels=1
INFO socket_0 subscribed to 1 instrument(s).
Sent: 3
Tick NSE:RELIANCE (1:2885) on NSECM in full mode
last_price=2951.0 last_quantity=25 volume=1204500 change=1.4089
last_trade_time=1790311528 exchange_timestamp=1790311529
best bid={'quantity': 100, 'price': 2950.0, 'orders': 3} best offer={'quantity': 200, 'price': 2951.0, 'orders': 4} levels=5
INFO socket_0 closed with status 1000: normal closure
INFO socket_0 stopped.
Gave up: False
examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalQuotesSocket/example_2_refused_token_and_stale_subscription.py
"""Shows a Wisdom Capital quotes socket replacing a refused market data token and clearing a subscription left from an earlier connection.
XTS refuses a dead market data token at the Engine.IO handshake with an answer mentioning `Invalid Token`. `WisdomCapitalQuotesSocket` takes that as a refused login, and its reconnect loop asks `WisdomCapitalMarketDataSession.replace` for a new token, which the API issues only if nobody replaced the refused one already. Separately, XTS answers a subscription for instruments still subscribed from an earlier connection with `e-session-0002 Instrument Already Subscribed`, whose prefix looks like a refused token; the socket instead drops that subscription with a `PUT` and asks again.
This program's first handshake is refused. The second succeeds, its touchline subscription is refused as already subscribed, then dropped and granted, and the depth subscription is granted. No real socket or HTTPS request is made: stand-in `urllib3` and `websocket` packages play the answers and the frames, and a stand-in `stock_brokers.api.wisdom_capital` module issues numbered market data tokens. The backoff is set to zero seconds.
Notice the market data login between the two handshakes, the `PUT` that drops the stale subscription, and the snapshot tick handed on once the subscription is granted.
Run it from the project root:
python examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalQuotesSocket/example_2_refused_token_and_stale_subscription.py
"""
import base64
import json
import logging
import sys
import types
from stock_brokers.websockets.wisdom_capital import (
WisdomCapitalMarketDataSession,
WisdomCapitalQuotesSocket,
)
class LoginRecord:
"""The Wisdom Capital login state every stand-in API object reads, standing in for the Redis `last_login` hash.
Attributes:
market_data_token (str | None): The market data token in force now.
interactive_user_id (str | None): The user id inside the interactive token in force now.
interactive_session (str | None): The session inside the interactive token in force now.
login_count (int): How many interactive logins the stand-in API has made.
market_data_logins (int): How many market data logins the stand-in API has made.
market_data_fails (bool): Whether the market data login fails.
"""
market_data_token = None
interactive_user_id = 'WC0001'
interactive_session = None
login_count = 0
market_data_logins = 0
market_data_fails = False
class StandInWisdomCapitalAPI:
"""A stand-in for `WisdomCapitalAPI` whose construction is a pretend login to both XTS applications.
Attributes:
market_data_session_error (Exception | None): The market data login failure, recorded rather than raised, as the real class does.
"""
def __init__(self):
"""Logs in, which here only issues numbered tokens.
Returns:
None: This method returns nothing.
"""
LoginRecord.login_count = LoginRecord.login_count + 1
LoginRecord.interactive_session = f'interactive-{LoginRecord.login_count}'
print(f'Stand-in login number {LoginRecord.login_count}')
self.market_data_session_error = None
if LoginRecord.market_data_fails:
self.market_data_session_error = RuntimeError('the market data login was refused: Invalid secret key')
elif LoginRecord.market_data_token is None:
self.log_in_to_market_data()
def log_in_to_market_data(self):
"""Issues the next numbered market data token.
Returns:
None: This method returns nothing.
"""
LoginRecord.market_data_logins = LoginRecord.market_data_logins + 1
LoginRecord.market_data_token = f'market-data-token-{LoginRecord.market_data_logins}'
def _current_login(self):
"""The interactive login in force now, its token a JWT carrying the user id as XTS signs it.
Returns:
dict: The access token.
"""
if LoginRecord.interactive_session is None:
return {
'access_token': None,
}
claims = {
'session': LoginRecord.interactive_session,
}
if LoginRecord.interactive_user_id is not None:
claims['userID'] = LoginRecord.interactive_user_id
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
return {
'access_token': f'header.{payload}.signature',
}
def market_data_session(self):
"""The market data token and user id in force now.
Returns:
dict: `access_token` and `user_id`.
"""
return {
'access_token': LoginRecord.market_data_token,
'user_id': 'WCMD01',
}
def replace_market_data_session(self, stale_access_token=None):
"""Logs in to market data again, unless the token in force is no longer the stale one.
Args:
stale_access_token (str | None): The token that was refused.
Returns:
dict: The market data session now in force.
"""
if LoginRecord.market_data_token == stale_access_token:
self.log_in_to_market_data()
print(f'Stand-in market data login issued {LoginRecord.market_data_token}')
else:
print('The market data token was already replaced; no login.')
return self.market_data_session()
class PinnedResponse:
"""A stand-in for the `urllib3` response of one XTS request.
Attributes:
status (int): The HTTP status.
data (bytes): The body.
"""
def __init__(self, status, text):
"""Keeps the answer.
Args:
status (int): The HTTP status.
text (str): The body as text.
Returns:
None: This method returns nothing.
"""
self.status = status
self.data = text.encode()
class PinnedPool:
"""A stand-in for `urllib3.HTTPSConnectionPool` that answers from a shared script.
Attributes:
answers (list): The answers still to give, shared by every pool.
"""
def __init__(self, answers):
"""Keeps the shared answers.
Args:
answers (list): The answers still to give, shared by every pool.
Returns:
None: This method returns nothing.
"""
self.answers = answers
def request(self, method, path, body=None, headers=None, timeout=None):
"""Prints the request and gives the next scripted answer.
Args:
method (str): The HTTP method.
path (str): The path, with its query string.
body (str | None): The JSON body.
headers (dict | None): The request headers.
timeout (int | None): The request timeout in seconds.
Returns:
PinnedResponse: The answer.
"""
shown = path.split('?')[0]
if body is not None:
code = json.loads(body).get('xtsMessageCode')
shown = f'{shown} for message code {code}'
answer = self.answers.pop(0)
print(f'HTTPS {method} {shown}: HTTP {answer.status}')
return answer
class StandInUrllib3:
"""A stand-in for the `urllib3` package whose pools answer from one script.
Attributes:
answers (list): The answers, in the order requests are made.
"""
def __init__(self, answers):
"""Keeps the answers.
Args:
answers (list): The answers, in the order requests are made.
Returns:
None: This method returns nothing.
"""
self.answers = answers
def HTTPSConnectionPool(self, host, port, assert_fingerprint=None):
"""Builds a pool, with the same signature as the real class.
Args:
host (str): The host.
port (int): The port.
assert_fingerprint (str | None): The certificate pin.
Returns:
PinnedPool: The pool.
"""
return PinnedPool(self.answers)
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`, where a frame may also be a function to call at that point.
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.split('?')[0]} with Engine.IO session {self.url.split('sid=')[1].split('&')[0]}")
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
if callable(frame):
frame()
continue
self.handlers['on_message'](self, frame)
self.handlers['on_close'](self, 1000, 'normal closure')
def send(self, data, opcode=None):
"""Prints a message the socket sends to the broker.
Args:
data (str | bytes): The message.
opcode (int | None): The frame type, when the socket names one.
Returns:
None: This method returns nothing.
"""
print(f'Sent: {data}')
def close(self, **options):
"""Notes that the socket asked to close.
Args:
**options (dict): Any close status the socket passes.
Returns:
None: This method returns nothing.
"""
self.closed = True
class ScriptedWebsocketModule:
"""A stand-in for the `websocket` package that hands out scripted connections in turn.
Attributes:
plans (list): What each connection does, in the order the socket connects.
when_finished (callable | None): Called when the last connection is handed out, so the socket stops after it.
"""
def __init__(self, plans):
"""Keeps the connection plans.
Args:
plans (list): What each connection does, in the order the socket connects.
Returns:
None: This method returns nothing.
"""
self.plans = plans
self.when_finished = None
def WebSocketApp(self, url, **handlers):
"""Builds the next scripted connection, closing the socket after the last one.
Args:
url (str): The feed URL.
**handlers (dict): The socket's `on_open`, `on_message`, `on_error` and `on_close` callbacks.
Returns:
ScriptedConnection: The connection.
"""
plan = self.plans.pop(0)
if not self.plans:
self.when_finished()
return ScriptedConnection(url, plan, handlers)
class RefusedTokenAndStaleSubscriptionExample:
"""Runs a Wisdom Capital quotes socket through a refused handshake and a stale subscription.
Attributes:
socket (WisdomCapitalQuotesSocket): The socket being shown.
"""
def __init__(self):
"""Builds the answers, 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.wisdom_capital')
api_module.WisdomCapitalAPI = StandInWisdomCapitalAPI
sys.modules['stock_brokers.api.wisdom_capital'] = api_module
refused = {
'type': 'error',
'code': 'e-session-0001',
'description': 'Invalid Token',
}
already_subscribed = {
'type': 'error',
'code': 'e-session-0002',
'description': 'Instrument Already Subscribed !',
}
success = {
'type': 'success',
}
answers = [
PinnedResponse(400, json.dumps(refused)),
self.handshake_answer('engine-session-2'),
PinnedResponse(400, json.dumps(already_subscribed)),
PinnedResponse(200, json.dumps(success)),
self.subscription_answer([
self.touchline(2950.5),
]),
self.subscription_answer([]),
]
sys.modules['urllib3'] = StandInUrllib3(answers)
plans = [
{
'frames': [
'3probe',
],
},
]
websocket_module = ScriptedWebsocketModule(plans)
sys.modules['websocket'] = websocket_module
logger = logging.getLogger('wisdom_capital.quotes')
tokens = [
'1:2885',
]
session = WisdomCapitalMarketDataSession(logger)
self.socket = WisdomCapitalQuotesSocket('socket_0', tokens, {}, session, self.print_ticks, logger)
self.socket.MIN_BACKOFF_SECONDS = 0
websocket_module.when_finished = self.socket.close
def handshake_answer(self, session_id):
"""An Engine.IO polling handshake answer, framed with its length prefix as XTS sends it.
Args:
session_id (str): The Engine.IO session id.
Returns:
PinnedResponse: The answer.
"""
handshake = {
'sid': session_id,
'upgrades': [
'websocket',
],
'pingInterval': 25000,
'pingTimeout': 20000,
}
text = json.dumps(handshake)
return PinnedResponse(200, f'{len(text) + 1}:0{text}2:40')
def touchline(self, last_price):
"""An XTS touchline for NSE RELIANCE (segment 1, instrument 2885), with XTS's misspelt last traded quantity.
Args:
last_price (float): The last traded price.
Returns:
dict: The touchline.
"""
return {
'ExchangeSegment': 1,
'ExchangeInstrumentID': 2885,
'LastTradedPrice': last_price,
'LastTradedQunatity': 25,
'TotalTradedQuantity': 1204500,
'AverageTradedPrice': 2949.0,
'Open': 2930.0,
'High': 2960.0,
'Low': 2925.0,
'Close': 2910.0,
'LastTradedTime': 1790311528 - 315513000,
'ExchangeTimeStamp': 1790311529 - 315513000,
'BidInfo': {
'Size': 100,
'Price': 2950.0,
'TotalOrders': 3,
},
'AskInfo': {
'Size': 120,
'Price': 2950.5,
'TotalOrders': 4,
},
}
def depth_event(self, last_price):
"""A socket.io market depth event whose quote nests its touchline, with padded zero rows.
Args:
last_price (float): The last traded price.
Returns:
str: The frame.
"""
bids = []
asks = []
for level in range(5):
bids.append({
'Size': 100 + level,
'Price': 2950.0 - level * 0.05,
'TotalOrders': 3,
})
asks.append({
'Size': 200 + level,
'Price': 2951.0 + level * 0.05,
'TotalOrders': 4,
})
bids.append({
'Size': 0,
'Price': 0,
'TotalOrders': 0,
})
quote = {
'ExchangeSegment': 1,
'ExchangeInstrumentID': 2885,
'Bids': bids,
'Asks': asks,
'Touchline': self.touchline(last_price),
}
event = [
'1502-json-full',
json.dumps(quote),
]
return '42' + json.dumps(event)
def subscription_answer(self, quotes):
"""A successful subscription answer carrying the current snapshot as JSON strings.
Args:
quotes (list): The quotes in the snapshot.
Returns:
PinnedResponse: The answer.
"""
encoded = []
for quote in quotes:
encoded.append(json.dumps(quote))
answer = {
'type': 'success',
'result': {
'listQuotes': encoded,
},
}
return PinnedResponse(200, json.dumps(answer))
def print_ticks(self, ticks):
"""Prints the main fields of each tick the socket handed on.
Args:
ticks (list): The ticks.
Returns:
None: This method returns nothing.
"""
for tick in ticks:
change = tick['change']
if change is not None:
change = round(change, 4)
print(f"Tick {tick['id']} ({tick['instrument_token']}) on {tick['exchange']} in {tick['mode']} mode")
print(f" last_price={tick['last_price']} last_quantity={tick['last_quantity']} volume={tick['volume']} change={change}")
print(f" last_trade_time={tick['last_trade_time']} exchange_timestamp={tick['exchange_timestamp']}")
if tick['depth']['buy']:
print(f" best bid={tick['depth']['buy'][0]} best offer={tick['depth']['sell'][0]} levels={len(tick['depth']['buy'])}")
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}')
print(f'Market data logins: {LoginRecord.market_data_logins}')
if __name__ == '__main__':
RefusedTokenAndStaleSubscriptionExample().run()
It prints:
Stand-in login number 1
HTTPS GET /apimarketdata/socket.io/: HTTP 400
ERROR socket_0 connection failed: RuntimeError: Engine.IO handshake failed with 400: {"type": "error", "code": "e-session-0001", "description": "Invalid Token"}
WARNING Replacing the Wisdom Capital market data token.
Stand-in market data login issued market-data-token-2
WARNING socket_0 disconnected. Reconnecting in 0 seconds.
HTTPS GET /apimarketdata/socket.io/: HTTP 200
Connecting to wss://trade.wisdomcapital.in/apimarketdata/socket.io/ with Engine.IO session engine-session-2
Sent: 2probe
Sent: 5
HTTPS POST /apimarketdata/instruments/subscription for message code 1501: HTTP 400
INFO socket_0: message code 1501 is still subscribed from an earlier connection. Dropping that subscription and asking again.
HTTPS PUT /apimarketdata/instruments/subscription for message code 1501: HTTP 200
HTTPS POST /apimarketdata/instruments/subscription for message code 1501: HTTP 200
HTTPS POST /apimarketdata/instruments/subscription for message code 1502: HTTP 200
Tick 1:2885 (1:2885) on NSECM in full mode
last_price=2950.5 last_quantity=25 volume=1204500 change=1.3918
last_trade_time=1790311528 exchange_timestamp=1790311529
best bid={'quantity': 100, 'price': 2950.0, 'orders': 3} best offer={'quantity': 120, 'price': 2950.5, 'orders': 4} levels=1
INFO socket_0 subscribed to 1 instrument(s).
INFO socket_0 closed with status 1000: normal closure
INFO socket_0 stopped.
Gave up: False
Market data logins: 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/wisdom_capital.py
Opens an Engine.IO session with the market data token in force now, attaches the websocket and blocks until it closes.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
When the Engine.IO handshake is refused. |
Source code in stock_brokers/websockets/wisdom_capital.py
Replaces the refused market data token, unless another socket or process already has.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Checks the pinned certificate, then begins the Engine.IO upgrade.
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/wisdom_capital.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/wisdom_capital.py
Stops this connection's heartbeat and 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. |
required |
message
|
str | None
|
The close reason. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/wisdom_capital.py
Handles Engine.IO control frames and hands on the quotes each socket.io event carries.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The connection the frame arrived on. |
required |
message
|
bytes | str
|
The frame. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/wisdom_capital.py
Subscribes the batch over REST for touchline and depth, and hands on the snapshot the answer carries.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The open connection, closed when the token is refused. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/wisdom_capital.py
Asks for one message code's subscription.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
instruments
|
list[dict]
|
The batch as XTS wants it, segment and instrument id per entry. |
required |
code
|
int
|
The XTS message code, touchline or market depth. |
required |
Returns:
| Type | Description |
|---|---|
tuple
|
The response and its decoded body. |
Source code in stock_brokers/websockets/wisdom_capital.py
Merges the quotes in a frame's socket.io event into their instruments' state.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message
|
str
|
The frame; only |
required |
Returns:
| Type | Description |
|---|---|
list[str]
|
The tokens whose state the event updated, each once, in order. |
Source code in stock_brokers/websockets/wisdom_capital.py
Merges an XTS quote into its instrument's state, a depth message's nested touchline included.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
quote
|
object
|
A decoded XTS quote. |
required |
Returns:
| Type | Description |
|---|---|
str | None
|
The instrument's token, or None for something that is not a quote. |
Source code in stock_brokers/websockets/wisdom_capital.py
Builds a normalized tick from an instrument's merged state.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
The instrument's |
required |
Returns:
| Type | Description |
|---|---|
dict
|
The tick. |
Source code in stock_brokers/websockets/wisdom_capital.py
Depth levels from an XTS bid or ask list, without the zero rows XTS pads the book with.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
rows
|
list | None
|
The depth rows. |
required |
Returns:
| Type | Description |
|---|---|
list[dict]
|
The levels. |
Source code in stock_brokers/websockets/wisdom_capital.py
An XTS timestamp on the true Unix epoch.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
timestamp
|
int | None
|
The timestamp XTS sent. |
required |
Returns:
| Type | Description |
|---|---|
int | None
|
The epoch, or None for a missing one. |
Source code in stock_brokers/websockets/wisdom_capital.py
Hands on the ticks that carry a price or a book.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
ticks
|
list[dict]
|
The ticks. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/wisdom_capital.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
Bases: BrokerWebsocket
Wisdom Capital's XTS interactive socket, which carries the account's order and position events.
Sets up the order updates socket.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
session
|
WisdomCapitalInteractiveSession
|
The shared interactive login. |
required |
on_updates
|
callable
|
Called on this socket's thread with an event's orders, its positions and the |
required |
logger
|
Logger
|
Where the connection reports. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example programs
examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalOrderUpdatesSocket/example_1_orders_positions_and_trades.py
"""Joins Wisdom Capital's XTS interactive namespace and prints the order and position events it hands on.
`WisdomCapitalOrderUpdatesSocket` joins the interactive namespace with the token every Wisdom Capital process shares, and the user id read from that token's JWT payload, adding `apiType=INTERACTIVE` so XTS pushes the account's order, trade and position events. It hands on an order event's orders and a position event's positions, each exactly as XTS sent them, and passes over a trade event because the order event that follows carries the fill. It answers a server ping `2` with `3`.
This program scripts one connection delivering an open order, a trade, a position, a server ping and the order filled. No real socket or HTTPS request is made: a stand-in `urllib3` package answers the Engine.IO handshake, a stand-in `websocket` package plays the frames and prints what the socket sends, and a stand-in `stock_brokers.api.wisdom_capital` module lets the real `WisdomCapitalInteractiveSession` supply a JWT token without logging in.
Notice that the trade event leaves no line, and that orders and positions reach the callback in separate calls.
Run it from the project root:
python examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalOrderUpdatesSocket/example_1_orders_positions_and_trades.py
"""
import base64
import json
import logging
import sys
import types
from stock_brokers.websockets.wisdom_capital import (
WisdomCapitalInteractiveSession,
WisdomCapitalOrderUpdatesSocket,
)
class LoginRecord:
"""The Wisdom Capital login state every stand-in API object reads, standing in for the Redis `last_login` hash.
Attributes:
market_data_token (str | None): The market data token in force now.
interactive_user_id (str | None): The user id inside the interactive token in force now.
interactive_session (str | None): The session inside the interactive token in force now.
login_count (int): How many interactive logins the stand-in API has made.
market_data_logins (int): How many market data logins the stand-in API has made.
market_data_fails (bool): Whether the market data login fails.
"""
market_data_token = None
interactive_user_id = 'WC0001'
interactive_session = None
login_count = 0
market_data_logins = 0
market_data_fails = False
class StandInWisdomCapitalAPI:
"""A stand-in for `WisdomCapitalAPI` whose construction is a pretend login to both XTS applications.
Attributes:
market_data_session_error (Exception | None): The market data login failure, recorded rather than raised, as the real class does.
"""
def __init__(self):
"""Logs in, which here only issues numbered tokens.
Returns:
None: This method returns nothing.
"""
LoginRecord.login_count = LoginRecord.login_count + 1
LoginRecord.interactive_session = f'interactive-{LoginRecord.login_count}'
print(f'Stand-in login number {LoginRecord.login_count}')
self.market_data_session_error = None
if LoginRecord.market_data_fails:
self.market_data_session_error = RuntimeError('the market data login was refused: Invalid secret key')
elif LoginRecord.market_data_token is None:
self.log_in_to_market_data()
def log_in_to_market_data(self):
"""Issues the next numbered market data token.
Returns:
None: This method returns nothing.
"""
LoginRecord.market_data_logins = LoginRecord.market_data_logins + 1
LoginRecord.market_data_token = f'market-data-token-{LoginRecord.market_data_logins}'
def _current_login(self):
"""The interactive login in force now, its token a JWT carrying the user id as XTS signs it.
Returns:
dict: The access token.
"""
if LoginRecord.interactive_session is None:
return {
'access_token': None,
}
claims = {
'session': LoginRecord.interactive_session,
}
if LoginRecord.interactive_user_id is not None:
claims['userID'] = LoginRecord.interactive_user_id
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
return {
'access_token': f'header.{payload}.signature',
}
def market_data_session(self):
"""The market data token and user id in force now.
Returns:
dict: `access_token` and `user_id`.
"""
return {
'access_token': LoginRecord.market_data_token,
'user_id': 'WCMD01',
}
def replace_market_data_session(self, stale_access_token=None):
"""Logs in to market data again, unless the token in force is no longer the stale one.
Args:
stale_access_token (str | None): The token that was refused.
Returns:
dict: The market data session now in force.
"""
if LoginRecord.market_data_token == stale_access_token:
self.log_in_to_market_data()
print(f'Stand-in market data login issued {LoginRecord.market_data_token}')
else:
print('The market data token was already replaced; no login.')
return self.market_data_session()
class PinnedResponse:
"""A stand-in for the `urllib3` response of one XTS request.
Attributes:
status (int): The HTTP status.
data (bytes): The body.
"""
def __init__(self, status, text):
"""Keeps the answer.
Args:
status (int): The HTTP status.
text (str): The body as text.
Returns:
None: This method returns nothing.
"""
self.status = status
self.data = text.encode()
class PinnedPool:
"""A stand-in for `urllib3.HTTPSConnectionPool` that answers from a shared script.
Attributes:
answers (list): The answers still to give, shared by every pool.
"""
def __init__(self, answers):
"""Keeps the shared answers.
Args:
answers (list): The answers still to give, shared by every pool.
Returns:
None: This method returns nothing.
"""
self.answers = answers
def request(self, method, path, body=None, headers=None, timeout=None):
"""Prints the request and gives the next scripted answer.
Args:
method (str): The HTTP method.
path (str): The path, with its query string.
body (str | None): The JSON body.
headers (dict | None): The request headers.
timeout (int | None): The request timeout in seconds.
Returns:
PinnedResponse: The answer.
"""
shown = path.split('?')[0]
if body is not None:
code = json.loads(body).get('xtsMessageCode')
shown = f'{shown} for message code {code}'
answer = self.answers.pop(0)
print(f'HTTPS {method} {shown}: HTTP {answer.status}')
return answer
class StandInUrllib3:
"""A stand-in for the `urllib3` package whose pools answer from one script.
Attributes:
answers (list): The answers, in the order requests are made.
"""
def __init__(self, answers):
"""Keeps the answers.
Args:
answers (list): The answers, in the order requests are made.
Returns:
None: This method returns nothing.
"""
self.answers = answers
def HTTPSConnectionPool(self, host, port, assert_fingerprint=None):
"""Builds a pool, with the same signature as the real class.
Args:
host (str): The host.
port (int): The port.
assert_fingerprint (str | None): The certificate pin.
Returns:
PinnedPool: The pool.
"""
return PinnedPool(self.answers)
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`, where a frame may also be a function to call at that point.
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.split('?')[0]} with Engine.IO session {self.url.split('sid=')[1].split('&')[0]}")
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
if callable(frame):
frame()
continue
self.handlers['on_message'](self, frame)
self.handlers['on_close'](self, 1000, 'normal closure')
def send(self, data, opcode=None):
"""Prints a message the socket sends to the broker.
Args:
data (str | bytes): The message.
opcode (int | None): The frame type, when the socket names one.
Returns:
None: This method returns nothing.
"""
print(f'Sent: {data}')
def close(self, **options):
"""Notes that the socket asked to close.
Args:
**options (dict): Any close status the socket passes.
Returns:
None: This method returns nothing.
"""
self.closed = True
class ScriptedWebsocketModule:
"""A stand-in for the `websocket` package that hands out scripted connections in turn.
Attributes:
plans (list): What each connection does, in the order the socket connects.
when_finished (callable | None): Called when the last connection is handed out, so the socket stops after it.
"""
def __init__(self, plans):
"""Keeps the connection plans.
Args:
plans (list): What each connection does, in the order the socket connects.
Returns:
None: This method returns nothing.
"""
self.plans = plans
self.when_finished = None
def WebSocketApp(self, url, **handlers):
"""Builds the next scripted connection, closing the socket after the last one.
Args:
url (str): The feed URL.
**handlers (dict): The socket's `on_open`, `on_message`, `on_error` and `on_close` callbacks.
Returns:
ScriptedConnection: The connection.
"""
plan = self.plans.pop(0)
if not self.plans:
self.when_finished()
return ScriptedConnection(url, plan, handlers)
class OrdersPositionsAndTradesExample:
"""Runs the Wisdom Capital order update socket against one scripted connection.
Attributes:
socket (WisdomCapitalOrderUpdatesSocket): The socket being shown.
"""
def __init__(self):
"""Builds the answers, 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.wisdom_capital')
api_module.WisdomCapitalAPI = StandInWisdomCapitalAPI
sys.modules['stock_brokers.api.wisdom_capital'] = api_module
sys.modules['urllib3'] = StandInUrllib3([
self.handshake_answer('engine-session-1'),
])
trade = [
'trade',
json.dumps({
'AppOrderID': 1100012345,
'LastTradedQuantity': 10,
}),
]
position = [
'position',
{
'TradingSymbol': 'INFY',
'Quantity': 10,
},
]
plans = [
{
'frames': [
'3probe',
self.order_event('New', 0),
'42' + json.dumps(trade),
'42' + json.dumps(position),
'2',
self.order_event('Filled', 10),
],
},
]
websocket_module = ScriptedWebsocketModule(plans)
sys.modules['websocket'] = websocket_module
logger = logging.getLogger('wisdom_capital.orders')
self.socket = WisdomCapitalOrderUpdatesSocket(WisdomCapitalInteractiveSession(logger), self.print_updates, logger)
websocket_module.when_finished = self.socket.close
def handshake_answer(self, session_id):
"""An Engine.IO polling handshake answer, framed with its length prefix as XTS sends it.
Args:
session_id (str): The Engine.IO session id.
Returns:
PinnedResponse: The answer.
"""
handshake = {
'sid': session_id,
'upgrades': [
'websocket',
],
'pingInterval': 25000,
'pingTimeout': 20000,
}
text = json.dumps(handshake)
return PinnedResponse(200, f'{len(text) + 1}:0{text}2:40')
def order_event(self, status, filled_quantity):
"""A socket.io order event for an INFY buy, its payload a JSON string as XTS sends it.
Args:
status (str): XTS's order status.
filled_quantity (int): How much has filled.
Returns:
str: The frame.
"""
order = {
'AppOrderID': 1100012345,
'OrderStatus': status,
'TradingSymbol': 'INFY',
'OrderSide': 'BUY',
'OrderQuantity': 10,
'CumulativeQuantity': filled_quantity,
}
event = [
'order',
json.dumps(order),
]
return '42' + json.dumps(event)
def print_updates(self, orders, positions, received_at):
"""Prints each order and position the socket handed on.
Args:
orders (list): The event's orders, exactly as XTS sent them.
positions (list): The event's positions, exactly as XTS sent them.
received_at (datetime.datetime): When the event arrived, which is not printed.
Returns:
None: This method returns nothing.
"""
for order in orders:
print(f"Order {order['AppOrderID']} {order['OrderStatus']}: {order['OrderSide']} {order['CumulativeQuantity']} of {order['OrderQuantity']} {order['TradingSymbol']}")
for position in positions:
print(f"Position {position['TradingSymbol']}: net quantity {position['Quantity']}")
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__':
OrdersPositionsAndTradesExample().run()
It prints:
Stand-in login number 1
HTTPS GET /interactive/socket.io/: HTTP 200
Connecting to wss://trade.wisdomcapital.in/interactive/socket.io/ with Engine.IO session engine-session-1
Sent: 2probe
Sent: 5
INFO Joined the interactive namespace. Waiting for order and position updates.
Order 1100012345 New: BUY 0 of 10 INFY
Position INFY: net quantity 10
Sent: 3
Order 1100012345 Filled: BUY 10 of 10 INFY
INFO Order updates socket closed with status 1000: normal closure
INFO Order updates socket stopped.
Gave up: False
examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalOrderUpdatesSocket/example_2_logged_out_by_a_new_login.py
"""Shows Wisdom Capital's order update socket reconnecting with the new token after XTS logs it out.
When another login replaces the interactive token, XTS sends the old connection a `logout` event and then leaves it open while delivering nothing. `WisdomCapitalOrderUpdatesSocket` closes it, and its own `run_forever` waits for the replacing login to reach the shared `last_login` hash before reconnecting with it; only if the token is still the logged-out one after ten seconds does it treat the session as refused and log in again.
In this program another process has already stored its new login by the time the logout arrives, so the wait ends at once and no login is made here. The second connection delivers one order. No real socket or HTTPS request is made: stand-in `urllib3` and `websocket` packages play the answers and frames, and a stand-in `stock_brokers.api.wisdom_capital` module holds the shared login. The backoff is set to zero seconds.
Notice the logout warning, that the second handshake carries a different Engine.IO session, and that only one login was made in total.
Run it from the project root:
python examples/stock_brokers/websockets/wisdom_capital/WisdomCapitalOrderUpdatesSocket/example_2_logged_out_by_a_new_login.py
"""
import base64
import json
import logging
import sys
import types
from stock_brokers.websockets.wisdom_capital import (
WisdomCapitalInteractiveSession,
WisdomCapitalOrderUpdatesSocket,
)
class LoginRecord:
"""The Wisdom Capital login state every stand-in API object reads, standing in for the Redis `last_login` hash.
Attributes:
market_data_token (str | None): The market data token in force now.
interactive_user_id (str | None): The user id inside the interactive token in force now.
interactive_session (str | None): The session inside the interactive token in force now.
login_count (int): How many interactive logins the stand-in API has made.
market_data_logins (int): How many market data logins the stand-in API has made.
market_data_fails (bool): Whether the market data login fails.
"""
market_data_token = None
interactive_user_id = 'WC0001'
interactive_session = None
login_count = 0
market_data_logins = 0
market_data_fails = False
class StandInWisdomCapitalAPI:
"""A stand-in for `WisdomCapitalAPI` whose construction is a pretend login to both XTS applications.
Attributes:
market_data_session_error (Exception | None): The market data login failure, recorded rather than raised, as the real class does.
"""
def __init__(self):
"""Logs in, which here only issues numbered tokens.
Returns:
None: This method returns nothing.
"""
LoginRecord.login_count = LoginRecord.login_count + 1
LoginRecord.interactive_session = f'interactive-{LoginRecord.login_count}'
print(f'Stand-in login number {LoginRecord.login_count}')
self.market_data_session_error = None
if LoginRecord.market_data_fails:
self.market_data_session_error = RuntimeError('the market data login was refused: Invalid secret key')
elif LoginRecord.market_data_token is None:
self.log_in_to_market_data()
def log_in_to_market_data(self):
"""Issues the next numbered market data token.
Returns:
None: This method returns nothing.
"""
LoginRecord.market_data_logins = LoginRecord.market_data_logins + 1
LoginRecord.market_data_token = f'market-data-token-{LoginRecord.market_data_logins}'
def _current_login(self):
"""The interactive login in force now, its token a JWT carrying the user id as XTS signs it.
Returns:
dict: The access token.
"""
if LoginRecord.interactive_session is None:
return {
'access_token': None,
}
claims = {
'session': LoginRecord.interactive_session,
}
if LoginRecord.interactive_user_id is not None:
claims['userID'] = LoginRecord.interactive_user_id
payload = base64.urlsafe_b64encode(json.dumps(claims).encode()).decode().rstrip('=')
return {
'access_token': f'header.{payload}.signature',
}
def market_data_session(self):
"""The market data token and user id in force now.
Returns:
dict: `access_token` and `user_id`.
"""
return {
'access_token': LoginRecord.market_data_token,
'user_id': 'WCMD01',
}
def replace_market_data_session(self, stale_access_token=None):
"""Logs in to market data again, unless the token in force is no longer the stale one.
Args:
stale_access_token (str | None): The token that was refused.
Returns:
dict: The market data session now in force.
"""
if LoginRecord.market_data_token == stale_access_token:
self.log_in_to_market_data()
print(f'Stand-in market data login issued {LoginRecord.market_data_token}')
else:
print('The market data token was already replaced; no login.')
return self.market_data_session()
class PinnedResponse:
"""A stand-in for the `urllib3` response of one XTS request.
Attributes:
status (int): The HTTP status.
data (bytes): The body.
"""
def __init__(self, status, text):
"""Keeps the answer.
Args:
status (int): The HTTP status.
text (str): The body as text.
Returns:
None: This method returns nothing.
"""
self.status = status
self.data = text.encode()
class PinnedPool:
"""A stand-in for `urllib3.HTTPSConnectionPool` that answers from a shared script.
Attributes:
answers (list): The answers still to give, shared by every pool.
"""
def __init__(self, answers):
"""Keeps the shared answers.
Args:
answers (list): The answers still to give, shared by every pool.
Returns:
None: This method returns nothing.
"""
self.answers = answers
def request(self, method, path, body=None, headers=None, timeout=None):
"""Prints the request and gives the next scripted answer.
Args:
method (str): The HTTP method.
path (str): The path, with its query string.
body (str | None): The JSON body.
headers (dict | None): The request headers.
timeout (int | None): The request timeout in seconds.
Returns:
PinnedResponse: The answer.
"""
shown = path.split('?')[0]
if body is not None:
code = json.loads(body).get('xtsMessageCode')
shown = f'{shown} for message code {code}'
answer = self.answers.pop(0)
print(f'HTTPS {method} {shown}: HTTP {answer.status}')
return answer
class StandInUrllib3:
"""A stand-in for the `urllib3` package whose pools answer from one script.
Attributes:
answers (list): The answers, in the order requests are made.
"""
def __init__(self, answers):
"""Keeps the answers.
Args:
answers (list): The answers, in the order requests are made.
Returns:
None: This method returns nothing.
"""
self.answers = answers
def HTTPSConnectionPool(self, host, port, assert_fingerprint=None):
"""Builds a pool, with the same signature as the real class.
Args:
host (str): The host.
port (int): The port.
assert_fingerprint (str | None): The certificate pin.
Returns:
PinnedPool: The pool.
"""
return PinnedPool(self.answers)
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`, where a frame may also be a function to call at that point.
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.split('?')[0]} with Engine.IO session {self.url.split('sid=')[1].split('&')[0]}")
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
if callable(frame):
frame()
continue
self.handlers['on_message'](self, frame)
self.handlers['on_close'](self, 1000, 'normal closure')
def send(self, data, opcode=None):
"""Prints a message the socket sends to the broker.
Args:
data (str | bytes): The message.
opcode (int | None): The frame type, when the socket names one.
Returns:
None: This method returns nothing.
"""
print(f'Sent: {data}')
def close(self, **options):
"""Notes that the socket asked to close.
Args:
**options (dict): Any close status the socket passes.
Returns:
None: This method returns nothing.
"""
self.closed = True
class ScriptedWebsocketModule:
"""A stand-in for the `websocket` package that hands out scripted connections in turn.
Attributes:
plans (list): What each connection does, in the order the socket connects.
when_finished (callable | None): Called when the last connection is handed out, so the socket stops after it.
"""
def __init__(self, plans):
"""Keeps the connection plans.
Args:
plans (list): What each connection does, in the order the socket connects.
Returns:
None: This method returns nothing.
"""
self.plans = plans
self.when_finished = None
def WebSocketApp(self, url, **handlers):
"""Builds the next scripted connection, closing the socket after the last one.
Args:
url (str): The feed URL.
**handlers (dict): The socket's `on_open`, `on_message`, `on_error` and `on_close` callbacks.
Returns:
ScriptedConnection: The connection.
"""
plan = self.plans.pop(0)
if not self.plans:
self.when_finished()
return ScriptedConnection(url, plan, handlers)
class LoggedOutByANewLoginExample:
"""Runs the Wisdom Capital order update socket through a logout and a reconnect.
Attributes:
socket (WisdomCapitalOrderUpdatesSocket): The socket being shown.
"""
def __init__(self):
"""Builds the answers, 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.wisdom_capital')
api_module.WisdomCapitalAPI = StandInWisdomCapitalAPI
sys.modules['stock_brokers.api.wisdom_capital'] = api_module
sys.modules['urllib3'] = StandInUrllib3([
self.handshake_answer('engine-session-1'),
self.handshake_answer('engine-session-2'),
])
logout = [
'logout',
{
'message': 'Logged in from another device',
},
]
plans = [
{
'frames': [
'3probe',
self.store_other_login,
'42' + json.dumps(logout),
],
},
{
'frames': [
'3probe',
self.order_event('Filled', 10),
],
},
]
websocket_module = ScriptedWebsocketModule(plans)
sys.modules['websocket'] = websocket_module
logger = logging.getLogger('wisdom_capital.orders')
self.socket = WisdomCapitalOrderUpdatesSocket(WisdomCapitalInteractiveSession(logger), self.print_updates, logger)
self.socket.MIN_BACKOFF_SECONDS = 0
websocket_module.when_finished = self.socket.close
def handshake_answer(self, session_id):
"""An Engine.IO polling handshake answer, framed with its length prefix as XTS sends it.
Args:
session_id (str): The Engine.IO session id.
Returns:
PinnedResponse: The answer.
"""
handshake = {
'sid': session_id,
'upgrades': [
'websocket',
],
'pingInterval': 25000,
'pingTimeout': 20000,
}
text = json.dumps(handshake)
return PinnedResponse(200, f'{len(text) + 1}:0{text}2:40')
def order_event(self, status, filled_quantity):
"""A socket.io order event for an INFY buy, its payload a JSON string as XTS sends it.
Args:
status (str): XTS's order status.
filled_quantity (int): How much has filled.
Returns:
str: The frame.
"""
order = {
'AppOrderID': 1100012345,
'OrderStatus': status,
'TradingSymbol': 'INFY',
'OrderSide': 'BUY',
'OrderQuantity': 10,
'CumulativeQuantity': filled_quantity,
}
event = [
'order',
json.dumps(order),
]
return '42' + json.dumps(event)
def store_other_login(self):
"""Stores another process's login, as the login that logged the socket out would, just before XTS sends the logout.
Returns:
None: This method returns nothing.
"""
LoginRecord.interactive_session = 'interactive-from-another-process'
def print_updates(self, orders, positions, received_at):
"""Prints each order and position the socket handed on.
Args:
orders (list): The event's orders, exactly as XTS sent them.
positions (list): The event's positions, exactly as XTS sent them.
received_at (datetime.datetime): When the event arrived, which is not printed.
Returns:
None: This method returns nothing.
"""
for order in orders:
print(f"Order {order['AppOrderID']} {order['OrderStatus']}: {order['OrderSide']} {order['CumulativeQuantity']} of {order['OrderQuantity']} {order['TradingSymbol']}")
for position in positions:
print(f"Position {position['TradingSymbol']}: net quantity {position['Quantity']}")
def run(self):
"""Runs the socket's own 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__':
LoggedOutByANewLoginExample().run()
It prints:
Stand-in login number 1
HTTPS GET /interactive/socket.io/: HTTP 200
Connecting to wss://trade.wisdomcapital.in/interactive/socket.io/ with Engine.IO session engine-session-1
Sent: 2probe
Sent: 5
INFO Joined the interactive namespace. Waiting for order and position updates.
WARNING Wisdom Capital logged this socket out: {"message": "Logged in from another device"}. Reconnecting with the login now in force.
INFO Order updates socket closed with status 1000: normal closure
WARNING Order updates socket disconnected. Reconnecting in 0 seconds.
HTTPS GET /interactive/socket.io/: HTTP 200
Connecting to wss://trade.wisdomcapital.in/interactive/socket.io/ with Engine.IO session engine-session-2
Sent: 2probe
Sent: 5
INFO Joined the interactive namespace. Waiting for order and position updates.
Order 1100012345 Filled: BUY 10 of 10 INFY
INFO Order updates socket closed with status 1000: normal closure
INFO Order updates socket stopped.
Gave up: False
Logins made: 1
Methods:
| Name | Description |
|---|---|
run_forever |
Connects, and reconnects with backoff, until closed or unable to recover, waiting after a logout for the login that replaced this one. |
close |
Closes the connection and stops reconnecting. |
Source code in stock_brokers/websockets/wisdom_capital.py
Connects, and reconnects with backoff, until closed or unable to recover, waiting after a logout for the login that replaced this one.
This is BrokerWebsocket.run_forever with one step added: after XTS logs the socket out, it waits for the replacing login to reach Redis, and treats a token that is still the logged-out one as a refused session.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example usage
Source code in stock_brokers/websockets/wisdom_capital.py
Waits for the login that logged this socket out to reach the shared last_login hash.
Returns:
| Type | Description |
|---|---|
bool
|
True when the shared login holds a token other than the one this connection used, and False when it still holds that token after |
Source code in stock_brokers/websockets/wisdom_capital.py
Opens an Engine.IO session with the shared login, attaches the websocket and blocks until it closes.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
No usable token is stored, or the Engine.IO handshake is refused. |
Source code in stock_brokers/websockets/wisdom_capital.py
Constructs the API again, without checking whether another process already replaced the token.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/wisdom_capital.py
Checks the pinned certificate, then begins the Engine.IO upgrade.
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/wisdom_capital.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/wisdom_capital.py
Stops this connection's heartbeat and 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. |
required |
message
|
str | None
|
The close reason. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/wisdom_capital.py
Handles Engine.IO control frames and hands on the orders and positions each socket.io event carries.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The connection the frame arrived on. |
required |
message
|
bytes | str
|
The frame. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/wisdom_capital.py
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 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 | |
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. |