master

background/finstitutions/trading/to_kafka.py

@@ -0,0 +1,248 @@
+"""
+
+    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)

utils_v2/database/async_mysql_v2.py

@@ -112,7 +112,10 @@ class AsyncMySQL:
         self.__kwargs = kwargs
         self.__min_pool_size = 10
         self.__max_pool_size = max(pool_size, self.__min_pool_size)
-        self.__pool = None
+        self.__pool: aiomysql = None
+
+        # For establishing connection:
+        self.__exclusive_semaphore = asyncio.Semaphore(1)
 
         # Minor adjustments for backward compatibility:
         self.__kwargs["db"] = self.__kwargs.pop("database")
@@ -123,15 +126,19 @@ class AsyncMySQL:
     def __del__(self):
         pass
 
-    async def connect(self):
+    async def connect(self) -> bool:
 
         """
         Establish a connection and create a pool of connections to call from.
-        :return: None.
+        :return: True if connected, else False
         """
 
         try:
 
+            # Ensure that only one connection attempt is being made at one time,
+            # and try to connect to the database:
+            async with self.__exclusive_semaphore:
+                if self.__pool is None:
                     self.__pool = await aiomysql.create_pool(
                         minsize = self.__min_pool_size,
                         maxsize = self.__max_pool_size,
@@ -139,10 +146,14 @@ class AsyncMySQL:
                         **self.__kwargs
                     )
 
+        # In case of any exception:
         except Exception as exception:
             self.__printer(exception)
             self.__pool = None
 
+        # Done here:
+        return False if self.__pool is None else True
+
     async def ensure_connection(self):
 
         """

utils_v2/nse/controllers/indices/constituents.py

@@ -243,12 +243,12 @@ class NSEIndexConstituents(AsyncNSEBase):
                     "totTradedVol": symbol["totalTradedVolume"],
                     "totTradedVal": symbol["totalTradedValue"],
                     "prevClose": symbol["previousClose"],
-                    "change": symbol["change"],
+                    "chg": symbol["change"],
-                    "pctChange": symbol["pChange"],
+                    "pChg": symbol["pChange"],
                     "yearHigh": symbol["yearHigh"],
                     "yearLow": symbol["yearLow"],
-                    "pctChange30d": symbol["perChange30d"],
+                    "pChg30d": symbol["perChange30d"],
-                    "pctChange365d": symbol["perChange365d"],
+                    "pChg365d": symbol["perChange365d"],
                     "ffmc": symbol["ffmc"]
                 } for symbol in raw_json["data"] if symbol["priority"] == 0
             ]

utils_v2/nse/controllers/indices/master.py

@@ -182,14 +182,14 @@ class NSEIndexMaster(AsyncNSEBase):
                     "low": idx["low"],
                     "close": idx["last"],
                     "prevClose": idx["previousClose"],
-                    "pctChange": idx["percentChange"],
+                    "pChg": idx["percentChange"],
                     "yearHigh": idx["yearHigh"],
                     "yearLow": idx["yearLow"],
                     "advances": idx.get("advances"),
                     "declines": idx.get("declines"),
                     "unchanged": idx.get("unchanged"),
-                    "pctChange30d": idx["perChange30d"],
+                    "pChg30d": idx["perChange30d"],
-                    "pctChange365d": idx["perChange365d"]
+                    "pChg365d": idx["perChange365d"]
                 } for idx in raw_json["data"]
             ]
 

utils_v2/nse/controllers/market/pre_market.py

