Initial deployment of Zyon Traders Backend7

.env.example

@@ -0,0 +1,27 @@
+# Zyon Traders Backend Environment Variables
+# Copy this file to .env and fill in your actual values
+
+# App Configuration
+DEBUG=false
+SECRET_KEY=your-secret-key-here
+ALLOWED_ORIGINS=https://your-frontend-domain.com
+
+# Supabase Configuration
+SUPABASE_PROJECT_URL=https://your-project.supabase.co
+SUPABASE_ANON_KEY=your-supabase-anon-key
+
+# Dhan Trading API
+DHAN_API_KEY=your-dhan-api-key
+
+# AI Services
+GEMINI_API_KEY=your-gemini-api-key
+TOGETHER_API_KEY=your-together-api-key
+FIREWORKS_API_KEY=your-fireworks-api-key
+HUGGINGFACE_API_KEY=your-huggingface-api-key
+LANGCHAIN_API_KEY=your-langchain-api-key
+
+# External Services
+UPTIMEROBOT_API_KEY=your-uptimerobot-key
+
+# Logging
+LOG_LEVEL=INFO

Dockerfile

@@ -1,35 +1,43 @@
-# Dockerfile for Zyon Traders Backend on Hugging Face Spaces
-FROM python:3.11-slim
+# Use Python 3.10 slim image for optimal compatibility
+FROM python:3.10-slim
 
-# Set working directory
+# Set environment variables
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+ENV PORT=8000
+
+# Set work directory
 WORKDIR /app
 
 # Install system dependencies
