🚀 Stock Streamer Pro - Real-Time Kafka-Based Stock Dynamic Content Streaming Platform

A dynamic, multi-user stock data streaming system built with Apache Kafka, Python Flask, and MySQL. Features real-time data visualization, admin approval workflows, and automatic topic management.

---

📋 Table of Contents

• Features
• System Architecture
• Tech Stack
• Installation
• Configuration
• Usage
• Project Structure
• API Endpoints

---

✨ Features

🎯 Core Functionality

• Dynamic Topic Management - Topics created on-demand and managed via Kafka Admin API
• Real-Time Streaming - Live stock price data streamed through Kafka
• Multi-User Support - User authentication and personalized subscriptions
• Admin Approval Workflow - Topics require admin approval before activation
• Auto-Activation - Producer automatically activates approved topics

📊 User Features

• User registration and authentication
• Subscribe/unsubscribe to stock topics
• Real-time price charts with Chart.js
• Live data feed with latest stock prices
• Multiple topic management
• Responsive modern UI

🔧 Admin Features

• 3-tab admin panel (Topics, Users, Subscriptions)
• Approve pending topics with company assignments
• Deactivate topics (auto-deletes from Kafka)
• View user subscriptions per topic
• View topic subscriptions per user
• Real-time subscription monitoring

---

🏗️ System Architecture

Component Flow

1. User → Requests topic via UI → Consumer API
2. Consumer → Creates pending topic in DB
3. Admin → Approves topic + assigns companies
4. Producer → Detects approval → Creates Kafka topic → Activates
5. Consumer → Starts consuming → Streams data to users
6. Admin → Can deactivate → Producer deletes Kafka topic

---

🛠️ Tech Stack

Component	Technology
Backend	Python 3.8+, Flask, Flask-Session, Flask-CORS
Message Broker	Apache Kafka 3.x
Database	MySQL 8.0
Frontend	HTML5, CSS3, JavaScript (ES6+)
Visualization	Chart.js 4.4.0
Admin Panel	Python Tkinter
Data Source	yfinance (Yahoo Finance API)

---

Python Packages

pip install flask flask-cors flask-session mysql-connector-python kafka-python yfinance

---

🚀 Installation

1. Clone Repository

git clone https://github.com/yourusername/stock-streamer-pro.git
cd stock-streamer-pro

2. Install Dependencies

pip install -r requirements.txt

3. Setup MySQL Database

CREATE DATABASE stock_streamer_db;
USE stock_streamer_db;

-- Topics Table
CREATE TABLE topics (
    id INT AUTO_INCREMENT PRIMARY KEY,
    topic_name VARCHAR(255) UNIQUE NOT NULL,
    companies TEXT,
    status ENUM('pending', 'approved', 'active', 'deactivate') NOT NULL DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Users Table
CREATE TABLE users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    username VARCHAR(50) UNIQUE NOT NULL,
    email VARCHAR(100) UNIQUE NOT NULL,
    password_hash VARCHAR(255) NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    last_login TIMESTAMP NULL
);

-- User Subscriptions Table
CREATE TABLE user_subscriptions (
    id INT AUTO_INCREMENT PRIMARY KEY,
    user_id INT NOT NULL,
    topic_id INT NOT NULL,
    subscribed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,     
    FOREIGN KEY (topic_id) REFERENCES topics(id) ON DELETE CASCADE,
    UNIQUE KEY (user_id, topic_id)
);

-- Create test user (password: password123)
INSERT INTO users (username, email, password_hash) 
VALUES ('testuser', 'test@example.com', SHA2('password123', 256));

4. Setup Kafka

# Start ZooKeeper
bin/zookeeper-server-start.sh config/zookeeper.properties

# Start Kafka Broker
bin/kafka-server-start.sh config/server.properties

Important Kafka Configuration (config/server.properties):

delete.topic.enable=true
auto.create.topics.enable=false

---

⚙️ Configuration

Update Configuration Files

1. db_utils.py

MYSQL_CONFIG = {
    "host": "YOUR_MYSQL_HOST",
    "user": "YOUR_MYSQL_USER",
    "password": "YOUR_MYSQL_PASSWORD",
    "database": "stock_streamer_db"
}

2. config.json

{
  "bootstrap_servers": ["YOUR_KAFKA_HOST:9092"],
  "db_config": {
    "host": "YOUR_MYSQL_HOST",
    "user": "YOUR_MYSQL_USER",
    "password": "YOUR_MYSQL_PASSWORD",
    "database": "stock_streamer_db"
  },
  "topic_check_interval": 5,
  "publish_interval": 10
}

3. index.html

const API_BASE = 'http://YOUR_CONSUMER_HOST:5000';

---

🎮 Usage

Start Services

Terminal 1/Node 1: Start Producer

python producer_main.py

Terminal 2/Node 2: Start Consumer API

python consumer.py

Terminal 3/Node 3: Start Admin Panel

python admin.py

Terminal 4/Node 4: Open Web UI

python3 -m http.server 3000
# Open user.html in browser
Browser : 'http://USER_IP:3000'

User Workflow

1. Register/Login

   • Open index.html in browser
   • Register new account or login with existing credentials

2. Request Topic

   • Enter topic name (e.g., "technology", "finance")
   • Click "Subscribe Now"
   • Status shows as "Pending Approval"

3. Admin Approves

   • Admin opens GUI panel
   • Selects pending topic
   • Enters companies (e.g., "GOOGL,MSFT,AAPL")
   • Clicks "Approve"

4. Auto-Activation

   • Producer detects approval
   • Creates Kafka topic
   • Marks as "Active" in database
   • Consumer starts streaming data

5. View Data

   • User sees "Streaming Active" status
   • Real-time chart updates
   • Live stock prices displayed

6. Unsubscribe

   • Click ❌ button on subscription
   • Removed from database
   • Can resubscribe anytime

Admin Workflow

Topics Tab:

• View all topics (pending, approved, active)
• Approve topics with company assignments
• Deactivate topics (triggers Kafka deletion)
• View subscribers for each topic

Users Tab:

• View all registered users
• See user details (last login, email)
• View subscriptions per user

Subscriptions Tab:

• Real-time subscription monitoring
• See which users subscribe to which topics
• Track subscription timestamps

---

📁 Project Structure

stock-streamer-pro/
│
├── producer/
│   ├── producer_main.py          # Main producer entry point
│   ├── topic_watcher.py           # Thread: Monitors DB for topic changes
│   ├── input_listener.py          # Thread: Fetches stock data
│   ├── publisher.py               # Thread: Publishes to Kafka
│   ├── db_utils.py                # Database helper functions
│   └── config.json                # Producer configuration
│
├── consumer/
│   └── consumer.py                # Flask API + Kafka consumers
│
├── admin/
│   └── admin.py                   # Tkinter admin GUI
│
├── ui/
│   └── user.html                 # User web interface
│
├── requirements.txt               # Python dependencies
└── README.md                      # This file

---

🌐 API Endpoints

Authentication

Method	Endpoint	Description
POST	/register	Register new user
POST	/login	User login
POST	/logout	User logout
GET	/current-user	Get logged-in user

Topics

Method	Endpoint	Description
POST	/request-topic	Subscribe to topic
POST	/unsubscribe-topic	Unsubscribe from topic
GET	/topic-status/<name>	Check topic status

Data

Method	Endpoint	Description
GET	/get-stock-data/<name>	Get latest stock data
GET	/health	Health check

---

👥 Authors

• Akshay R - @Akshay10258
• Amogh Shetty - @amoghshetty09
• Amrutha - @amrutha0307
• Yashas J - @Yashtech-714