""" AUTHOR: Khushal P Soonderji DATE: Monday, 30th Dec. 2024 OBJECTIVE: To get live updates from Zerodha and push them to Kafka. REFERENCES: N01. YouTube Webinar: https://www.youtube.com/watch?v=9vzd289Eedk 02. Official Example (GitHub): https://github.com/zerodha/pykiteconnect/blob/master/examples/threaded_ticker.py DOWNLOADS: N/A """ # ***************************************************************************************************************** # ***** **** # *** IMPORT *** # ***** **** # ***************************************************************************************************************** # To make sibling directories accessible for imports: import sys sys.path.append(".") sys.path.append("..") # System-level activities: import io import os # My utils: from utils_v2.string import json from utils_v2.system import files from utils_v2.queue.kafka import ProducerKafka, ConsumerKafka # To make HTTP calls: import httpx # To work with date and time: import datetime import time # Models: 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 # ***************************************************************************************************************** # ***** **** # *** MACROS / ONE-TIME INIT *** # ***** **** # ***************************************************************************************************************** # --- Nothing Yet # ***************************************************************************************************************** # ***** **** # *** VARIABLES *** # ***** **** # ***************************************************************************************************************** # For user management: accounts = {} # For Zerodha and related to ticks: INSTRUMENT_TOKENS = [] INSTRUMENT_LOOKUP = {} # 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 metrics: tick_count = 0 ticks_since_flush = 0 # ***************************************************************************************************************** # ***** **** # *** FUNCTIONS *** # ***** **** # ***************************************************************************************************************** def flush_kafka(): print("FLUSHING!") kafka_producer.flush() def to_kafka(tick: TradingTick) -> bool: global tick_count global ticks_since_flush 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 # --------------------------------------------------------------------------------------------------------------------- def on_connect(ws, response): 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") # --------------------------------------------------------------------------------------------------------------------- def on_ticks(ws, ticks): # 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 ''}") # --------------------------------------------------------------------------------------------------------------------- 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] # Create an instance of Zerodha's Kite connection: kite = KiteConnect(api_key = api_key) kite.set_access_token(access_token) # 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 = "MCX") # 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] ] # Create the lookup: for i in instruments: INSTRUMENT_TOKENS.append(i.brokerToken) INSTRUMENT_LOOKUP[i.brokerToken] = i.model_dump() # Start the websocket with Zerodha: kite_ws = KiteTicker( api_key = api_key, access_token = access_token ) # Assign the callbacks: kite_ws.on_connect = on_connect kite_ws.on_ticks = on_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. kite_ws.connect(threaded = True) # --------------------------------------------------------------------------------------------------------------------- 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 # ***************************************************************************************************************** # ***** **** # *** MAIN PROGRAM *** # ***** **** # ***************************************************************************************************************** if __name__ == "__main__": init() main()