added my functionalities

.gitignore

@@ -28,8 +28,14 @@ checkpoints.json
 # CSV Files (except symbols.csv)
 *.csv
 !symbols.csv
+
+*.png
+*.txt
+*.py
+=======
 !all_symbols.csv
 
+
 # Environments 
 .env 
 .venv 

.sample.env

@@ -204,4 +204,3 @@ CSRF_TIME_LIMIT = ''
 # Examples: 'instance1_session', 'user1_session', 'app_session', etc.
 SESSION_COOKIE_NAME = 'session'
 CSRF_COOKIE_NAME = 'csrf_token'
-

blueprints/apikey.py

@@ -39,7 +39,8 @@ def manage_api_key():
 
         # Generate new API key
         api_key = generate_api_key()
-
+        logger.info(f"Generated API key for user: {api_key}")
+
         # Store the API key (auth_db will handle both hashing and encryption)
         key_id = upsert_api_key(user_id, api_key)
 

blueprints/core.py

@@ -44,6 +44,7 @@ def setup():
 
             # Automatically generate and save API key
             api_key = generate_api_key()
+            logger.info(f"Generated API key for user: {api_key}")
             key_id = upsert_api_key(username, api_key)
             if not key_id:
                 logger.error(f"Failed to create API key for user {username}")

broker/flattrade/api/data.py

@@ -41,7 +41,7 @@ def get_api_response(endpoint, auth, method="POST", payload=None):
     data = response.text
 
     # Print raw response for debugging
-    logger.info(f"Raw Response: {data}")
+    # logger.info(f"Raw Response: {data}")
 
     try:
         return json.loads(data)

broker/flattrade/streaming/flattrade_adapter.py

@@ -228,6 +228,14 @@ def normalize(data: Dict[str, Any], msg_type: str) -> Dict[str, Any]:
         return result
 
 
+# RedPanda/Kafka imports
+try:
+    from kafka import KafkaProducer
+    from kafka.errors import KafkaError
+    KAFKA_AVAILABLE = True
+except ImportError:
+    KAFKA_AVAILABLE = False
+
 class FlattradeWebSocketAdapter(BaseBrokerWebSocketAdapter):
     """Flattrade WebSocket adapter with improved structure and error handling"""
 
@@ -256,8 +264,17 @@ def _setup_connection_management(self):
         """Initialize connection management"""
         self.running = False
         self.connected = False
+
+        # Kafka/RedPanda specific attributes        
+        self.kafka_enabled = self.redpanda_enabled and KAFKA_AVAILABLE
+        self.kafka_publish_lock = threading.Lock()
         self.lock = threading.Lock()
         self.reconnect_attempts = 0
+
+        if self.kafka_enabled:
+            self.logger.info("Kafka publishing enabled for Flattrade adapter")
+        else:
+            self.logger.info("Kafka publishing disabled - using ZMQ only")
 
     def _setup_normalizers(self):
         """Initialize data normalizers"""
@@ -567,7 +584,7 @@ def _resubscribe_all(self):
                 elif mode == Config.MODE_DEPTH:
                     depth_scrips.add(scrip)
                     self.ws_subscription_refs[scrip]['depth_count'] += 1
-
+                    
             # Resubscribe in batches
             if touchline_scrips:
                 scrip_list = '#'.join(touchline_scrips)
@@ -603,6 +620,27 @@ def _on_message(self, ws, message):
             self.logger.error(f"JSON decode error: {e}, message: {message}")
         except Exception as e:
             self.logger.error(f"Message processing error: {e}", exc_info=True)
+
+    def publish_market_data(self, topic: str, data: dict) -> None:
+        # --- 1) ZMQ Publish (existing) ---
+        if self.zmq_enabled:
+            self.logger.info(f"[ZMQ PUBLISH] Topic: {topic} | Data: {data}")
+            self.logger.debug(f"[DEBUG] ZMQ publish call for topic: {topic}, data keys: {list(data.keys())}")
+            super().publish_market_data(topic, data)
+
+        # --- 2) Kafka Publish (new) ---
+        if self.kafka_enabled and self.kafka_producer:
+            try:
+                with self.kafka_publish_lock:
+                    # The KafkaProducer was set up in BaseBrokerWebSocketAdapter._setup_redpanda()
+                    # We assume the topic already exists or auto‐creation is enabled.
+                    # You can also prefix or map your ZMQ topic to a Kafka topic namespace here.
+                    self.kafka_producer.send("tick_data", key=topic, value=data)
+                    # Optionally flush immediately (costly):
+                    # self.kafka_producer.flush()
+                self.logger.info(f"[KAFKA PUBLISH] Topic: {"tick_data"} | Key:{topic} | Data: {data}")
+            except KafkaError as e:
+                self.logger.error(f"[KAFKA ERROR] Failed to publish to Kafka topic {"tick_data"}:{topic}: {e}")
 
     def _process_market_message(self, data: Dict[str, Any]) -> None:
         """Process market data messages with better error handling"""

