#!/usr/bin/env python3 """ Market API Router - Implements cryptocurrency market endpoints Handles GET /api/market/price, GET /api/market/ohlc, POST /api/sentiment/analyze, and WebSocket /ws """ from fastapi import APIRouter, HTTPException, Query, WebSocket, WebSocketDisconnect from fastapi.responses import JSONResponse from typing import Optional, Dict, Any, List from pydantic import BaseModel, Field from datetime import datetime import logging import json import asyncio import time # Import services from backend.services.coingecko_client import coingecko_client from backend.services.binance_client import BinanceClient from backend.services.ai_service_unified import UnifiedAIService from backend.services.market_data_aggregator import market_data_aggregator from backend.services.sentiment_aggregator import sentiment_aggregator from backend.services.hf_dataset_aggregator import hf_dataset_aggregator logger = logging.getLogger(__name__) router = APIRouter(tags=["Market API"]) # WebSocket connection manager class WebSocketManager: """Manages WebSocket connections and subscriptions""" def __init__(self): self.active_connections: Dict[str, WebSocket] = {} self.subscriptions: Dict[str, List[str]] = {} # client_id -> [symbols] self.price_streams: Dict[str, asyncio.Task] = {} async def connect(self, websocket: WebSocket, client_id: str): """Accept WebSocket connection""" await websocket.accept() self.active_connections[client_id] = websocket self.subscriptions[client_id] = [] logger.info(f"WebSocket client {client_id} connected") async def disconnect(self, client_id: str): """Disconnect WebSocket client""" if client_id in self.active_connections: del self.active_connections[client_id] if client_id in self.subscriptions: del self.subscriptions[client_id] if client_id in self.price_streams: self.price_streams[client_id].cancel() del self.price_streams[client_id] logger.info(f"WebSocket client {client_id} disconnected") async def subscribe(self, client_id: str, symbol: str): """Subscribe client to symbol updates""" if client_id not in self.subscriptions: self.subscriptions[client_id] = [] if symbol.upper() not in self.subscriptions[client_id]: self.subscriptions[client_id].append(symbol.upper()) logger.info(f"Client {client_id} subscribed to {symbol.upper()}") async def send_message(self, client_id: str, message: Dict[str, Any]): """Send message to specific client""" if client_id in self.active_connections: try: await self.active_connections[client_id].send_json(message) except Exception as e: logger.error(f"Error sending message to {client_id}: {e}") await self.disconnect(client_id) async def broadcast_to_subscribers(self, symbol: str, data: Dict[str, Any]): """Broadcast data to all clients subscribed to symbol""" symbol_upper = symbol.upper() for client_id, symbols in self.subscriptions.items(): if symbol_upper in symbols: await self.send_message(client_id, data) # Global WebSocket manager instance ws_manager = WebSocketManager() # Binance client instance binance_client = BinanceClient() # AI service instance ai_service = UnifiedAIService() # ============================================================================ # GET /api/market/price # ============================================================================ @router.get("/api/market/price") async def get_market_price( symbol: str = Query(..., description="Cryptocurrency symbol (e.g., BTC, ETH)") ): """ Fetch the current market price of a specific cryptocurrency. Uses ALL free market data providers with intelligent fallback: CoinGecko, CoinPaprika, CoinCap, Binance, CoinLore, Messari, CoinStats Returns: - If symbol is valid: current price with timestamp - If symbol is invalid: 404 error """ try: symbol_upper = symbol.upper() # Use market data aggregator with automatic fallback to ALL free providers price_data = await market_data_aggregator.get_price(symbol_upper) return { "symbol": price_data.get("symbol", symbol_upper), "price": price_data.get("price", 0), "source": price_data.get("source", "unknown"), "timestamp": price_data.get("timestamp", int(time.time() * 1000)) // 1000 } except HTTPException: raise except Exception as e: logger.error(f"Error fetching price for {symbol}: {e}") raise HTTPException( status_code=502, detail=f"Error fetching price data: {str(e)}" ) # ============================================================================ # GET /api/market/ohlc # ============================================================================ @router.get("/api/market/ohlc") async def get_market_ohlc( symbol: str = Query(..., description="Cryptocurrency symbol (e.g., BTC, ETH)"), timeframe: str = Query("1h", description="Timeframe (1h, 4h, 1d)") ): """ Fetch historical OHLC (Open, High, Low, Close) data for a cryptocurrency. Uses multiple sources with fallback: 1. Binance Public API (real-time) 2. HuggingFace Datasets (linxy/CryptoCoin - 26 symbols) 3. HuggingFace Datasets (WinkingFace/CryptoLM - BTC, ETH, SOL, XRP) Returns: - If symbol and timeframe are valid: OHLC data array - If invalid: 404 error """ try: symbol_upper = symbol.upper() # Validate timeframe valid_timeframes = ["1m", "5m", "15m", "30m", "1h", "4h", "1d", "1w"] if timeframe not in valid_timeframes: raise HTTPException( status_code=400, detail=f"Invalid timeframe '{timeframe}'. Valid timeframes: {', '.join(valid_timeframes)}" ) # Try Binance first (real-time data) try: ohlcv_data = await binance_client.get_ohlcv(symbol_upper, timeframe, limit=100) if ohlcv_data and len(ohlcv_data) > 0: # Format response ohlc_list = [] for item in ohlcv_data: ohlc_list.append({ "open": item.get("open", 0), "high": item.get("high", 0), "low": item.get("low", 0), "close": item.get("close", 0), "timestamp": item.get("timestamp", int(time.time())) }) logger.info(f"✅ Binance: Fetched OHLC for {symbol_upper}/{timeframe}") return { "symbol": symbol_upper, "timeframe": timeframe, "ohlc": ohlc_list, "source": "binance" } except Exception as e: logger.warning(f"⚠️ Binance failed for {symbol_upper}/{timeframe}: {e}") # Fallback to HuggingFace Datasets (historical data) try: hf_ohlcv_data = await hf_dataset_aggregator.get_ohlcv(symbol_upper, timeframe, limit=100) if hf_ohlcv_data and len(hf_ohlcv_data) > 0: # Format response ohlc_list = [] for item in hf_ohlcv_data: ohlc_list.append({ "open": item.get("open", 0), "high": item.get("high", 0), "low": item.get("low", 0), "close": item.get("close", 0), "timestamp": item.get("timestamp", int(time.time())) }) logger.info(f"✅ HuggingFace Datasets: Fetched OHLC for {symbol_upper}/{timeframe}") return { "symbol": symbol_upper, "timeframe": timeframe, "ohlc": ohlc_list, "source": "huggingface" } except Exception as e: logger.warning(f"⚠️ HuggingFace Datasets failed for {symbol_upper}/{timeframe}: {e}") # No data found from any source raise HTTPException( status_code=404, detail=f"No OHLC data found for symbol '{symbol}' with timeframe '{timeframe}' from any source (Binance, HuggingFace)" ) except HTTPException: raise except Exception as e: logger.error(f"Error fetching OHLC data: {e}") raise HTTPException( status_code=502, detail=f"Error fetching OHLC data: {str(e)}" ) # ============================================================================ # POST /api/sentiment/analyze # ============================================================================ class SentimentAnalyzeRequest(BaseModel): """Request model for sentiment analysis""" text: str = Field(..., description="Text to analyze for sentiment", min_length=1) @router.post("/api/sentiment/analyze") async def analyze_sentiment(request: SentimentAnalyzeRequest): """ Analyze the sentiment of a given text (Bullish, Bearish, Neutral). Returns: - If text is valid: sentiment analysis result - If text is missing or invalid: 400 error """ try: if not request.text or len(request.text.strip()) == 0: raise HTTPException( status_code=400, detail="Text parameter is required and cannot be empty" ) # Use AI service for sentiment analysis try: result = await ai_service.analyze_sentiment( text=request.text, category="crypto", use_ensemble=True ) # Map sentiment to required format label = result.get("label", "neutral").lower() confidence = result.get("confidence", 0.5) # Map label to sentiment if "bullish" in label or "positive" in label: sentiment = "Bullish" score = confidence if confidence > 0.5 else 0.6 elif "bearish" in label or "negative" in label: sentiment = "Bearish" score = 1 - confidence if confidence < 0.5 else 0.4 else: sentiment = "Neutral" score = 0.5 return { "sentiment": sentiment, "score": score, "confidence": confidence } except Exception as e: logger.error(f"Error analyzing sentiment: {e}") # Fallback to simple keyword-based analysis text_lower = request.text.lower() positive_words = ['bullish', 'buy', 'moon', 'pump', 'up', 'gain', 'profit', 'good', 'great', 'strong'] negative_words = ['bearish', 'sell', 'dump', 'down', 'loss', 'crash', 'bad', 'fear', 'weak', 'drop'] pos_count = sum(1 for word in positive_words if word in text_lower) neg_count = sum(1 for word in negative_words if word in text_lower) if pos_count > neg_count: sentiment = "Bullish" elif neg_count > pos_count: sentiment = "Bearish" else: sentiment = "Neutral" return { "sentiment": sentiment, "score": 0.65 if sentiment == "Bullish" else (0.35 if sentiment == "Bearish" else 0.5), "confidence": 0.6 } except HTTPException: raise except Exception as e: logger.error(f"Error in sentiment analysis: {e}") raise HTTPException( status_code=502, detail=f"Error analyzing sentiment: {str(e)}" ) # ============================================================================ # WebSocket /ws # ============================================================================ async def stream_price_updates(client_id: str, symbol: str): """Stream price updates for a subscribed symbol""" symbol_upper = symbol.upper() while client_id in ws_manager.active_connections: try: # Get current price try: market_data = await coingecko_client.get_market_prices(symbols=[symbol_upper], limit=1) if market_data and len(market_data) > 0: coin = market_data[0] price = coin.get("price", 0) else: # Fallback to Binance ticker = await binance_client.get_ticker(f"{symbol_upper}USDT") price = float(ticker.get("lastPrice", 0)) if ticker else 0 except Exception as e: logger.warning(f"Error fetching price for {symbol_upper}: {e}") price = 0 # Send update to client await ws_manager.send_message(client_id, { "symbol": symbol_upper, "price": price, "timestamp": int(time.time()) }) # Wait 5 seconds before next update await asyncio.sleep(5) except asyncio.CancelledError: break except Exception as e: logger.error(f"Error in price stream for {symbol_upper}: {e}") await asyncio.sleep(5) @router.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): """ WebSocket endpoint for real-time cryptocurrency data updates. Connection: - Clients connect to receive real-time data - Send subscription messages to subscribe to specific symbols Subscription Message: { "type": "subscribe", "symbol": "BTC" } Unsubscribe Message: { "type": "unsubscribe", "symbol": "BTC" } Ping Message: { "type": "ping" } """ client_id = f"client_{int(time.time() * 1000)}_{id(websocket)}" try: await ws_manager.connect(websocket, client_id) # Send welcome message await websocket.send_json({ "type": "connected", "client_id": client_id, "message": "Connected to cryptocurrency data WebSocket", "timestamp": int(time.time()) }) # Handle incoming messages while True: try: # Receive message with timeout data = await asyncio.wait_for(websocket.receive_text(), timeout=30.0) try: message = json.loads(data) msg_type = message.get("type", "").lower() if msg_type == "subscribe": symbol = message.get("symbol", "").upper() if not symbol: await websocket.send_json({ "type": "error", "error": "Symbol is required for subscription", "timestamp": int(time.time()) }) continue await ws_manager.subscribe(client_id, symbol) # Start price streaming task if not already running task_key = f"{client_id}_{symbol}" if task_key not in ws_manager.price_streams: task = asyncio.create_task(stream_price_updates(client_id, symbol)) ws_manager.price_streams[task_key] = task await websocket.send_json({ "type": "subscribed", "symbol": symbol, "message": f"Subscribed to {symbol} updates", "timestamp": int(time.time()) }) elif msg_type == "unsubscribe": symbol = message.get("symbol", "").upper() if symbol in ws_manager.subscriptions.get(client_id, []): ws_manager.subscriptions[client_id].remove(symbol) task_key = f"{client_id}_{symbol}" if task_key in ws_manager.price_streams: ws_manager.price_streams[task_key].cancel() del ws_manager.price_streams[task_key] await websocket.send_json({ "type": "unsubscribed", "symbol": symbol, "message": f"Unsubscribed from {symbol} updates", "timestamp": int(time.time()) }) elif msg_type == "ping": await websocket.send_json({ "type": "pong", "timestamp": int(time.time()) }) else: await websocket.send_json({ "type": "error", "error": f"Unknown message type: {msg_type}", "timestamp": int(time.time()) }) except json.JSONDecodeError: await websocket.send_json({ "type": "error", "error": "Invalid JSON format", "timestamp": int(time.time()) }) except asyncio.TimeoutError: # Send heartbeat await websocket.send_json({ "type": "heartbeat", "timestamp": int(time.time()), "status": "alive" }) except WebSocketDisconnect: logger.info(f"WebSocket client {client_id} disconnected normally") await ws_manager.disconnect(client_id) except Exception as e: logger.error(f"WebSocket error for {client_id}: {e}", exc_info=True) try: await websocket.send_json({ "type": "error", "error": f"Server error: {str(e)}", "timestamp": int(time.time()) }) except: pass await ws_manager.disconnect(client_id) finally: await ws_manager.disconnect(client_id)