Upload 2 files

app.py

@@ -10,14 +10,12 @@ from contextlib import asynccontextmanager
 
 from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import JSONResponse, HTMLResponse, FileResponse
-from fastapi.staticfiles import StaticFiles
+from fastapi.responses import JSONResponse
 
 # Import API modules
 from api.endpoints import router as api_router
 from api.websocket import router as websocket_router, manager as ws_manager
 from api.pool_endpoints import router as pool_router
-from api.data_endpoints import router as data_router
 
 # Import new WebSocket service routers
 from api.ws_unified_router import router as ws_unified_router, start_all_websocket_streams
@@ -170,15 +168,6 @@ async def lifespan(app: FastAPI):
         task_scheduler.start()
         logger.info("Task scheduler started successfully")
 
-        # 7. Start WebSocket data broadcaster
-        logger.info("Starting WebSocket data broadcaster...")
-        try:
-            from api.ws_data_broadcaster import broadcaster
-            asyncio.create_task(broadcaster.start_broadcasting())
-            logger.info("WebSocket data broadcaster started")
-        except Exception as e:
-            logger.warning(f"Could not start WebSocket data broadcaster: {e}")
-
         # Log startup summary
         logger.info("=" * 80)
         logger.info("Crypto API Monitoring System started successfully")
@@ -196,26 +185,17 @@ async def lifespan(app: FastAPI):
         logger.info("Shutting down Crypto API Monitoring System...")
         logger.info("=" * 80)
 
-        # 1. Stop WebSocket data broadcaster
-        logger.info("Stopping WebSocket data broadcaster...")
-        try:
-            from api.ws_data_broadcaster import broadcaster
-            await broadcaster.stop_broadcasting()
-            logger.info("WebSocket data broadcaster stopped")
-        except Exception as e:
-            logger.warning(f"Error stopping WebSocket data broadcaster: {e}")
-
-        # 2. Stop task scheduler
+        # 1. Stop task scheduler
         logger.info("Stopping task scheduler...")
         task_scheduler.stop()
         logger.info("Task scheduler stopped")
 
-        # 3. Stop WebSocket background tasks
+        # 2. Stop WebSocket background tasks
         logger.info("Stopping WebSocket background tasks...")
         await ws_manager.stop_background_tasks()
         logger.info("WebSocket background tasks stopped")
 
-        # 4. Close all WebSocket connections
+        # 3. Close all WebSocket connections
         logger.info("Closing WebSocket connections...")
         await ws_manager.close_all_connections()
         logger.info("WebSocket connections closed")
@@ -329,12 +309,6 @@ app.include_router(
     tags=["Pool Management"]
 )
 
-# Include Data endpoints router (cryptocurrency data)
-app.include_router(
-    data_router,
-    tags=["Crypto Data"]
-)
-
 # Include HF router (if available)
 if HF_ROUTER_AVAILABLE:
     try:
@@ -374,26 +348,10 @@ logger.info("All WebSocket service routers included successfully")
 # Root Endpoints
 # ============================================================================
 
