Kotak's two websockets: the HSM market feed that Kotak's own Python SDK (neo_api_client) streams from, and the account's order and position update stream.
The login.
KotakSession holds the KotakAPI both sockets read the access token, the session id and the account's host from, afresh on every connect, so a login made by any process is picked up.
The quote sockets log in again one at a time, and only when no other socket or process has already replaced the session that failed; the order socket logs in again whenever it is refused, without that check, as it always has.
Quotes.
KotakQuotesSocket connects to wss://mlhsm.kotaksecurities.com.
Every request and response is a binary frame that begins with its length in two big-endian bytes and then a one-byte frame type.
The connection request (type 1) carries the access token, the session id and the source JS_API as length-prefixed fields.
Kotak answers with a type 1 frame whose status is K for accepted, and with the number of data frames after which it wants an acknowledgement.
Every instrument is subscribed twice, as a scrip topic (sf|EXCHANGE|TOKEN) for its prices and quantities and as a depth topic (dp|EXCHANGE|TOKEN) for its five-level book, up to 100 topics a subscription request (type 4) on channel 1.
Data frames (type 6) begin with a four-byte message number when acknowledgements were asked for, and after that many data frames the socket sends an acknowledgement (type 3) carrying the last message number.
Each data frame holds packets that are either a snapshot (S, 83) - topic id, topic name, the numeric fields as four-byte signed integers in field order, and string fields by id - or an update (U, 85) - topic id and the numeric fields again, where -2147483648 means unchanged.
A topic's latest values are kept, and a whole tick is built for an instrument every time either of its topics changes, once its scrip snapshot has arrived.
A snapshot's trading symbol (string field 54) names an instrument that had no name, and that name is handed to on_name before the frame's ticks.
Prices are integers divided by the topic's multiplier (scrip field 23, depth field 32) times ten to its precision (scrip field 24, depth field 33).
The quote socket gives up only after six more failed connects, not on the first refusal after logging in again.
Order updates.
KotakOrderUpdatesSocket connects to wss://<host>/realtime, where the host is the base_url Kotak assigned the session at login.
After connecting it sends Kotak's connection frame, a raw unquoted string that is deliberately not JSON: {type:cn,Authorization:<access token>,Sid:<session id>,src:WEB}.
Kotak acknowledges it with {"ak":"ok","type":"cn",...} or refuses it with {"ak":"nk","type":"failed to connect","msg":...}; a refusal discards whatever else the same message carried.
From then on the connection carries {"type": "order", "data": {...}} and {"type": "position", "data": {...}} messages with Kotak's abbreviated fields.
Neither socket writes Redis.
The quotes socket hands names to on_name and each data frame's ticks to on_ticks, and the order socket hands each message's orders and positions to on_updates; the scripts in bin/kotak/ do the writing.
Classes:
| Name | Description |
|---|---|
KotakTruncatedFrameError |
A binary frame ended before a field it declared. |
KotakFrameReader |
A cursor over one binary frame from Kotak's HSM feed, reading big-endian fields in order. |
KotakFeedRequests |
Builds the binary requests a client sends to Kotak's HSM feed. |
KotakFeedTopic |
The latest values of one subscribed topic, a scrip or a depth topic of one instrument. |
KotakInstrumentQuote |
One instrument's scrip and depth topics, turned into a normalized tick. |
KotakSession |
The Kotak login every socket authenticates with, logged in again by one socket at a time. |
KotakQuotesSocket |
One Kotak HSM feed websocket carrying the scrip and depth topics of a batch of instruments. |
KotakOrderUpdatesSocket |
Kotak's order and position update websocket, on the host the session was assigned. |
Bases: ValueError
A binary frame ended before a field it declared.
Example programs
examples/stock_brokers/websockets/kotak/KotakTruncatedFrameError/example_1_raising_and_catching.py
"""Raises `KotakTruncatedFrameError` by reading past the end of a frame and catches it.
`KotakTruncatedFrameError` means a binary frame from Kotak's HSM feed ended before a field it declared. `KotakFrameReader` raises it, and the quotes socket catches it for the whole frame, logs a warning and hands on the ticks of the packets it had already read. It is a `ValueError`, so code that treats any malformed value alike can catch it that way.
This program reads a text field whose declared length is longer than the bytes that follow. It needs no stand-ins.
Notice the message naming the frame's length, the size asked for and the position.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakTruncatedFrameError/example_1_raising_and_catching.py
"""
from stock_brokers.websockets.kotak import (
KotakFrameReader,
KotakTruncatedFrameError,
)
class RaisingAndCatchingExample:
"""Reads a text field that runs past the end of its frame.
Attributes:
reader (KotakFrameReader): The reader over the short frame.
"""
def __init__(self):
"""Builds a frame whose text field declares 12 bytes but carries 4.
Returns:
None: This method returns nothing.
"""
self.reader = KotakFrameReader(bytes([12]) + b'RELI')
def run(self):
"""Reads the field and prints the error.
Returns:
None: This method returns nothing.
"""
size = self.reader.read_unsigned(1)
try:
self.reader.read_text(size)
except KotakTruncatedFrameError as error:
print(f'Caught {type(error).__name__}: {error}')
print(f'It is a ValueError: {isinstance(error, ValueError)}')
if __name__ == '__main__':
RaisingAndCatchingExample().run()
It prints:
examples/stock_brokers/websockets/kotak/KotakTruncatedFrameError/example_2_caught_as_a_value_error.py
"""Catches `KotakTruncatedFrameError` as a plain `ValueError` while parsing a snapshot packet cut short.
Because `KotakTruncatedFrameError` subclasses `ValueError`, a caller that already handles malformed values can catch it without naming it. This program parses the start of a snapshot packet, as the quotes socket does, from bytes that stop halfway through the topic name.
It raises the error on purpose with a short packet and catches it as `ValueError`. It needs no stand-ins.
Notice that the topic id was read before the error, and that the class name still shows which error it was.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakTruncatedFrameError/example_2_caught_as_a_value_error.py
"""
from stock_brokers.websockets.kotak import (
KotakFrameReader,
KotakTruncatedFrameError,
)
class CaughtAsAValueErrorExample:
"""Parses a snapshot packet that stops inside its topic name.
Attributes:
reader (KotakFrameReader): The reader over the short packet.
"""
def __init__(self):
"""Builds the start of a snapshot packet whose topic name is cut short.
Returns:
None: This method returns nothing.
"""
name = b'sf|nse_cm|2885'
packet = bytes([83]) + (7).to_bytes(4, 'big', signed=True) + bytes([len(name)]) + name[:6]
self.reader = KotakFrameReader(packet)
def parse(self):
"""Reads the packet type, the topic id and the topic name.
Returns:
tuple: The packet type, the topic id and the topic name.
Raises:
KotakTruncatedFrameError: When the packet ends inside the topic name.
"""
packet_type = self.reader.read_unsigned(1)
topic_id = self.reader.read_signed_integer()
print(f'Read packet type {packet_type} for topic id {topic_id}')
topic_name = self.reader.read_text(self.reader.read_unsigned(1))
return packet_type, topic_id, topic_name
def run(self):
"""Parses the packet and catches the error as a ValueError.
Returns:
None: This method returns nothing.
"""
try:
self.parse()
except ValueError as error:
print(f'Caught as ValueError: {type(error).__name__}: {error}')
print(f'Is a KotakTruncatedFrameError: {isinstance(error, KotakTruncatedFrameError)}')
if __name__ == '__main__':
CaughtAsAValueErrorExample().run()
It prints:
A cursor over one binary frame from Kotak's HSM feed, reading big-endian fields in order.
Attributes:
| Name | Type | Description |
|---|---|---|
position |
int
|
The offset of the next unread byte. |
Starts reading a frame at its first byte.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
data
|
bytes
|
The binary frame. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example programs
examples/stock_brokers/websockets/kotak/KotakFrameReader/example_1_reading_a_connection_response.py
"""Reads a Kotak HSM connection response field by field with `KotakFrameReader`.
`KotakFrameReader` is a cursor over one binary frame from Kotak's HSM feed. Each read takes the next bytes and moves `position` past them: `read_unsigned` reads a big-endian unsigned integer of a given width, `read_signed_integer` a four-byte signed one, and `read_text` a string of single-byte characters. Every frame starts with its length in two bytes and its type in one.
This program builds the connection response Kotak sends when it accepts a session, with an acknowledgement interval of 50, and reads it the way the quotes socket does. It needs no stand-ins.
Notice how `position` advances with each read and ends at the frame's length plus the two bytes of the length itself.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakFrameReader/example_1_reading_a_connection_response.py
"""
from stock_brokers.websockets.kotak import (
KotakFrameReader,
)
class ReadingAConnectionResponseExample:
"""Reads one connection response.
Attributes:
reader (KotakFrameReader): The reader being shown.
"""
def __init__(self):
"""Builds the frame and a reader over it.
Returns:
None: This method returns nothing.
"""
status_field = bytes([1]) + (1).to_bytes(2, 'big') + b'K'
acknowledgement_field = bytes([2]) + (4).to_bytes(2, 'big') + (50).to_bytes(4, 'big')
content = bytes([1, 2]) + status_field + acknowledgement_field
frame = len(content).to_bytes(2, 'big') + content
print(f'Frame: {frame.hex()}')
self.reader = KotakFrameReader(frame)
def run(self):
"""Reads each field and prints it with the reader's position.
Returns:
None: This method returns nothing.
"""
length = self.reader.read_unsigned(2)
frame_type = self.reader.read_unsigned(1)
field_count = self.reader.read_unsigned(1)
print(f'length={length} type={frame_type} fields={field_count} position={self.reader.position}')
field_id = self.reader.read_unsigned(1)
status = self.reader.read_text(self.reader.read_unsigned(2))
print(f'field {field_id}: status={status!r} position={self.reader.position}')
field_id = self.reader.read_unsigned(1)
acknowledge_every = self.reader.read_unsigned(self.reader.read_unsigned(2))
print(f'field {field_id}: acknowledge every {acknowledge_every} data frames position={self.reader.position}')
if __name__ == '__main__':
ReadingAConnectionResponseExample().run()
It prints:
examples/stock_brokers/websockets/kotak/KotakFrameReader/example_2_signed_values_and_the_end_of_a_frame.py
"""Reads signed integers, including Kotak's unchanged marker, and runs a `KotakFrameReader` past the end of its frame.
Kotak's numeric fields are four-byte signed integers, and -2147483648, the smallest one, means that a field in an update packet is unchanged. `read_signed_integer` reads them. Reading past the end of the frame raises `KotakTruncatedFrameError`, which says how long the frame was and where the read started, instead of returning short data.
This program reads a price, a negative change and the unchanged marker, then asks for four more bytes than are left. It needs no stand-ins.
Notice that the failed read leaves `position` where it was.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakFrameReader/example_2_signed_values_and_the_end_of_a_frame.py
"""
from stock_brokers.websockets.kotak import (
KotakFrameReader,
KotakTruncatedFrameError,
)
class SignedValuesAndTheEndOfAFrameExample:
"""Reads signed values and then reads too far.
Attributes:
reader (KotakFrameReader): The reader being shown.
"""
def __init__(self):
"""Builds a frame of three signed integers and two spare bytes.
Returns:
None: This method returns nothing.
"""
values = [
295050,
-1250,
-2147483648,
]
data = b''
for value in values:
data = data + value.to_bytes(4, 'big', signed=True)
self.reader = KotakFrameReader(data + b'\x00\x07')
def run(self):
"""Reads the three values, then a fourth that is not there.
Returns:
None: This method returns nothing.
"""
for _ in range(3):
print(f'Read {self.reader.read_signed_integer()}')
try:
self.reader.read_signed_integer()
except KotakTruncatedFrameError as error:
print(f'KotakTruncatedFrameError: {error}')
print(f'Position after the failed read: {self.reader.position}')
if __name__ == '__main__':
SignedValuesAndTheEndOfAFrameExample().run()
It prints:
Methods:
| Name | Description |
|---|---|
read_unsigned |
Reads an unsigned big-endian integer. |
read_signed_integer |
Reads a four-byte signed big-endian integer. |
read_text |
Reads a string of single-byte characters. |
Source code in stock_brokers/websockets/kotak.py
The next bytes, moving past them.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
size
|
int
|
How many bytes to read. |
required |
Returns:
| Type | Description |
|---|---|
bytes
|
The bytes read. |
Raises:
| Type | Description |
|---|---|
KotakTruncatedFrameError
|
The frame has fewer than |
Source code in stock_brokers/websockets/kotak.py
Reads an unsigned big-endian integer.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
size
|
int
|
Its width in bytes. |
required |
Returns:
| Type | Description |
|---|---|
int
|
The integer. |
Raises:
| Type | Description |
|---|---|
KotakTruncatedFrameError
|
The frame ends first. |
Example usage
Source code in stock_brokers/websockets/kotak.py
Reads a four-byte signed big-endian integer.
Returns:
| Type | Description |
|---|---|
int
|
The integer. |
Raises:
| Type | Description |
|---|---|
KotakTruncatedFrameError
|
The frame ends first. |
Example usage
From example_2_signed_values_and_the_end_of_a_frame.py
print(f'Read {self.reader.read_signed_integer()}')
Source code in stock_brokers/websockets/kotak.py
Reads a string of single-byte characters.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
size
|
int
|
Its length in bytes. |
required |
Returns:
| Type | Description |
|---|---|
str
|
The string. |
Raises:
| Type | Description |
|---|---|
KotakTruncatedFrameError
|
The frame ends first. |
Example usage
From example_1_reading_a_connection_response.py
status = self.reader.read_text(self.reader.read_unsigned(2))
Source code in stock_brokers/websockets/kotak.py
Builds the binary requests a client sends to Kotak's HSM feed.
Example programs
examples/stock_brokers/websockets/kotak/KotakFeedRequests/example_1_connection_and_subscription.py
"""Builds the connection request and the scrip and depth subscriptions a Kotak quotes socket sends.
`KotakFeedRequests` builds the binary requests a client sends to Kotak's HSM feed. Each is a frame: its length in two bytes, a type byte, a field count, and fields of an id, a two-byte length and the value. `connection` (type 1) carries the access token, the session id and the source `JS_API`, and `subscription` (type 4) lists up to 100 topic names, `sf|...` for scrip or `dp|...` for depth, on channel 1.
This program builds the connection request and both subscriptions for two instruments, prints them as hex, and reads the topic names back out of the scrip subscription with `KotakFrameReader`. It needs no stand-ins.
Notice that each instrument becomes the topic `sf|nse_cm|2885` or `sf|nse_cm|1594` in the scrip subscription.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakFeedRequests/example_1_connection_and_subscription.py
"""
from stock_brokers.websockets.kotak import (
KotakFeedRequests,
KotakFrameReader,
)
class ConnectionAndSubscriptionExample:
"""Builds and reads back Kotak feed requests.
Attributes:
requests (KotakFeedRequests): The request builder being shown.
instrument_tokens (list): The instruments to subscribe.
"""
def __init__(self):
"""Builds the request builder and the instruments.
Returns:
None: This method returns nothing.
"""
self.requests = KotakFeedRequests()
self.instrument_tokens = [
'nse_cm|2885',
'nse_cm|1594',
]
def run(self):
"""Prints each request and the topic names inside the scrip subscription.
Returns:
None: This method returns nothing.
"""
print(f"Connection: {self.requests.connection('kotak-token-1', 'kotak-sid-1').hex()}")
scrip = self.requests.subscription('sf', self.instrument_tokens)
print(f'Scrip subscription: {scrip.hex()}')
print(f"Depth subscription: {self.requests.subscription('dp', self.instrument_tokens).hex()}")
reader = KotakFrameReader(scrip)
reader.read_unsigned(2)
print(f'Frame type {reader.read_unsigned(1)} with {reader.read_unsigned(1)} fields')
reader.read_unsigned(1)
reader.read_unsigned(2)
for _ in range(reader.read_unsigned(2)):
print(f' topic {reader.read_text(reader.read_unsigned(1))}')
if __name__ == '__main__':
ConnectionAndSubscriptionExample().run()
It prints:
Connection: 0029010301000d6b6f74616b2d746f6b656e2d3102000b6b6f74616b2d7369642d310300064a535f415049
Scrip subscription: 0029040201002000020e73667c6e73655f636d7c323838350e73667c6e73655f636d7c3135393402000101
Depth subscription: 0029040201002000020e64707c6e73655f636d7c323838350e64707c6e73655f636d7c3135393402000101
Frame type 4 with 2 fields
topic sf|nse_cm|2885
topic sf|nse_cm|1594
examples/stock_brokers/websockets/kotak/KotakFeedRequests/example_2_acknowledgements.py
"""Builds the acknowledgements a Kotak quotes socket sends after every few data frames.
Kotak's connection response says after how many data frames it wants an acknowledgement. The quotes socket counts data frames and, at that count, sends `KotakFeedRequests.acknowledgement` (type 3) carrying the message number of the last data frame, as a four-byte signed field.
This program builds acknowledgements for three message numbers and reads each one back with `KotakFrameReader`. It needs no stand-ins.
Notice that every acknowledgement is ten bytes long and that only the last four change.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakFeedRequests/example_2_acknowledgements.py
"""
from stock_brokers.websockets.kotak import (
KotakFeedRequests,
KotakFrameReader,
)
class AcknowledgementsExample:
"""Builds and reads back acknowledgement frames.
Attributes:
requests (KotakFeedRequests): The request builder being shown.
"""
def __init__(self):
"""Builds the request builder.
Returns:
None: This method returns nothing.
"""
self.requests = KotakFeedRequests()
def run(self):
"""Prints each acknowledgement and the message number read back from it.
Returns:
None: This method returns nothing.
"""
message_numbers = [
2,
50,
100000,
]
for message_number in message_numbers:
frame = self.requests.acknowledgement(message_number)
reader = KotakFrameReader(frame)
reader.read_unsigned(2)
frame_type = reader.read_unsigned(1)
reader.read_unsigned(1)
reader.read_unsigned(1)
reader.read_unsigned(2)
print(f'{frame.hex()} is type {frame_type} acknowledging message {reader.read_signed_integer()}')
if __name__ == '__main__':
AcknowledgementsExample().run()
It prints:
Methods:
| Name | Description |
|---|---|
connection |
Builds the connection request. |
subscription |
Builds a subscription to up to 100 topics on channel 1. |
acknowledgement |
Builds the acknowledgement of data frames up to a message number. |
Builds the connection request.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
access_token
|
str
|
The REST access token from Kotak's login. |
required |
session_id
|
str
|
The session id from Kotak's login. |
required |
Returns:
| Type | Description |
|---|---|
bytes
|
The frame. |
Example usage
From example_1_connection_and_subscription.py
print(f"Connection: {self.requests.connection('kotak-token-1', 'kotak-sid-1').hex()}")
Source code in stock_brokers/websockets/kotak.py
Builds a subscription to up to 100 topics on channel 1.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
topic_prefix
|
str
|
|
required |
instrument_tokens
|
list[str]
|
The |
required |
Returns:
| Type | Description |
|---|---|
bytes
|
The frame. |
Example usage
From example_1_connection_and_subscription.py
scrip = self.requests.subscription('sf', self.instrument_tokens)
From example_1_connection_and_subscription.py
print(f"Depth subscription: {self.requests.subscription('dp', self.instrument_tokens).hex()}")
Source code in stock_brokers/websockets/kotak.py
Builds the acknowledgement of data frames up to a message number.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
message_number
|
int
|
The message number of the last data frame received. |
required |
Returns:
| Type | Description |
|---|---|
bytes
|
The frame. |
Example usage
Source code in stock_brokers/websockets/kotak.py
Joins a frame type and its fields behind the frame's length.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
frame_type
|
int
|
The frame type byte. |
required |
fields
|
list[bytes]
|
The encoded fields, each already carrying its id and length. |
required |
Returns:
| Type | Description |
|---|---|
bytes
|
The whole frame. |
Source code in stock_brokers/websockets/kotak.py
Encodes one field as its id, a two-byte length and its bytes.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
field_id
|
int
|
The field id. |
required |
value
|
bytes
|
The field's value. |
required |
Returns:
| Type | Description |
|---|---|
bytes
|
The encoded field. |
Source code in stock_brokers/websockets/kotak.py
The latest values of one subscribed topic, a scrip or a depth topic of one instrument.
Attributes:
| Name | Type | Description |
|---|---|---|
kind |
str
|
|
instrument_token |
str
|
The instrument's |
strings |
dict[int, str]
|
The topic's string fields by field id, such as its trading symbol. |
Creates a topic from its snapshot's name and strings.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
kind
|
str
|
|
required |
instrument_token
|
str
|
The instrument's |
required |
strings
|
dict[int, str]
|
The snapshot's string fields by field id. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example programs
examples/stock_brokers/websockets/kotak/KotakFeedTopic/example_1_scrip_snapshot_and_update.py
"""Applies a scrip snapshot and an update to a `KotakFeedTopic` and reads its numbers and prices.
`KotakFeedTopic` keeps the latest values of one subscribed topic. `apply` takes a snapshot's or an update's numeric fields in field order, ignoring -2147483648, which Kotak sends for an unchanged field. `number` returns a field as sent, and `price` divides a price field by the topic's multiplier times ten to its precision, which for a scrip topic are fields 23 and 24.
This program applies a snapshot of an NSE equity with prices in paise (multiplier 1, precision 2), then an update that changes only the last price (field 5). It needs no stand-ins.
Notice that the volume (field 4) keeps its snapshot value after the update, and that a field never sent reads as None.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakFeedTopic/example_1_scrip_snapshot_and_update.py
"""
from stock_brokers.websockets.kotak import (
KotakFeedTopic,
)
class ScripSnapshotAndUpdateExample:
"""Updates one scrip topic and reads it.
Attributes:
topic (KotakFeedTopic): The topic being shown.
"""
def __init__(self):
"""Builds a scrip topic for RELIANCE with its trading symbol.
Returns:
None: This method returns nothing.
"""
strings = {
54: 'RELIANCE-EQ',
}
self.topic = KotakFeedTopic('sf', 'nse_cm|2885', strings)
def run(self):
"""Applies the snapshot and the update and prints the fields after each.
Returns:
None: This method returns nothing.
"""
snapshot = []
for _ in range(25):
snapshot.append(0)
snapshot[4] = 1204500
snapshot[5] = 295050
snapshot[21] = 291000
snapshot[23] = 1
snapshot[24] = 2
self.topic.apply(snapshot)
print(f'{self.topic.kind} topic for {self.topic.instrument_token} named {self.topic.strings[54]}')
print(f'After the snapshot: last price {self.topic.price(5)}, close {self.topic.price(21)}, volume {self.topic.number(4)}')
unchanged = -2147483648
update = [
unchanged,
unchanged,
unchanged,
unchanged,
unchanged,
295175,
]
self.topic.apply(update)
print(f'After the update: last price {self.topic.price(5)}, volume {self.topic.number(4)}')
print(f'A field never sent: {self.topic.number(40)} and its price {self.topic.price(40)}')
if __name__ == '__main__':
ScripSnapshotAndUpdateExample().run()
It prints:
examples/stock_brokers/websockets/kotak/KotakFeedTopic/example_2_depth_and_currency_scale.py
"""Reads prices from a depth topic and from a currency scrip topic, each with its own scale.
A depth topic keeps its multiplier and precision in fields 32 and 33 instead of the scrip topic's 23 and 24. `KotakFeedTopic.price` picks the right pair from the topic's kind, falls back to a multiplier of 1 and a precision of 2 when the feed has not sent them, and rounds to the precision so the division leaves no floating point noise.
This program builds a depth topic with a best bid of 2950.00 and a currency scrip topic with a precision of 4, and a scrip topic that never sent its scale. It needs no stand-ins.
Notice that the currency price keeps four decimal places, and that the topic without a scale reads its integer as paise.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakFeedTopic/example_2_depth_and_currency_scale.py
"""
from stock_brokers.websockets.kotak import (
KotakFeedTopic,
)
class DepthAndCurrencyScaleExample:
"""Reads prices from topics with different scales.
Attributes:
depth (KotakFeedTopic): A depth topic with a precision of 2.
currency (KotakFeedTopic): A currency scrip topic with a precision of 4.
unscaled (KotakFeedTopic): A scrip topic that has not sent its scale.
"""
def __init__(self):
"""Builds and fills the three topics.
Returns:
None: This method returns nothing.
"""
self.depth = KotakFeedTopic('dp', 'nse_cm|2885', {})
depth_values = []
for _ in range(34):
depth_values.append(0)
depth_values[2] = 295000
depth_values[12] = 100
depth_values[32] = 1
depth_values[33] = 2
self.depth.apply(depth_values)
self.currency = KotakFeedTopic('sf', 'cde_fo|1234', {})
currency_values = []
for _ in range(25):
currency_values.append(0)
currency_values[5] = 835125
currency_values[23] = 1
currency_values[24] = 4
self.currency.apply(currency_values)
self.unscaled = KotakFeedTopic('sf', 'nse_cm|1594', {})
unscaled_values = [
0,
0,
0,
0,
0,
150025,
]
self.unscaled.apply(unscaled_values)
def run(self):
"""Prints a price from each topic.
Returns:
None: This method returns nothing.
"""
print(f'Depth best bid: {self.depth.price(2)} for quantity {self.depth.number(12)}')
print(f'Currency last price: {self.currency.price(5)}')
print(f'Last price without a scale: {self.unscaled.price(5)}')
if __name__ == '__main__':
DepthAndCurrencyScaleExample().run()
It prints:
Methods:
| Name | Description |
|---|---|
apply |
Takes a snapshot's or an update's numeric fields, keeping earlier values where a field is unchanged. |
number |
A numeric field as sent. |
price |
A price field in rupees, scaled by the topic's multiplier and precision. |
Source code in stock_brokers/websockets/kotak.py
Takes a snapshot's or an update's numeric fields, keeping earlier values where a field is unchanged.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
numbers
|
list[int]
|
The numeric fields in field order. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example usage
Source code in stock_brokers/websockets/kotak.py
A numeric field as sent.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
index
|
int
|
The field index. |
required |
Returns:
| Type | Description |
|---|---|
int | None
|
The value, or None when the feed has not sent the field. |
Example usage
From example_1_scrip_snapshot_and_update.py
print(f'After the snapshot: last price {self.topic.price(5)}, close {self.topic.price(21)}, volume {self.topic.number(4)}')
From example_1_scrip_snapshot_and_update.py
print(f'After the update: last price {self.topic.price(5)}, volume {self.topic.number(4)}')
A price field in rupees, scaled by the topic's multiplier and precision.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
index
|
int
|
The field index. |
required |
Returns:
| Type | Description |
|---|---|
float | None
|
The price, or None when the feed has not sent the field. |
Example usage
From example_1_scrip_snapshot_and_update.py
print(f'After the snapshot: last price {self.topic.price(5)}, close {self.topic.price(21)}, volume {self.topic.number(4)}')
From example_1_scrip_snapshot_and_update.py
print(f'After the update: last price {self.topic.price(5)}, volume {self.topic.number(4)}')
From example_1_scrip_snapshot_and_update.py
print(f'A field never sent: {self.topic.number(40)} and its price {self.topic.price(40)}')
Source code in stock_brokers/websockets/kotak.py
One instrument's scrip and depth topics, turned into a normalized tick.
Attributes:
| Name | Type | Description |
|---|---|---|
instrument_token |
str
|
The instrument's |
scrip |
KotakFeedTopic | None
|
The scrip topic, or None before its snapshot. |
depth |
KotakFeedTopic | None
|
The depth topic, or None before its snapshot. |
Creates an instrument with neither topic yet.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
instrument_token
|
str
|
The instrument's |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example programs
examples/stock_brokers/websockets/kotak/KotakInstrumentQuote/example_1_tick_from_the_scrip_topic.py
"""Builds a normalized tick from a Kotak instrument's scrip topic, and shows there is none before it.
`KotakInstrumentQuote` pairs one instrument's scrip topic and depth topic and turns their latest values into a normalized tick with `tick`. There is no tick until the scrip snapshot has arrived, because the scrip carries the prices. Without a depth topic the tick is in quote mode with an empty book. The quotes socket passes the instrument's trading symbol as the name, and a tick without one is keyed by its `EXCHANGE|TOKEN`.
This program asks for a tick before and after attaching a scrip topic, with and without a name. It needs no stand-ins.
Notice that the percentage change is computed from the last price and the close, and that the trade and feed times are the epochs the scrip carried.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakInstrumentQuote/example_1_tick_from_the_scrip_topic.py
"""
from stock_brokers.websockets.kotak import (
KotakFeedTopic,
KotakInstrumentQuote,
)
class TickFromTheScripTopicExample:
"""Builds ticks for one instrument before and after its scrip topic.
Attributes:
quote (KotakInstrumentQuote): The instrument being shown.
"""
def __init__(self):
"""Builds the instrument with no topics.
Returns:
None: This method returns nothing.
"""
self.quote = KotakInstrumentQuote('nse_cm|2885')
def scrip_topic(self):
"""A scrip topic of RELIANCE with a quote in paise.
Returns:
KotakFeedTopic: The topic.
"""
topic = KotakFeedTopic('sf', 'nse_cm|2885', {})
values = []
for _ in range(25):
values.append(0)
values[2] = 1790311529
values[3] = 1790311528
values[4] = 1204500
values[5] = 295050
values[14] = 292500
values[15] = 296000
values[20] = 293000
values[21] = 291000
values[23] = 1
values[24] = 2
topic.apply(values)
return topic
def print_tick(self, tick):
"""Prints the main fields of a tick, or that there is none.
Args:
tick (dict | None): The tick.
Returns:
None: This method returns nothing.
"""
if tick is None:
print('No tick yet')
return
print(f"Tick {tick['id']} in {tick['mode']} mode: last_price={tick['last_price']} change={round(tick['change'], 4)} ohlc={tick['ohlc']}")
print(f" last_trade_time={tick['last_trade_time']} exchange_timestamp={tick['exchange_timestamp']} depth={tick['depth']}")
def run(self):
"""Prints the tick before the scrip topic, then with and without a name.
Returns:
None: This method returns nothing.
"""
self.print_tick(self.quote.tick('RELIANCE-EQ'))
self.quote.scrip = self.scrip_topic()
self.print_tick(self.quote.tick('RELIANCE-EQ'))
self.print_tick(self.quote.tick(None))
print(f'Instrument {self.quote.instrument_token} has depth: {self.quote.depth is not None}')
if __name__ == '__main__':
TickFromTheScripTopicExample().run()
It prints:
No tick yet
Tick RELIANCE-EQ in quote mode: last_price=2950.5 change=1.3918 ohlc={'open': 2930.0, 'high': 2960.0, 'low': 2925.0, 'close': 2910.0}
last_trade_time=1790311528 exchange_timestamp=1790311529 depth={'buy': [], 'sell': []}
Tick nse_cm|2885 in quote mode: last_price=2950.5 change=1.3918 ohlc={'open': 2930.0, 'high': 2960.0, 'low': 2925.0, 'close': 2910.0}
last_trade_time=1790311528 exchange_timestamp=1790311529 depth={'buy': [], 'sell': []}
Instrument nse_cm|2885 has depth: False
examples/stock_brokers/websockets/kotak/KotakInstrumentQuote/example_2_full_tick_with_depth.py
"""Adds a depth topic to a Kotak instrument and builds a full tick with its five-level book.
Once a Kotak instrument has a depth topic, `KotakInstrumentQuote.tick` reads up to five levels on each side from it, best first, and the tick switches to full mode. A level whose price and quantity were never sent is left out.
This program attaches a scrip topic and a depth topic that sent only three buy levels and two sell levels. It needs no stand-ins.
Notice the book's quantities and order counts, and that the missing levels are absent rather than empty.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakInstrumentQuote/example_2_full_tick_with_depth.py
"""
from stock_brokers.websockets.kotak import (
KotakFeedTopic,
KotakInstrumentQuote,
)
class FullTickWithDepthExample:
"""Builds a full tick from scrip and depth topics.
Attributes:
quote (KotakInstrumentQuote): The instrument being shown.
"""
def __init__(self):
"""Builds the instrument with both topics.
Returns:
None: This method returns nothing.
"""
self.quote = KotakInstrumentQuote('nse_cm|2885')
self.quote.scrip = self.scrip_topic()
depth = KotakFeedTopic('dp', 'nse_cm|2885', {})
unchanged = -2147483648
values = []
for _ in range(34):
values.append(unchanged)
for level in range(3):
values[2 + level] = 295000 - level * 5
values[12 + level] = 100 + level
values[22 + level] = 3 + level
for level in range(2):
values[7 + level] = 295100 + level * 5
values[17 + level] = 200 + level
values[27 + level] = 4 + level
values[32] = 1
values[33] = 2
depth.apply(values)
self.quote.depth = depth
def scrip_topic(self):
"""A scrip topic of RELIANCE with a quote in paise.
Returns:
KotakFeedTopic: The topic.
"""
topic = KotakFeedTopic('sf', 'nse_cm|2885', {})
values = []
for _ in range(25):
values.append(0)
values[2] = 1790311529
values[3] = 1790311528
values[4] = 1204500
values[5] = 295050
values[14] = 292500
values[15] = 296000
values[20] = 293000
values[21] = 291000
values[23] = 1
values[24] = 2
topic.apply(values)
return topic
def print_tick(self, tick):
"""Prints the main fields of a tick, or that there is none.
Args:
tick (dict | None): The tick.
Returns:
None: This method returns nothing.
"""
if tick is None:
print('No tick yet')
return
print(f"Tick {tick['id']} in {tick['mode']} mode: last_price={tick['last_price']} change={round(tick['change'], 4)} ohlc={tick['ohlc']}")
print(f" last_trade_time={tick['last_trade_time']} exchange_timestamp={tick['exchange_timestamp']} depth={tick['depth']}")
def run(self):
"""Prints the full tick.
Returns:
None: This method returns nothing.
"""
self.print_tick(self.quote.tick('RELIANCE-EQ'))
if __name__ == '__main__':
FullTickWithDepthExample().run()
It prints:
Tick RELIANCE-EQ in full mode: last_price=2950.5 change=1.3918 ohlc={'open': 2930.0, 'high': 2960.0, 'low': 2925.0, 'close': 2910.0}
last_trade_time=1790311528 exchange_timestamp=1790311529 depth={'buy': [{'quantity': 100, 'price': 2950.0, 'orders': 3}, {'quantity': 101, 'price': 2949.95, 'orders': 4}, {'quantity': 102, 'price': 2949.9, 'orders': 5}], 'sell': [{'quantity': 200, 'price': 2951.0, 'orders': 4}, {'quantity': 201, 'price': 2951.05, 'orders': 5}]}
Methods:
| Name | Description |
|---|---|
tick |
Builds the normalized tick from the latest topic values. |
Source code in stock_brokers/websockets/kotak.py
Builds the normalized tick from the latest topic values.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str | None
|
The instrument's trading symbol, or None when it has none. |
required |
Returns:
| Type | Description |
|---|---|
dict | None
|
The tick, or None before the scrip snapshot has arrived. |
Example usage
Source code in stock_brokers/websockets/kotak.py
One side of the order book.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
price_start
|
int
|
The field index of the best level's price. |
required |
quantity_start
|
int
|
The field index of the best level's quantity. |
required |
orders_start
|
int
|
The field index of the best level's order count. |
required |
Returns:
| Type | Description |
|---|---|
list[dict]
|
Up to five levels of |
Source code in stock_brokers/websockets/kotak.py
The Kotak login every socket authenticates with, logged in again by one socket at a time.
Constructs KotakAPI, 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/kotak/KotakSession/example_1_login_and_session_id.py
"""Reads a Kotak session's stored login and logs in again only when the failed session id is still current.
`KotakSession` holds one `KotakAPI` and reads the stored login afresh on every connect. `current_login` returns the whole stored login, which the order socket needs for the host Kotak assigned the session, and `credentials` returns the access token and session id the quotes socket sends. `log_in_again` compares session ids: it logs in only when the session id a refused socket used is still the one in force.
Constructing the real `KotakAPI` logs in to Kotak, so the program registers a stand-in module under `stock_brokers.api.kotak`. Each construction counts as one login and issues the next numbered token and session id, and `_current_login` answers from a shared record standing in for the Redis `last_login` hash.
Notice the host in the stored login, and that the second request with the stale session id makes no login.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakSession/example_1_login_and_session_id.py
"""
import logging
import sys
import types
from stock_brokers.websockets.kotak import (
KotakSession,
)
class LoginRecord:
"""The Kotak login state every stand-in API object reads, standing in for the Redis `last_login` hash.
Attributes:
access_token (str | None): The access token in force now.
session_id (str | None): The session id in force now.
login_count (int): How many logins the stand-in API has made.
"""
access_token = None
session_id = None
login_count = 0
class StandInKotakAPI:
"""A stand-in for `KotakAPI` whose construction is a pretend login."""
def __init__(self):
"""Logs in, which here only issues the next numbered access token and session id.
Returns:
None: This method returns nothing.
"""
LoginRecord.login_count = LoginRecord.login_count + 1
LoginRecord.access_token = f'kotak-token-{LoginRecord.login_count}'
LoginRecord.session_id = f'kotak-sid-{LoginRecord.login_count}'
print(f'Stand-in login number {LoginRecord.login_count}')
def _current_login(self):
"""The login in force now, as the real class reads it from Redis.
Returns:
dict | None: The access token, session id and host, or None before any login.
"""
if LoginRecord.access_token is None:
return None
return {
'access_token': LoginRecord.access_token,
'sid': LoginRecord.session_id,
'base_url': 'https://e21.kotaksecurities.com/',
}
class LoginAndSessionIdExample:
"""Reads a Kotak session's login and asks it to log in again twice with the same session id.
Attributes:
session (KotakSession): 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.kotak')
api_module.KotakAPI = StandInKotakAPI
sys.modules['stock_brokers.api.kotak'] = api_module
self.session = KotakSession(logging.getLogger('kotak.session'))
def run(self):
"""Prints the login and credentials around two login requests.
Returns:
None: This method returns nothing.
"""
print(f'Stored login: {self.session.current_login()}')
access_token, failed_session_id = self.session.credentials()
print(f'Refused with {access_token} and {failed_session_id}')
self.session.log_in_again(failed_session_id)
self.session.log_in_again(failed_session_id)
print(f'Credentials now: {self.session.credentials()}')
print(f'Logins made: {LoginRecord.login_count}')
if __name__ == '__main__':
LoginAndSessionIdExample().run()
It prints:
Stand-in login number 1
Stored login: {'access_token': 'kotak-token-1', 'sid': 'kotak-sid-1', 'base_url': 'https://e21.kotaksecurities.com/'}
Refused with kotak-token-1 and kotak-sid-1
WARNING Logging in to Kotak again.
Stand-in login number 2
Credentials now: ('kotak-token-2', 'kotak-sid-2')
Logins made: 2
examples/stock_brokers/websockets/kotak/KotakSession/example_2_logging_in_without_checking.py
"""Logs in to Kotak again unconditionally, as the order update socket does after a refusal.
The order update socket calls `KotakSession.log_in_again_without_checking` when Kotak refuses its connection frame. That logs in even when another process has already replaced the session, unlike `log_in_again`, which would skip the login in that case.
Constructing the real `KotakAPI` logs in to Kotak, so the program registers a stand-in module under `stock_brokers.api.kotak` whose constructions count as logins. It replaces the session as another process would, shows that `log_in_again` with the old session id does nothing, and then calls `log_in_again_without_checking`.
Notice that only the unconditional call makes a login.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakSession/example_2_logging_in_without_checking.py
"""
import logging
import sys
import types
from stock_brokers.websockets.kotak import (
KotakSession,
)
class LoginRecord:
"""The Kotak login state every stand-in API object reads, standing in for the Redis `last_login` hash.
Attributes:
access_token (str | None): The access token in force now.
session_id (str | None): The session id in force now.
login_count (int): How many logins the stand-in API has made.
"""
access_token = None
session_id = None
login_count = 0
class StandInKotakAPI:
"""A stand-in for `KotakAPI` whose construction is a pretend login."""
def __init__(self):
"""Logs in, which here only issues the next numbered access token and session id.
Returns:
None: This method returns nothing.
"""
LoginRecord.login_count = LoginRecord.login_count + 1
LoginRecord.access_token = f'kotak-token-{LoginRecord.login_count}'
LoginRecord.session_id = f'kotak-sid-{LoginRecord.login_count}'
print(f'Stand-in login number {LoginRecord.login_count}')
def _current_login(self):
"""The login in force now, as the real class reads it from Redis.
Returns:
dict | None: The access token, session id and host, or None before any login.
"""
if LoginRecord.access_token is None:
return None
return {
'access_token': LoginRecord.access_token,
'sid': LoginRecord.session_id,
'base_url': 'https://e21.kotaksecurities.com/',
}
class LoggingInWithoutCheckingExample:
"""Compares the checked and unchecked logins of a Kotak session.
Attributes:
session (KotakSession): 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.kotak')
api_module.KotakAPI = StandInKotakAPI
sys.modules['stock_brokers.api.kotak'] = api_module
self.session = KotakSession(logging.getLogger('kotak.session'))
def run(self):
"""Replaces the session elsewhere, then tries both kinds of login.
Returns:
None: This method returns nothing.
"""
first_session_id = self.session.credentials()[1]
LoginRecord.session_id = 'kotak-sid-from-another-process'
print(f'Another process replaced {first_session_id}.')
self.session.log_in_again(first_session_id)
print(f'After log_in_again: {self.session.credentials()}')
self.session.log_in_again_without_checking()
print(f'After log_in_again_without_checking: {self.session.credentials()}')
print(f'Logins made: {LoginRecord.login_count}')
if __name__ == '__main__':
LoggingInWithoutCheckingExample().run()
It prints:
Methods:
| Name | Description |
|---|---|
current_login |
The stored login in force now, which may be one another process has just obtained. |
credentials |
The access token and session id in force now. |
log_in_again |
Logs in again, unless another socket or process already replaced the session that failed. |
log_in_again_without_checking |
Logs in again whether or not the session was already replaced. |
Source code in stock_brokers/websockets/kotak.py
The stored login in force now, which may be one another process has just obtained.
Returns:
| Type | Description |
|---|---|
dict
|
The login, with |
Example usage
Source code in stock_brokers/websockets/kotak.py
The access token and session id in force now.
Returns:
| Type | Description |
|---|---|
tuple
|
The access token and the session id, either of which may be None. |
Example usage
Source code in stock_brokers/websockets/kotak.py
Logs in again, unless another socket or process already replaced the session that failed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
stale_session_id
|
str | None
|
The session id 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/kotak.py
Logs in again whether or not the session was already replaced.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Raises:
| Type | Description |
|---|---|
Exception
|
Whatever |
Example usage
Source code in stock_brokers/websockets/kotak.py
Logs in by constructing KotakAPI again; the caller holds the lock.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Raises:
| Type | Description |
|---|---|
Exception
|
Whatever |
Source code in stock_brokers/websockets/kotak.py
Bases: BrokerWebsocket
One Kotak HSM feed websocket carrying the scrip and depth topics 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 |
instrument_tokens
|
list[str]
|
The batch of |
required |
names
|
dict[str, str]
|
|
required |
session
|
KotakSession
|
The shared login. |
required |
on_name
|
callable
|
Called with an instrument and its trading symbol when a snapshot names an unnamed instrument, on this socket's thread. |
required |
on_ticks
|
callable
|
Called with each data frame's ticks, on this socket's thread. |
required |
logger
|
Logger
|
Where the connection reports. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example programs
examples/stock_brokers/websockets/kotak/KotakQuotesSocket/example_1_scrip_and_depth_topics.py
"""Runs a Kotak HSM quotes socket through authentication, scrip and depth snapshots, an update and an acknowledgement.
`KotakQuotesSocket` sends a binary connection request with the access token and session id when the connection opens. Kotak accepts with a `K` and says after how many data frames it wants an acknowledgement, and the socket then subscribes every instrument twice: a scrip topic (`sf|...`) for prices and quantities and a depth topic (`dp|...`) for the five-level book. Data frames hold snapshot packets, which name a topic and carry every numeric field, and update packets, which carry the numeric fields again with -2147483648 for unchanged. The socket keeps each topic's latest values and builds a whole tick for an instrument whenever either of its topics changes.
This program subscribes `nse_cm|2885`. Kotak accepts with an acknowledgement interval of 2, the first data frame carries the scrip and depth snapshots, and the second an update that changes only the last price and volume. No real socket is opened: a stand-in `websocket` package plays the frames and prints each binary frame the socket sends as hex, and a stand-in `stock_brokers.api.kotak` module lets the real `KotakSession` supply the token and session id without logging in.
Notice the connection request and the two subscription frames, that the snapshot's trading symbol names the instrument through `on_name` before the ticks, that the update's tick keeps the depth and the open, high, low and close, and the acknowledgement frame (type 3) after the second data frame.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakQuotesSocket/example_1_scrip_and_depth_topics.py
"""
import logging
import sys
import types
from stock_brokers.websockets.kotak import (
KotakQuotesSocket,
KotakSession,
)
class LoginRecord:
"""The Kotak login state every stand-in API object reads, standing in for the Redis `last_login` hash.
Attributes:
access_token (str | None): The access token in force now.
session_id (str | None): The session id in force now.
login_count (int): How many logins the stand-in API has made.
"""
access_token = None
session_id = None
login_count = 0
class StandInKotakAPI:
"""A stand-in for `KotakAPI` whose construction is a pretend login."""
def __init__(self):
"""Logs in, which here only issues the next numbered access token and session id.
Returns:
None: This method returns nothing.
"""
LoginRecord.login_count = LoginRecord.login_count + 1
LoginRecord.access_token = f'kotak-token-{LoginRecord.login_count}'
LoginRecord.session_id = f'kotak-sid-{LoginRecord.login_count}'
print(f'Stand-in login number {LoginRecord.login_count}')
def _current_login(self):
"""The login in force now, as the real class reads it from Redis.
Returns:
dict | None: The access token, session id and host, or None before any login.
"""
if LoginRecord.access_token is None:
return None
return {
'access_token': LoginRecord.access_token,
'sid': LoginRecord.session_id,
'base_url': 'https://e21.kotaksecurities.com/',
}
class HandshakeRefused(Exception):
"""A refused websocket handshake, carrying the HTTP status the way `websocket-client` does.
Attributes:
status_code (int): The HTTP status the broker answered with.
"""
def __init__(self, status_code):
"""Keeps the status.
Args:
status_code (int): The HTTP status the broker answered with.
Returns:
None: This method returns nothing.
"""
super().__init__(f'Handshake status {status_code}')
self.status_code = status_code
class ScriptedConnection:
"""A stand-in for `websocket.WebSocketApp` that either refuses the handshake or plays frames to the socket's handlers.
Attributes:
url (str): The URL the socket asked for.
plan (dict): Either a `refuse_with` status or a list of `frames`.
handlers (dict): The socket's callbacks, by name.
closed (bool): Whether the socket asked to close, after which no more frames are delivered.
"""
def __init__(self, url, plan, handlers):
"""Keeps what the socket passed in.
Args:
url (str): The URL the socket asked for.
plan (dict): Either a `refuse_with` status or a list of `frames`.
handlers (dict): The socket's callbacks, by name.
Returns:
None: This method returns nothing.
"""
self.url = url
self.plan = plan
self.handlers = handlers
self.closed = False
def run_forever(self, **options):
"""Plays the plan and returns when the pretend connection closes.
Args:
**options (dict): The ping and TLS settings the socket asks for.
Returns:
None: This method returns nothing.
"""
print(f'Connecting to {self.url}')
if 'refuse_with' in self.plan:
self.handlers['on_error'](self, HandshakeRefused(self.plan['refuse_with']))
self.handlers['on_close'](self, None, None)
return
self.handlers['on_open'](self)
for frame in self.plan['frames']:
if self.closed:
break
self.handlers['on_message'](self, frame)
self.handlers['on_close'](self, 1000, 'normal closure')
def send(self, data, opcode=None):
"""Prints a message the socket sends to the broker.
Args:
data (str | bytes): The message.
opcode (int | None): The frame type, when the socket names one.
Returns:
None: This method returns nothing.
"""
if isinstance(data, bytes):
print(f'Sent binary frame of type {data[2]}: {data.hex()}')
else:
print(f'Sent: {data}')
def close(self, **options):
"""Notes that the socket asked to close.
Args:
**options (dict): Any close status the socket passes.
Returns:
None: This method returns nothing.
"""
self.closed = True
class FrameOpcodes:
"""The one websocket frame opcode the Kotak quotes socket names.
Attributes:
OPCODE_BINARY (int): The binary frame opcode.
"""
OPCODE_BINARY = 2
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.
ABNF (FrameOpcodes): The frame opcodes, as `websocket.ABNF` holds them.
"""
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
self.ABNF = FrameOpcodes()
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 HsmFeedFrames:
"""Builds the binary frames Kotak's HSM feed sends, big endian."""
def framed(self, frame_type, body):
"""A frame: its length, its type and its body.
Args:
frame_type (int): The frame type.
body (bytes): Everything after the type.
Returns:
bytes: The frame.
"""
content = bytes([frame_type]) + body
return len(content).to_bytes(2, 'big') + content
def status_field(self, status):
"""A status field: id 1, a two-byte length and the status.
Args:
status (str): The status letter.
Returns:
bytes: The field.
"""
return bytes([1]) + len(status).to_bytes(2, 'big') + status.encode('latin-1')
def connection_response(self, status, acknowledge_every):
"""A connection response with a status and the acknowledgement interval.
Args:
status (str): `K` for accepted.
acknowledge_every (int): After how many data frames Kotak wants an acknowledgement.
Returns:
bytes: The frame.
"""
acknowledgement = bytes([2]) + (4).to_bytes(2, 'big') + acknowledge_every.to_bytes(4, 'big')
return self.framed(1, bytes([2]) + self.status_field(status) + acknowledgement)
def subscription_response(self):
"""An accepted subscription response.
Returns:
bytes: The frame.
"""
return self.framed(4, bytes([1]) + self.status_field('K'))
def numbers(self, values):
"""Numeric fields, four signed bytes each, with None as Kotak's unchanged marker.
Args:
values (list): The values.
Returns:
bytes: The packed values.
"""
packed = bytes([len(values)])
for value in values:
if value is None:
value = -2147483648
packed = packed + value.to_bytes(4, 'big', signed=True)
return packed
def snapshot(self, topic_id, topic_name, values, trading_symbol):
"""A snapshot packet carrying the trading symbol as string field 54.
Args:
topic_id (int): The topic id updates refer to.
topic_name (str): The topic name, such as `sf|nse_cm|2885`.
values (list): The numeric fields in field order.
trading_symbol (str): The trading symbol.
Returns:
bytes: The packet, after its two-byte length.
"""
name = topic_name.encode('latin-1')
symbol = trading_symbol.encode('latin-1')
body = bytes([83]) + topic_id.to_bytes(4, 'big', signed=True) + bytes([len(name)]) + name + self.numbers(values)
body = body + bytes([1, 54, len(symbol)]) + symbol
return len(body).to_bytes(2, 'big') + body
def update(self, topic_id, values):
"""An update packet.
Args:
topic_id (int): The topic id.
values (list): The numeric fields in field order.
Returns:
bytes: The packet, after its two-byte length.
"""
body = bytes([85]) + topic_id.to_bytes(4, 'big', signed=True) + self.numbers(values)
return len(body).to_bytes(2, 'big') + body
def data(self, message_number, packets):
"""A data frame, with a message number when acknowledgements were asked for.
Args:
message_number (int | None): The message number an acknowledgement refers to, or None when acknowledgements were not asked for.
packets (list): The packets.
Returns:
bytes: The frame.
"""
body = b''
if message_number is not None:
body = message_number.to_bytes(4, 'big', signed=True)
body = body + len(packets).to_bytes(2, 'big')
for packet in packets:
body = body + packet
return self.framed(6, body)
def scrip_values(self, last_price, volume):
"""The 25 scrip fields of an NSE equity, with prices in paise, a multiplier of 1 and a precision of 2.
Args:
last_price (int): The last price in paise.
volume (int): The day's volume.
Returns:
list: The fields in field order.
"""
values = []
for _ in range(25):
values.append(None)
values[2] = 1790311529
values[3] = 1790311528
values[4] = volume
values[5] = last_price
values[6] = 25
values[7] = 54000
values[8] = 61000
values[13] = last_price - 150
values[14] = 292500
values[15] = 296000
values[20] = 293000
values[21] = 291000
values[22] = 0
values[23] = 1
values[24] = 2
return values
def depth_values(self):
"""The 34 depth fields of five levels a side, with prices in paise.
Returns:
list: The fields in field order.
"""
values = []
for _ in range(34):
values.append(None)
for level in range(5):
values[2 + level] = 295000 - level * 5
values[7 + level] = 295100 + level * 5
values[12 + level] = 100 + level
values[17 + level] = 200 + level
values[22 + level] = 3 + level
values[27 + level] = 4 + level
values[32] = 1
values[33] = 2
return values
class ScripAndDepthTopicsExample:
"""Runs a Kotak quotes socket against one scripted HSM connection.
Attributes:
socket (KotakQuotesSocket): The socket being shown.
"""
def __init__(self):
"""Builds the frames, the stand-ins, the session and the socket.
Returns:
None: This method returns nothing.
"""
logging.basicConfig(stream=sys.stdout, format='%(levelname)s %(message)s', level=logging.INFO)
api_module = types.ModuleType('stock_brokers.api.kotak')
api_module.KotakAPI = StandInKotakAPI
sys.modules['stock_brokers.api.kotak'] = api_module
frames = HsmFeedFrames()
update_values = [
None,
None,
None,
None,
1205000,
295100,
]
plans = [
{
'frames': [
frames.connection_response('K', 2),
frames.subscription_response(),
frames.data(1, [
frames.snapshot(7, 'sf|nse_cm|2885', frames.scrip_values(295050, 1204500), 'RELIANCE-EQ'),
frames.snapshot(8, 'dp|nse_cm|2885', frames.depth_values(), 'RELIANCE-EQ'),
]),
frames.data(2, [
frames.update(7, update_values),
]),
],
},
]
websocket_module = ScriptedWebsocketModule(plans)
sys.modules['websocket'] = websocket_module
logger = logging.getLogger('kotak.quotes')
instrument_tokens = [
'nse_cm|2885',
]
self.names = {}
session = KotakSession(logger)
self.socket = KotakQuotesSocket('socket_0', instrument_tokens, self.names, session, self.print_name, self.print_ticks, logger)
websocket_module.when_finished = self.socket.close
def print_name(self, instrument_token, trading_symbol):
"""Prints an instrument a snapshot named.
Args:
instrument_token (str): The instrument's `EXCHANGE|TOKEN`.
trading_symbol (str): Its trading symbol.
Returns:
None: This method returns nothing.
"""
print(f'Name learned: {instrument_token} is {trading_symbol}')
def print_ticks(self, ticks):
"""Prints the main fields of each tick the socket built.
Args:
ticks (list): The normalized ticks from one data frame.
Returns:
None: This method returns nothing.
"""
for tick in ticks:
change = tick['change']
if change is not None:
change = round(change, 4)
print(f"Tick {tick['id']} ({tick['instrument_token']}) in {tick['mode']} mode")
print(f" last_price={tick['last_price']} volume={tick['volume']} change={change}")
print(f" ohlc={tick['ohlc']}")
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__':
ScripAndDepthTopicsExample().run()
It prints:
Stand-in login number 1
Connecting to wss://mlhsm.kotaksecurities.com
Sent binary frame of type 1: 0029010301000d6b6f74616b2d746f6b656e2d3102000b6b6f74616b2d7369642d310300064a535f415049
INFO socket_0 opened. Authenticating.
Sent binary frame of type 4: 001a040201001100010e73667c6e73655f636d7c3238383502000101
Sent binary frame of type 4: 001a040201001100010e64707c6e73655f636d7c3238383502000101
INFO socket_0 authenticated and subscribed to 1 instrument(s) with depth, acknowledging every 2 data frame(s).
Name learned: nse_cm|2885 is RELIANCE-EQ
Tick RELIANCE-EQ (nse_cm|2885) in full mode
last_price=2950.5 volume=1204500 change=1.3918
ohlc={'open': 2930.0, 'high': 2960.0, 'low': 2925.0, 'close': 2910.0}
best bid={'quantity': 100, 'price': 2950.0, 'orders': 3} best offer={'quantity': 200, 'price': 2951.0, 'orders': 4} levels=5
Sent binary frame of type 3: 0009030101000400000002
Tick RELIANCE-EQ (nse_cm|2885) in full mode
last_price=2951.0 volume=1205000 change=1.4089
ohlc={'open': 2930.0, 'high': 2960.0, 'low': 2925.0, 'close': 2910.0}
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/kotak/KotakQuotesSocket/example_2_refused_session_and_truncated_frame.py
"""Shows a Kotak quotes socket logging in again after a refused session, then surviving a truncated data frame.
When Kotak answers the connection request with a status other than `K`, `KotakQuotesSocket` closes the connection as a refused login, and its reconnect loop asks the shared `KotakSession` to log in again unless another socket or process already replaced the session. A data frame that ends before a field it declared raises `KotakTruncatedFrameError` inside the socket, which logs a warning and still hands on the ticks of the packets it had read.
This program's first connection is refused with status `N`. The second is accepted without acknowledgements being asked for (an interval of 0), and its data frame holds a whole scrip snapshot followed by an update packet cut short. No real socket is opened: a stand-in `websocket` package plays the frames, and a stand-in `stock_brokers.api.kotak` module counts each `KotakAPI` construction as one login. The backoff is set to zero seconds.
Notice the new token and session id in the second connection request, the warning about the truncated frame, and that the snapshot's tick still arrives in quote mode because no depth topic was sent.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakQuotesSocket/example_2_refused_session_and_truncated_frame.py
"""
import logging
import sys
import types
from stock_brokers.websockets.kotak import (
KotakQuotesSocket,
KotakSession,
)
class LoginRecord:
"""The Kotak login state every stand-in API object reads, standing in for the Redis `last_login` hash.
Attributes:
access_token (str | None): The access token in force now.
session_id (str | None): The session id in force now.
login_count (int): How many logins the stand-in API has made.
"""
access_token = None
session_id = None
login_count = 0
class StandInKotakAPI:
"""A stand-in for `KotakAPI` whose construction is a pretend login."""
def __init__(self):
"""Logs in, which here only issues the next numbered access token and session id.
Returns:
None: This method returns nothing.
"""
LoginRecord.login_count = LoginRecord.login_count + 1
LoginRecord.access_token = f'kotak-token-{LoginRecord.login_count}'
LoginRecord.session_id = f'kotak-sid-{LoginRecord.login_count}'
print(f'Stand-in login number {LoginRecord.login_count}')
def _current_login(self):
"""The login in force now, as the real class reads it from Redis.
Returns:
dict | None: The access token, session id and host, or None before any login.
"""
if LoginRecord.access_token is None:
return None
return {
'access_token': LoginRecord.access_token,
'sid': LoginRecord.session_id,
'base_url': 'https://e21.kotaksecurities.com/',
}
class HandshakeRefused(Exception):
"""A refused websocket handshake, carrying the HTTP status the way `websocket-client` does.
Attributes:
status_code (int): The HTTP status the broker answered with.
"""
def __init__(self, status_code):
"""Keeps the status.
Args:
status_code (int): The HTTP status the broker answered with.
Returns:
None: This method returns nothing.
"""
super().__init__(f'Handshake status {status_code}')
self.status_code = status_code
class ScriptedConnection:
"""A stand-in for `websocket.WebSocketApp` that either refuses the handshake or plays frames to the socket's handlers.
Attributes:
url (str): The URL the socket asked for.
plan (dict): Either a `refuse_with` status or a list of `frames`.
handlers (dict): The socket's callbacks, by name.
closed (bool): Whether the socket asked to close, after which no more frames are delivered.
"""
def __init__(self, url, plan, handlers):
"""Keeps what the socket passed in.
Args:
url (str): The URL the socket asked for.
plan (dict): Either a `refuse_with` status or a list of `frames`.
handlers (dict): The socket's callbacks, by name.
Returns:
None: This method returns nothing.
"""
self.url = url
self.plan = plan
self.handlers = handlers
self.closed = False
def run_forever(self, **options):
"""Plays the plan and returns when the pretend connection closes.
Args:
**options (dict): The ping and TLS settings the socket asks for.
Returns:
None: This method returns nothing.
"""
print(f'Connecting to {self.url}')
if 'refuse_with' in self.plan:
self.handlers['on_error'](self, HandshakeRefused(self.plan['refuse_with']))
self.handlers['on_close'](self, None, None)
return
self.handlers['on_open'](self)
for frame in self.plan['frames']:
if self.closed:
break
self.handlers['on_message'](self, frame)
self.handlers['on_close'](self, 1000, 'normal closure')
def send(self, data, opcode=None):
"""Prints a message the socket sends to the broker.
Args:
data (str | bytes): The message.
opcode (int | None): The frame type, when the socket names one.
Returns:
None: This method returns nothing.
"""
if isinstance(data, bytes):
print(f'Sent binary frame of type {data[2]}: {data.hex()}')
else:
print(f'Sent: {data}')
def close(self, **options):
"""Notes that the socket asked to close.
Args:
**options (dict): Any close status the socket passes.
Returns:
None: This method returns nothing.
"""
self.closed = True
class FrameOpcodes:
"""The one websocket frame opcode the Kotak quotes socket names.
Attributes:
OPCODE_BINARY (int): The binary frame opcode.
"""
OPCODE_BINARY = 2
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.
ABNF (FrameOpcodes): The frame opcodes, as `websocket.ABNF` holds them.
"""
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
self.ABNF = FrameOpcodes()
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 HsmFeedFrames:
"""Builds the binary frames Kotak's HSM feed sends, big endian."""
def framed(self, frame_type, body):
"""A frame: its length, its type and its body.
Args:
frame_type (int): The frame type.
body (bytes): Everything after the type.
Returns:
bytes: The frame.
"""
content = bytes([frame_type]) + body
return len(content).to_bytes(2, 'big') + content
def status_field(self, status):
"""A status field: id 1, a two-byte length and the status.
Args:
status (str): The status letter.
Returns:
bytes: The field.
"""
return bytes([1]) + len(status).to_bytes(2, 'big') + status.encode('latin-1')
def connection_response(self, status, acknowledge_every):
"""A connection response with a status and the acknowledgement interval.
Args:
status (str): `K` for accepted.
acknowledge_every (int): After how many data frames Kotak wants an acknowledgement.
Returns:
bytes: The frame.
"""
acknowledgement = bytes([2]) + (4).to_bytes(2, 'big') + acknowledge_every.to_bytes(4, 'big')
return self.framed(1, bytes([2]) + self.status_field(status) + acknowledgement)
def subscription_response(self):
"""An accepted subscription response.
Returns:
bytes: The frame.
"""
return self.framed(4, bytes([1]) + self.status_field('K'))
def numbers(self, values):
"""Numeric fields, four signed bytes each, with None as Kotak's unchanged marker.
Args:
values (list): The values.
Returns:
bytes: The packed values.
"""
packed = bytes([len(values)])
for value in values:
if value is None:
value = -2147483648
packed = packed + value.to_bytes(4, 'big', signed=True)
return packed
def snapshot(self, topic_id, topic_name, values, trading_symbol):
"""A snapshot packet carrying the trading symbol as string field 54.
Args:
topic_id (int): The topic id updates refer to.
topic_name (str): The topic name, such as `sf|nse_cm|2885`.
values (list): The numeric fields in field order.
trading_symbol (str): The trading symbol.
Returns:
bytes: The packet, after its two-byte length.
"""
name = topic_name.encode('latin-1')
symbol = trading_symbol.encode('latin-1')
body = bytes([83]) + topic_id.to_bytes(4, 'big', signed=True) + bytes([len(name)]) + name + self.numbers(values)
body = body + bytes([1, 54, len(symbol)]) + symbol
return len(body).to_bytes(2, 'big') + body
def update(self, topic_id, values):
"""An update packet.
Args:
topic_id (int): The topic id.
values (list): The numeric fields in field order.
Returns:
bytes: The packet, after its two-byte length.
"""
body = bytes([85]) + topic_id.to_bytes(4, 'big', signed=True) + self.numbers(values)
return len(body).to_bytes(2, 'big') + body
def data(self, message_number, packets):
"""A data frame, with a message number when acknowledgements were asked for.
Args:
message_number (int | None): The message number an acknowledgement refers to, or None when acknowledgements were not asked for.
packets (list): The packets.
Returns:
bytes: The frame.
"""
body = b''
if message_number is not None:
body = message_number.to_bytes(4, 'big', signed=True)
body = body + len(packets).to_bytes(2, 'big')
for packet in packets:
body = body + packet
return self.framed(6, body)
def scrip_values(self, last_price, volume):
"""The 25 scrip fields of an NSE equity, with prices in paise, a multiplier of 1 and a precision of 2.
Args:
last_price (int): The last price in paise.
volume (int): The day's volume.
Returns:
list: The fields in field order.
"""
values = []
for _ in range(25):
values.append(None)
values[2] = 1790311529
values[3] = 1790311528
values[4] = volume
values[5] = last_price
values[6] = 25
values[7] = 54000
values[8] = 61000
values[13] = last_price - 150
values[14] = 292500
values[15] = 296000
values[20] = 293000
values[21] = 291000
values[22] = 0
values[23] = 1
values[24] = 2
return values
def depth_values(self):
"""The 34 depth fields of five levels a side, with prices in paise.
Returns:
list: The fields in field order.
"""
values = []
for _ in range(34):
values.append(None)
for level in range(5):
values[2 + level] = 295000 - level * 5
values[7 + level] = 295100 + level * 5
values[12 + level] = 100 + level
values[17 + level] = 200 + level
values[22 + level] = 3 + level
values[27 + level] = 4 + level
values[32] = 1
values[33] = 2
return values
class RefusedSessionAndTruncatedFrameExample:
"""Runs a Kotak quotes socket through a refused session and a truncated frame.
Attributes:
socket (KotakQuotesSocket): The socket being shown.
"""
def __init__(self):
"""Builds the frames, the stand-ins, the session and the socket.
Returns:
None: This method returns nothing.
"""
logging.basicConfig(stream=sys.stdout, format='%(levelname)s %(message)s', level=logging.INFO)
api_module = types.ModuleType('stock_brokers.api.kotak')
api_module.KotakAPI = StandInKotakAPI
sys.modules['stock_brokers.api.kotak'] = api_module
frames = HsmFeedFrames()
update = frames.update(3, [
None,
None,
None,
None,
1205000,
295100,
])
truncated = frames.data(None, [
frames.snapshot(3, 'sf|nse_cm|2885', frames.scrip_values(295050, 1204500), 'RELIANCE-EQ'),
update[:9],
])
plans = [
{
'frames': [
frames.connection_response('N', 0),
],
},
{
'frames': [
frames.connection_response('K', 0),
truncated,
],
},
]
websocket_module = ScriptedWebsocketModule(plans)
sys.modules['websocket'] = websocket_module
logger = logging.getLogger('kotak.quotes')
instrument_tokens = [
'nse_cm|2885',
]
names = {
'nse_cm|2885': 'RELIANCE-EQ',
}
session = KotakSession(logger)
self.socket = KotakQuotesSocket('socket_0', instrument_tokens, names, session, self.print_name, self.print_ticks, logger)
self.socket.MIN_BACKOFF_SECONDS = 0
websocket_module.when_finished = self.socket.close
def print_name(self, instrument_token, trading_symbol):
"""Prints an instrument a snapshot named.
Args:
instrument_token (str): The instrument's `EXCHANGE|TOKEN`.
trading_symbol (str): Its trading symbol.
Returns:
None: This method returns nothing.
"""
print(f'Name learned: {instrument_token} is {trading_symbol}')
def print_ticks(self, ticks):
"""Prints the main fields of each tick the socket built.
Args:
ticks (list): The normalized ticks from one data frame.
Returns:
None: This method returns nothing.
"""
for tick in ticks:
change = tick['change']
if change is not None:
change = round(change, 4)
print(f"Tick {tick['id']} ({tick['instrument_token']}) in {tick['mode']} mode")
print(f" last_price={tick['last_price']} volume={tick['volume']} change={change}")
print(f" ohlc={tick['ohlc']}")
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 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__':
RefusedSessionAndTruncatedFrameExample().run()
It prints:
Stand-in login number 1
Connecting to wss://mlhsm.kotaksecurities.com
Sent binary frame of type 1: 0029010301000d6b6f74616b2d746f6b656e2d3102000b6b6f74616b2d7369642d310300064a535f415049
INFO socket_0 opened. Authenticating.
ERROR socket_0: Kotak refused the session (status 'N').
INFO socket_0 closed with status 1000: normal closure
WARNING Logging in to Kotak again.
Stand-in login number 2
WARNING socket_0 disconnected. Reconnecting in 0 seconds.
Connecting to wss://mlhsm.kotaksecurities.com
Sent binary frame of type 1: 0029010301000d6b6f74616b2d746f6b656e2d3202000b6b6f74616b2d7369642d320300064a535f415049
INFO socket_0 opened. Authenticating.
Sent binary frame of type 4: 001a040201001100010e73667c6e73655f636d7c3238383502000101
Sent binary frame of type 4: 001a040201001100010e64707c6e73655f636d7c3238383502000101
INFO socket_0 authenticated and subscribed to 1 instrument(s) with depth, acknowledging every 0 data frame(s).
Tick RELIANCE-EQ (nse_cm|2885) in quote mode
last_price=2950.5 volume=1204500 change=1.3918
ohlc={'open': 2930.0, 'high': 2960.0, 'low': 2925.0, 'close': 2910.0}
WARNING socket_0 skipped the rest of a truncated frame of type 6: frame of 151 bytes ended reading 4 at 150
INFO socket_0 closed with status 1000: normal closure
INFO socket_0 stopped.
Gave up: False
Logins made: 2
Methods:
| Name | Description |
|---|---|
close |
Closes the connection and stops reconnecting. |
run_forever |
Connects, and reconnects with backoff, until closed or unable to recover. |
Source code in stock_brokers/websockets/kotak.py
Gives up only once six connects in a row have failed, as Kotak's quotes loop always has.
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/kotak.py
Opens the websocket and blocks until it closes; the credentials are sent once it opens.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/kotak.py
Logs in again through the shared session, which skips the login when the session was already replaced.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/kotak.py
Sends one binary frame.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The open connection. |
required |
frame
|
bytes
|
The frame. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/kotak.py
Sends the connection request with the login in force now; subscribing waits for Kotak's acceptance.
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/kotak.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/kotak.py
Dispatches a binary frame by its type.
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/kotak.py
Reads the connection response, then subscribes or reports the refusal.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The connection. |
required |
reader
|
KotakFrameReader
|
The frame, positioned after its type. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Raises:
| Type | Description |
|---|---|
KotakTruncatedFrameError
|
The frame ends before a declared field. |
Source code in stock_brokers/websockets/kotak.py
Subscribes every instrument's scrip and depth topics, 100 at a time.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The accepted connection. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/kotak.py
Reads a subscription response and reports a refusal.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
reader
|
KotakFrameReader
|
The frame, positioned after its type. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Raises:
| Type | Description |
|---|---|
KotakTruncatedFrameError
|
The frame ends before a declared field. |
Source code in stock_brokers/websockets/kotak.py
Acknowledges data frames when asked, applies each packet to its topic, and hands on the changed instruments' ticks, even when the frame is cut short.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The connection. |
required |
reader
|
KotakFrameReader
|
The frame, positioned after its type. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Raises:
| Type | Description |
|---|---|
KotakTruncatedFrameError
|
The frame ends before a declared field. |
Source code in stock_brokers/websockets/kotak.py
Reads a snapshot packet and registers its topic, naming its instrument when it had no name.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
reader
|
KotakFrameReader
|
The frame, positioned after the packet type. |
required |
Returns:
| Type | Description |
|---|---|
KotakFeedTopic | None
|
The topic, or None when its name is not a scrip or depth topic. |
Raises:
| Type | Description |
|---|---|
KotakTruncatedFrameError
|
The frame ends before a declared field. |
Source code in stock_brokers/websockets/kotak.py
Reads an update packet and applies it to its topic.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
reader
|
KotakFrameReader
|
The frame, positioned after the packet type. |
required |
Returns:
| Type | Description |
|---|---|
KotakFeedTopic | None
|
The topic, or None when no snapshot has registered its id on this connection. |
Raises:
| Type | Description |
|---|---|
KotakTruncatedFrameError
|
The frame ends before a declared field. |
Source code in stock_brokers/websockets/kotak.py
Hands on the latest tick of each changed instrument that has had its scrip snapshot.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
instrument_tokens
|
list[str]
|
The instruments whose topics changed, in arrival order. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/kotak.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
Reports that the connection closed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The connection that closed. |
required |
status_code
|
int | None
|
The close status the broker sent. |
required |
message
|
str | None
|
The close reason the broker sent. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/base.py
Bases: BrokerWebsocket
Kotak's order and position update websocket, on the host the session was assigned.
Sets up the order updates socket.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
session
|
KotakSession
|
The login. |
required |
on_updates
|
callable
|
Called on this socket's thread with a message's order and position |
required |
logger
|
Logger
|
Where the connection reports. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example programs
examples/stock_brokers/websockets/kotak/KotakOrderUpdatesSocket/example_1_orders_and_positions.py
"""Connects Kotak's order update socket to the session's own host and prints the orders and positions it hands on.
`KotakOrderUpdatesSocket` connects to `wss://<host>/realtime`, where the host is the `base_url` Kotak assigned the session at login, and sends Kotak's connection frame: a raw string that looks like JSON but deliberately is not, with the access token and session id. Kotak acknowledges it with `{"ak": "ok", "type": "cn"}`, and from then on sends `order` and `position` messages whose `data` carries Kotak's abbreviated fields, handed on exactly as sent.
This program scripts one connection delivering the acknowledgement, an open order, a position and the order completed. No real socket is opened: a stand-in `websocket` package plays the messages and prints what the socket sends, and a stand-in `stock_brokers.api.kotak` module lets the real `KotakSession` supply the host, token and session id without logging in.
Notice the URL built from the session's host, the unquoted connection frame, and that orders and positions reach the callback as separate lists.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakOrderUpdatesSocket/example_1_orders_and_positions.py
"""
import json
import logging
import sys
import types
from stock_brokers.websockets.kotak import (
KotakOrderUpdatesSocket,
KotakSession,
)
class LoginRecord:
"""The Kotak login state every stand-in API object reads, standing in for the Redis `last_login` hash.
Attributes:
access_token (str | None): The access token in force now.
session_id (str | None): The session id in force now.
login_count (int): How many logins the stand-in API has made.
"""
access_token = None
session_id = None
login_count = 0
class StandInKotakAPI:
"""A stand-in for `KotakAPI` whose construction is a pretend login."""
def __init__(self):
"""Logs in, which here only issues the next numbered access token and session id.
Returns:
None: This method returns nothing.
"""
LoginRecord.login_count = LoginRecord.login_count + 1
LoginRecord.access_token = f'kotak-token-{LoginRecord.login_count}'
LoginRecord.session_id = f'kotak-sid-{LoginRecord.login_count}'
print(f'Stand-in login number {LoginRecord.login_count}')
def _current_login(self):
"""The login in force now, as the real class reads it from Redis.
Returns:
dict | None: The access token, session id and host, or None before any login.
"""
if LoginRecord.access_token is None:
return None
return {
'access_token': LoginRecord.access_token,
'sid': LoginRecord.session_id,
'base_url': 'https://e21.kotaksecurities.com/',
}
class HandshakeRefused(Exception):
"""A refused websocket handshake, carrying the HTTP status the way `websocket-client` does.
Attributes:
status_code (int): The HTTP status the broker answered with.
"""
def __init__(self, status_code):
"""Keeps the status.
Args:
status_code (int): The HTTP status the broker answered with.
Returns:
None: This method returns nothing.
"""
super().__init__(f'Handshake status {status_code}')
self.status_code = status_code
class ScriptedConnection:
"""A stand-in for `websocket.WebSocketApp` that either refuses the handshake or plays frames to the socket's handlers.
Attributes:
url (str): The URL the socket asked for.
plan (dict): Either a `refuse_with` status or a list of `frames`.
handlers (dict): The socket's callbacks, by name.
closed (bool): Whether the socket asked to close, after which no more frames are delivered.
"""
def __init__(self, url, plan, handlers):
"""Keeps what the socket passed in.
Args:
url (str): The URL the socket asked for.
plan (dict): Either a `refuse_with` status or a list of `frames`.
handlers (dict): The socket's callbacks, by name.
Returns:
None: This method returns nothing.
"""
self.url = url
self.plan = plan
self.handlers = handlers
self.closed = False
def run_forever(self, **options):
"""Plays the plan and returns when the pretend connection closes.
Args:
**options (dict): The ping and TLS settings the socket asks for.
Returns:
None: This method returns nothing.
"""
print(f'Connecting to {self.url}')
if 'refuse_with' in self.plan:
self.handlers['on_error'](self, HandshakeRefused(self.plan['refuse_with']))
self.handlers['on_close'](self, None, None)
return
self.handlers['on_open'](self)
for frame in self.plan['frames']:
if self.closed:
break
self.handlers['on_message'](self, frame)
self.handlers['on_close'](self, 1000, 'normal closure')
def send(self, data, opcode=None):
"""Prints a message the socket sends to the broker.
Args:
data (str | bytes): The message.
opcode (int | None): The frame type, when the socket names one.
Returns:
None: This method returns nothing.
"""
if isinstance(data, bytes):
print(f'Sent binary frame of type {data[2]}: {data.hex()}')
else:
print(f'Sent: {data}')
def close(self, **options):
"""Notes that the socket asked to close.
Args:
**options (dict): Any close status the socket passes.
Returns:
None: This method returns nothing.
"""
self.closed = True
class FrameOpcodes:
"""The one websocket frame opcode the Kotak quotes socket names.
Attributes:
OPCODE_BINARY (int): The binary frame opcode.
"""
OPCODE_BINARY = 2
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.
ABNF (FrameOpcodes): The frame opcodes, as `websocket.ABNF` holds them.
"""
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
self.ABNF = FrameOpcodes()
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 OrdersAndPositionsExample:
"""Runs the Kotak order update socket against one scripted connection.
Attributes:
socket (KotakOrderUpdatesSocket): The socket being shown.
"""
def __init__(self):
"""Builds the messages, the stand-ins, the session and the socket.
Returns:
None: This method returns nothing.
"""
logging.basicConfig(stream=sys.stdout, format='%(levelname)s %(message)s', level=logging.INFO)
api_module = types.ModuleType('stock_brokers.api.kotak')
api_module.KotakAPI = StandInKotakAPI
sys.modules['stock_brokers.api.kotak'] = api_module
acknowledgement = {
'ak': 'ok',
'type': 'cn',
'task': 'cn',
}
position = {
'type': 'position',
'data': {
'trdSym': 'INFY-EQ',
'exSeg': 'nse_cm',
'flBuyQty': 10,
'flSellQty': 0,
},
}
plans = [
{
'frames': [
json.dumps(acknowledgement),
json.dumps(self.order('open', 0)),
json.dumps(position),
json.dumps(self.order('complete', 10)),
],
},
]
websocket_module = ScriptedWebsocketModule(plans)
sys.modules['websocket'] = websocket_module
logger = logging.getLogger('kotak.orders')
self.socket = KotakOrderUpdatesSocket(KotakSession(logger), self.print_updates, logger)
websocket_module.when_finished = self.socket.close
def print_updates(self, orders, positions, received_at):
"""Prints each order and position the socket handed on.
Args:
orders (list): The message's order `data` objects, exactly as Kotak sent them.
positions (list): The message's position `data` objects, exactly as Kotak sent them.
received_at (datetime.datetime): When the message arrived, which is not printed.
Returns:
None: This method returns nothing.
"""
for order in orders:
print(f"Order {order['nOrdNo']} {order['ordSt']}: {order['trnsTp']} {order['fldQty']} of {order['qty']} {order['trdSym']}")
for position in positions:
print(f"Position {position['trdSym']}: bought {position['flBuyQty']} sold {position['flSellQty']}")
def order(self, status, filled_quantity):
"""One Kotak order update for an INFY buy, with Kotak's abbreviated fields.
Args:
status (str): Kotak's order status.
filled_quantity (int): How much has filled.
Returns:
dict: The message.
"""
return {
'type': 'order',
'data': {
'nOrdNo': '260925000012345',
'ordSt': status,
'trdSym': 'INFY-EQ',
'exSeg': 'nse_cm',
'trnsTp': 'B',
'qty': 10,
'fldQty': filled_quantity,
'prc': '1500.00',
},
}
def run(self):
"""Runs the socket's reconnect loop until the scripted connection ends.
Returns:
None: This method returns nothing.
"""
self.socket.run_forever()
print(f'Gave up: {self.socket.gave_up}')
if __name__ == '__main__':
OrdersAndPositionsExample().run()
It prints:
Stand-in login number 1
Connecting to wss://e21.kotaksecurities.com/realtime
Sent: {type:cn,Authorization:kotak-token-1,Sid:kotak-sid-1,src:WEB}
INFO Order updates socket opened. Waiting for Kotak's acknowledgement.
INFO Order updates socket authenticated. Waiting for updates.
Order 260925000012345 open: B 0 of 10 INFY-EQ
Position INFY-EQ: bought 10 sold 0
Order 260925000012345 complete: B 10 of 10 INFY-EQ
INFO Order updates socket closed with status 1000: normal closure
INFO Order updates socket stopped.
Gave up: False
examples/stock_brokers/websockets/kotak/KotakOrderUpdatesSocket/example_2_refused_connection_frame.py
"""Shows Kotak's order update socket logging in again, without checking, after Kotak refuses its connection frame.
Kotak refuses a dead session's connection frame with `{"ak": "nk", "type": "failed to connect", ...}`. `KotakOrderUpdatesSocket` then discards whatever else the same message carried, closes the connection, and its reconnect loop calls `KotakSession.log_in_again_without_checking`, which logs in whether or not another process already replaced the session.
This program's first connection answers with a JSON list holding the refusal and an order, which is discarded. The second connection is acknowledged and delivers the order completed. No real socket is opened: a stand-in `websocket` package plays the messages, and a stand-in `stock_brokers.api.kotak` module counts each `KotakAPI` construction as one login. The backoff is set to zero seconds.
Notice that the order in the refused message never reaches the callback, and that the second connection frame carries the new token and session id.
Run it from the project root:
python examples/stock_brokers/websockets/kotak/KotakOrderUpdatesSocket/example_2_refused_connection_frame.py
"""
import json
import logging
import sys
import types
from stock_brokers.websockets.kotak import (
KotakOrderUpdatesSocket,
KotakSession,
)
class LoginRecord:
"""The Kotak login state every stand-in API object reads, standing in for the Redis `last_login` hash.
Attributes:
access_token (str | None): The access token in force now.
session_id (str | None): The session id in force now.
login_count (int): How many logins the stand-in API has made.
"""
access_token = None
session_id = None
login_count = 0
class StandInKotakAPI:
"""A stand-in for `KotakAPI` whose construction is a pretend login."""
def __init__(self):
"""Logs in, which here only issues the next numbered access token and session id.
Returns:
None: This method returns nothing.
"""
LoginRecord.login_count = LoginRecord.login_count + 1
LoginRecord.access_token = f'kotak-token-{LoginRecord.login_count}'
LoginRecord.session_id = f'kotak-sid-{LoginRecord.login_count}'
print(f'Stand-in login number {LoginRecord.login_count}')
def _current_login(self):
"""The login in force now, as the real class reads it from Redis.
Returns:
dict | None: The access token, session id and host, or None before any login.
"""
if LoginRecord.access_token is None:
return None
return {
'access_token': LoginRecord.access_token,
'sid': LoginRecord.session_id,
'base_url': 'https://e21.kotaksecurities.com/',
}
class HandshakeRefused(Exception):
"""A refused websocket handshake, carrying the HTTP status the way `websocket-client` does.
Attributes:
status_code (int): The HTTP status the broker answered with.
"""
def __init__(self, status_code):
"""Keeps the status.
Args:
status_code (int): The HTTP status the broker answered with.
Returns:
None: This method returns nothing.
"""
super().__init__(f'Handshake status {status_code}')
self.status_code = status_code
class ScriptedConnection:
"""A stand-in for `websocket.WebSocketApp` that either refuses the handshake or plays frames to the socket's handlers.
Attributes:
url (str): The URL the socket asked for.
plan (dict): Either a `refuse_with` status or a list of `frames`.
handlers (dict): The socket's callbacks, by name.
closed (bool): Whether the socket asked to close, after which no more frames are delivered.
"""
def __init__(self, url, plan, handlers):
"""Keeps what the socket passed in.
Args:
url (str): The URL the socket asked for.
plan (dict): Either a `refuse_with` status or a list of `frames`.
handlers (dict): The socket's callbacks, by name.
Returns:
None: This method returns nothing.
"""
self.url = url
self.plan = plan
self.handlers = handlers
self.closed = False
def run_forever(self, **options):
"""Plays the plan and returns when the pretend connection closes.
Args:
**options (dict): The ping and TLS settings the socket asks for.
Returns:
None: This method returns nothing.
"""
print(f'Connecting to {self.url}')
if 'refuse_with' in self.plan:
self.handlers['on_error'](self, HandshakeRefused(self.plan['refuse_with']))
self.handlers['on_close'](self, None, None)
return
self.handlers['on_open'](self)
for frame in self.plan['frames']:
if self.closed:
break
self.handlers['on_message'](self, frame)
self.handlers['on_close'](self, 1000, 'normal closure')
def send(self, data, opcode=None):
"""Prints a message the socket sends to the broker.
Args:
data (str | bytes): The message.
opcode (int | None): The frame type, when the socket names one.
Returns:
None: This method returns nothing.
"""
if isinstance(data, bytes):
print(f'Sent binary frame of type {data[2]}: {data.hex()}')
else:
print(f'Sent: {data}')
def close(self, **options):
"""Notes that the socket asked to close.
Args:
**options (dict): Any close status the socket passes.
Returns:
None: This method returns nothing.
"""
self.closed = True
class FrameOpcodes:
"""The one websocket frame opcode the Kotak quotes socket names.
Attributes:
OPCODE_BINARY (int): The binary frame opcode.
"""
OPCODE_BINARY = 2
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.
ABNF (FrameOpcodes): The frame opcodes, as `websocket.ABNF` holds them.
"""
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
self.ABNF = FrameOpcodes()
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 RefusedConnectionFrameExample:
"""Runs the Kotak order update socket through a refusal and a good connection.
Attributes:
socket (KotakOrderUpdatesSocket): The socket being shown.
"""
def __init__(self):
"""Builds the messages, the stand-ins, the session and the socket.
Returns:
None: This method returns nothing.
"""
logging.basicConfig(stream=sys.stdout, format='%(levelname)s %(message)s', level=logging.INFO)
api_module = types.ModuleType('stock_brokers.api.kotak')
api_module.KotakAPI = StandInKotakAPI
sys.modules['stock_brokers.api.kotak'] = api_module
refusal = {
'ak': 'nk',
'type': 'failed to connect',
'msg': 'Invalid session',
}
acknowledgement = {
'ak': 'ok',
'type': 'cn',
}
refused_batch = [
refusal,
self.order('open', 0),
]
plans = [
{
'frames': [
json.dumps(refused_batch),
],
},
{
'frames': [
json.dumps(acknowledgement),
json.dumps(self.order('complete', 10)),
],
},
]
websocket_module = ScriptedWebsocketModule(plans)
sys.modules['websocket'] = websocket_module
logger = logging.getLogger('kotak.orders')
self.socket = KotakOrderUpdatesSocket(KotakSession(logger), self.print_updates, logger)
self.socket.MIN_BACKOFF_SECONDS = 0
websocket_module.when_finished = self.socket.close
def print_updates(self, orders, positions, received_at):
"""Prints each order and position the socket handed on.
Args:
orders (list): The message's order `data` objects, exactly as Kotak sent them.
positions (list): The message's position `data` objects, exactly as Kotak sent them.
received_at (datetime.datetime): When the message arrived, which is not printed.
Returns:
None: This method returns nothing.
"""
for order in orders:
print(f"Order {order['nOrdNo']} {order['ordSt']}: {order['trnsTp']} {order['fldQty']} of {order['qty']} {order['trdSym']}")
for position in positions:
print(f"Position {position['trdSym']}: bought {position['flBuyQty']} sold {position['flSellQty']}")
def order(self, status, filled_quantity):
"""One Kotak order update for an INFY buy, with Kotak's abbreviated fields.
Args:
status (str): Kotak's order status.
filled_quantity (int): How much has filled.
Returns:
dict: The message.
"""
return {
'type': 'order',
'data': {
'nOrdNo': '260925000012345',
'ordSt': status,
'trdSym': 'INFY-EQ',
'exSeg': 'nse_cm',
'trnsTp': 'B',
'qty': 10,
'fldQty': filled_quantity,
'prc': '1500.00',
},
}
def run(self):
"""Runs the socket's reconnect loop until the scripted connections run out.
Returns:
None: This method returns nothing.
"""
self.socket.run_forever()
print(f'Gave up: {self.socket.gave_up}')
print(f'Logins made: {LoginRecord.login_count}')
if __name__ == '__main__':
RefusedConnectionFrameExample().run()
It prints:
Stand-in login number 1
Connecting to wss://e21.kotaksecurities.com/realtime
Sent: {type:cn,Authorization:kotak-token-1,Sid:kotak-sid-1,src:WEB}
INFO Order updates socket opened. Waiting for Kotak's acknowledgement.
ERROR Kotak refused the order update connection: Invalid session
INFO Order updates socket closed with status 1000: normal closure
WARNING Logging in to Kotak again.
Stand-in login number 2
WARNING Order updates socket disconnected. Reconnecting in 0 seconds.
Connecting to wss://e21.kotaksecurities.com/realtime
Sent: {type:cn,Authorization:kotak-token-2,Sid:kotak-sid-2,src:WEB}
INFO Order updates socket opened. Waiting for Kotak's acknowledgement.
INFO Order updates socket authenticated. Waiting for updates.
Order 260925000012345 complete: B 10 of 10 INFY-EQ
INFO Order updates socket closed with status 1000: normal closure
INFO Order updates socket stopped.
Gave up: False
Logins made: 2
Methods:
| Name | Description |
|---|---|
close |
Closes the connection and stops reconnecting. |
run_forever |
Connects, and reconnects with backoff, until closed or unable to recover. |
Source code in stock_brokers/websockets/kotak.py
Opens the websocket on the session's own host and blocks until it closes.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
When the stored login has no host. |
Source code in stock_brokers/websockets/kotak.py
Logs in again, without checking whether another process already replaced the session.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Sends the raw connection frame with the token and session id in force now.
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/kotak.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/kotak.py
Hands a message's orders and positions on, and treats a refused connection frame as a dead session.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The connection the message arrived on. |
required |
message
|
bytes | str
|
The message: one JSON object or a JSON list of them. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Source code in stock_brokers/websockets/kotak.py
Closes the connection and stops reconnecting.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example usage
Connects, and reconnects with backoff, until closed or unable to recover.
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |
Example usage
Source code in stock_brokers/websockets/base.py
Whether a socket that has already logged in again gives up on this failure.
The loop asks only when the connection was refused or has failed six times in a row. By default the answer is yes, so a refusal straight after logging in again ends the socket. A broker whose refusals are not always about the login overrides this to give up only after six more failed connects.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
failed_connects
|
int
|
How many connects in a row have failed since the last login. |
required |
Returns:
| Type | Description |
|---|---|
bool
|
True to give up now. |
Source code in stock_brokers/websockets/base.py
Reports that the connection closed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
websocket_connection
|
WebSocketApp
|
The connection that closed. |
required |
status_code
|
int | None
|
The close status the broker sent. |
required |
message
|
str | None
|
The close reason the broker sent. |
required |
Returns:
| Type | Description |
|---|---|
None
|
This method returns nothing. |