@@ -207,14 +207,14 @@ class NSEPreMarket(AsyncNSEBase):
                 raw_symbol_detail = raw_symbol_data["detail"]["preOpenMarket"]
                 formatted_data["data"].append({
                     "symbol": raw_symbol_metadata["symbol"],
-                    "marketCap": raw_symbol_metadata["marketCap"],
+                    "ffmc": raw_symbol_metadata["marketCap"],
                     "trigger": raw_symbol_metadata["purpose"],
                     "yearHigh": raw_symbol_metadata["yearHigh"],
                     "yearLow": raw_symbol_metadata["yearLow"],
                     "prevClose": raw_symbol_metadata["previousClose"],
                     "premarketPrice": raw_symbol_metadata["iep"],
                     "chg": raw_symbol_metadata["change"],
-                    "pctChg": raw_symbol_metadata["pChange"],
+                    "pChg": raw_symbol_metadata["pChange"],
                     "totalTradedVolume": raw_symbol_detail["totalTradedVolume"],
                     "totalBuyVolume": raw_symbol_detail["totalBuyQuantity"],
                     "totalSellVolume": raw_symbol_detail["totalSellQuantity"],
@@ -223,7 +223,7 @@ class NSEPreMarket(AsyncNSEBase):
             # Data sorting (descending order of percent change):
             formatted_data["data"] = sorted(
                 formatted_data["data"],
-                key = lambda x: x["pctChg"],
+                key = lambda x: x["pChg"],
                 reverse = True
             )
 

utils_v2/queue/async_kafka.py

@@ -473,7 +473,7 @@ if __name__ == "__main__":
 
     import asyncio
     import time
-    from data_models.kafka_message import KafkaMessage
+    # from data_models.kafka_message import KafkaMessage
 
     ssl_ctx = get_ssl_context(
         ca_file = r"/home/developer/PycharmProjects/utils/cred/kafka/cert_authority.pem",
@@ -497,13 +497,13 @@ if __name__ == "__main__":
 
         while True:
             messages = await consumer.consume(count = 1)
-            if len(messages) > 0: print("MESSAGE:", json.to_string(messages[0], default=str))
+            if len(messages) > 0: print("MESSAGE:", json.to_string(messages[0], default = str))
             await asyncio.sleep(1.0)
 
     async def producer_test():
 
         producer = ProducerKafka(
-            topic = "kft_file_upload",
+            topic = "tickers",
             bootstrap_servers = "del.ditscentre.in:9092",
             security_protocol = "SSL",
             ssl_context = ssl_ctx
@@ -512,43 +512,11 @@ if __name__ == "__main__":
         print("READY!")
 
         while True:
-            my_msg = KafkaMessage(
+            success = await producer.produce({"name": "Bhopli", "color": "orange"})
-                data = {
+            print("PRODUCED:", success)
-                    "accepted": False,
+            await asyncio.sleep(0.25)
-                    "reason": "low resolution"
-                },
-                media = {
-                    "name": "pikachu_poster.jpg",
-                    "ext": "jpg",
-                    "url": "https://nexcom.ditscentre.in/utils/files/small/download/66ded1c1c1c05139a618b5ff",
-                    "attr": {
-                        "user": "SarangKabir",
-                        "project": "ACE-PGP",
-                        "id": 173,
-                        "campaignActivityId": "25",
-                        "idCampaign": 49,
-                        "phoneNo": "7977821877"
-                    }
-                },
-                appId = "aceWockhardt",
-                proc = {
-                    "name": "_assessImg",
-                    "attr": {
-                        "blurThreshold": 0.25,
-                        "clarityThreshold": 0.65,
-                        "nsfwThreshold": 0.25,
-                        "minWidth": 512,
-                        "minHeight": 512
-                    }
-                },
-                ack = None
-            )
-            success = await producer.produce(my_msg.model_dump())
-            print("produced...")
-            time.sleep(1.0)
-            break
 
         await producer.close()
 
 
-    asyncio.run(consumer_test())
+    asyncio.run(producer_test())

utils_v2/queue/kafka.py

@@ -81,7 +81,7 @@ import ssl
 #   *****************************************************************************************************************
 
 
-def create_config(
+def create_producer_config(
         bootstrap_servers: str | List[str],
         group_id: str | None = None,
         security_protocol: Literal["PLAINTEXT", "SSL"] = "PLAINTEXT",
@@ -142,8 +142,8 @@ class ProducerKafka:
 
     def __init__(
             self,
-            topic: str,
             config: dict,
+            topic: str,
             serializer = None,
             debug: bool = True,
             debug_prefix = "Kafka (P) | "
@@ -151,8 +151,8 @@ class ProducerKafka:
 
         """
         Create a Kafka Producer.
-        :param topic: The topic to produce on.
         :param config: The configuration as expected by Confluent-Kafka library.
+        :param topic: The topic to produce on.
         :param serializer: The serializer to use.
         :param debug: Whether, or not, you want to print the debug strings.
         :param debug_prefix: The prefix to use while debugging.
@@ -175,6 +175,46 @@ class ProducerKafka:
     def disable_debug(self):
         self.__printer.disable()
 
+    @staticmethod
+    def create_config(
+            bootstrap_servers: str | List[str],
+            security_protocol: Literal["PLAINTEXT", "SSL"] = "PLAINTEXT",
+            ca_file: str | None = None,
+            cert_file: str | None = None,
+            key_file: str | None = None,
+            client_id: str | int | None = None
+    ):
+
+        """
+        Creates the config required for Confluent-Kafka's library.
+        :param bootstrap_servers: The addresses of the Kafka brokers.
+        :param security_protocol: What sort of security protocol to use.
+        :param ca_file: Needed for 'SSL' security protocol.
+        :param cert_file: Needed for 'SSL' security protocol.
+        :param key_file: Needed for 'SSL' security protocol.
+        :param client_id: An identifier for one producer. Useful for debugging later.
+        :return: The dictionary that needs to be passed as the 'conf' param when creating the producer.
+        """
+
+        # Start with the bare minimum:
+        if not isinstance(bootstrap_servers, list): bootstrap_servers = [bootstrap_servers]
+        config = {
+            "bootstrap.servers": ",".join(bootstrap_servers),
+            "security.protocol": security_protocol
+        }
+
+        # Add the SSL security details:
+        if security_protocol == "SSL":
+            config["ssl.ca.location"] = ca_file
+            config["ssl.certificate.location"] = cert_file
+            config["ssl.key.location"] = key_file
+
+        # Add an identifier for debugging:
+        if client_id: config["client.id"] = client_id
+
+        # Done here:
+        return config
+
     def connect(self) -> bool:
 
         """
@@ -270,242 +310,204 @@ class ProducerKafka:
 # ---------------------------------------------------------------------------------------------------------------------
 
 
-# class ConsumerKafka:
+class ConsumerKafka:
-#
-#     def __init__(
-#             self,
-#             topic,
-#             serializer = None,
-#             debug = True,
-#             debug_prefix = "Kafka (C) | ",
-#             **kwargs
-#     ):
-#
-#         """
-#         Create a Kafka Consumer.
-#         :param topic: The topic to consumer on.
-#         :param serializer: The serializer to use.
-#         :param debug: Whether, or not, you want to print the debug strings.
-#         :param debug_prefix: The prefix to use while debugging.
-#         :param kwargs: Any configuration parameters for the Kafka instances.
-#         """
-#
-#         # Initialize the debugger:
-#         self.__printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
-#         if not debug: self.__printer.disable()
-#
-#         # initialize the Kafka producer:
-#         self.__topic = topic
-#         self.__kwargs = kwargs
-#         self.__consumer = None
-#         self.__connected = False
-#         self.__serializer = serializer or JSONSerializer()
-#
-#         # For establishing connection:
-#         self.__exclusive_semaphore = asyncio.Semaphore(1)
-#
-#     def enable_debug(self):
-#         self.__printer.enable()
-#
-#     def disable_debug(self):
-#         self.__printer.disable()
-#
-#     async def connect(self):
-#
-#         """
-#         Connects to the Kafka server if not connected.
-#         :return: True or False based on the success of the operation.
-#         """
-#
-#         async with self.__exclusive_semaphore:
-#             if not self.__connected:
-#                 try:
-#                     self.__consumer = AIOKafkaConsumer(self.__topic, **self.__kwargs)
-#                     await self.__consumer.start()
-#                     self.__connected = True
-#                 except Exception as exception: self.__printer(exception)
-#             return self.__connected
-#
-#     async def ensure_connection(self):
-#
-#         """
-#         Connects to the Kafka server if not connected.
-#         :return: True or False based on the success of the operation.
-#         """
-#
-#         if not self.__connected: await self.connect()
-#         return self.__connected
-#
-#     async def close(self):
-#
-#         """
-#         Terminates the connection.
-#         :return: None.
-#         """
-#
-#         if self.__connected:
-#             try:
-#                 await self.__consumer.stop()
-#                 self.__printer("Consumer closed!")
-#                 self.__connected = False
-#             except Exception as exception: self.__printer(exception)
-#
-#     async def consume(self, count = 1, timeout = 0.05, encoding = "utf-8"):
-#
-#         """
-#         Get messages from the Kafka server.
-#         :param count: The number of messages to get from the Kafka server.
-#         :param timeout: The time in seconds to wait for retrieval.
-#         :param encoding: The encoding to use.
-#         :return: The messages that were received. If no messages are available, an empty list will be returned.
-#         """
-#
-#         # Ensure connectivity to the server.
-#         # If not connected, return with failure immediately.
-#         if not await self.ensure_connection(): return []
-#
-#         # Make a variable that will hold the final results:
-#         messages = []
-#
-#         try:
-#
-#             # Read some messages:
-#             results = await self.__consumer.getmany(
-#                 max_records = max(1, count),
-#                 timeout_ms = int(timeout * 1_000)
-#             )
-#
-#             # Format the received messages:
-#             if results:
-#                 for topic_partition, records in results.items():
-#                     for record in records:
-#                         record_dict = record.__dict__
-#                         record_dict["value"] = self.__serializer.deserialize(
-#                             data = record_dict["value"],
-#                             encoding = encoding
-#                         )
-#                         messages.append(record_dict)
-#
-#         # Debugging print if something went wrong:
-#         except Exception as exception: self.__printer(exception)
-#
-#         # Done here:
-#         return messages
 
+    def __init__(
+            self,
+            config: dict,
+            topic: str,
+            serializer = None,
+            debug = True,
+            debug_prefix = "Kafka (C) | ",
+            **kwargs
+    ):
 
-# ---------------------------------------------------------------------------------------------------------------------
+        """
+        Create a Kafka Consumer.
+        :param topic: the topic to consume on.
+        :param serializer: The serializer to use.
+        :param debug: Whether, or not, you want to print the debug strings.
+        :param debug_prefix: The prefix to use while debugging.
+        :param kwargs: Any configuration parameters for the Kafka instances.
+        """
 
+        # Initialize the debugger:
+        self.__printer = IceCreamDebugger(prefix = debug_prefix, includeContext = True)
+        if not debug: self.__printer.disable()
 
-# class BidirectionalKafka:
+        # initialize the Kafka consumer:
-#
+        self.__config = config
-#     # The 'roles' that the instance can take.
+        self.__topic = topic
-#     # The master talks on the channel (topic) that the slave listens on and vice versa.
+        self.__kwargs = kwargs
-#     # Master-Slave is only for deciding who talks on which channel and who listens on which.
+        self.__consumer = None
-#     # In a two-party system, one must be the master, the other must be the slave.
+        self.__connected = False
-#     # There are no extra privileges that the master enjoys. The naming convention was borrowed from common protocols
+        self.__serializer = serializer or JSONSerializer()
-#     # used in electronics (like I2C).
+
-#     ROLE_MASTER = 1
+    def enable_debug(self):
-#     ROLE_SLAVE = 0
+        self.__printer.enable()
-#
+
-#     def __init__(
+    def disable_debug(self):
-#             self,
+        self.__printer.disable()
-#             role,
+
-#             topic,
+    @staticmethod
-#             ack_topic: str = None,
+    def create_config(
-#             group: str = None,
+            bootstrap_servers: str | List[str],
-#             serializer = None,
+            group_id: str,
-#             debug = True,
+            auto_offset_reset: Literal["latest", "earliest"] = "latest",
-#             debug_prefix = "Kafka (B) | ",
+            security_protocol: Literal["PLAINTEXT", "SSL"] = "PLAINTEXT",
-#             **kwargs
+            ca_file: str | None = None,
-#     ):
+            cert_file: str | None = None,
-#
+            key_file: str | None = None,
-#         """
+            client_id: str | int | None = None
-#         Creates a walkie-talkie type setup to use Kafka in a bidirectional manner. Fo more information on all the
+    ):
-#         individual methods, please read through the doc-strings of the component classes 'ProducerKafka', and
+
-#         'ConsumerKafka'.
+        """
-#         :param role: Select from "ROLE_MASTER" and "ROLE_SLAVE". Between the two parties that are talking, one will be
+        Creates the config required for Confluent-Kafka's library.
-#             the master and the other will be the slave. The channel that the master uses to speak will the one the slave
+        :param bootstrap_servers: The addresses of the Kafka brokers.
-#             uses to listen, and vice versa.
+        :param group_id: When a set of consumers are working on one topic in a group such that you want only one of them
-#         :param topic: The topic to communicate on. Will be the same between the master and the slave.
+            to read a particular message.
-#         :param ack_topic: Explicitly provide this for the second channel, or it will be created from the name of the
+        :param auto_offset_reset: Use this to influence the behaviour of how the Kafka consumer will read messages when
-#             topic itself. Will be the same between the master and the slave.
+            it first connects to the broker. It could either want to read the earliest (oldest) messages or the latest
-#         :param group: The group to assign the instance to.
+            (newest) messages from the queue.
-#         :param debug: Whether, or not, you want to print the debug strings.
+        :param security_protocol: What sort of security protocol to use.
-#         :param debug_prefix: The prefix to use while debugging.
+        :param ca_file: Needed for 'SSL' security protocol.
-#         :param kwargs: Any configuration parameters for the Kafka instances.
+        :param cert_file: Needed for 'SSL' security protocol.
-#         """
+        :param key_file: Needed for 'SSL' security protocol.
-#
+        :param client_id: An identifier for one producer. Useful for debugging later.
-#         # Not down the basic variables:
+        :return: The dictionary that needs to be passed as the 'conf' param when creating the producer.
-#         self.__role = role
+        """
-#         self.__topic = topic
+
-#         self.__ack_topic = ack_topic or topic + "Ack"
+        # Start with the bare minimum:
-#         self.__group = group
+        if not isinstance(bootstrap_servers, list): bootstrap_servers = [bootstrap_servers]
-#
+        config = {
-#         # In case the current instance is the master,
+            "bootstrap.servers": ",".join(bootstrap_servers),
-#         # it will talk on "topic", and listen on "ack_topic":
+            "security.protocol": security_protocol
-#         if self.__role == self.ROLE_MASTER:
+        }
-#             self.__producer_kwargs = kwargs.copy()
+
-#             self.__producer = ProducerKafka(
+        # Add the SSL security details:
-#                 topic = self.__topic,
+        if security_protocol == "SSL":
-#                 serializer = serializer,
+            config["ssl.ca.location"] = ca_file
-#                 debug = debug,
+            config["ssl.certificate.location"] = cert_file
-#                 debug_prefix = debug_prefix.strip() + " (P) | ",
+            config["ssl.key.location"] = key_file
-#                 **self.__producer_kwargs
+
-#             )
+        # Add an identifier for debugging:
-#             self.__consumer_kwargs = kwargs.copy()
+        if client_id: config["client.id"] = client_id
-#             self.__consumer_kwargs["group_id"] = self.__group
+
-#             self.__consumer = ConsumerKafka(
+        # Consumer-specific:
-#                 topic = self.__ack_topic,
+        config["group.id"] = group_id
-#                 group = group,
+        config["auto.offset.reset"] = auto_offset_reset
-#                 serializer = serializer,
+
-#                 debug = debug,
+        # Done here:
-#                 debug_prefix = debug_prefix.strip() + " (C) | ",
+        return config
-#                 **self.__consumer_kwargs
+
-#             )
+    def connect(self):
-#
+
-#         # On the other hand, if the current instance is a slave,
+        """
-#         # It will listen on "topic", and talk on "ack_topic":
+        Connects to the Kafka server if not connected.
-#         else:
+        :return: True or False based on the success of the operation.
-#             self.__producer_kwargs = kwargs.copy()
+        """
-#             self.__producer = ProducerKafka(
+
-#                 topic = self.__ack_topic,
+        if not self.__connected:
-#                 serializer = serializer,
+            try:
-#                 debug = debug,
+                if not isinstance(self.__topic, list): self.__topic = [self.__topic]
-#                 debug_prefix = debug_prefix.strip() + " (P) | ",
+                self.__consumer = Consumer(self.__config)
-#                 **self.__producer_kwargs
+                self.__consumer.subscribe(self.__topic)
-#             )
+                self.__connected = True
-#             self.__consumer_kwargs = kwargs.copy()
+            except Exception as exception: self.__printer(exception)
-#             self.__consumer_kwargs["group_id"] = self.__group
+        return self.__connected
-#             self.__consumer = ConsumerKafka(
+
-#                 topic = self.__topic,
+    def ensure_connection(self):
-#                 group = group,
+
-#                 serializer = serializer,
+        """
-#                 debug = debug,
+        Connects to the Kafka server if not connected.
-#                 debug_prefix = debug_prefix.strip() + " (C) | ",
+        :return: True or False based on the success of the operation.
-#                 **self.__consumer_kwargs
+        """
-#             )
+
-#
+        if not self.__connected: self.connect()
-#     def enable_debug(self):
+        return self.__connected
-#         self.__producer.enable_debug()
+
-#         self.__consumer.enable_debug()
+    def close(self):
-#
+
-#     def disable_debug(self):
+        """
-#         self.__producer.disable_debug()
+        Terminates the connection.
-#         self.__consumer.disable_debug()
+        :return: None.
-#
+        """
-#     async def ensure_connection(self):
+
-#         await self.__producer.ensure_connection()
+        if self.__connected:
-#         await self.__consumer.ensure_connection()
+            try:
-#
+                self.__consumer.close()
-#     async def close(self):
+                self.__printer("Consumer closed!")
-#         await self.__producer.close()
+                self.__connected = False
-#         await self.__consumer.close()
+            except Exception as exception: self.__printer(exception)
-#
+
-#     async def produce(self, message, encoding = "utf-8"):
+    def __consume_one(
-#         return await self.__producer.produce(message, encoding = encoding)
+            self,
-#
+            timeout = 0.05,
-#     async def consume(self, count = 1, timeout = 0.05, encoding = "utf-8"):
+            encoding = "utf-8"
-#         return await self.__consumer.consume(count = count, timeout = timeout, encoding = encoding)
+    ) -> dict | None:
+
+        """
+        Here we consume exactly one message from the broker. If we need multiple messages (a batch of messages), we call
+        this method as many times as needed.
+        :param timeout: The time in seconds to wait for retrieval.
+        :param encoding: The encoding to use.
+        :return: A dict that holds the details of the message. Null if no message was fetched.
+        """
+
+        message = self.__consumer.poll(timeout = timeout)
+        if message is None or message.error(): return None
+        else:
+            message_value = message.value()
+            if message_value is not None: message_value = self.__serializer.deserialize(message_value, encoding = encoding)
+            return {
+                "topic": message.topic(),
+                "partition": message.partition(),
+                "offset": message.offset(),
+                "key": message.key().decode("utf-8") if message.key() else None,
+                "value": message_value,
+                "timestamp": message.timestamp()[1],
+                "headers": message.headers()
+            }
+
+    def consume(self, count = 1, timeout = 0.05, encoding = "utf-8"):
+
+        """
+        Get messages from the Kafka broker.
+        :param count: The number of messages to get from the Kafka server.
+        :param timeout: The time in seconds to wait for retrieval.
+        :param encoding: The encoding to use.
+        :return: The messages that were received. If no messages are available, an empty list will be returned.
+        """
+
+        # Make some variables:
+        start_time = time.time()
+        messages = []
+
+        # Ensure connectivity to the server.
+        # If not connected, return with failure immediately.
+        if not self.ensure_connection(): return messages
+
+        # Get into an indefinite loop.
+        while True:
+
+            # If timed-out, break out of the loop:
+            elapsed_time = time.time() - start_time
+            if elapsed_time > timeout: break
+
+            # Get the next message:
+            message = self.__consume_one(
+                timeout = timeout - elapsed_time,
+                encoding = encoding
+            )
+
+            # If a message was fetched in the timeout, append it to the list of messages.
+            # Break out of the loop if you have reached the needed no. of messages:
+            if message:
+                messages.append(message)
+                if len(messages) >= count: break
+
+        # Done here:
+        return messages
 
 
 #   *****************************************************************************************************************
@@ -528,12 +530,13 @@ if __name__ == "__main__":
         # Create and connect the producer:
         producer = ProducerKafka(
             topic = "kft_file_upload",
-            config = create_config(
+            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"
+                key_file = r"../../creds/kafka/privkey.pem",
+                client_id = 123
             )
         )
         producer.connect()
@@ -557,6 +560,29 @@ if __name__ == "__main__":
         # Ensure a graceful close:
         producer.close()
 
+    def consumer_test():
+
+        # Create and connect the producer:
+        consumer = ConsumerKafka(
+            topic = "tickers",
+            config = ConsumerKafka.create_config(
+                bootstrap_servers = "del.ditscentre.in:9092",
+                group_id = "test-group",
+                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",
+            )
+        )
+        consumer.connect()
+        print("CONSUMER READY!")
+
+        for _ in range(5):
+            messages = consumer.consume(count = 3, timeout = 2.5)
+            print(json.to_string(messages, default = str))
+
+        consumer.close()
 
     # Run the test code:
-    producer_test(count = 5)
+    # producer_test(count = 5)
+    consumer_test()