broker/flattrade/streaming/flattrade_dummy_api.py

@@ -0,0 +1,202 @@
+import asyncio
+import websockets
+import json
+import random
+from datetime import datetime
+
+# Mock database
+valid_tokens = {
+    "valid_token_123": {"user_id": "FZ15709", "client_id": "74eb594de4a944558aeacd623a714d16"}
+}
+
+# Initialize market data variables
+last_price = 1300.0
+volume = 300
+open_price = 1296.0
+high = 1302.0
+low = 1294.0
+close = 1299.5
+
+last_price_2 = 425.0
+volume_2 = 1000
+open_price_2 = 422.0
+high_2 = 430.0
+low_2 = 420.0
+close_2 = 426.0
+
+async def handle_connection(websocket):
+    print("Client connected")
+    authenticated = False
+    subscribed = False
+
+    try:
+        # Create a task for sending market data
+        market_data_task = None
+
+        async for message in websocket:
+            try:
+                data = json.loads(message)
+                print(f"Received: {data}")
+
+                # Authentication handling
+                if data.get("t") == "c" and not authenticated:
+                    response = {
+                        "t": "ck",
+                        "s": "OK",
+                        "uid": "FZ15709"
+                    }
+                    authenticated = True
+                    await websocket.send(json.dumps(response))
+                    print("Authentication successful")
+
+                # Subscription handling
+                elif data.get("t") == "t" and authenticated and not subscribed:
+                    response = {
+                        "t": "tk",
+                        "e": "NSE",
+                        "tk": "2885",
+                        #"ts": "RELIANCE-EQ",
+                        "ti": "1",
+                        "ls": "1",
+                        "lp": str(last_price),
+                        "pc": "0.5",
+                        "v": str(volume),
+                        "o": str(open_price),
+                        "h": str(high),
+                        "l": str(low),
+                        "c": str(close),
+                        "ap": str(last_price)
+                    }
+                    response_2 = {
+                        "t": "tk",
+                        "e": "NSE",
+                        "tk": "11536",
+                        "ts": "TCS-EQ",
+                        "ti": "1",
+                        "ls": "1",
+                        "lp": str(last_price_2),
+                        "pc": "0.5",
+                        "v": str(volume_2),
+                        "o": str(open_price_2),
+                        "h": str(high_2),
+                        "l": str(low_2),
+                        "c": str(close_2),
+                        "ap": str(last_price_2)
+                    }
+                    subscribed = True
+                    await websocket.send(json.dumps(response))
+                    await websocket.send(json.dumps(response_2))            
+                    print("Subscribed")
+
+                    # Start sending market data after subscription
+                    market_data_task = asyncio.create_task(send_market_data(websocket))
+
+                elif data.get("t") == "u" and subscribed:
+                    if market_data_task:
+                        market_data_task.cancel()
+                        try:
+                            await market_data_task
+                        except asyncio.CancelledError:
+                            pass
+
+                    response = {
+                            "t": "uk",
+                            "k": "NSE|2885#NSE|11536"
+                        }
+                    subscribed = False
+                    await websocket.send(json.dumps(response))
+                    print("Unsubscribed")
+
+
+                # Unknown message type
+                else:
+                    if not authenticated:
+                        response = {
+                            "t": "error",
+                            "emsg": "Not authenticated"
+                        }
+                        await websocket.send(json.dumps(response))
+                        print("Not Authenticated")
+
+            except json.JSONDecodeError:
+                response = {
+                    "t": "error",
+                    "emsg": "Invalid JSON"
+                }
+                await websocket.send(json.dumps(response))
+
+    except websockets.exceptions.ConnectionClosed:
+        print("Client disconnected")
+        if market_data_task:
+            market_data_task.cancel()
+
+async def send_market_data(websocket):
+    """Continuously send market data updates"""
+    global last_price, volume, open_price, high, low, close
+    global last_price_2, volume_2, open_price_2, high_2, low_2, close_2
+
+    print("Starting market data stream")
+    while True:
+        try:
+            # Update market data
+            last_price += round(random.uniform(-2, 2), 2)
+            volume += random.randint(100, 1000)
+            open_price += round(random.uniform(-2, 2), 2)
+            high += round(random.uniform(-2, 2), 2)
+            low += round(random.uniform(-2, 2), 2)
+            close += round(random.uniform(-2, 2), 2)
+
+            last_price_2 += round(random.uniform(-2, 2), 2)
+            volume_2 += random.randint(100, 1000)
+            open_price_2 += round(random.uniform(-2, 2), 2)
+            high_2 += round(random.uniform(-2, 2), 2)
+            low_2 += round(random.uniform(-2, 2), 2)
+            close_2 += round(random.uniform(-2, 2), 2)
+
+            touchline_data = {
+                "t": "tf",
+                "e": "NSE",
+                "tk": "2885",
+                "lp": str(last_price),
+                "pc": "0.5",
+                "v": str(volume),
+                "o": str(open_price),
+                "h": str(high),
+                "l": str(low),
+                "c": str(close),
+                "ap": str(last_price)
+            }
+
+            touchline_data_2 = {
+                "t": "tf",
+                "e": "NSE",
+                "tk": "11536",
+                "lp": str(last_price_2),
+                "pc": "0.5",
+                "v": str(volume_2),
+                "o": str(open_price_2),
+                "h": str(high_2),
+                "l": str(low_2),
+                "c": str(close_2),
+                "ap": str(last_price_2)
+            }
+
+            await websocket.send(json.dumps(touchline_data))
+            await websocket.send(json.dumps(touchline_data_2))
+            print("Sent market data update")
+            await asyncio.sleep(2)
+
+        except websockets.exceptions.ConnectionClosed:
+            print("Connection closed while sending market data")
+            break
+        except Exception as e:
+            print(f"Error in market data stream: {e}")
+            break
+
+async def main():
+    async with websockets.serve(handle_connection, "localhost", 8766):
+        print("Dummy WebSocket server running on ws://localhost:8766")
+        await asyncio.Future()  # Run forever
+
+if __name__ == "__main__":
+    asyncio.run(main())

