khushalps/api_utils_converse_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

(20250101) A more formal Tick-In script has been created.

Browse Source
This commit is contained in:
khushal
2025-01-01 09:36:06 +00:00
parent c668cfa4c8
commit 6691a68b3a
5 changed files with 442 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
background/finstitutions/trading/to_kafka.py
+409 -149
View File
@@ -6,7 +6,7 @@
DATE:
Monday, 30th Dec. 2024
Wednesday, 1st Jan., 2025.
OBJECTIVE:
@@ -42,23 +42,41 @@ import os
# My utils:
from utils_v2.string import json
from utils_v2.string import regex
from utils_v2.system import files
from utils_v2.database.async_mongo_v2 import AsyncMongo
from utils_v2.cache.async_redis_cache_v2 import AsyncRedisCache
from utils_v2.queue.kafka.controllers.kafka import ProducerKafka, ConsumerKafka
from utils_v2.serialization.json_serializer import JSONSerializer
# To make HTTP calls:
import httpx
import socket
# To work with date and time:
import datetime
import time
# Controllers:
from controllers_v2.finstitutions.trading.all_trading import AllTradingController
# Models:
from models.core.auth_token import CoreAuthTokenModel
from models.finstitutions.trading.symbols import TradingSymbol
from models.finstitutions.trading.ticks import TradingTick
# To work with Zerodha's Kite platform:
from kiteconnect import KiteConnect, KiteTicker
# To work with datatypes:
from typing import List
# For debugging:
from icecream import IceCreamDebugger
# For asynchronous operations:
import asyncio
# *****************************************************************************************************************
# ***** ****
@@ -67,7 +85,22 @@ from kiteconnect import KiteConnect, KiteTicker
# *****************************************************************************************************************
# --- Nothing Yet
# To make API calls:
http_client = httpx.Client(
limits = httpx.Limits(
max_connections = 100, # ............ Maximum number of connections allowed in the pool.
max_keepalive_connections = 50, # ... Maximum number of connections that can be kept alive.
),
timeout = httpx.Timeout(
pool = 120.0, # .... Time to wait for a free connection from the pool.
connect = 2.5, # ... Time to wait for establishing a connection to the server.
write = 10.0, # .... Time to wait for sending data.
read = 9.9 # ....... Time to wait for receiving data.
)
)
# For debugging:
printer = IceCreamDebugger(prefix = "Tick-In (Sful) | ", includeContext = True)
# *****************************************************************************************************************
@@ -77,49 +110,23 @@ from kiteconnect import KiteConnect, KiteTicker
# *****************************************************************************************************************
# For user management:
accounts = {}
# To identify this process:
SERVER_HOSTNAME = str(socket.gethostname())
TOKEN_KEY = None
AUTH_TOKEN = None
# For Zerodha and related to ticks:
INSTRUMENT_TOKENS = []
INSTRUMENT_LOOKUP = {}
# Pertaining to the behaviour of this script:
SCRIPT_DATA = {}
# For Kafka:
cwd = files.get_cwd()
parent_dir = cwd
kafka_consumer = ConsumerKafka(
topic = "tick-listners",
config = ConsumerKafka.create_config(
bootstrap_servers = "del.ditscentre.in:9092",
group_id = "123",
security_protocol = "SSL",
ca_file = r"../../../creds/kafka/cert_authority.pem",
cert_file = r"../../../creds/kafka/fullchain.pem",
key_file = r"../../../creds/kafka/privkey.pem"
# ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"),
# cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"),
# key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem")
),
debug = False
)
kafka_producer = ProducerKafka(
topic = "tickers",
config = ProducerKafka.create_config(
bootstrap_servers = "del.ditscentre.in:9092",
security_protocol = "SSL",
ca_file = r"../../../creds/kafka/cert_authority.pem",
cert_file = r"../../../creds/kafka/fullchain.pem",
key_file = r"../../../creds/kafka/privkey.pem"
# ca_file = os.path.join(parent_dir, "creds", "kafka", "cert_authority.pem"),
# cert_file = os.path.join(parent_dir, "creds", "kafka", "fullchain.pem"),
# key_file = os.path.join(parent_dir, "creds", "kafka", "privkey.pem")
),
debug = False
)
# For kafka:
kafka_producer: ProducerKafka | None = None
kafka_consumer: ConsumerKafka | None = None
TOTAL_TICK_COUNT = 0
TICKS_SINCE_FLUSH = 0
# For metrics:
tick_count = 0
ticks_since_flush = 0
# For Zerodha-Kite:
ZERODHA_INSTRUMENT_TOKENS = []
ZERODHA_INSTRUMENT_LOOKUP = {}
# *****************************************************************************************************************
@@ -129,129 +136,368 @@ ticks_since_flush = 0
# *****************************************************************************************************************
def flush_kafka():
print("FLUSHING!")
kafka_producer.flush()
def ticks_to_kafka(ticks: List[TradingTick]) -> int:
"""
This function simply takes the ticks received from the stockbroker's servers and sends them to the internal Kafka
queue for various end consumption use cases.
:param ticks: The modeled ticks that must be pushed to the Kafka queue.
:return: The count of the ticks that were successfully pushed to the queue.
"""
# Declare the required global variables:
global TOTAL_TICK_COUNT
global TICKS_SINCE_FLUSH
# Start with basic variables:
total_count = len(ticks)
success_count = 0
failure_count = 0
# Push out the ticks to the queue:
for tick in ticks:
# Flush the existing messages if needed:
TOTAL_TICK_COUNT += 1
TICKS_SINCE_FLUSH += 1
if TICKS_SINCE_FLUSH > 50_000:
kafka_producer.flush()
TICKS_SINCE_FLUSH = 0
# Push this one tick to the queue:
success = kafka_producer.produce(value = tick.summary)
if success: success_count += 1
else: failure_count += 1
# Debugging print:
ticks_str = f"TICKS: {total_count:6,} | PRODUCED: {success_count:6,} | TOTAL: {TOTAL_TICK_COUNT:12,}"
printer(ticks_str)
# Done here:
return success_count
# ---------------------------------------------------------------------------------------------------------------------
def to_kafka(tick: TradingTick) -> bool:
def on_zerodha_kite_connect(ws, response) -> None:
global tick_count
global ticks_since_flush
"""
To do something when we get connected to Zerodha's Kite API successfully.
:param ws: The websocket object that is connected to Zerodha.
:param response: Some input from Zerodha's library.
:return: None
"""
tick_count += 1
ticks_since_flush += 1
if ticks_since_flush >= 50_000:
flush_kafka()
ticks_since_flush = 0
success = False
summary = tick.summary
summary["messageType"] = "ticks"
success = kafka_producer.produce(value = summary)
if not success:
print("ERROR ON TICK NO.:", tick_count)
flush_kafka()
return success
printer("Zerodha Kite WS connected.")
ws.subscribe(ZERODHA_INSTRUMENT_TOKENS)
ws.set_mode(ws.MODE_FULL, ZERODHA_INSTRUMENT_TOKENS)
printer("Zerodha Kite instruments subscribed.", len(ZERODHA_INSTRUMENT_TOKENS))
# ---------------------------------------------------------------------------------------------------------------------
def on_connect(ws, response):
def on_zerodha_kite_ticks(ws, ticks) -> None:
print("\n\n")
print("ON CONNECT:")
print("Successfully connected. Response: {}".format(response))
ws.subscribe(INSTRUMENT_TOKENS)
ws.set_mode(ws.MODE_FULL, INSTRUMENT_TOKENS)
print(f"Subscribed to {len(INSTRUMENT_TOKENS):,} tokens in 'Full' mode.")
print("\n\n")
"""
The function that gets called when Zerodha's tick updates come in.
:param ws: The websocket object that is connected to Zerodha.
:param ticks: The actual tick data received from Zerodha's Kite platform.
:return: None
"""
# Model the raw input ticks:
ticks = TradingTick.from_zerodha_kite(
ticks = ticks,
instrument_lookup = ZERODHA_INSTRUMENT_LOOKUP
)
# Send the ticks to Kafka:
success_count = ticks_to_kafka(ticks)
# ---------------------------------------------------------------------------------------------------------------------
def on_ticks(ws, ticks):
def setup_zerodha_kite_feed(
auth_token: CoreAuthTokenModel,
total_instruments: List[dict]
) -> bool:
# print("TICK SAMPLE:", json.to_string(ticks, default=str))
ticks = TradingTick.from_zerodha_kite(ticks = ticks, instrument_lookup = INSTRUMENT_LOOKUP)
# print("TICK SAMPLE:", json.to_string(ticks[0].model_dump(), default=str))
# print("TICK SAMPLE:", json.to_string(ticks[0].summary, default=str))
results = [to_kafka(tick) for tick in ticks]
success = sum(results)
print(f"TICKS: {len(ticks): <6,} | PRODUCED: {success: <6,} | TOTAL: {tick_count: >10,}{' | FAILURE(S)!' if success < len(results)else ''}")
"""
Sets up the websocket for Zerodha's Kite platform.
:param auth_token: The auth-token model to use to se the feed up.
:param total_instruments: The list of dicts that describe the instruments we need to subscribe to.
:return: True if successful, else False.
"""
# Declare the required global variables:
global ZERODHA_INSTRUMENT_TOKENS
global ZERODHA_INSTRUMENT_LOOKUP
# ---------------------------------------------------------------------------------------------------------------------
# Filter the input instruments. This has to be tuned to match what you get from the API.
valid_broker_symbols = []
valid_broker_tokens = []
for i in total_instruments:
symbol = i["symbol"]
broker_token = i["broker_token"]
if broker_token is not None and i["source"] == "zerodha":
valid_broker_symbols.append(symbol)
valid_broker_tokens.append(str(broker_token))
def init():
# Global vars:
global INSTRUMENT_TOKENS
global INSTRUMENT_LOOKUP
# Load Zerodha credentials:
creds = json.from_file(r"../../../creds/zerodha/api.json")
# creds = json.from_file(os.path.join(parent_dir, "creds", "zerodha", "api.json"))
api_key = creds["apiKey"]
access_token = creds["accessToken"]
# Get the instruments of interest:
response = httpx.post(url = r"https://api.thecaoffice.com/markets/watchlist/distincts")
instruments_of_interest = response.json()["data"]["rs0"]
print("TOTAL INSTR. OF INTEREST:", len(instruments_of_interest))
symbols_of_interest = [i["symbol"] for i in instruments_of_interest]
# Debugging print:
printer(len(total_instruments), len(valid_broker_symbols), len(valid_broker_tokens))
# Create an instance of Zerodha's Kite connection:
kite = KiteConnect(api_key = api_key)
kite.set_access_token(access_token)
kite = KiteConnect(api_key = auth_token.auth["apiKey"])
kite.set_access_token(auth_token.token["accessToken"])
# Get the entire list of instruments:
instruments = []
# instruments += kite.instruments(exchange = "NSE")
# instruments += kite.instruments(exchange = "NFO")
# instruments += kite.instruments(exchange = "BSE")
# instruments += kite.instruments(exchange = "BFO")
instruments += kite.instruments(exchange = "NSE")
instruments += kite.instruments(exchange = "NFO")
instruments += kite.instruments(exchange = "BSE")
instruments += kite.instruments(exchange = "BFO")
instruments += kite.instruments(exchange = "MCX")
# instruments += kite.instruments(exchange = "CDS")
# instruments += kite.instruments(exchange = "BCD")
instruments += kite.instruments(exchange = "CDS")
instruments += kite.instruments(exchange = "BCD")
# # Pick the instruments of interest:
# instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments[:1000]]
# instruments = [
# TradingSymbol.from_zerodha_kite(i) for i in instruments
# if i["tradingsymbol"] in symbols_of_interest or i["name"] in symbols_of_interest
# ]
instruments = [
TradingSymbol.from_zerodha_kite(i) for i in instruments
if i["instrument_token"] in [109760007]
]
# Convert the loaded instruments to their modelled form:
instruments = [TradingSymbol.from_zerodha_kite(i) for i in instruments]
# Create the lookup:
for i in instruments:
INSTRUMENT_TOKENS.append(i.brokerToken)
INSTRUMENT_LOOKUP[i.brokerToken] = i.model_dump()
if str(i.brokerToken) in valid_broker_tokens:
ZERODHA_INSTRUMENT_TOKENS.append(i.brokerToken)
ZERODHA_INSTRUMENT_LOOKUP[i.brokerToken] = i.model_dump()
printer("Zerodha instruments loaded.")
# Start the websocket with Zerodha:
# If there are no symbols or too many symbols, we return with failure:
if not (1 <= len(ZERODHA_INSTRUMENT_TOKENS) <= 3_000):
printer("Zerodha must have min. 1 and max. 3,000 symbols in one WS.", len(ZERODHA_INSTRUMENT_TOKENS))
return False
# Create the websocket to Zerodha:
kite_ws = KiteTicker(
api_key = api_key,
access_token = access_token
api_key = auth_token.auth["apiKey"],
access_token = auth_token.token["accessToken"]
)
# Assign the callbacks:
kite_ws.on_connect = on_connect
kite_ws.on_ticks = on_ticks
kite_ws.on_connect = on_zerodha_kite_connect
kite_ws.on_ticks = on_zerodha_kite_ticks
# If you choose to go threaded, you will need to work purely with callbacks.
# You will need to have an infinite loop in the main thread.
# Run the websocket in a background thread and release the main thread:
kite_ws.connect(threaded = True)
# Done here:
return True
# ---------------------------------------------------------------------------------------------------------------------
async def get_auth_token(
script_cred: dict,
token_key: str,
debug: bool
) -> CoreAuthTokenModel | None:
"""
We need the full auth-token model to run the tick-ingestion script. This is a separate special coroutine that will
use the existing async code to fetch the full document from the database.
:param script_cred: The credentials of the script (needed to connect to the database).
:param token_key: The token 'key' whose full auth-token model needs to be fetched.
:param debug: Whether, or not, you would like to print debugging details.
:return: The full auth-token of the account if found, or None if not found.
"""
# ┳┳┓
# ┃┃┃┏┓┏┓┏┓┏┓
# ┛ ┗┗┛┛┗┗┫┗┛
# ┛
data_mongo = AsyncMongo(
connection_string = script_cred["mongoDb"]["data"]["connectionString"],
database_name = script_cred["mongoDb"]["data"]["dbName"],
max_connections = script_cred["mongoDb"]["data"]["poolSize"],
debug = debug
)
await data_mongo.connect()
# ┳┓ ┓• ┏┓ ┓
# ┣┫┏┓┏┫┓┏ ━━ ┃ ┏┓┏┣┓┏┓
# ┛┗┗ ┗┻┗┛ ┗┛┗┻┗┛┗┗
general_cache = AsyncRedisCache(
connection_string = script_cred["redisCache"]["general"]["connectionString"],
serializer = JSONSerializer(),
debug = debug,
debug_prefix = "General Cache | "
)
await general_cache.connect()
# ┏┓ ┓┓ ┓┏┏┓
# ┃ ┏┓┏┓╋┏┓┏┓┃┃┏┓┏┓┏ ┃┃┏┛
# ┗┛┗┛┛┗┗┛ ┗┛┗┗┗ ┛ ┛ ┗┛┗━
controller = AllTradingController(
cache = general_cache,
http_client = httpx.AsyncClient(timeout = 10.0),
alert_url = SCRIPT_DATA["alerts"]["url"],
debug = debug
)
# Fetch and return the auth-token model for the given key:
return await controller.get_token_from_key(
mongo_data_conn = data_mongo,
token_key = token_key
)
# ---------------------------------------------------------------------------------------------------------------------
def init(
script_id: str,
token_key: str,
debug: bool
) -> bool:
# Declare the required global variables:
global TOKEN_KEY
global AUTH_TOKEN
global SCRIPT_DATA
global kafka_producer
global kafka_consumer
# Basic stuff:
TOKEN_KEY = token_key
if not debug: printer.disable()
# ┏┓ ┓ ┓ ┳┓
# ┃ ┏┓┏┓┏┫ ┏┓┏┓┏┫ ┃┃┏┓╋┏┓
# ┗┛┛ ┗ ┗┻ ┗┻┛┗┗┻ ┻┛┗┻┗┗┻
# Get the script credentials:
response = http_client.get(
url = r"https://nexcom.ditscentre.in/internal/cred/get",
headers = {"X-Script-Id": script_id}
)
if response.status_code not in [200]:
print("FATAL: SCRIPT CREDENTIALS LOADING FAILED!")
return False
script_cred = response.json().get("data")
# Get the script data:
response = http_client.get(
url = r"https://nexcom.ditscentre.in/internal/data/get",
headers = {"X-Script-Id": script_id}
)
if response.status_code not in [200]:
print("FATAL: SCRIPT DATA LOADING FAILED!")
return False
SCRIPT_DATA = response.json().get("data")
# Done with this step:
printer("Cred and Data loaded.")
# ┓┏┓ ┏┓ ┏┓┓•
# ┃┫ ┏┓╋┃┏┏┓ ┃ ┃┓┏┓┏┓╋┏
# ┛┗┛┗┻┛┛┗┗┻ ┗┛┗┗┗ ┛┗┗┛
# Create the producer that will produce tick-by-tick data that it receives from the stockbroker's server:
producer_creds = script_cred["kafka"]["producer"]
kafka_producer = ProducerKafka(
config = ProducerKafka.create_config(
bootstrap_servers = producer_creds["config"]["bootstrapServers"],
security_protocol = producer_creds["config"].get("securityProtocol", "PLAINTEXT"),
ca_file = producer_creds["config"].get("caFile"),
cert_file = producer_creds["config"].get("certFile"),
key_file = producer_creds["config"].get("keyFile"),
client_id = f"{SERVER_HOSTNAME}_{TOKEN_KEY}"
),
topic = producer_creds["topic"],
serializer = JSONSerializer(),
debug = debug
)
if not kafka_producer.connect():
print("FATAL: KAFKA PRODUCER NOT CREATED!")
return False
printer("Kafka producer ready.")
# Create the consumer that will listen to changes in watchlist:
consumer_creds = script_cred["kafka"]["consumer"]
kafka_consumer = ConsumerKafka(
config = ConsumerKafka.create_config(
bootstrap_servers = consumer_creds["config"]["bootstrapServers"],
group_id = f"{SERVER_HOSTNAME}_{TOKEN_KEY}",
auto_offset_reset = consumer_creds["config"].get("autoOffsetReset", "latest"),
security_protocol = consumer_creds["config"].get("securityProtocol", "PLAINTEXT"),
ca_file = consumer_creds["config"].get("caFile"),
cert_file = consumer_creds["config"].get("certFile"),
key_file = consumer_creds["config"].get("keyFile"),
client_id = f"{SERVER_HOSTNAME}_{TOKEN_KEY}"
),
topic = consumer_creds["topic"],
serializer = JSONSerializer(),
debug = debug
)
if not kafka_consumer.connect():
print("FATAL: KAFKA CONSUMER NOT CREATED!")
return False
printer("Kafka consumer ready.")
# ┏┓ ┓ ┏┳┓ ┓
# ┣┫┓┏╋┣┓━━ ┃ ┏┓┃┏┏┓┏┓
# ┛┗┗┻┗┛┗ ┻ ┗┛┛┗┗ ┛┗
AUTH_TOKEN = asyncio.run(get_auth_token(
script_cred = script_cred,
token_key = token_key,
debug = debug
))
if not AUTH_TOKEN:
print("FATAL: AUTH-TOKEN FETCHING FAILED!")
return False
printer("Aut-token fetched.")
# ┳┓ ┏┓ ┓
# ┃┃┏┓╋┏┓━━┣ ┏┓┏┓┏┫
# ┻┛┗┻┗┗┻ ┻ ┗ ┗ ┗┻
# First we load the tokens of interest that we need to monitor:
response = http_client.post(
url = r"https://api.thecaoffice.com/markets/watchlist/distincts",
json = {"tokenKey": token_key}
)
instruments_of_interest = response.json()["data"]["rs0"]
# Set the feed up by the stockbroker:
live_feed = False
if AUTH_TOKEN.client == "zerodhaKite":
live_feed = setup_zerodha_kite_feed(
auth_token = AUTH_TOKEN,
total_instruments = instruments_of_interest
)
else: print(f"FATAL: INVALID TRADING CLIENT ({AUTH_TOKEN.client})!")
# If our live feed setup failed:
if not live_feed:
print("FATAL: LIVE FEED SETUP FAILED!")
return False
printer("Live feed ready.")
# ┳┓
# ┃┃┏┓┏┓┏┓
# ┻┛┗┛┛┗┗
# If everything went well, we return with success:
printer("Initialization done.")
return True
# ---------------------------------------------------------------------------------------------------------------------
@@ -259,28 +505,7 @@ def init():
def main():
while True:
messages = kafka_consumer.consume(count = 1, timeout = 5.0)
for message in messages:
token_key = message["tokenKey"]
account = None
# Create the websocket:
if accounts[token_key] not in list(accounts.keys()):
pass
account = accounts[token_key]
# Subscribe/unsubscribe:
ws = account["ws"]
if message["action"] == "subscribe":
ws.subscribe(message["brokerTokens"])
ws.set_mode(ws.MODE_FULL, message["brokerTokens"])
pass
elif message["action"] == "unsubscribe":
ws.unsubscribe(message["brokerTokens"])
pass
pass
# *****************************************************************************************************************
@@ -292,5 +517,40 @@ def main():
if __name__ == "__main__":
init()
main()
# To get args. from the terminal:
import argparse
# Get the config. from the command-line:
parser = argparse.ArgumentParser(
description = (
"Create one stateful process that gets tick-by-tick updates from stock brokers "
"and produces them on the common Kafka broker. This process will be dedicated to one account."
)
)
parser.add_argument(
"-s", "--script-id",
dest = "script_id",
type = str,
help = "The id of this script (will affect the loaded config)."
)
parser.add_argument(
"-t", "--token-key", "--token-id",
dest = "token_key",
type = str,
help = "The 'key' to use to retrieve the auth-token for accessing the broker account."
)
parser.add_argument(
"-d", "--debug",
dest = "debug",
action = "store_true",
help = "Whether, or not, you want to see debugging messages in the terminal.",
default = False
)
args = parser.parse_args()
# Run the main script:
if init(
script_id = args.script_id,
token_key = args.token_key,
debug = args.debug
): main()