-@app.get("/", response_class=HTMLResponse, tags=["Root"])
+@app.get("/", tags=["Root"])
 async def root():
     """
-    Serve the Vidya HTML UI dashboard
-
-    Returns:
-        HTML dashboard interface
-    """
-    try:
-        with open("index.html", "r", encoding="utf-8") as f:
-            return HTMLResponse(content=f.read())
-    except Exception as e:
-        logger.error(f"Error serving index.html: {e}")
-        return HTMLResponse(content=f"<h1>Error loading dashboard</h1><p>{str(e)}</p>", status_code=500)
-
-
-@app.get("/api-info", tags=["Root"])
-async def api_info():
-    """
-    API information and available endpoints
+    Root endpoint with API information and available endpoints
 
     Returns:
         API information and endpoint listing

hf_client.py

@@ -0,0 +1,273 @@
+from __future__ import annotations
+
+from typing import List, Dict, Any, Optional
+import os
+import time
+from functools import lru_cache
+from collections import deque
+from datetime import datetime
+
+ENABLE_SENTIMENT = os.getenv("ENABLE_SENTIMENT", "true").lower() in ("1", "true", "yes")
+SOCIAL_MODEL = os.getenv("SENTIMENT_SOCIAL_MODEL", "ElKulako/cryptobert")
+NEWS_MODEL = os.getenv("SENTIMENT_NEWS_MODEL", "kk08/CryptoBERT")
+
+
+@lru_cache(maxsize=4)
+def _pl(model_name: str):
+    if not ENABLE_SENTIMENT:
+        return None
+    from transformers import pipeline
+    return pipeline("sentiment-analysis", model=model_name)
+
+
+def _label_to_score(lbl: str) -> float:
+    l = (lbl or "").lower()
+    if "positive" in l or "bullish" in l:
+        return 1.0
+    if "negative" in l or "bearish" in l:
+        return -1.0
+    return 0.0
+
+
+def analyze_social_sentiment(text: str) -> Dict[str, Any]:
+    """Analyze social media text sentiment using CryptoBERT."""
+    if not ENABLE_SENTIMENT or not text or not text.strip():
+        return {"sentiment": "neutral", "score": 0.0, "confidence": 0.0}
+    
+    pipe = _pl(SOCIAL_MODEL)
+    if pipe is None:
+        return {"sentiment": "neutral", "score": 0.0, "confidence": 0.0}
+    
+    try:
+        result = pipe(text[:512])[0]
+        label = result.get("label", "NEUTRAL")
+        confidence = result.get("score", 0.0)
+        score = _label_to_score(label)
+        
+        return {
+            "sentiment": label.lower(),
+            "score": score,
+            "confidence": confidence
+        }
+    except Exception as e:
+        return {"sentiment": "neutral", "score": 0.0, "confidence": 0.0, "error": str(e)}
+
+
+def analyze_news_sentiment(text: str) -> Dict[str, Any]:
+    """Analyze news text sentiment using CryptoBERT."""
+    if not ENABLE_SENTIMENT or not text or not text.strip():
+        return {"sentiment": "neutral", "score": 0.0, "confidence": 0.0}
+    
+    pipe = _pl(NEWS_MODEL)
+    if pipe is None:
+        return {"sentiment": "neutral", "score": 0.0, "confidence": 0.0}
+    
+    try:
+        result = pipe(text[:512])[0]
+        label = result.get("label", "NEUTRAL")
+        confidence = result.get("score", 0.0)
+        score = _label_to_score(label)
+        
+        return {
+            "sentiment": label.lower(),
+            "score": score,
+            "confidence": confidence
+        }
+    except Exception as e:
+        return {"sentiment": "neutral", "score": 0.0, "confidence": 0.0, "error": str(e)}
+
+
+def batch_analyze_sentiment(
+    texts: List[str],
+    model_type: str = "social"
+) -> List[Dict[str, Any]]:
+    """Analyze multiple texts in batch."""
+    if not ENABLE_SENTIMENT or not texts:
+        return [{"sentiment": "neutral", "score": 0.0, "confidence": 0.0} for _ in texts]
+    
+    model_name = SOCIAL_MODEL if model_type == "social" else NEWS_MODEL
+    pipe = _pl(model_name)
+    
+    if pipe is None:
+        return [{"sentiment": "neutral", "score": 0.0, "confidence": 0.0} for _ in texts]
+    
+    results = []
+    for text in texts:
+        if not text or not text.strip():
+            results.append({"sentiment": "neutral", "score": 0.0, "confidence": 0.0})
+            continue
+        
+        try:
+            result = pipe(text[:512])[0]
+            label = result.get("label", "NEUTRAL")
+            confidence = result.get("score", 0.0)
+            score = _label_to_score(label)
+            
+            results.append({
+                "sentiment": label.lower(),
+                "score": score,
+                "confidence": confidence
+            })
+        except Exception as e:
+            results.append({
+                "sentiment": "neutral",
+                "score": 0.0,
+                "confidence": 0.0,
+                "error": str(e)
+            })
+    
+    return results
+
+
+class HFClient:
+    """HuggingFace Client for sentiment analysis with usage tracking."""
+    
+    def __init__(self, max_history: int = 100):
+        """Initialize HFClient with usage tracking.
+        
+        Args:
+            max_history: Maximum number of recent results to keep in history
+        """
+        self.max_history = max_history
+        self._history = deque(maxlen=max_history)
+        self._stats = {
+            "total_requests": 0,
+            "successful_requests": 0,
+            "failed_requests": 0,
+            "total_latency_ms": 0.0,
+            "model_usage": {}
+        }
+    
+    def analyze_sentiment(
+        self,
+        text: str,
+        model_type: str = "social",
+        metadata: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
+        """Analyze sentiment with tracking.
+        
+        Args:
+            text: Text to analyze
+            model_type: Type of model to use ("social" or "news")
+            metadata: Optional metadata to attach to result
+            
+        Returns:
+            Sentiment analysis result with metadata
+        """
+        start_time = time.time()
+        self._stats["total_requests"] += 1
+        
+        # Track model usage
+        model_name = SOCIAL_MODEL if model_type == "social" else NEWS_MODEL
+        if model_name not in self._stats["model_usage"]:
+            self._stats["model_usage"][model_name] = 0
+        self._stats["model_usage"][model_name] += 1
+        
+        try:
+            # Perform analysis
+            if model_type == "social":
+                result = analyze_social_sentiment(text)
+            else:
+                result = analyze_news_sentiment(text)
+            
+            # Calculate latency
+            latency_ms = (time.time() - start_time) * 1000
+            self._stats["total_latency_ms"] += latency_ms
+            
+            # Track success
+            if "error" not in result:
+                self._stats["successful_requests"] += 1
+            else:
+                self._stats["failed_requests"] += 1
+            
+            # Add metadata
+            enriched_result = {
+                **result,
+                "timestamp": datetime.utcnow().isoformat(),
+                "model_type": model_type,
+                "model_name": model_name,
+                "latency_ms": round(latency_ms, 2),
+                "metadata": metadata or {}
+            }
+            
+            # Add to history
+            self._history.append(enriched_result)
+            
+            return enriched_result
+            
+        except Exception as e:
+            self._stats["failed_requests"] += 1
+            latency_ms = (time.time() - start_time) * 1000
+            self._stats["total_latency_ms"] += latency_ms
+            
+            error_result = {
+                "sentiment": "neutral",
+                "score": 0.0,
+                "confidence": 0.0,
+                "error": str(e),
+                "timestamp": datetime.utcnow().isoformat(),
+                "model_type": model_type,
+                "model_name": model_name,
+                "latency_ms": round(latency_ms, 2),
+                "metadata": metadata or {}
+            }
+            
+            self._history.append(error_result)
+            return error_result
+    
+    def get_usage_stats(self) -> Dict[str, Any]:
+        """Get usage statistics.
+        
+        Returns:
+            Dictionary containing usage statistics
+        """
+        total_requests = self._stats["total_requests"]
+        avg_latency = (
+            self._stats["total_latency_ms"] / total_requests 
+            if total_requests > 0 
+            else 0.0
+        )
+        success_rate = (
+            self._stats["successful_requests"] / total_requests * 100 
+            if total_requests > 0 
+            else 0.0
+        )
+        
+        return {
+            "total_requests": total_requests,
+            "successful_requests": self._stats["successful_requests"],
+            "failed_requests": self._stats["failed_requests"],
+            "success_rate_percent": round(success_rate, 2),
+            "avg_latency_ms": round(avg_latency, 2),
+            "model_usage": self._stats["model_usage"],
+            "history_size": len(self._history),
+            "max_history": self.max_history
+        }
+    
+    def get_recent_results(self, limit: Optional[int] = None) -> List[Dict[str, Any]]:
+        """Get recent analysis results.
+        
+        Args:
+            limit: Maximum number of results to return (None for all)
+            
+        Returns:
+            List of recent results
+        """
+        if limit is None or limit >= len(self._history):
+            return list(self._history)
+        return list(self._history)[-limit:]
+    
+    def clear_history(self) -> None:
+        """Clear the results history."""
+        self._history.clear()
+    
+    def reset_stats(self) -> None:
+        """Reset all usage statistics."""
+        self._stats = {
+            "total_requests": 0,
+            "successful_requests": 0,
+            "failed_requests": 0,
+            "total_latency_ms": 0.0,
+            "model_usage": {}
+        }
+        self.clear_history()