"""

    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.system import files
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:
cwd = files.get_cwd()
parent_dir = cwd
kafka_producer = ProducerKafka(
    topic = "tickers",
    config = create_config(
        bootstrap_servers = "del.ditscentre.in:9092",
        # buffer_memory = 3_35_54_432,
        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
    print(json.to_string(summary))
    print(json.to_string(tick.model_dump(), default=str))
    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 main():

    # 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)


#   *****************************************************************************************************************
#  *****                                                                                                          ****
# ***                                               MAIN PROGRAM                                                    ***
#  *****                                                                                                          ****
#   *****************************************************************************************************************


if __name__ == "__main__":

    main()
    while True: time.sleep(3_600.00)