broker/flattrade/streaming/flattrade_websocket.py

@@ -15,6 +15,8 @@ class FlattradeWebSocket:
 
     # Connection constants
     WS_URL = "wss://piconnect.flattrade.in/PiConnectWSTp/"
+    #WS_URL = "ws://localhost:8766"
+
     CONNECTION_TIMEOUT = 15
     THREAD_JOIN_TIMEOUT = 5
 

broker/flattrade/streaming/kafkaconsum.py

@@ -0,0 +1,17 @@
+from kafka import KafkaConsumer
+import json
+
+consumer = KafkaConsumer(
+    #'tick_raw',
+    'tick_data',
+    bootstrap_servers='localhost:9092',
+    group_id='test-group',
+    auto_offset_reset='earliest',
+    key_deserializer=lambda k: k.decode('utf-8') if k else None,
+    value_deserializer=lambda v: json.loads(v.decode('utf-8'))
+)
+
+print("Listening for messages...")
+
+for msg in consumer:
+    print(f"{msg.key} => {msg.value}")

broker/flattrade/streaming/kafkaprod.py

@@ -0,0 +1,20 @@
+from kafka import KafkaProducer
+import json
+
+producer = KafkaProducer(
+    bootstrap_servers='localhost:9092',
+    key_serializer=lambda k: k.encode('utf-8'),
+    value_serializer=lambda v: json.dumps(v).encode('utf-8')
+)
+
+# Sample tick data
+tick_data = {
+    "symbol": "RELIANCE",
+    "ltp": 2810.50,
+    "volume": 100,
+    "timestamp": 1720458436000
+}
+
+producer.send('tick_raw', key='RELIANCE', value=tick_data)
+producer.flush()
+print("Tick sent!")