190 lines
6.7 KiB
Python
190 lines
6.7 KiB
Python
"""
|
|
|
|
AUTHOR:
|
|
|
|
Khushal P Soonderji
|
|
|
|
DATE:
|
|
|
|
Tuesday, 24th 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.queue.kafka import ProducerKafka, create_config
|
|
|
|
# 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 Zerodha and related to ticks:
|
|
INSTRUMENT_TOKENS = []
|
|
INSTRUMENT_LOOKUP = {}
|
|
|
|
# For Kafka:
|
|
kafka_producer = ProducerKafka(
|
|
topic = "tickers",
|
|
config = 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"
|
|
)
|
|
)
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** FUNCTIONS ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
def to_kafka(tick: TradingTick) -> bool:
|
|
|
|
success = False
|
|
summary = tick.summary
|
|
success = kafka_producer.produce(value = summary)
|
|
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):
|
|
|
|
ticks = TradingTick.from_zerodha_kite(ticks = ticks, instrument_lookup = INSTRUMENT_LOOKUP)
|
|
results = [to_kafka(tick) for tick in ticks]
|
|
success = sum(results)
|
|
print(f"TICKS: {len(ticks): <4} | PRODUCED: {success: <4}{' | FAILURE(S)!' if success < len(results)else ''}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------------------------------
|
|
|
|
|
|
def main():
|
|
|
|
# Global vars:
|
|
global INSTRUMENT_TOKENS
|
|
global INSTRUMENT_LOOKUP
|
|
|
|
# Load Zerodha credentials:
|
|
creds = json.from_file(r"../../creds/zerodha/api.json")
|
|
api_key = creds["apiKey"]
|
|
access_token = creds["accessToken"]
|
|
|
|
# Create an instance of Zerodha's Kite connection:
|
|
kite = KiteConnect(api_key = api_key)
|
|
kite.set_access_token(access_token)
|
|
|
|
# Get a list of instruments to work with:
|
|
instruments = kite.instruments(exchange = "MCX")
|
|
instruments = instruments[:100]
|
|
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()
|
|
|
|
# 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)
|
|
|
|
|
|
# *****************************************************************************************************************
|
|
# ***** ****
|
|
# *** MAIN PROGRAM ***
|
|
# ***** ****
|
|
# *****************************************************************************************************************
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
main()
|
|
while True: time.sleep(3_600.00)
|