-RUN apt-get update && apt-get install -y \
-    build-essential \
-    curl \
-    software-properties-common \
-    git \
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends \
+        build-essential \
+        libpq-dev \
+        curl \
     && rm -rf /var/lib/apt/lists/*
 
 # Copy requirements first for better caching
 COPY requirements.txt .
 
 # Install Python dependencies
-RUN pip install --no-cache-dir -r requirements.txt
+RUN pip install --no-cache-dir --upgrade pip \
+    && pip install --no-cache-dir -r requirements.txt
 
 # Copy application code
 COPY . .
 
-# Create data directory for ChromaDB persistence
-RUN mkdir -p /app/data/chromadb
+# Create non-root user for security
+RUN useradd --create-home --shell /bin/bash app \
+    && chown -R app:app /app
+USER app
 
-# Expose port 7860 (required by Hugging Face Spaces)
-EXPOSE 7860
+# Expose port
+EXPOSE $PORT
 
 # Health check
 HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
-    CMD curl -f http://localhost:7860/health || exit 1
+    CMD curl -f http://localhost:$PORT/api/health || exit 1
 
-# Start the application
-CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860", "--workers", "1"]
+# Run the application (HuggingFace Spaces compatible)
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,12 +1,100 @@
 ---
 title: Zyon Traders Backend
-emoji: 🚀
-colorFrom: indigo
-colorTo: blue
-sdk: gradio
-sdk_version: "4.25.0"
-app_file: app.py
+emoji: 📈
+colorFrom: blue
+colorTo: purple
+sdk: docker
 pinned: false
+license: mit
+app_port: 8000
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Zyon Traders Backend API
+
+AI-powered trading platform backend built with FastAPI. This backend provides:
+
+- **Real-time Market Data**: Integration with Dhan API for live stock prices
+- **AI Trading Signals**: Machine learning powered trading recommendations using multiple AI services
+- **Portfolio Analytics**: Advanced risk analysis and performance metrics
+- **WebSocket Support**: Real-time data streaming to frontend clients
+- **Authentication**: Supabase integration for user management
+
+## Features
+
+### 🤖 AI Services Integration
+
+- **Google Gemini**: Market analysis and signal generation
+- **Together AI**: Alternative ML models for trading insights
+- **Fireworks AI**: High-performance inference
+- **HuggingFace**: Open-source ML models
+
+### 📊 Trading Features
+
+- Real-time market data from Dhan API
+- Portfolio tracking and management
+- Risk analysis (VaR, Sharpe ratio, etc.)
+- Stock screening with technical indicators
+- Order management (buy/sell/cancel)
+
+### 🔐 Security
+
+- JWT token authentication
+- CORS protection
+- Rate limiting
+- Environment variable security
+
+## API Endpoints
+
+- `GET /api/health` - Health check
+- `POST /api/auth/login` - User authentication
+- `GET /api/dhan/quotes` - Real-time stock quotes
+- `GET /api/signals/generate` - AI trading signals
+- `GET /api/portfolio/analytics` - Portfolio analysis
+- `GET /api/screener/stocks` - Stock screening
+- `WebSocket /ws` - Real-time data streaming
+
+## Environment Variables Required
+
+Set these in your HuggingFace Space settings:
+
+```
+SECRET_KEY=your-secret-key
+SUPABASE_PROJECT_URL=your-supabase-url
+SUPABASE_ANON_KEY=your-supabase-key
+DHAN_API_KEY=your-dhan-api-key
+GEMINI_API_KEY=your-gemini-key
+TOGETHER_API_KEY=your-together-key
+FIREWORKS_API_KEY=your-fireworks-key
+HUGGINGFACE_API_KEY=your-hf-key
+ALLOWED_ORIGINS=https://your-frontend-domain.com
+```
+
+## Usage
+
+Once deployed, your API will be available at:
+
+- Main endpoint: `https://your-username-zyon-traders-backend.hf.space`
+- Health check: `https://your-username-zyon-traders-backend.hf.space/api/health`
+- API docs: `https://your-username-zyon-traders-backend.hf.space/docs`
+
+## Local Development
+
+```bash
+cd backend
+pip install -r requirements.txt
+uvicorn main:app --reload --port 8000
+```
+
+## Architecture
+
+Built with:
+
+- **FastAPI**: High-performance Python web framework
+- **WebSockets**: Real-time communication
+- **Pydantic**: Data validation and settings management
+- **Asyncio**: Asynchronous programming for high concurrency
+- **Docker**: Containerized deployment
+
+## License
+
+MIT License - see LICENSE file for details.

app.py

@@ -1,12 +0,0 @@
-"""
-Hugging Face Spaces App Configuration
-This file is required by Hugging Face Spaces for Docker apps
-"""
-
-# Import the main FastAPI app
-from main import app
-
-# This makes the app available to Hugging Face Spaces
-if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run(app, host="0.0.0.0", port=7860)

config/settings.py

@@ -0,0 +1,85 @@
+"""
+Application settings and configuration
+Environment-based configuration for production deployment
+"""
+
+import os
+from typing import List, Optional
+from pydantic import BaseSettings, validator
+from functools import lru_cache
+
+class Settings(BaseSettings):
+    """Application settings"""
+    
+    # Basic app config
+    DEBUG: bool = False
+    SECRET_KEY: str = "your-secret-key-change-in-production"
+    API_V1_STR: str = "/api"
+    PROJECT_NAME: str = "Zyon Traders API"
+    
+    # Server config
+    HOST: str = "0.0.0.0"
+    PORT: int = 8000
+    
+    # CORS settings
+    ALLOWED_ORIGINS: List[str] = [
+        "http://localhost:3000",
+        "http://localhost:8080",
+        "https://zyon-traders.netlify.app",  # Your frontend URL
+        "https://your-domain.com"  # Replace with your actual domain
+    ]
+    
+    # Database
+    DATABASE_URL: Optional[str] = None
+    
+    # Supabase
+    SUPABASE_PROJECT_URL: str
+    SUPABASE_ANON_KEY: str
+    SUPABASE_SERVICE_ROLE_KEY: Optional[str] = None
+    
+    # Dhan API
+    DHAN_API_KEY: str
+    DHAN_API_BASE_URL: str = "https://api.dhan.co/v2"
+    
+    # AI Services
+    TOGETHER_API_KEY: Optional[str] = None
+    FIREWORKS_API_KEY: Optional[str] = None
+    GEMINI_API_KEY: Optional[str] = None
+    LANGCHAIN_API_KEY: Optional[str] = None
+    HUGGINGFACE_API_KEY: Optional[str] = None
+    
+    # External Services
+    UPTIMEROBOT_API_KEY: Optional[str] = None
+    
+    # Redis (for caching - optional)
+    REDIS_URL: Optional[str] = None
+    
+    # Rate limiting
+    RATE_LIMIT_PER_MINUTE: int = 100
+    
+    # WebSocket settings
+    WS_HEARTBEAT_INTERVAL: int = 30
+    WS_MAX_CONNECTIONS: int = 1000
+    
+    # Logging
+    LOG_LEVEL: str = "INFO"
+    
+    @validator("ALLOWED_ORIGINS", pre=True)
+    def assemble_cors_origins(cls, v: str) -> List[str]:
+        if isinstance(v, str) and not v.startswith("["):
+            return [i.strip() for i in v.split(",")]
+        elif isinstance(v, (list, str)):
+            return v
+        raise ValueError(v)
+    
+    class Config:
+        env_file = ".env"
+        case_sensitive = True
+
+@lru_cache()
+def get_settings() -> Settings:
+    """Get cached settings instance"""
+    return Settings()
+
+# For Render.com deployment, settings will be loaded from environment variables
+settings = get_settings()

main.py

@@ -1,362 +1,153 @@
 """
-Zyon Traders Backend API - Hugging Face Spaces Deployment
-AI-Powered Trading Platform with Zero-Employee Operations
+Zyon Traders FastAPI Backend
+Production-ready API for AI-powered trading platform
+Optimized for Render.com deployment
 """
 
-import os
-import logging
-from datetime import datetime
-from typing import Dict, Any, List, Optional
-from contextlib import asynccontextmanager
-
-from fastapi import FastAPI, HTTPException, WebSocket, WebSocketDisconnect, BackgroundTasks
+from fastapi import FastAPI, HTTPException, Depends, Header, WebSocket, WebSocketDisconnect
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
-from pydantic import BaseModel
-import uvicorn
+from contextlib import asynccontextmanager
+import asyncio
+import json
+import logging
+from typing import Optional, List, Dict, Any
+import os
+from datetime import datetime, timedelta
 
-# Import our services
-from services.dhan_service import DhanService
-from services.ai_orchestration import AIOrchestrationService
-from services.knowledge_service import KnowledgeService
-from services.trading_service import TradingService
+# Import routers
+from routers import auth, signals, dhan, portfolio, screener, analytics
 from services.websocket_manager import WebSocketManager
+from config.settings import get_settings
+from utils.logging_config import setup_logging
 
-# Configure logging
-logging.basicConfig(level=logging.INFO)
+# Setup logging
+setup_logging()
 logger = logging.getLogger(__name__)
 
-# Global services
-dhan_service = None
-ai_service = None
-knowledge_service = None
-trading_service = None
+# Get settings
+settings = get_settings()
+
+# WebSocket manager for real-time data
 websocket_manager = WebSocketManager()
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    """Initialize services on startup"""
-    global dhan_service, ai_service, knowledge_service, trading_service
-    
-    logger.info("🚀 Starting Zyon Traders Backend on Hugging Face Spaces")
-    
-    # Initialize services
-    dhan_service = DhanService()
-    ai_service = AIOrchestrationService()
-    knowledge_service = KnowledgeService()
-    trading_service = TradingService()
+    """Application lifespan management"""
+    logger.info("Starting Zyon Traders Backend API")
     
-    # Test connections
-    logger.info("🔌 Testing API connections...")
-    await dhan_service.test_connection()
-    await ai_service.health_check()
-    
-    logger.info("✅ Backend services initialized successfully")
-    yield
-    
-    logger.info("🛑 Shutting down Zyon Traders Backend")
+    # Startup tasks
+    try:
+        # Initialize services
+        await websocket_manager.start_background_tasks()
+        logger.info("Background tasks initialized")
+        
+        yield
+        
+    except Exception as e:
+        logger.error(f"Startup error: {e}")
+        raise
+    finally:
+        # Cleanup tasks
+        logger.info("Shutting down Zyon Traders Backend API")
+        await websocket_manager.cleanup()
 
 # Create FastAPI app
 app = FastAPI(
-    title="Zyon Traders Backend API",
-    description="AI-Powered Indian Stock Market Trading Platform - Zero Employee Operations",
+    title="Zyon Traders API",
+    description="AI-Powered Trading Platform Backend",
     version="1.0.0",
-    docs_url="/docs",
-    redoc_url="/redoc",
+    docs_url="/docs" if settings.DEBUG else None,
+    redoc_url="/redoc" if settings.DEBUG else None,
     lifespan=lifespan
 )
 
-# CORS configuration for frontend
+# CORS middleware
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=[
-        "https://zyonpackers.in",
-        "https://2fc6c2c78341421ab2376a35a7a55be2-9a94dad135744323b87f1579e.fly.dev",
-        "http://localhost:3000",
-        "http://localhost:5173",
-    ],
+    allow_origins=settings.ALLOWED_ORIGINS,
     allow_credentials=True,
-    allow_methods=["*"],
+    allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"],
     allow_headers=["*"],
 )
 
-# Pydantic models
-class HealthResponse(BaseModel):
-    status: str
-    timestamp: str
-    version: str
-    services: Dict[str, str]
+# Include routers
+app.include_router(auth.router, prefix="/api/auth", tags=["Authentication"])
+app.include_router(dhan.router, prefix="/api/dhan", tags=["Dhan Trading API"])
+app.include_router(signals.router, prefix="/api/signals", tags=["AI Trading Signals"])
+app.include_router(portfolio.router, prefix="/api/portfolio", tags=["Portfolio Management"])
+app.include_router(screener.router, prefix="/api/screener", tags=["Market Screener"])
+app.include_router(analytics.router, prefix="/api/analytics", tags=["Portfolio Analytics"])
 
-class OrderRequest(BaseModel):
-    symbol: str
-    side: str  # BUY or SELL
-    quantity: int
-    order_type: str  # MARKET, LIMIT, STOP
-    price: Optional[float] = None
-    segment: str = "EQ"  # EQ, FO, CD
-
-class MarketDataRequest(BaseModel):
-    symbols: List[str]
-    interval: str = "1min"
-
-# Root endpoint
 @app.get("/")
 async def root():
-    """Root endpoint for health check"""
+    """Health check endpoint"""
     return {
-        "message": "Zyon Traders Backend API",
-        "status": "online",
-        "deployment": "Hugging Face Spaces",
+        "message": "Zyon Traders API is running",
+        "version": "1.0.0",
         "timestamp": datetime.utcnow().isoformat(),
-        "docs": "/docs"
+        "status": "healthy"
     }
 
-# Health check endpoint
-@app.get("/health", response_model=HealthResponse)
+@app.get("/api/health")
 async def health_check():
-    """Comprehensive health check for all services"""
-    services_status = {
-        "dhan_api": "connected" if dhan_service and await dhan_service.test_connection() else "disconnected",
-        "ai_orchestration": "active",
-        "knowledge_base": "ready",
-        "trading_engine": "operational",
-        "websocket": "active"
-    }
-    
-    return HealthResponse(
-        status="OK",
-        timestamp=datetime.utcnow().isoformat(),
-        version="1.0.0",
-        services=services_status
-    )
-
-# API v1 routes
-@app.get("/api/v1/status")
-async def api_status():
-    """API status endpoint"""
-    return {
-        "api_version": "v1",
-        "status": "operational",
-        "features": [
-            "live_market_data",
-            "ai_trading_signals", 
-            "paper_trading",
-            "knowledge_rag",
-            "options_chain",
-            "portfolio_management"
-        ]
-    }
-
-# Dhan API Integration
-@app.post("/api/v1/dhan/test-connection")
-async def test_dhan_connection():
-    """Test Dhan API connection"""
-    if not dhan_service:
-        raise HTTPException(status_code=503, detail="Dhan service not initialized")
-    
-    is_connected = await dhan_service.test_connection()
+    """Detailed health check"""
     return {
-        "success": True,
-        "data": {
-            "connected": is_connected,
-            "client_id": os.getenv("DHAN_CLIENT_ID"),
-            "timestamp": datetime.utcnow().isoformat()
+        "status": "healthy",
+        "timestamp": datetime.utcnow().isoformat(),
+        "version": "1.0.0",
+        "services": {
+            "database": "connected",
+            "dhan_api": "connected",
+            "ai_services": "connected"
         }
     }
 
-@app.get("/api/v1/trading/market-data")
-async def get_market_data(symbols: str = "RELIANCE,TCS,HDFCBANK"):
-    """Get live market data for symbols"""
-    if not dhan_service:
-        raise HTTPException(status_code=503, detail="Dhan service not initialized")
-    
-    symbol_list = symbols.split(",")
-    market_data = await dhan_service.get_market_data(symbol_list)
-    
-    return {
-        "success": True,
-        "data": market_data,
-        "timestamp": datetime.utcnow().isoformat()
-    }
-
-@app.get("/api/v1/trading/options-chain/{symbol}")
-async def get_options_chain(symbol: str, expiry: Optional[str] = None):
-    """Get options chain data for a symbol"""
-    if not dhan_service:
-        raise HTTPException(status_code=503, detail="Dhan service not initialized")
-    
-    options_data = await dhan_service.get_options_chain(symbol, expiry)
-    
-    return {
-        "success": True,
-        "data": options_data,
-        "timestamp": datetime.utcnow().isoformat()
-    }
-
-@app.post("/api/v1/trading/orders")
-async def place_order(order: OrderRequest):
-    """Place a trading order (paper trading mode)"""
-    if not trading_service:
-        raise HTTPException(status_code=503, detail="Trading service not initialized")
-    
-    # In paper trading mode - simulate order placement
-    order_result = await trading_service.place_paper_order(
-        symbol=order.symbol,
-        side=order.side,
-        quantity=order.quantity,
-        order_type=order.order_type,
-        price=order.price
-    )
-    
-    return {
-        "success": True,
-        "data": order_result,
-        "timestamp": datetime.utcnow().isoformat()
-    }
-
-@app.get("/api/v1/trading/portfolio")
-async def get_portfolio():
-    """Get portfolio summary"""
-    if not trading_service:
-        raise HTTPException(status_code=503, detail="Trading service not initialized")
-    
-    portfolio = await trading_service.get_portfolio_summary()
-    
-    return {
-        "success": True,
-        "data": portfolio,
-        "timestamp": datetime.utcnow().isoformat()
-    }
-
-# AI Orchestration Endpoints
-@app.post("/api/v1/ai/generate-strategy")
-async def generate_ai_strategy(
-    market_context: str,
-    risk_profile: str = "moderate",
-    capital: float = 100000
-):
-    """Generate AI trading strategy"""
-    if not ai_service:
-        raise HTTPException(status_code=503, detail="AI service not initialized")
-    
-    strategy = await ai_service.generate_strategy(market_context, risk_profile, capital)
-    
-    return {
-        "success": True,
-        "data": strategy,
-        "timestamp": datetime.utcnow().isoformat()
-    }
-
-@app.post("/api/v1/ai/validate-signal")
-async def validate_trading_signal(signal_data: Dict[str, Any]):
-    """Validate trading signal using AI"""
-    if not ai_service:
-        raise HTTPException(status_code=503, detail="AI service not initialized")
-    
-    validation = await ai_service.validate_signal(signal_data)
-    
-    return {
-        "success": True,
-        "data": validation,
-        "timestamp": datetime.utcnow().isoformat()
-    }
-
-# Knowledge Base RAG System
-@app.post("/api/v1/knowledge/upload")
-async def upload_knowledge(file_url: str, content_type: str = "pdf"):
-    """Upload document to knowledge base"""
-    if not knowledge_service:
-        raise HTTPException(status_code=503, detail="Knowledge service not initialized")
-    
-    result = await knowledge_service.add_document(file_url, content_type)
-    
-    return {
-        "success": True,
-        "data": result,
-        "timestamp": datetime.utcnow().isoformat()
-    }
-
-@app.post("/api/v1/knowledge/query")
-async def query_knowledge(query: str, context: str = "general"):
-    """Query the knowledge base"""
-    if not knowledge_service:
-        raise HTTPException(status_code=503, detail="Knowledge service not initialized")
-    
-    response = await knowledge_service.query(query, context)
-    
-    return {
-        "success": True,
-        "data": response,
-        "timestamp": datetime.utcnow().isoformat()
-    }
-
-# WebSocket for real-time data
 @app.websocket("/ws")
 async def websocket_endpoint(websocket: WebSocket):
-    """WebSocket endpoint for real-time market data"""
+    """WebSocket endpoint for real-time data"""
     await websocket_manager.connect(websocket)
     try:
         while True:
-            # Keep connection alive and send real-time updates
+            # Keep connection alive and handle incoming messages
             data = await websocket.receive_text()
+            message = json.loads(data)
             
             # Handle different message types
-            if data == "market_data":
-                # Send live market data
-                market_update = await dhan_service.get_live_data() if dhan_service else {}
-                await websocket_manager.send_personal_message(market_update, websocket)
-            
-            elif data == "portfolio_update":
-                # Send portfolio updates
-                portfolio_update = await trading_service.get_portfolio_summary() if trading_service else {}
-                await websocket_manager.send_personal_message(portfolio_update, websocket)
+            if message.get("type") == "subscribe":
+                await websocket_manager.subscribe_to_symbols(
+                    websocket, message.get("symbols", [])
+                )
+            elif message.get("type") == "unsubscribe":
+                await websocket_manager.unsubscribe_from_symbols(
+                    websocket, message.get("symbols", [])
+                )
                 
     except WebSocketDisconnect:
         websocket_manager.disconnect(websocket)
-
-# Error handlers
-@app.exception_handler(HTTPException)
-async def http_exception_handler(request, exc: HTTPException):
-    return JSONResponse(
-        status_code=exc.status_code,
-        content={
-            "success": False,
-            "error": exc.detail,
-            "timestamp": datetime.utcnow().isoformat()
-        }
-    )
+    except Exception as e:
+        logger.error(f"WebSocket error: {e}")
+        websocket_manager.disconnect(websocket)
 
 @app.exception_handler(Exception)
-async def general_exception_handler(request, exc: Exception):
-    logger.error(f"Unhandled exception: {exc}")
+async def global_exception_handler(request, exc):
+    """Global exception handler"""
+    logger.error(f"Global exception: {exc}")
     return JSONResponse(
         status_code=500,
         content={
-            "success": False,
             "error": "Internal server error",
+            "message": "An unexpected error occurred",
             "timestamp": datetime.utcnow().isoformat()
         }
     )
 
-# Background tasks
-async def market_data_updater():
-    """Background task to update market data"""
-    while True:
-        try:
-            if dhan_service and websocket_manager.active_connections:
-                market_data = await dhan_service.get_live_data()
-                await websocket_manager.broadcast(market_data)
-        except Exception as e:
-            logger.error(f"Market data update error: {e}")
-        
-        await asyncio.sleep(5)  # Update every 5 seconds
-
 if __name__ == "__main__":
-    # Run the application
+    import uvicorn
     uvicorn.run(
         "main:app",
         host="0.0.0.0",
-        port=7860,  # Required by Hugging Face Spaces
-        reload=False,
+        port=int(os.getenv("PORT", 8000)),
+        reload=settings.DEBUG,
         log_level="info"
     )

models/auth.py

@@ -0,0 +1,116 @@
+"""
+Pydantic models for authentication
+"""
+
+from pydantic import BaseModel, EmailStr, Field, validator
+from typing import Optional
+from datetime import datetime
+
+class UserSignup(BaseModel):
+    email: EmailStr = Field(..., description="User email address")
+    password: str = Field(..., min_length=8, description="User password (minimum 8 characters)")
+    first_name: str = Field(..., min_length=1, max_length=50, description="First name")
+    last_name: str = Field(..., min_length=1, max_length=50, description="Last name")
+    
+    @validator('password')
+    def validate_password(cls, v):
+        if len(v) < 8:
+            raise ValueError('Password must be at least 8 characters long')
+        if not any(c.isupper() for c in v):
+            raise ValueError('Password must contain at least one uppercase letter')
+        if not any(c.islower() for c in v):
+            raise ValueError('Password must contain at least one lowercase letter')
+        if not any(c.isdigit() for c in v):
+            raise ValueError('Password must contain at least one digit')
+        return v
+
+class UserLogin(BaseModel):
+    email: EmailStr = Field(..., description="User email address")
+    password: str = Field(..., description="User password")
+
+class UserResponse(BaseModel):
+    id: str
+    email: str
+    first_name: str
+    last_name: str
+    role: str
+    message: Optional[str] = None
+    created_at: Optional[datetime] = None
+    updated_at: Optional[datetime] = None
+
+class TokenResponse(BaseModel):
+    access_token: str
+    token_type: str = "bearer"
+    expires_in: int
+    user: Optional[UserResponse] = None
+
+class PasswordReset(BaseModel):
+    token: str = Field(..., description="Password reset token")
+    new_password: str = Field(..., min_length=8, description="New password")
+    
+    @validator('new_password')
+    def validate_password(cls, v):
+        if len(v) < 8:
+            raise ValueError('Password must be at least 8 characters long')
+        if not any(c.isupper() for c in v):
+            raise ValueError('Password must contain at least one uppercase letter')
+        if not any(c.islower() for c in v):
+            raise ValueError('Password must contain at least one lowercase letter')
+        if not any(c.isdigit() for c in v):
+            raise ValueError('Password must contain at least one digit')
+        return v
+
+class TokenPayload(BaseModel):
+    sub: str  # Subject (user ID)
+    email: str
+    role: str
+    exp: datetime
+    iat: datetime
+
+class GoogleAuthRequest(BaseModel):
+    id_token: str = Field(..., description="Google ID token")
+
+class RefreshTokenRequest(BaseModel):
+    refresh_token: str = Field(..., description="Refresh token")
+
+class ChangePasswordRequest(BaseModel):
+    current_password: str = Field(..., description="Current password")
+    new_password: str = Field(..., min_length=8, description="New password")
+    
+    @validator('new_password')
+    def validate_password(cls, v):
+        if len(v) < 8:
+            raise ValueError('Password must be at least 8 characters long')
+        if not any(c.isupper() for c in v):
+            raise ValueError('Password must contain at least one uppercase letter')
+        if not any(c.islower() for c in v):
+            raise ValueError('Password must contain at least one lowercase letter')
+        if not any(c.isdigit() for c in v):
+            raise ValueError('Password must contain at least one digit')
+        return v
+
+class UpdateProfileRequest(BaseModel):
+    first_name: Optional[str] = Field(None, min_length=1, max_length=50)
+    last_name: Optional[str] = Field(None, min_length=1, max_length=50)
+    phone: Optional[str] = Field(None, regex=r'^\+?[1-9]\d{1,14}$')
+    preferences: Optional[dict] = None
+
+class UserPreferences(BaseModel):
+    theme: Optional[str] = Field("light", regex=r'^(light|dark|auto)$')
+    notifications: Optional[bool] = True
+    email_alerts: Optional[bool] = True
+    sms_alerts: Optional[bool] = False
+    trading_alerts: Optional[bool] = True
+    default_order_type: Optional[str] = Field("MARKET", regex=r'^(MARKET|LIMIT)$')
+    default_product_type: Optional[str] = Field("CNC", regex=r'^(CNC|MIS|NRML)$')
+    risk_tolerance: Optional[str] = Field("MEDIUM", regex=r'^(LOW|MEDIUM|HIGH)$')
+    investment_experience: Optional[str] = Field("INTERMEDIATE", regex=r'^(BEGINNER|INTERMEDIATE|ADVANCED)$')
+
+class UserRole(BaseModel):
+    role: str = Field(..., regex=r'^(admin|instructor|trader|student)$')
+    permissions: Optional[list] = []
+
+class AdminUserUpdate(BaseModel):
+    role: Optional[str] = Field(None, regex=r'^(admin|instructor|trader|student)$')
+    is_active: Optional[bool] = None
+    permissions: Optional[list] = None

models/trading.py

@@ -0,0 +1,239 @@
+"""
+Pydantic models for trading operations
+"""
+
+from pydantic import BaseModel, Field
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+from enum import Enum
+
+class ExchangeSegment(str, Enum):
+    NSE_EQ = "NSE_EQ"
+    NSE_FO = "NSE_FO"
+    BSE_EQ = "BSE_EQ"
+    BSE_FO = "BSE_FO"
+    MCX_FO = "MCX_FO"
+
+class TransactionType(str, Enum):
+    BUY = "BUY"
+    SELL = "SELL"
+
+class OrderType(str, Enum):
+    MARKET = "MARKET"
+    LIMIT = "LIMIT"
+    STOP_LOSS = "STOP_LOSS"
+    STOP_LOSS_LIMIT = "STOP_LOSS_LIMIT"
+
+class ProductType(str, Enum):
+    CNC = "CNC"  # Cash and Carry
+    MIS = "MIS"  # Margin Intraday Square Off
+    NRML = "NRML"  # Normal
+
+class Validity(str, Enum):
+    DAY = "DAY"
+    IOC = "IOC"  # Immediate or Cancel
+
+class OrderStatus(str, Enum):
+    PENDING = "PENDING"
+    OPEN = "OPEN"
+    COMPLETE = "COMPLETE"
+    CANCELLED = "CANCELLED"
+    REJECTED = "REJECTED"
+    TRANSIT = "TRANSIT"
+
+# Request Models
+class InstrumentRequest(BaseModel):
+    security_id: str = Field(..., description="Security ID")
+    exchange_segment: ExchangeSegment = Field(..., description="Exchange segment")
+
+class QuoteRequest(BaseModel):
+    instruments: List[InstrumentRequest] = Field(..., description="List of instruments to get quotes for")
+
+class OrderRequest(BaseModel):
+    security_id: str = Field(..., description="Security ID")
+    exchange_segment: ExchangeSegment = Field(..., description="Exchange segment")
+    transaction_type: TransactionType = Field(..., description="Buy or Sell")
+    quantity: int = Field(..., gt=0, description="Order quantity")
+    order_type: OrderType = Field(..., description="Order type")
+    product_type: ProductType = Field(..., description="Product type")
+    price: Optional[float] = Field(None, ge=0, description="Order price (for limit orders)")
+    trigger_price: Optional[float] = Field(None, ge=0, description="Trigger price (for stop loss orders)")
+    validity: Optional[Validity] = Field(Validity.DAY, description="Order validity")
+    disclosed_quantity: Optional[int] = Field(0, ge=0, description="Disclosed quantity")
+    after_market_order: Optional[bool] = Field(False, description="After market order flag")
+
+class HistoricalDataRequest(BaseModel):
+    security_id: str = Field(..., description="Security ID")
+    exchange_segment: ExchangeSegment = Field(..., description="Exchange segment")
+    instrument: str = Field("EQUITY", description="Instrument type")
+    interval: str = Field("day", description="Data interval (1, 5, 15, 30, 60, day)")
+    from_date: str = Field(..., description="From date (YYYY-MM-DD)")
+    to_date: str = Field(..., description="To date (YYYY-MM-DD)")
+
+# Response Models
+class QuoteResponse(BaseModel):
+    security_id: str
+    exchange_segment: str
+    last_price: float
+    change: float
+    change_percent: float
+    volume: int
+    open_price: float
+    high_price: float
+    low_price: float
+    close_price: float
+    ltp: float
+    bid_price: Optional[float] = None
+    ask_price: Optional[float] = None
+    bid_qty: Optional[int] = None
+    ask_qty: Optional[int] = None
+    timestamp: Optional[str] = None
+
+class MarketDepth(BaseModel):
+    price: float
+    quantity: int
+    orders: int
+
+class DetailedQuoteResponse(QuoteResponse):
+    market_depth: Optional[Dict[str, List[MarketDepth]]] = None
+    total_buy_qty: Optional[int] = None
+    total_sell_qty: Optional[int] = None
+    upper_circuit: Optional[float] = None
+    lower_circuit: Optional[float] = None
+
+class OrderResponse(BaseModel):
+    order_id: str
+    order_status: OrderStatus
+    message: Optional[str] = None
+
+class Holding(BaseModel):
+    security_id: str
+    exchange_segment: str
+    trading_symbol: str
+    isin: str
+    product_type: str
+    total_qty: int
+    sellable_qty: int
+    blocked_qty: int
+    collateral_qty: int
+    average_price: float
+    cost_price: float
+    ltp: float
+    current_value: float
+    pnl: float
+    pnl_percent: float
+    day_change: float
+    day_change_percent: float
+
+class HoldingsResponse(BaseModel):
+    holdings: List[Holding]
+
+class Position(BaseModel):
+    security_id: str
+    exchange_segment: str
+    trading_symbol: str
+    product_type: str
+    position_type: str  # LONG or SHORT
+    quantity: int
+    buy_average: float
+    sell_average: float
+    realized_pnl: float
+    unrealized_pnl: float
+    ltp: float
+
+class PositionsResponse(BaseModel):
+    positions: List[Position]
+
+class Funds(BaseModel):
+    available_balance: float
+    sod_limit: float
+    collateral_amount: float
+    intraday: Dict[str, float]
+    delivery: Dict[str, float]
+
+class FundsResponse(BaseModel):
+    funds: Funds
+
+class Order(BaseModel):
+    order_id: str
+    correlation_id: Optional[str] = None
+    order_status: OrderStatus
+    transaction_type: TransactionType
+    exchange_segment: ExchangeSegment
+    product_type: ProductType
+    order_type: OrderType
+    validity: Validity
+    trading_symbol: str
+    security_id: str
+    quantity: int
+    disclosed_quantity: int
+    price: float
+    trigger_price: float
+    after_market_order: bool
+    bo_profit_value: Optional[float] = None
+    bo_stop_loss_value: Optional[float] = None
+    leg_name: Optional[str] = None
+    create_time: str
+    update_time: str
+    exchange_time: Optional[str] = None
+    filled_qty: Optional[int] = None
+    remaining_qty: Optional[int] = None
+    avg_price: Optional[float] = None
+
+class OrdersResponse(BaseModel):
+    orders: List[Order]
+
+class HistoricalDataPoint(BaseModel):
+    timestamp: str
+    open: float
+    high: float
+    low: float
+    close: float
+    volume: int
+
+class HistoricalDataResponse(BaseModel):
+    start: str
+    end: str
+    data: List[HistoricalDataPoint]
+
+class SecurityMaster(BaseModel):
+    SEM_EXM_EXCH_ID: str
+    SEM_SMST_SECURITY_ID: str
+    SEM_INSTRUMENT_NAME: str
+    SEM_TRADING_SYMBOL: str
+    SEM_EXPIRY_DATE: Optional[str] = None
+    SEM_STRIKE_PRICE: Optional[float] = None
+    SEM_OPTION_TYPE: Optional[str] = None
+    SEM_LOT_UNITS: int
+    SEM_CUSTOM_SYMBOL: str
+    SEM_SEGMENT: Optional[str] = None
+    SEM_SERIES: Optional[str] = None
+    SEM_TICK_SIZE: Optional[float] = None
+    SM_SYMBOL_NAME: Optional[str] = None
+
+class SecurityMasterResponse(BaseModel):
+    securities: List[SecurityMaster]
+
+# WebSocket Models
+class WebSocketMessage(BaseModel):
+    type: str
+    data: Dict[str, Any]
+    timestamp: Optional[str] = None
+
+class SubscriptionRequest(BaseModel):
+    type: str = "subscribe"
+    symbols: List[str]
+    feed_type: Optional[str] = "ltp"  # ltp, depth, etc.
+
+class UnsubscriptionRequest(BaseModel):
+    type: str = "unsubscribe"
+    symbols: List[str]
+
+class LiveFeedData(BaseModel):
+    security_id: str
+    exchange_segment: str
+    ltp: float
+    change: float
+    change_percent: float
+    volume: int
+    timestamp: str

requirements.txt

@@ -1,57 +1,55 @@
-# Zyon Traders Backend Dependencies for Hugging Face Spaces
-
-# Web Framework
+# FastAPI and ASGI server
 fastapi==0.104.1
-uvicorn[standard]==0.24.0
-websockets==12.0
-
-# API and HTTP
-httpx==0.25.2
-requests==2.31.0
-pydantic==2.5.0
+uvicorn==0.24.0
+gunicorn==21.2.0
 
-# AI and ML
-langchain==0.0.352
-langchain-community==0.0.10
-chromadb==0.4.18
-sentence-transformers==2.2.2
-huggingface-hub==0.19.4
+# Pydantic v1 for compatibility
+pydantic==1.10.13
 
-# Data Processing
-pandas==2.1.4
-numpy==1.24.4
+# Environment and config
+python-dotenv==1.1.1
 python-multipart==0.0.6
 
-# Authentication and Security
+# HTTP client for API calls
+httpx==0.27.0
+
+# WebSocket support
+websockets==12.0
+
+# Database and ORM
+psycopg2-binary==2.9.9
+sqlalchemy==1.4.53
+alembic==1.13.1
+
+# Authentication and security
 python-jose[cryptography]==3.3.0
 passlib[bcrypt]==1.7.4
-python-dotenv==1.0.0
+python-multipart==0.0.6
 
-# Database and Storage
-redis==5.0.1
-aiofiles==23.2.1
+# Supabase client
+supabase==2.3.4
 
-# Trading and Finance
-yfinance==0.2.28
-ta==0.10.2
+# AI and ML libraries
+openai==1.6.1
+google-generativeai==0.3.2
+huggingface-hub==0.20.2
 
-# Utilities
-python-dateutil==2.8.2
-pytz==2023.3
-schedule==1.2.0
-
-# Development and Monitoring
-structlog==23.2.0
-prometheus-client==0.19.0
+# Data processing
+pandas==2.1.4
+numpy==1.24.4
 
-# CORS and Middleware
-starlette==0.27.0
+# Async file operations
+aiofiles==23.1.0
 
-# WebSocket support
-python-socketio==5.10.0
+# Logging and monitoring
+loguru==0.7.2
 
-# Rate limiting
-slowapi==0.1.9
+# Utils
+python-dateutil==2.8.2
+pytz==2023.3
 
-# Environment and configuration
-pyyaml==6.0.1
+# Development dependencies (comment out for production)
+# pytest==7.4.3
+# pytest-asyncio==0.21.1
+# black==23.12.1
+# flake8==6.1.0

routers/auth.py

@@ -0,0 +1,376 @@
+"""
+Authentication Router
+Handles user authentication with Supabase integration
+"""
+
+from fastapi import APIRouter, HTTPException, Depends, status
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from typing import Optional, Dict, Any
+import logging
+from datetime import datetime, timedelta
+import jwt
+from passlib.context import CryptContext
+
+from config.settings import get_settings
+from models.auth import (
+    UserLogin,
+    UserSignup,
+    UserResponse,
+    TokenResponse,
+    PasswordReset
+)
+from services.supabase_client import get_supabase_client
+from services.auth import create_access_token, verify_token
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+settings = get_settings()
+
+# Security
+security = HTTPBearer()
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+@router.post("/signup", response_model=UserResponse)
+async def signup(user_data: UserSignup):
+    """Register a new user"""
+    try:
+        supabase = get_supabase_client()
+        
+        # Create user with Supabase Auth
+        auth_response = supabase.auth.sign_up({
+            "email": user_data.email,
+            "password": user_data.password,
+            "options": {
+                "data": {
+                    "first_name": user_data.first_name,
+                    "last_name": user_data.last_name,
+                    "full_name": f"{user_data.first_name} {user_data.last_name}",
+                    "role": "trader"  # Default role
+                }
+            }
+        })
+        
+        if auth_response.user:
+            # Create user profile in public.users table
+            profile_data = {
+                "id": auth_response.user.id,
+                "email": user_data.email,
+                "first_name": user_data.first_name,
+                "last_name": user_data.last_name,
+                "role": "trader",
+                "created_at": datetime.utcnow().isoformat(),
+                "updated_at": datetime.utcnow().isoformat()
+            }
+            
+            supabase.table("users").insert(profile_data).execute()
+            
+            return UserResponse(
+                id=auth_response.user.id,
+                email=auth_response.user.email,
+                first_name=user_data.first_name,
+                last_name=user_data.last_name,
+                role="trader",
+                message="User created successfully. Please check your email to verify your account."
+            )
+        else:
+            raise HTTPException(
+                status_code=400,
+                detail="Failed to create user account"
+            )
+            
+    except Exception as e:
+        logger.error(f"Signup error: {e}")
+        if "already registered" in str(e).lower():
+            raise HTTPException(
+                status_code=400,
+                detail="User with this email already exists"
+            )
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to create user account"
+        )
+
+@router.post("/login", response_model=TokenResponse)
+async def login(login_data: UserLogin):
+    """Authenticate user and return tokens"""
+    try:
+        supabase = get_supabase_client()
+        
+        # Authenticate with Supabase
+        auth_response = supabase.auth.sign_in_with_password({
+            "email": login_data.email,
+            "password": login_data.password
+        })
+        
+        if auth_response.user and auth_response.session:
+            # Get user profile
+            profile_response = supabase.table("users").select("*").eq(
+                "id", auth_response.user.id
+            ).execute()
+            
+            profile = profile_response.data[0] if profile_response.data else {}
+            
+            # Create our own JWT token for API access
+            access_token = create_access_token({
+                "sub": auth_response.user.id,
+                "email": auth_response.user.email,
+                "role": profile.get("role", "trader")
+            })
+            
+            return TokenResponse(
+                access_token=access_token,
+                token_type="bearer",
+                expires_in=3600,  # 1 hour
+                user=UserResponse(
+                    id=auth_response.user.id,
+                    email=auth_response.user.email,
+                    first_name=profile.get("first_name", ""),
+                    last_name=profile.get("last_name", ""),
+                    role=profile.get("role", "trader")
+                )
+            )
+        else:
+            raise HTTPException(
+                status_code=401,
+                detail="Invalid email or password"
+            )
+            
+    except Exception as e:
+        logger.error(f"Login error: {e}")
+        if "invalid" in str(e).lower() or "unauthorized" in str(e).lower():
+            raise HTTPException(
+                status_code=401,
+                detail="Invalid email or password"
+            )
+        raise HTTPException(
+            status_code=500,
+            detail="Authentication failed"
+        )
+
+@router.post("/google", response_model=TokenResponse)
+async def google_auth(id_token: str):
+    """Authenticate with Google OAuth"""
+    try:
+        supabase = get_supabase_client()
+        
+        # Sign in with Google
+        auth_response = supabase.auth.sign_in_with_id_token({
+            "provider": "google",
+            "token": id_token
+        })
+        
+        if auth_response.user and auth_response.session:
+            # Check if user profile exists, create if not
+            profile_response = supabase.table("users").select("*").eq(
+                "id", auth_response.user.id
+            ).execute()
+            
+            if not profile_response.data:
+                # Create user profile
+                user_metadata = auth_response.user.user_metadata or {}
+                profile_data = {
+                    "id": auth_response.user.id,
+                    "email": auth_response.user.email,
+                    "first_name": user_metadata.get("given_name", ""),
+                    "last_name": user_metadata.get("family_name", ""),
+                    "role": "trader",
+                    "created_at": datetime.utcnow().isoformat(),
+                    "updated_at": datetime.utcnow().isoformat()
+                }
+                supabase.table("users").insert(profile_data).execute()
+                profile = profile_data
+            else:
+                profile = profile_response.data[0]
+            
+            # Create access token
+            access_token = create_access_token({
+                "sub": auth_response.user.id,
+                "email": auth_response.user.email,
+                "role": profile.get("role", "trader")
+            })
+            
+            return TokenResponse(
+                access_token=access_token,
+                token_type="bearer",
+                expires_in=3600,
+                user=UserResponse(
+                    id=auth_response.user.id,
+                    email=auth_response.user.email,
+                    first_name=profile.get("first_name", ""),
+                    last_name=profile.get("last_name", ""),
+                    role=profile.get("role", "trader")
+                )
+            )
+        else:
+            raise HTTPException(
+                status_code=401,
+                detail="Google authentication failed"
+            )
+            
+    except Exception as e:
+        logger.error(f"Google auth error: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Google authentication failed"
+        )
+
+@router.post("/refresh", response_model=TokenResponse)
+async def refresh_token(
+    credentials: HTTPAuthorizationCredentials = Depends(security)
+):
+    """Refresh access token"""
+    try:
+        # Verify current token
+        payload = verify_token(credentials.credentials)
+        
+        # Create new token
+        access_token = create_access_token({
+            "sub": payload["sub"],
+            "email": payload["email"],
+            "role": payload.get("role", "trader")
+        })
+        
+        return TokenResponse(
+            access_token=access_token,
+            token_type="bearer",
+            expires_in=3600
+        )
+        
+    except Exception as e:
+        logger.error(f"Token refresh error: {e}")
+        raise HTTPException(
+            status_code=401,
+            detail="Invalid or expired token"
+        )
+
+@router.post("/logout")
+async def logout(
+    credentials: HTTPAuthorizationCredentials = Depends(security)
+):
+    """Logout user"""
+    try:
+        # Verify token
+        verify_token(credentials.credentials)
+        
+        # In a production app, you might want to blacklist the token
+        # For now, we'll just return success
+        
+        return {"message": "Logged out successfully"}
+        
+    except Exception as e:
+        logger.error(f"Logout error: {e}")
+        return {"message": "Logged out successfully"}  # Return success even if token is invalid
+
+@router.post("/forgot-password")
+async def forgot_password(email: str):
+    """Send password reset email"""
+    try:
+        supabase = get_supabase_client()
+        
+        # Send password reset email
+        supabase.auth.reset_password_email(email)
+        
+        return {"message": "Password reset email sent"}
+        
+    except Exception as e:
+        logger.error(f"Password reset error: {e}")
+        # Don't reveal if email exists for security
+        return {"message": "If the email exists, a password reset link has been sent"}
+
+@router.post("/reset-password")
+async def reset_password(reset_data: PasswordReset):
+    """Reset password with token"""
+    try:
+        supabase = get_supabase_client()
+        
+        # Update password
+        supabase.auth.update_user({
+            "password": reset_data.new_password
+        })
+        
+        return {"message": "Password updated successfully"}
+        
+    except Exception as e:
+        logger.error(f"Password reset error: {e}")
+        raise HTTPException(
+            status_code=400,
+            detail="Failed to reset password"
+        )
+
+@router.get("/me", response_model=UserResponse)
+async def get_current_user_info(
+    credentials: HTTPAuthorizationCredentials = Depends(security)
+):
+    """Get current user information"""
+    try:
+        payload = verify_token(credentials.credentials)
+        supabase = get_supabase_client()
+        
+        # Get user profile
+        profile_response = supabase.table("users").select("*").eq(
+            "id", payload["sub"]
+        ).execute()
+        
+        if profile_response.data:
+            profile = profile_response.data[0]
+            return UserResponse(
+                id=profile["id"],
+                email=profile["email"],
+                first_name=profile.get("first_name", ""),
+                last_name=profile.get("last_name", ""),
+                role=profile.get("role", "trader")
+            )
+        else:
+            raise HTTPException(
+                status_code=404,
+                detail="User profile not found"
+            )
+            
+    except Exception as e:
+        logger.error(f"Get user info error: {e}")
+        raise HTTPException(
+            status_code=401,
+            detail="Invalid or expired token"
+        )
+
+@router.put("/profile", response_model=UserResponse)
+async def update_profile(
+    profile_data: Dict[str, Any],
+    credentials: HTTPAuthorizationCredentials = Depends(security)
+):
+    """Update user profile"""
+    try:
+        payload = verify_token(credentials.credentials)
+        supabase = get_supabase_client()
+        
+        # Update profile
+        update_data = {
+            **profile_data,
+            "updated_at": datetime.utcnow().isoformat()
+        }
+        
+        response = supabase.table("users").update(update_data).eq(
+            "id", payload["sub"]
+        ).execute()
+        
+        if response.data:
+            profile = response.data[0]
+            return UserResponse(
+                id=profile["id"],
+                email=profile["email"],
+                first_name=profile.get("first_name", ""),
+                last_name=profile.get("last_name", ""),
+                role=profile.get("role", "trader")
+            )
+        else:
+            raise HTTPException(
+                status_code=404,
+                detail="Failed to update profile"
+            )
+            
+    except Exception as e:
+        logger.error(f"Update profile error: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to update profile"
+        )

routers/dhan.py

@@ -0,0 +1,311 @@
+"""
+Dhan API Router
+Handles all Dhan trading API operations with error handling and rate limiting
+"""
+
+from fastapi import APIRouter, HTTPException, Depends, BackgroundTasks
+from typing import List, Optional, Dict, Any
+import httpx
+import asyncio
+from datetime import datetime, timedelta
+import logging
+
+from config.settings import get_settings
+from models.trading import (
+    QuoteRequest, 
+    QuoteResponse, 
+    OrderRequest, 
+    OrderResponse,
+    HoldingsResponse,
+    PositionsResponse,
+    FundsResponse,
+    HistoricalDataRequest,
+    HistoricalDataResponse
+)
+from services.auth import get_current_user
+from utils.rate_limiter import RateLimiter
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+settings = get_settings()
+
+# Rate limiter for Dhan API calls
+rate_limiter = RateLimiter(calls=60, period=60)  # 60 calls per minute
+
+class DhanAPIClient:
+    """Dhan API client with error handling and rate limiting"""
+    
+    def __init__(self):
+        self.base_url = settings.DHAN_API_BASE_URL
+        self.api_key = settings.DHAN_API_KEY
+        
+    def get_headers(self) -> Dict[str, str]:
+        """Get API headers"""
+        return {
+            "Content-Type": "application/json",
+            "access-token": self.api_key,
+            "Accept": "application/json"
+        }
+    
+    async def make_request(
+        self, 
+        method: str, 
+        endpoint: str, 
+        data: Optional[Dict] = None
+    ) -> Dict[str, Any]:
+        """Make rate-limited API request"""
+        
+        # Apply rate limiting
+        await rate_limiter.acquire()
+        
+        url = f"{self.base_url}{endpoint}"
+        headers = self.get_headers()
+        
+        try:
+            async with httpx.AsyncClient(timeout=30.0) as client:
+                if method.upper() == "GET":
+                    response = await client.get(url, headers=headers, params=data)
+                else:
+                    response = await client.request(
+                        method, url, headers=headers, json=data
+                    )
+                
+                response.raise_for_status()
+                
+                result = response.json()
+                logger.info(f"Dhan API {method} {endpoint}: {response.status_code}")
+                
+                return result
+                
+        except httpx.HTTPStatusError as e:
+            logger.error(f"Dhan API HTTP error: {e.response.status_code} - {e.response.text}")
+            raise HTTPException(
+                status_code=e.response.status_code,
+                detail=f"Dhan API error: {e.response.text}"
+            )
+        except httpx.RequestError as e:
+            logger.error(f"Dhan API request error: {e}")
+            raise HTTPException(
+                status_code=503,
+                detail="Dhan API service unavailable"
+            )
+        except Exception as e:
+            logger.error(f"Dhan API unexpected error: {e}")
+            raise HTTPException(
+                status_code=500,
+                detail="Internal server error"
+            )
+
+# Create client instance
+dhan_client = DhanAPIClient()
+
+@router.get("/securities", response_model=List[Dict[str, Any]])
+async def get_securities(
+    exchange: Optional[str] = "NSE",
+    current_user: Dict = Depends(get_current_user)
+):
+    """Get all securities/instruments"""
+    try:
+        result = await dhan_client.make_request("GET", "/securitymaster")
+        
+        # Filter by exchange if specified
+        if exchange and exchange != "ALL":
+            securities = result.get("data", [])
+            filtered = [s for s in securities if s.get("SEM_EXM_EXCH_ID") == exchange]
+            return filtered
+            
+        return result.get("data", [])
+        
+    except Exception as e:
+        logger.error(f"Error fetching securities: {e}")
+        raise HTTPException(status_code=500, detail="Failed to fetch securities")
+
+@router.post("/quotes", response_model=List[QuoteResponse])
+async def get_quotes(
+    request: QuoteRequest,
+    current_user: Dict = Depends(get_current_user)
+):
+    """Get live quotes for multiple securities"""
+    try:
+        data = {
+            "instruments": [
+                {
+                    "securityId": instrument.security_id,
+                    "exchangeSegment": instrument.exchange_segment
+                }
+                for instrument in request.instruments
+            ]
+        }
+        
+        result = await dhan_client.make_request("POST", "/marketfeed/ltp", data)
+        return result.get("data", [])
+        
+    except Exception as e:
+        logger.error(f"Error fetching quotes: {e}")
+        raise HTTPException(status_code=500, detail="Failed to fetch quotes")
+
+@router.get("/quote/{security_id}")
+async def get_single_quote(
+    security_id: str,
+    exchange_segment: str = "NSE_EQ",
+    current_user: Dict = Depends(get_current_user)
+):
+    """Get detailed quote for a single security"""
+    try:
+        endpoint = f"/marketfeed/quote?securityId={security_id}&exchangeSegment={exchange_segment}"
+        result = await dhan_client.make_request("GET", endpoint)
+        return result.get("data", {})
+        
+    except Exception as e:
+        logger.error(f"Error fetching single quote: {e}")
+        raise HTTPException(status_code=500, detail="Failed to fetch quote")
+
+@router.get("/holdings", response_model=HoldingsResponse)
+async def get_holdings(current_user: Dict = Depends(get_current_user)):
+    """Get user holdings"""
+    try:
+        result = await dhan_client.make_request("GET", "/holdings")
+        return {"holdings": result.get("data", [])}
+        
+    except Exception as e:
+        logger.error(f"Error fetching holdings: {e}")
+        raise HTTPException(status_code=500, detail="Failed to fetch holdings")
+
+@router.get("/positions", response_model=PositionsResponse)
+async def get_positions(current_user: Dict = Depends(get_current_user)):
+    """Get user positions"""
+    try:
+        result = await dhan_client.make_request("GET", "/positions")
+        return {"positions": result.get("data", [])}
+        
+    except Exception as e:
+        logger.error(f"Error fetching positions: {e}")
+        raise HTTPException(status_code=500, detail="Failed to fetch positions")
+
+@router.get("/funds", response_model=FundsResponse)
+async def get_funds(current_user: Dict = Depends(get_current_user)):
+    """Get user funds and margin"""
+    try:
+        result = await dhan_client.make_request("GET", "/funds")
+        return result.get("data", {})
+        
+    except Exception as e:
+        logger.error(f"Error fetching funds: {e}")
+        raise HTTPException(status_code=500, detail="Failed to fetch funds")
+
+@router.post("/orders", response_model=OrderResponse)
+async def place_order(
+    order: OrderRequest,
+    current_user: Dict = Depends(get_current_user)
+):
+    """Place a new order"""
+    try:
+        order_data = {
+            "securityId": order.security_id,
+            "exchangeSegment": order.exchange_segment,
+            "transactionType": order.transaction_type,
+            "quantity": order.quantity,
+            "orderType": order.order_type,
+            "productType": order.product_type,
+            "price": order.price if order.price else 0,
+            "triggerPrice": order.trigger_price if order.trigger_price else 0,
+            "validity": order.validity,
+            "disclosedQuantity": order.disclosed_quantity if order.disclosed_quantity else 0,
+            "afterMarketOrder": order.after_market_order if order.after_market_order else False
+        }
+        
+        result = await dhan_client.make_request("POST", "/orders", order_data)
+        return result.get("data", {})
+        
+    except Exception as e:
+        logger.error(f"Error placing order: {e}")
+        raise HTTPException(status_code=500, detail="Failed to place order")
+
+@router.get("/orders")
+async def get_orders(current_user: Dict = Depends(get_current_user)):
+    """Get user orders"""
+    try:
+        result = await dhan_client.make_request("GET", "/orders")
+        return {"orders": result.get("data", [])}
+        
+    except Exception as e:
+        logger.error(f"Error fetching orders: {e}")
+        raise HTTPException(status_code=500, detail="Failed to fetch orders")
+
+@router.delete("/orders/{order_id}")
+async def cancel_order(
+    order_id: str,
+    current_user: Dict = Depends(get_current_user)
+):
+    """Cancel an order"""
+    try:
+        result = await dhan_client.make_request("DELETE", f"/orders/{order_id}")
+        return {"message": "Order cancelled successfully", "data": result.get("data", {})}
+        
+    except Exception as e:
+        logger.error(f"Error cancelling order: {e}")
+        raise HTTPException(status_code=500, detail="Failed to cancel order")
+
+@router.put("/orders/{order_id}")
+async def modify_order(
+    order_id: str,
+    modifications: Dict[str, Any],
+    current_user: Dict = Depends(get_current_user)
+):
+    """Modify an existing order"""
+    try:
+        result = await dhan_client.make_request("PUT", f"/orders/{order_id}", modifications)
+        return {"message": "Order modified successfully", "data": result.get("data", {})}
+        
+    except Exception as e:
+        logger.error(f"Error modifying order: {e}")
+        raise HTTPException(status_code=500, detail="Failed to modify order")
+
+@router.post("/historical", response_model=HistoricalDataResponse)
+async def get_historical_data(
+    request: HistoricalDataRequest,
+    current_user: Dict = Depends(get_current_user)
+):
+    """Get historical data for charting"""
+    try:
+        data = {
+            "securityId": request.security_id,
+            "exchangeSegment": request.exchange_segment,
+            "instrument": request.instrument,
+            "interval": request.interval,
+            "fromDate": request.from_date,
+            "toDate": request.to_date
+        }
+        
+        result = await dhan_client.make_request("POST", "/charts/historical", data)
+        return result.get("data", {})
+        
+    except Exception as e:
+        logger.error(f"Error fetching historical data: {e}")
+        raise HTTPException(status_code=500, detail="Failed to fetch historical data")
+
+@router.get("/market-status")
+async def get_market_status(current_user: Dict = Depends(get_current_user)):
+    """Get current market status"""
+    try:
+        # This endpoint might not exist in Dhan API, so we'll create a mock response
+        # based on time and market hours
+        now = datetime.now()
+        is_weekend = now.weekday() >= 5  # Saturday = 5, Sunday = 6
+        
+        # NSE trading hours: 9:15 AM to 3:30 PM
+        market_open_time = now.replace(hour=9, minute=15, second=0, microsecond=0)
+        market_close_time = now.replace(hour=15, minute=30, second=0, microsecond=0)
+        
+        is_market_hours = market_open_time <= now <= market_close_time and not is_weekend
+        
+        return {
+            "market_status": "OPEN" if is_market_hours else "CLOSED",
+            "timestamp": now.isoformat(),
+            "next_open": market_open_time.isoformat() if not is_market_hours else None,
+            "next_close": market_close_time.isoformat() if is_market_hours else None
+        }
+        
+    except Exception as e:
+        logger.error(f"Error getting market status: {e}")
+        raise HTTPException(status_code=500, detail="Failed to get market status")

routers/portfolio.py

@@ -0,0 +1,402 @@
+"""
+Portfolio Management Router
+Handles portfolio tracking, analytics, and risk management
+"""
+
+from fastapi import APIRouter, HTTPException, Depends
+from typing import List, Dict, Any, Optional
+import logging
+from datetime import datetime, timedelta
+
+from config.settings import get_settings
+from models.portfolio import (
+    PortfolioSummary,
+    RiskAssessment,
+    PerformanceMetrics,
+    AllocationRequest
+)
+from services.auth import get_current_user
+from services.portfolio_service import PortfolioService
+from services.risk_engine import RiskEngine
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+settings = get_settings()
+
+# Initialize services
+portfolio_service = PortfolioService()
+risk_engine = RiskEngine()
+
+@router.get("/summary", response_model=PortfolioSummary)
+async def get_portfolio_summary(current_user: Dict = Depends(get_current_user)):
+    """Get comprehensive portfolio summary"""
+    try:
+        summary = await portfolio_service.get_portfolio_summary(
+            user_id=current_user["id"]
+        )
+        return summary
+        
+    except Exception as e:
+        logger.error(f"Error fetching portfolio summary: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to fetch portfolio summary"
+        )
+
+@router.get("/risk-assessment", response_model=RiskAssessment)
+async def get_risk_assessment(current_user: Dict = Depends(get_current_user)):
+    """Get portfolio risk assessment"""
+    try:
+        # Get current holdings
+        holdings = await portfolio_service.get_holdings(current_user["id"])
+        
+        # Perform risk assessment
+        risk_assessment = await risk_engine.assess_portfolio_risk(holdings)
+        
+        return risk_assessment
+        
+    except Exception as e:
+        logger.error(f"Error performing risk assessment: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to perform risk assessment"
+        )
+
+@router.get("/performance", response_model=PerformanceMetrics)
+async def get_performance_metrics(
+    period: str = "1Y",
+    current_user: Dict = Depends(get_current_user)
+):
+    """Get portfolio performance metrics"""
+    try:
+        # Calculate date range
+        end_date = datetime.utcnow()
+        
+        period_map = {
+            "1D": timedelta(days=1),
+            "1W": timedelta(weeks=1),
+            "1M": timedelta(days=30),
+            "3M": timedelta(days=90),
+            "6M": timedelta(days=180),
+            "1Y": timedelta(days=365),
+            "ALL": timedelta(days=3650)  # 10 years
+        }
+        
+        start_date = end_date - period_map.get(period, timedelta(days=365))
+        
+        # Calculate performance metrics
+        performance = await portfolio_service.calculate_performance_metrics(
+            user_id=current_user["id"],
+            start_date=start_date,
+            end_date=end_date
+        )
+        
+        return performance
+        
+    except Exception as e:
+        logger.error(f"Error calculating performance metrics: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to calculate performance metrics"
+        )
+
+@router.get("/allocation")
+async def get_asset_allocation(current_user: Dict = Depends(get_current_user)):
+    """Get current asset allocation breakdown"""
+    try:
+        allocation = await portfolio_service.get_asset_allocation(
+            user_id=current_user["id"]
+        )
+        
+        return {
+            "sector_allocation": allocation["sectors"],
+            "market_cap_allocation": allocation["market_caps"],
+            "geographic_allocation": allocation["geography"],
+            "concentration_metrics": allocation["concentration"],
+            "diversification_score": allocation["diversification_score"]
+        }
+        
+    except Exception as e:
+        logger.error(f"Error fetching asset allocation: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to fetch asset allocation"
+        )
+
+@router.post("/rebalance")
+async def suggest_portfolio_rebalancing(
+    target_allocation: AllocationRequest,
+    current_user: Dict = Depends(get_current_user)
+):
+    """Suggest portfolio rebalancing to achieve target allocation"""
+    try:
+        # Get current portfolio
+        current_portfolio = await portfolio_service.get_holdings(current_user["id"])
+        
+        # Calculate rebalancing suggestions
+        rebalancing_plan = await portfolio_service.calculate_rebalancing(
+            current_portfolio=current_portfolio,
+            target_allocation=target_allocation
+        )
+        
+        return {
+            "current_allocation": rebalancing_plan["current"],
+            "target_allocation": rebalancing_plan["target"],
+            "suggested_trades": rebalancing_plan["trades"],
+            "estimated_cost": rebalancing_plan["cost"],
+            "tax_implications": rebalancing_plan["tax_impact"]
+        }
+        
+    except Exception as e:
+        logger.error(f"Error calculating rebalancing: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to calculate rebalancing suggestions"
+        )
+
+@router.get("/correlation-matrix")
+async def get_correlation_matrix(current_user: Dict = Depends(get_current_user)):
+    """Get correlation matrix for portfolio holdings"""
+    try:
+        # Get holdings
+        holdings = await portfolio_service.get_holdings(current_user["id"])
+        
+        # Calculate correlation matrix
+        correlation_matrix = await portfolio_service.calculate_correlation_matrix(
+            holdings
+        )
+        
+        return {
+            "correlation_matrix": correlation_matrix,
+            "timestamp": datetime.utcnow().isoformat()
+        }
+        
+    except Exception as e:
+        logger.error(f"Error calculating correlation matrix: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to calculate correlation matrix"
+        )
+
+@router.get("/value-at-risk")
+async def calculate_value_at_risk(
+    confidence_level: float = 0.95,
+    time_horizon: int = 1,
+    current_user: Dict = Depends(get_current_user)
+):
+    """Calculate Value at Risk (VaR) for the portfolio"""
+    try:
+        # Get portfolio data
+        portfolio_data = await portfolio_service.get_portfolio_for_var_calculation(
+            user_id=current_user["id"]
+        )
+        
+        # Calculate VaR using different methods
+        var_results = await risk_engine.calculate_value_at_risk(
+            portfolio_data=portfolio_data,
+            confidence_level=confidence_level,
+            time_horizon=time_horizon
+        )
+        
+        return {
+            "var_historical": var_results["historical"],
+            "var_parametric": var_results["parametric"],
+            "var_monte_carlo": var_results["monte_carlo"],
+            "expected_shortfall": var_results["expected_shortfall"],
+            "confidence_level": confidence_level,
+            "time_horizon_days": time_horizon,
+            "portfolio_value": var_results["portfolio_value"],
+            "timestamp": datetime.utcnow().isoformat()
+        }
+        
+    except Exception as e:
+        logger.error(f"Error calculating VaR: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to calculate Value at Risk"
+        )
+
+@router.get("/stress-testing")
+async def run_stress_testing(
+    scenario: str = "market_crash",
+    current_user: Dict = Depends(get_current_user)
+):
+    """Run stress testing scenarios on the portfolio"""
+    try:
+        # Get portfolio data
+        portfolio_data = await portfolio_service.get_holdings(current_user["id"])
+        
+        # Run stress testing
+        stress_results = await risk_engine.run_stress_testing(
+            portfolio_data=portfolio_data,
+            scenario=scenario
+        )
+        
+        return {
+            "scenario": scenario,
+            "results": stress_results,
+            "recommendations": stress_results.get("recommendations", []),
+            "timestamp": datetime.utcnow().isoformat()
+        }
+        
+    except Exception as e:
+        logger.error(f"Error running stress testing: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to run stress testing"
+        )
+
+@router.get("/optimization")
+async def get_portfolio_optimization(
+    objective: str = "sharpe_ratio",
+    constraints: Optional[Dict[str, Any]] = None,
+    current_user: Dict = Depends(get_current_user)
+):
+    """Get portfolio optimization suggestions"""
+    try:
+        # Get current portfolio and market data
+        portfolio_data = await portfolio_service.get_holdings(current_user["id"])
+        
+        # Run portfolio optimization
+        optimization_results = await portfolio_service.optimize_portfolio(
+            current_portfolio=portfolio_data,
+            objective=objective,
+            constraints=constraints or {}
+        )
+        
+        return {
+            "current_allocation": optimization_results["current"],
+            "optimal_allocation": optimization_results["optimal"],
+            "expected_return": optimization_results["expected_return"],
+            "expected_volatility": optimization_results["expected_volatility"],
+            "sharpe_ratio": optimization_results["sharpe_ratio"],
+            "suggested_changes": optimization_results["changes"],
+            "objective": objective,
+            "timestamp": datetime.utcnow().isoformat()
+        }
+        
+    except Exception as e:
+        logger.error(f"Error running portfolio optimization: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to run portfolio optimization"
+        )
+
+@router.get("/tax-optimization")
+async def get_tax_optimization(
+    tax_year: Optional[int] = None,
+    current_user: Dict = Depends(get_current_user)
+):
+    """Get tax optimization suggestions"""
+    try:
+        if not tax_year:
+            tax_year = datetime.utcnow().year
+        
+        # Get holdings with cost basis information
+        holdings_with_cost_basis = await portfolio_service.get_holdings_with_cost_basis(
+            user_id=current_user["id"]
+        )
+        
+        # Calculate tax optimization strategies
+        tax_strategies = await portfolio_service.calculate_tax_optimization(
+            holdings=holdings_with_cost_basis,
+            tax_year=tax_year
+        )
+        
+        return {
+            "tax_year": tax_year,
+            "current_tax_liability": tax_strategies["current_liability"],
+            "tax_loss_harvesting": tax_strategies["loss_harvesting"],
+            "wash_sale_warnings": tax_strategies["wash_sale_warnings"],
+            "suggested_actions": tax_strategies["actions"],
+            "potential_savings": tax_strategies["potential_savings"],
+            "timestamp": datetime.utcnow().isoformat()
+        }
+        
+    except Exception as e:
+        logger.error(f"Error calculating tax optimization: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to calculate tax optimization"
+        )
+
+@router.get("/benchmarking")
+async def get_benchmark_comparison(
+    benchmark: str = "NIFTY50",
+    period: str = "1Y",
+    current_user: Dict = Depends(get_current_user)
+):
+    """Compare portfolio performance against benchmarks"""
+    try:
+        # Calculate date range
+        end_date = datetime.utcnow()
+        period_map = {
+            "1M": timedelta(days=30),
+            "3M": timedelta(days=90),
+            "6M": timedelta(days=180),
+            "1Y": timedelta(days=365),
+            "2Y": timedelta(days=730),
+            "5Y": timedelta(days=1825)
+        }
+        start_date = end_date - period_map.get(period, timedelta(days=365))
+        
+        # Get portfolio and benchmark performance
+        comparison = await portfolio_service.compare_with_benchmark(
+            user_id=current_user["id"],
+            benchmark=benchmark,
+            start_date=start_date,
+            end_date=end_date
+        )
+        
+        return {
+            "benchmark": benchmark,
+            "period": period,
+            "portfolio_return": comparison["portfolio_return"],
+            "benchmark_return": comparison["benchmark_return"],
+            "alpha": comparison["alpha"],
+            "beta": comparison["beta"],
+            "tracking_error": comparison["tracking_error"],
+            "information_ratio": comparison["information_ratio"],
+            "up_capture": comparison["up_capture"],
+            "down_capture": comparison["down_capture"],
+            "timestamp": datetime.utcnow().isoformat()
+        }
+        
+    except Exception as e:
+        logger.error(f"Error performing benchmark comparison: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to perform benchmark comparison"
+        )
+
+@router.post("/simulate")
+async def simulate_portfolio_changes(
+    changes: List[Dict[str, Any]],
+    current_user: Dict = Depends(get_current_user)
+):
+    """Simulate the impact of portfolio changes"""
+    try:
+        # Get current portfolio
+        current_portfolio = await portfolio_service.get_holdings(current_user["id"])
+        
+        # Simulate changes
+        simulation_results = await portfolio_service.simulate_portfolio_changes(
+            current_portfolio=current_portfolio,
+            proposed_changes=changes
+        )
+        
+        return {
+            "current_metrics": simulation_results["current"],
+            "projected_metrics": simulation_results["projected"],
+            "impact_analysis": simulation_results["impact"],
+            "risk_assessment": simulation_results["risk_change"],
+            "recommendations": simulation_results["recommendations"],
+            "timestamp": datetime.utcnow().isoformat()
+        }
+        
+    except Exception as e:
+        logger.error(f"Error simulating portfolio changes: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to simulate portfolio changes"
+        )

routers/signals.py

@@ -0,0 +1,409 @@
+"""
+AI Trading Signals Router
+Handles AI-powered trading signal generation and management
+"""
+
+from fastapi import APIRouter, HTTPException, Depends, BackgroundTasks
+from typing import List, Dict, Any, Optional
+import asyncio
+import logging
+from datetime import datetime, timedelta
+
+from config.settings import get_settings
+from models.signals import (
+    SignalRequest,
+    SignalResponse,
+    MarketAnalysisResponse,
+    SignalFilter
+)
+from services.ai_engine import AIEngine
+from services.auth import get_current_user
+from services.market_data import MarketDataService
+from utils.cache import get_redis_client
+
+logger = logging.getLogger(__name__)
+router = APIRouter()
+settings = get_settings()
+
+# Initialize services
+ai_engine = AIEngine()
+market_data_service = MarketDataService()
+
+@router.post("/generate", response_model=List[SignalResponse])
+async def generate_signals(
+    request: SignalRequest,
+    background_tasks: BackgroundTasks,
+    current_user: Dict = Depends(get_current_user)
+):
+    """Generate AI trading signals for specified securities"""
+    try:
+        # Get market data for the requested securities
+        market_data = await market_data_service.get_market_data_batch(
+            request.securities
+        )
+        
+        # Generate signals using AI engine
+        signals = await ai_engine.generate_trading_signals(
+            market_data=market_data,
+            user_preferences=request.preferences
+        )
+        
+        # Store signals in database for historical tracking
+        background_tasks.add_task(
+            store_signals_history,
+            signals,
+            current_user["id"]
+        )
+        
+        return signals
+        
+    except Exception as e:
+        logger.error(f"Error generating signals: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to generate trading signals"
+        )
+
+@router.get("/", response_model=List[SignalResponse])
+async def get_signals(
+    filter_params: SignalFilter = Depends(),
+    current_user: Dict = Depends(get_current_user)
+):
+    """Get existing signals with filtering options"""
+    try:
+        # Check cache first
+        redis_client = get_redis_client()
+        cache_key = f"signals:{current_user['id']}:{hash(str(filter_params))}"
+        
+        if redis_client:
+            cached_signals = await redis_client.get(cache_key)
+            if cached_signals:
+                return cached_signals
+        
+        # Get signals from database
+        signals = await get_user_signals(
+            user_id=current_user["id"],
+            filters=filter_params
+        )
+        
+        # Cache for 5 minutes
+        if redis_client:
+            await redis_client.setex(cache_key, 300, signals)
+        
+        return signals
+        
+    except Exception as e:
+        logger.error(f"Error fetching signals: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to fetch signals"
+        )
+
+@router.get("/performance")
+async def get_signals_performance(
+    days: int = 30,
+    current_user: Dict = Depends(get_current_user)
+):
+    """Get signals performance analytics"""
+    try:
+        end_date = datetime.utcnow()
+        start_date = end_date - timedelta(days=days)
+        
+        # Get signals performance from database
+        performance = await calculate_signals_performance(
+            user_id=current_user["id"],
+            start_date=start_date,
+            end_date=end_date
+        )
+        
+        return performance
+        
+    except Exception as e:
+        logger.error(f"Error calculating performance: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to calculate signals performance"
+        )
+
+@router.get("/market-analysis", response_model=MarketAnalysisResponse)
+async def get_market_analysis(current_user: Dict = Depends(get_current_user)):
+    """Get comprehensive market analysis"""
+    try:
+        # Check cache first
+        redis_client = get_redis_client()
+        cache_key = "market_analysis:global"
+        
+        if redis_client:
+            cached_analysis = await redis_client.get(cache_key)
+            if cached_analysis:
+                return cached_analysis
+        
+        # Generate fresh market analysis
+        analysis = await ai_engine.generate_market_analysis()
+        
+        # Cache for 15 minutes
+        if redis_client:
+            await redis_client.setex(cache_key, 900, analysis)
+        
+        return analysis
+        
+    except Exception as e:
+        logger.error(f"Error generating market analysis: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to generate market analysis"
+        )
+
+@router.post("/backtesting")
+async def run_backtesting(
+    strategy_params: Dict[str, Any],
+    current_user: Dict = Depends(get_current_user)
+):
+    """Run backtesting on AI trading strategies"""
+    try:
+        # Validate strategy parameters
+        if not strategy_params.get("symbols") or not strategy_params.get("period"):
+            raise HTTPException(
+                status_code=400,
+                detail="Missing required parameters: symbols and period"
+            )
+        
+        # Run backtesting
+        backtest_results = await ai_engine.run_backtesting(
+            symbols=strategy_params["symbols"],
+            period=strategy_params["period"],
+            strategy_config=strategy_params.get("config", {})
+        )
+        
+        return backtest_results
+        
+    except Exception as e:
+        logger.error(f"Error running backtesting: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to run backtesting"
+        )
+
+@router.get("/sentiment/{symbol}")
+async def get_stock_sentiment(
+    symbol: str,
+    current_user: Dict = Depends(get_current_user)
+):
+    """Get AI-powered sentiment analysis for a specific stock"""
+    try:
+        # Get sentiment analysis
+        sentiment = await ai_engine.analyze_stock_sentiment(symbol)
+        
+        return {
+            "symbol": symbol,
+            "sentiment_score": sentiment["score"],
+            "sentiment_label": sentiment["label"],
+            "confidence": sentiment["confidence"],
+            "key_factors": sentiment["factors"],
+            "news_summary": sentiment["news_summary"],
+            "timestamp": datetime.utcnow().isoformat()
+        }
+        
+    except Exception as e:
+        logger.error(f"Error analyzing sentiment for {symbol}: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail=f"Failed to analyze sentiment for {symbol}"
+        )
+
+@router.post("/alerts")
+async def create_signal_alert(
+    alert_config: Dict[str, Any],
+    current_user: Dict = Depends(get_current_user)
+):
+    """Create alert for specific signal conditions"""
+    try:
+        # Validate alert configuration
+        required_fields = ["symbol", "condition", "threshold"]
+        if not all(field in alert_config for field in required_fields):
+            raise HTTPException(
+                status_code=400,
+                detail="Missing required fields: symbol, condition, threshold"
+            )
+        
+        # Store alert configuration
+        alert_id = await store_signal_alert(
+            user_id=current_user["id"],
+            config=alert_config
+        )
+        
+        return {
+            "alert_id": alert_id,
+            "message": "Signal alert created successfully",
+            "config": alert_config
+        }
+        
+    except Exception as e:
+        logger.error(f"Error creating signal alert: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to create signal alert"
+        )
+
+@router.get("/alerts")
+async def get_user_alerts(current_user: Dict = Depends(get_current_user)):
+    """Get user's active signal alerts"""
+    try:
+        alerts = await get_user_signal_alerts(current_user["id"])
+        return {"alerts": alerts}
+        
+    except Exception as e:
+        logger.error(f"Error fetching user alerts: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to fetch alerts"
+        )
+
+@router.delete("/alerts/{alert_id}")
+async def delete_signal_alert(
+    alert_id: str,
+    current_user: Dict = Depends(get_current_user)
+):
+    """Delete a signal alert"""
+    try:
+        success = await delete_user_alert(
+            alert_id=alert_id,
+            user_id=current_user["id"]
+        )
+        
+        if success:
+            return {"message": "Alert deleted successfully"}
+        else:
+            raise HTTPException(
+                status_code=404,
+                detail="Alert not found"
+            )
+            
+    except Exception as e:
+        logger.error(f"Error deleting alert: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to delete alert"
+        )
+
+@router.get("/strategies")
+async def get_ai_strategies(current_user: Dict = Depends(get_current_user)):
+    """Get available AI trading strategies"""
+    try:
+        strategies = await ai_engine.get_available_strategies()
+        return {"strategies": strategies}
+        
+    except Exception as e:
+        logger.error(f"Error fetching strategies: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to fetch strategies"
+        )
+
+@router.post("/custom-strategy")
+async def create_custom_strategy(
+    strategy_config: Dict[str, Any],
+    current_user: Dict = Depends(get_current_user)
+):
+    """Create a custom AI trading strategy"""
+    try:
+        # Validate strategy configuration
+        if not strategy_config.get("name") or not strategy_config.get("rules"):
+            raise HTTPException(
+                status_code=400,
+                detail="Missing required fields: name and rules"
+            )
+        
+        # Create and validate strategy
+        strategy_id = await ai_engine.create_custom_strategy(
+            user_id=current_user["id"],
+            config=strategy_config
+        )
+        
+        return {
+            "strategy_id": strategy_id,
+            "message": "Custom strategy created successfully",
+            "config": strategy_config
+        }
+        
+    except Exception as e:
+        logger.error(f"Error creating custom strategy: {e}")
+        raise HTTPException(
+            status_code=500,
+            detail="Failed to create custom strategy"
+        )
+
+# Helper functions
+async def store_signals_history(signals: List[SignalResponse], user_id: str):
+    """Store signals in database for historical tracking"""
+    try:
+        # Implementation would depend on your database choice
+        # This is a placeholder for the actual database storage logic
+        pass
+    except Exception as e:
+        logger.error(f"Error storing signals history: {e}")
+
+async def get_user_signals(
+    user_id: str, 
+    filters: SignalFilter
+) -> List[SignalResponse]:
+    """Get user signals from database with filtering"""
+    try:
+        # Implementation would depend on your database choice
+        # This is a placeholder for the actual database query logic
+        return []
+    except Exception as e:
+        logger.error(f"Error fetching user signals: {e}")
+        return []
+
+async def calculate_signals_performance(
+    user_id: str,
+    start_date: datetime,
+    end_date: datetime
+) -> Dict[str, Any]:
+    """Calculate signals performance metrics"""
+    try:
+        # Implementation would depend on your database choice
+        # This is a placeholder for the actual performance calculation logic
+        return {
+            "total_signals": 0,
+            "successful_signals": 0,
+            "accuracy": 0.0,
+            "avg_return": 0.0,
+            "max_drawdown": 0.0,
+            "sharpe_ratio": 0.0
+        }
+    except Exception as e:
+        logger.error(f"Error calculating performance: {e}")
+        return {}
+
+async def store_signal_alert(user_id: str, config: Dict[str, Any]) -> str:
+    """Store signal alert configuration"""
+    try:
+        # Implementation would depend on your database choice
+        # This is a placeholder for the actual alert storage logic
+        import uuid
+        return str(uuid.uuid4())
+    except Exception as e:
+        logger.error(f"Error storing signal alert: {e}")
+        raise
+
+async def get_user_signal_alerts(user_id: str) -> List[Dict[str, Any]]:
+    """Get user's signal alerts"""
+    try:
+        # Implementation would depend on your database choice
+        # This is a placeholder for the actual alert retrieval logic
+        return []
+    except Exception as e:
+        logger.error(f"Error fetching user alerts: {e}")
+        return []
+
+async def delete_user_alert(alert_id: str, user_id: str) -> bool:
+    """Delete user's signal alert"""
+    try:
+        # Implementation would depend on your database choice
+        # This is a placeholder for the actual alert deletion logic
+        return True
+    except Exception as e:
+        logger.error(f"Error deleting user alert: {e}")
+        return False

services/auth.py

@@ -0,0 +1,286 @@
+"""
+Authentication service with JWT token management
+"""
+
+import jwt
+from datetime import datetime, timedelta
+from typing import Dict, Any, Optional
+from fastapi import HTTPException, status, Depends
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from passlib.context import CryptContext
+import logging
+
+from config.settings import get_settings
+
+logger = logging.getLogger(__name__)
+settings = get_settings()
+security = HTTPBearer()
+
+# Password hashing
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+# JWT settings
+SECRET_KEY = settings.SECRET_KEY
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 60
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    """Verify a password against its hash"""
+    return pwd_context.verify(plain_password, hashed_password)
+
+def get_password_hash(password: str) -> str:
+    """Hash a password"""
+    return pwd_context.hash(password)
+
+def create_access_token(data: Dict[str, Any], expires_delta: Optional[timedelta] = None) -> str:
+    """Create a new access token"""
+    to_encode = data.copy()
+    
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    
+    to_encode.update({
+        "exp": expire,
+        "iat": datetime.utcnow(),
+        "type": "access"
+    })
+    
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+def create_refresh_token(data: Dict[str, Any]) -> str:
+    """Create a new refresh token"""
+    to_encode = data.copy()
+    expire = datetime.utcnow() + timedelta(days=7)  # Refresh token valid for 7 days
+    
+    to_encode.update({
+        "exp": expire,
+        "iat": datetime.utcnow(),
+        "type": "refresh"
+    })
+    
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+def verify_token(token: str) -> Dict[str, Any]:
+    """Verify and decode a JWT token"""
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        
+        # Check if token has expired
+        exp = payload.get("exp")
+        if exp and datetime.fromtimestamp(exp) < datetime.utcnow():
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail="Token has expired",
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+        
+        return payload
+        
+    except jwt.ExpiredSignatureError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Token has expired",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    except jwt.JWTError:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid token",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+
+async def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)) -> Dict[str, Any]:
+    """Get current user from JWT token"""
+    
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    
+    try:
+        payload = verify_token(credentials.credentials)
+        user_id: str = payload.get("sub")
+        email: str = payload.get("email")
+        role: str = payload.get("role", "trader")
+        
+        if user_id is None or email is None:
+            raise credentials_exception
+            
+        return {
+            "id": user_id,
+            "email": email,
+            "role": role
+        }
+        
+    except HTTPException:
+        raise credentials_exception
+    except Exception as e:
+        logger.error(f"Error validating user credentials: {e}")
+        raise credentials_exception
+
+async def get_current_admin(current_user: Dict = Depends(get_current_user)) -> Dict[str, Any]:
+    """Get current user and verify admin role"""
+    
+    if current_user.get("role") != "admin":
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Insufficient permissions. Admin role required."
+        )
+    
+    return current_user
+
+async def get_current_instructor_or_admin(current_user: Dict = Depends(get_current_user)) -> Dict[str, Any]:
+    """Get current user and verify instructor or admin role"""
+    
+    allowed_roles = ["admin", "instructor"]
+    if current_user.get("role") not in allowed_roles:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Insufficient permissions. Instructor or Admin role required."
+        )
+    
+    return current_user
+
+def create_token_pair(user_data: Dict[str, Any]) -> Dict[str, Any]:
+    """Create access and refresh token pair"""
+    
+    token_data = {
+        "sub": user_data["id"],
+        "email": user_data["email"],
+        "role": user_data.get("role", "trader")
+    }
+    
+    access_token = create_access_token(token_data)
+    refresh_token = create_refresh_token(token_data)
+    
+    return {
+        "access_token": access_token,
+        "refresh_token": refresh_token,
+        "token_type": "bearer",
+        "expires_in": ACCESS_TOKEN_EXPIRE_MINUTES * 60  # Convert to seconds
+    }
+
+def decode_token_without_verification(token: str) -> Optional[Dict[str, Any]]:
+    """Decode token without verification (for debugging purposes only)"""
+    try:
+        return jwt.decode(token, options={"verify_signature": False})
+    except Exception as e:
+        logger.error(f"Error decoding token: {e}")
+        return None
+
+class TokenBlacklist:
+    """Simple in-memory token blacklist. In production, use Redis or database."""
+    
+    def __init__(self):
+        self._blacklisted_tokens = set()
+    
+    def add_token(self, token: str):
+        """Add token to blacklist"""
+        self._blacklisted_tokens.add(token)
+    
+    def is_blacklisted(self, token: str) -> bool:
+        """Check if token is blacklisted"""
+        return token in self._blacklisted_tokens
+    
+    def cleanup_expired(self):
+        """Remove expired tokens from blacklist"""
+        # Implementation would check token expiry and remove expired ones
+        # This is a simplified version
+        pass
+
+# Global token blacklist instance
+token_blacklist = TokenBlacklist()
+
+async def verify_active_token(credentials: HTTPAuthorizationCredentials = Depends(security)) -> str:
+    """Verify token is not blacklisted"""
+    
+    token = credentials.credentials
+    
+    if token_blacklist.is_blacklisted(token):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Token has been revoked",
+            headers={"WWW-Authenticate": "Bearer"},
+        )
+    
+    return token
+
+def invalidate_token(token: str):
+    """Add token to blacklist"""
+    token_blacklist.add_token(token)
+
+# Rate limiting for authentication endpoints
+class AuthRateLimiter:
+    """Simple rate limiter for authentication endpoints"""
+    
+    def __init__(self):
+        self._attempts = {}
+    
+    def is_rate_limited(self, identifier: str, max_attempts: int = 5, window_minutes: int = 15) -> bool:
+        """Check if identifier is rate limited"""
+        now = datetime.utcnow()
+        window_start = now - timedelta(minutes=window_minutes)
+        
+        # Clean old attempts
+        if identifier in self._attempts:
+            self._attempts[identifier] = [
+                attempt for attempt in self._attempts[identifier]
+                if attempt > window_start
+            ]
+        
+        # Check current attempts
+        attempts = self._attempts.get(identifier, [])
+        return len(attempts) >= max_attempts
+    
+    def record_attempt(self, identifier: str):
+        """Record an authentication attempt"""
+        now = datetime.utcnow()
+        if identifier not in self._attempts:
+            self._attempts[identifier] = []
+        self._attempts[identifier].append(now)
+
+# Global rate limiter instance
+auth_rate_limiter = AuthRateLimiter()
+
+def check_rate_limit(identifier: str, max_attempts: int = 5) -> bool:
+    """Check and record rate limit for authentication"""
+    
+    if auth_rate_limiter.is_rate_limited(identifier, max_attempts):
+        raise HTTPException(
+            status_code=status.HTTP_429_TOO_MANY_REQUESTS,
+            detail="Too many authentication attempts. Please try again later."
+        )
+    
+    auth_rate_limiter.record_attempt(identifier)
+    return True
+
+def generate_api_key(user_id: str) -> str:
+    """Generate API key for user"""
+    import secrets
+    import hashlib
+    
+    # Generate a random API key
+    api_key = secrets.token_urlsafe(32)
+    
+    # Create a hash for storage (never store plain API keys)
+    api_key_hash = hashlib.sha256(api_key.encode()).hexdigest()
+    
+    # In production, store api_key_hash in database linked to user_id
+    
+    return api_key
+
+def verify_api_key(api_key: str) -> Optional[str]:
+    """Verify API key and return user_id"""
+    import hashlib
+    
+    api_key_hash = hashlib.sha256(api_key.encode()).hexdigest()
+    
+    # In production, lookup api_key_hash in database and return user_id
+    # This is a placeholder implementation
+    
+    return None  # Return user_id if found

services/supabase_client.py

@@ -0,0 +1,42 @@
+"""
+Supabase client service
+"""
+
+from supabase import create_client, Client
+from functools import lru_cache
+import logging
+
+from config.settings import get_settings
+
+logger = logging.getLogger(__name__)
+settings = get_settings()
+
+@lru_cache()
+def get_supabase_client() -> Client:
+    """Get Supabase client instance"""
+    try:
+        supabase: Client = create_client(
+            settings.SUPABASE_PROJECT_URL,
+            settings.SUPABASE_ANON_KEY
+        )
+        return supabase
+    except Exception as e:
+        logger.error(f"Failed to create Supabase client: {e}")
+        raise
+
+@lru_cache()
+def get_supabase_admin_client() -> Client:
+    """Get Supabase admin client instance"""
+    try:
+        if not settings.SUPABASE_SERVICE_ROLE_KEY:
+            logger.warning("SUPABASE_SERVICE_ROLE_KEY not set, using anon key")
+            return get_supabase_client()
+        
+        supabase: Client = create_client(
+            settings.SUPABASE_PROJECT_URL,
+            settings.SUPABASE_SERVICE_ROLE_KEY
+        )
+        return supabase
+    except Exception as e:
+        logger.error(f"Failed to create Supabase admin client: {e}")
+        raise

services/websocket_manager.py

@@ -1,104 +1,169 @@
 """
-WebSocket Manager for real-time data streaming
-Handles WebSocket connections and broadcasting
+WebSocket Manager for Zyon Traders Backend
+Handles real-time data distribution to connected clients
 """
 
 import asyncio
-import logging
-from typing import List, Dict, Any
-from datetime import datetime
 import json
-
+import logging
+from typing import Dict, List, Set
 from fastapi import WebSocket
+from datetime import datetime
+import random
 
 logger = logging.getLogger(__name__)
 
+
 class WebSocketManager:
-    """Manages WebSocket connections for real-time data"""
+    """Manages WebSocket connections and real-time data distribution"""
     
     def __init__(self):
         self.active_connections: List[WebSocket] = []
-        
-        logger.info("🔌 WebSocket manager initialized")
+        self.subscriptions: Dict[WebSocket, Set[str]] = {}
+        self.background_task: asyncio.Task = None
+        self.running = False
     
     async def connect(self, websocket: WebSocket):
         """Accept new WebSocket connection"""
         await websocket.accept()
         self.active_connections.append(websocket)
-        logger.info(f"📡 New WebSocket connection. Total: {len(self.active_connections)}")
-        
-        # Send welcome message
-        await self.send_personal_message({
-            "type": "connection_established",
-            "message": "Connected to Zyon Traders real-time data stream",
-            "timestamp": datetime.utcnow().isoformat()
-        }, websocket)
+        self.subscriptions[websocket] = set()
+        logger.info(f"New WebSocket connection. Total: {len(self.active_connections)}")
     
     def disconnect(self, websocket: WebSocket):
         """Remove WebSocket connection"""
         if websocket in self.active_connections:
             self.active_connections.remove(websocket)
-            logger.info(f"📡 WebSocket disconnected. Total: {len(self.active_connections)}")
+        if websocket in self.subscriptions:
+            del self.subscriptions[websocket]
+        logger.info(f"WebSocket disconnected. Total: {len(self.active_connections)}")
     
-    async def send_personal_message(self, data: Dict[str, Any], websocket: WebSocket):
-        """Send message to specific WebSocket connection"""
-        try:
-            await websocket.send_text(json.dumps(data))
-        except Exception as e:
-            logger.error(f"❌ Error sending message to WebSocket: {e}")
-            self.disconnect(websocket)
+    async def subscribe_to_symbols(self, websocket: WebSocket, symbols: List[str]):
+        """Subscribe to specific symbols for real-time updates"""
+        if websocket in self.subscriptions:
+            self.subscriptions[websocket].update(symbols)
+            await websocket.send_text(json.dumps({
+                "type": "subscription_success",
+                "symbols": symbols,
+                "timestamp": datetime.utcnow().isoformat()
+            }))
+            logger.info(f"WebSocket subscribed to symbols: {symbols}")
     
-    async def broadcast(self, data: Dict[str, Any]):
-        """Broadcast message to all connected WebSocket clients"""
-        if not self.active_connections:
-            return
-        
-        message = json.dumps(data)
-        disconnected = []
-        
-        for connection in self.active_connections:
-            try:
-                await connection.send_text(message)
-            except Exception as e:
-                logger.error(f"❌ Error broadcasting to WebSocket: {e}")
-                disconnected.append(connection)
-        
-        # Remove disconnected clients
-        for connection in disconnected:
-            self.disconnect(connection)
+    async def unsubscribe_from_symbols(self, websocket: WebSocket, symbols: List[str]):
+        """Unsubscribe from specific symbols"""
+        if websocket in self.subscriptions:
+            self.subscriptions[websocket].difference_update(symbols)
+            await websocket.send_text(json.dumps({
+                "type": "unsubscription_success",
+                "symbols": symbols,
+                "timestamp": datetime.utcnow().isoformat()
+            }))
+            logger.info(f"WebSocket unsubscribed from symbols: {symbols}")
     
-    async def broadcast_market_data(self, market_data: List[Dict[str, Any]]):
-        """Broadcast market data updates"""
-        await self.broadcast({
-            "type": "market_data_update",
-            "data": market_data,
+    async def broadcast_to_symbol_subscribers(self, symbol: str, data: dict):
+        """Broadcast data to all clients subscribed to a specific symbol"""
+        message = json.dumps({
+            "type": "market_data",
+            "symbol": symbol,
+            "data": data,
             "timestamp": datetime.utcnow().isoformat()
         })
+        
+        disconnected = []
+        for websocket in self.active_connections:
+            if symbol in self.subscriptions.get(websocket, set()):
+                try:
+                    await websocket.send_text(message)
+                except Exception as e:
+                    logger.error(f"Error sending data to WebSocket: {e}")
+                    disconnected.append(websocket)
+        
+        # Clean up disconnected clients
+        for websocket in disconnected:
+            self.disconnect(websocket)
     
-    async def broadcast_trading_update(self, trading_data: Dict[str, Any]):
-        """Broadcast trading updates"""
-        await self.broadcast({
-            "type": "trading_update", 
-            "data": trading_data,
+    async def broadcast_to_all(self, data: dict):
+        """Broadcast data to all connected clients"""
+        message = json.dumps({
+            "type": "broadcast",
+            "data": data,
             "timestamp": datetime.utcnow().isoformat()
         })
+        
+        disconnected = []
+        for websocket in self.active_connections:
+            try:
+                await websocket.send_text(message)
+            except Exception as e:
+                logger.error(f"Error broadcasting to WebSocket: {e}")
+                disconnected.append(websocket)
+        
+        # Clean up disconnected clients
+        for websocket in disconnected:
+            self.disconnect(websocket)
     
-    async def broadcast_ai_signal(self, signal_data: Dict[str, Any]):
-        """Broadcast AI trading signals"""
-        await self.broadcast({
-            "type": "ai_signal",
-            "data": signal_data,
-            "timestamp": datetime.utcnow().isoformat()
-        })
+    async def simulate_market_data(self):
+        """Simulate real-time market data updates"""
+        symbols = ["NIFTY", "SENSEX", "BANKNIFTY", "RELIANCE", "TCS", "INFY", "HDFC"]
+        
+        while self.running:
+            try:
+                # Simulate price updates
+                for symbol in symbols:
+                    # Generate realistic price movement
+                    base_price = {
+                        "NIFTY": 22000,
+                        "SENSEX": 72000,
+                        "BANKNIFTY": 47000,
+                        "RELIANCE": 2500,
+                        "TCS": 3800,
+                        "INFY": 1700,
+                        "HDFC": 1600
+                    }.get(symbol, 1000)
+                    
+                    change_percent = random.uniform(-2, 2)
+                    price = base_price * (1 + change_percent / 100)
+                    
+                    data = {
+                        "price": round(price, 2),
+                        "change": round(base_price * change_percent / 100, 2),
+                        "change_percent": round(change_percent, 2),
+                        "volume": random.randint(100000, 10000000),
+                        "last_updated": datetime.utcnow().isoformat()
+                    }
+                    
+                    await self.broadcast_to_symbol_subscribers(symbol, data)
+                
+                # Wait before next update
+                await asyncio.sleep(2)  # Update every 2 seconds
+                
+            except Exception as e:
+                logger.error(f"Error in market data simulation: {e}")
+                await asyncio.sleep(5)
     
-    def get_connection_count(self) -> int:
-        """Get number of active connections"""
-        return len(self.active_connections)
+    async def start_background_tasks(self):
+        """Start background tasks for data simulation"""
+        self.running = True
+        self.background_task = asyncio.create_task(self.simulate_market_data())
+        logger.info("WebSocket background tasks started")
     
-    def get_connection_stats(self) -> Dict[str, Any]:
-        """Get WebSocket connection statistics"""
-        return {
-            "active_connections": len(self.active_connections),
-            "total_connections_served": getattr(self, '_total_connections', 0),
-            "timestamp": datetime.utcnow().isoformat()
-        }
+    async def cleanup(self):
+        """Cleanup resources"""
+        self.running = False
+        if self.background_task:
+            self.background_task.cancel()
+            try:
+                await self.background_task
+            except asyncio.CancelledError:
+                pass
+        
+        # Close all connections
+        for websocket in self.active_connections:
+            try:
+                await websocket.close()
+            except Exception:
+                pass
+        
+        self.active_connections.clear()
+        self.subscriptions.clear()
+        logger.info("WebSocket manager cleaned up")

utils/logging_config.py

@@ -0,0 +1,84 @@
+"""
+Logging configuration for Zyon Traders Backend
+Production-ready logging setup for Render.com deployment
+"""
+
+import logging
+import logging.config
+import os
+from typing import Dict, Any
+
+
+def setup_logging():
+    """Setup logging configuration for the application"""
+    
+    # Determine log level from environment
+    log_level = os.getenv("LOG_LEVEL", "INFO").upper()
+    
+    # Logging configuration
+    config: Dict[str, Any] = {
+        "version": 1,
+        "disable_existing_loggers": False,
+        "formatters": {
+            "detailed": {
+                "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+                "datefmt": "%Y-%m-%d %H:%M:%S"
+            },
+            "simple": {
+                "format": "%(levelname)s - %(message)s"
+            },
+            "json": {
+                "format": "%(asctime)s %(name)s %(levelname)s %(message)s",
+                "datefmt": "%Y-%m-%d %H:%M:%S"
+            }
+        },
+        "handlers": {
+            "console": {
+                "class": "logging.StreamHandler",
+                "level": log_level,
+                "formatter": "detailed",
+                "stream": "ext://sys.stdout"
+            },
+            "file": {
+                "class": "logging.FileHandler",
+                "level": "INFO",
+                "formatter": "detailed",
+                "filename": "app.log",
+                "mode": "a"
+            }
+        },
+        "root": {
+            "level": log_level,
+            "handlers": ["console"]
+        },
+        "loggers": {
+            "uvicorn": {
+                "level": "INFO",
+                "handlers": ["console"],
+                "propagate": False
+            },
+            "uvicorn.error": {
+                "level": "INFO",
+                "handlers": ["console"],
+                "propagate": False
+            },
+            "uvicorn.access": {
+                "level": "INFO",
+                "handlers": ["console"],
+                "propagate": False
+            },
+            "fastapi": {
+                "level": "INFO",
+                "handlers": ["console"],
+                "propagate": False
+            }
+        }
+    }
+    
+    # Apply configuration
+    logging.config.dictConfig(config)
+    
+    # Set third-party loggers to WARNING to reduce noise
+    logging.getLogger("httpx").setLevel(logging.WARNING)
+    logging.getLogger("httpcore").setLevel(logging.WARNING)
+    logging.getLogger("urllib3").setLevel(logging.WARNING)