diff --git "a/final/hf_unified_server.py" "b/final/hf_unified_server.py" --- "a/final/hf_unified_server.py" +++ "b/final/hf_unified_server.py" @@ -1,2575 +1,2575 @@ -"""Unified HuggingFace Space API Server leveraging shared collectors and AI helpers.""" - -import asyncio -import time -import os -import sys -import io - -# Fix encoding for Windows console (must be done before any print/logging) -if sys.platform == "win32": - try: - sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') - sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace') - except Exception: - pass # If already wrapped, ignore - -# Set environment variables to force PyTorch and avoid TensorFlow/Keras issues -os.environ.setdefault('TRANSFORMERS_NO_ADVISORY_WARNINGS', '1') -os.environ.setdefault('TRANSFORMERS_VERBOSITY', 'error') -os.environ.setdefault('TF_CPP_MIN_LOG_LEVEL', '3') # Suppress TensorFlow warnings -# Force PyTorch as default framework -os.environ.setdefault('TRANSFORMERS_FRAMEWORK', 'pt') - -from datetime import datetime, timedelta -from fastapi import Body, FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect -from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import FileResponse, JSONResponse, HTMLResponse -from fastapi.staticfiles import StaticFiles -from starlette.websockets import WebSocketState -from typing import Any, Dict, List, Optional, Union -from statistics import mean -import logging -import random -import json -from pathlib import Path -import httpx - - -from ai_models import ( - analyze_chart_points, - analyze_crypto_sentiment, - analyze_market_text, - get_model_info, - initialize_models, - registry_status, -) -from backend.services.local_resource_service import LocalResourceService -from collectors.aggregator import ( - CollectorError, - MarketDataCollector, - NewsCollector, - ProviderStatusCollector, -) -from config import COIN_SYMBOL_MAPPING, get_settings - -# Setup logging -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) - -# Create FastAPI app -app = FastAPI( - title="Cryptocurrency Data & Analysis API", - description="Complete API for cryptocurrency data, market analysis, and trading signals", - version="3.0.0" -) - -# CORS -app.add_middleware( - CORSMiddleware, - allow_origins=["*"], - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], -) - -# Runtime state -START_TIME = time.time() -cache = {"ohlcv": {}, "prices": {}, "market_data": {}, "providers": [], "last_update": None} -settings = get_settings() -market_collector = MarketDataCollector() -news_collector = NewsCollector() -provider_collector = ProviderStatusCollector() - -# Load providers config -WORKSPACE_ROOT = Path(__file__).parent -PROVIDERS_CONFIG_PATH = settings.providers_config_path -FALLBACK_RESOURCE_PATH = WORKSPACE_ROOT / "crypto_resources_unified_2025-11-11.json" -LOG_DIR = WORKSPACE_ROOT / "logs" -APL_REPORT_PATH = WORKSPACE_ROOT / "PROVIDER_AUTO_DISCOVERY_REPORT.json" - -# Ensure log directory exists -LOG_DIR.mkdir(parents=True, exist_ok=True) - -# Database path (managed by DatabaseManager in the admin API) -DB_PATH = WORKSPACE_ROOT / "data" / "api_monitor.db" - -def tail_log_file(path: Path, max_lines: int = 200) -> List[str]: - """Return the last max_lines from a log file, if it exists.""" - if not path.exists(): - return [] - try: - with path.open("r", encoding="utf-8", errors="ignore") as f: - lines = f.readlines() - return lines[-max_lines:] - except Exception as e: - logger.error(f"Error reading log file {path}: {e}") - return [] - - -def load_providers_config(): - """Load providers from providers_config_extended.json""" - try: - if PROVIDERS_CONFIG_PATH.exists(): - with open(PROVIDERS_CONFIG_PATH, 'r', encoding='utf-8') as f: - config = json.load(f) - providers = config.get('providers', {}) - logger.info(f"Loaded {len(providers)} providers from providers_config_extended.json") - return providers - else: - logger.warning(f"providers_config_extended.json not found at {PROVIDERS_CONFIG_PATH}") - return {} - except Exception as e: - logger.error(f"Error loading providers config: {e}") - return {} - -# Load providers at startup -PROVIDERS_CONFIG = load_providers_config() -local_resource_service = LocalResourceService(FALLBACK_RESOURCE_PATH) - -HF_SAMPLE_NEWS = [ - { - "title": "Bitcoin holds key liquidity zone", - "source": "Fallback Ledger", - "sentiment": "positive", - "sentiment_score": 0.64, - "entities": ["BTC"], - "summary": "BTC consolidates near resistance with steady inflows", - }, - { - "title": "Ethereum staking demand remains resilient", - "source": "Fallback Ledger", - "sentiment": "neutral", - "sentiment_score": 0.12, - "entities": ["ETH"], - "summary": "Validator queue shortens as fees stabilize around L2 adoption", - }, - { - "title": "Solana ecosystem sees TVL uptick", - "source": "Fallback Ledger", - "sentiment": "positive", - "sentiment_score": 0.41, - "entities": ["SOL"], - "summary": "DeFi protocols move to Solana as mempool congestion drops", - }, -] - -# Mount static files (CSS, JS) -try: - static_path = WORKSPACE_ROOT / "static" - if static_path.exists(): - app.mount("/static", StaticFiles(directory=str(static_path)), name="static") - logger.info(f"Static files mounted from {static_path}") - else: - logger.warning(f"Static directory not found: {static_path}") -except Exception as e: - logger.error(f"Error mounting static files: {e}") - -# Mount api-resources for frontend access -try: - api_resources_path = WORKSPACE_ROOT / "api-resources" - if api_resources_path.exists(): - app.mount("/api-resources", StaticFiles(directory=str(api_resources_path)), name="api-resources") - logger.info(f"API resources mounted from {api_resources_path}") - else: - logger.warning(f"API resources directory not found: {api_resources_path}") -except Exception as e: - logger.error(f"Error mounting API resources: {e}") - -# ============================================================================ -# Helper utilities & Data Fetching Functions -# ============================================================================ - -def _normalize_asset_symbol(symbol: str) -> str: - symbol = (symbol or "").upper() - suffixes = ("USDT", "USD", "BTC", "ETH", "BNB") - for suffix in suffixes: - if symbol.endswith(suffix) and len(symbol) > len(suffix): - return symbol[: -len(suffix)] - return symbol - - -def _format_price_record(record: Dict[str, Any]) -> Dict[str, Any]: - price = record.get("price") or record.get("current_price") - change_pct = record.get("change_24h") or record.get("price_change_percentage_24h") - change_abs = None - if price is not None and change_pct is not None: - try: - change_abs = float(price) * float(change_pct) / 100.0 - except (TypeError, ValueError): - change_abs = None - - return { - "id": record.get("id") or record.get("symbol", "").lower(), - "symbol": record.get("symbol", "").upper(), - "name": record.get("name"), - "current_price": price, - "market_cap": record.get("market_cap"), - "market_cap_rank": record.get("rank"), - "total_volume": record.get("volume_24h") or record.get("total_volume"), - "price_change_24h": change_abs, - "price_change_percentage_24h": change_pct, - "high_24h": record.get("high_24h"), - "low_24h": record.get("low_24h"), - "last_updated": record.get("last_updated"), - } - - -async def fetch_binance_ohlcv(symbol: str = "BTCUSDT", interval: str = "1h", limit: int = 100): - """Fetch OHLCV data from Binance via the shared collector.""" - - try: - candles = await market_collector.get_ohlcv(symbol, interval, limit) - return [ - { - **candle, - "timestamp": int(datetime.fromisoformat(candle["timestamp"]).timestamp() * 1000), - "datetime": candle["timestamp"], - } - for candle in candles - ] - except CollectorError as exc: - logger.error("Error fetching OHLCV: %s", exc) - fallback_symbol = _normalize_asset_symbol(symbol) - fallback = local_resource_service.get_ohlcv(fallback_symbol, interval, limit) - if fallback: - return fallback - return [] - - -async def fetch_coingecko_prices(symbols: Optional[List[str]] = None, limit: int = 10): - """Fetch price snapshots using the shared market collector.""" - - source = "coingecko" - try: - if symbols: - tasks = [market_collector.get_coin_details(_normalize_asset_symbol(sym)) for sym in symbols] - results = await asyncio.gather(*tasks, return_exceptions=True) - coins: List[Dict[str, Any]] = [] - for result in results: - if isinstance(result, Exception): - continue - coins.append(_format_price_record(result)) - if coins: - return coins, source - else: - top = await market_collector.get_top_coins(limit=limit) - formatted = [_format_price_record(entry) for entry in top] - if formatted: - return formatted, source - except CollectorError as exc: - logger.error("Error fetching aggregated prices: %s", exc) - - fallback = ( - local_resource_service.get_prices_for_symbols([sym for sym in symbols or []]) - if symbols - else local_resource_service.get_top_prices(limit) - ) - if fallback: - return fallback, "local-fallback" - return [], source - - -async def fetch_binance_ticker(symbol: str): - """Provide ticker-like information sourced from CoinGecko market data.""" - - try: - coin = await market_collector.get_coin_details(_normalize_asset_symbol(symbol)) - except CollectorError as exc: - logger.error("Unable to load ticker for %s: %s", symbol, exc) - coin = None - - if coin: - price = coin.get("price") - change_pct = coin.get("change_24h") or 0.0 - change_abs = price * change_pct / 100 if price is not None and change_pct is not None else None - return { - "symbol": symbol.upper(), - "price": price, - "price_change_24h": change_abs, - "price_change_percent_24h": change_pct, - "high_24h": coin.get("high_24h"), - "low_24h": coin.get("low_24h"), - "volume_24h": coin.get("volume_24h"), - "quote_volume_24h": coin.get("volume_24h"), - }, "binance" - - fallback_symbol = _normalize_asset_symbol(symbol) - fallback = local_resource_service.get_ticker_snapshot(fallback_symbol) - if fallback: - fallback["symbol"] = symbol.upper() - return fallback, "local-fallback" - return None, "binance" - - -# ============================================================================ -# Core Endpoints -# ============================================================================ - -@app.get("/health") -async def health(): - """System health check using shared collectors.""" - - async def _safe_call(coro): - try: - data = await coro - return {"status": "ok", "count": len(data) if hasattr(data, "__len__") else 1} - except Exception as exc: # pragma: no cover - network heavy - return {"status": "error", "detail": str(exc)} - - market_task = asyncio.create_task(_safe_call(market_collector.get_top_coins(limit=3))) - news_task = asyncio.create_task(_safe_call(news_collector.get_latest_news(limit=3))) - providers_task = asyncio.create_task(_safe_call(provider_collector.get_providers_status())) - - market_status, news_status, providers_status = await asyncio.gather( - market_task, news_task, providers_task - ) - - ai_status = registry_status() - service_states = { - "market_data": market_status, - "news": news_status, - "providers": providers_status, - "ai_models": ai_status, - } - - degraded = any(state.get("status") != "ok" for state in (market_status, news_status, providers_status)) - overall = "healthy" if not degraded else "degraded" - - return { - "status": overall, - "service": "cryptocurrency-data-api", - "timestamp": datetime.utcnow().isoformat(), - "version": app.version, - "providers_loaded": market_status.get("count", 0), - "services": service_states, - } - - -@app.get("/info") -async def info(): - """System information""" - hf_providers = [p for p in PROVIDERS_CONFIG.keys() if "huggingface_space" in p] - - return { - "service": "Cryptocurrency Data & Analysis API", - "version": app.version, - "endpoints": { - "core": ["/health", "/info", "/api/providers"], - "data": ["/api/ohlcv", "/api/crypto/prices/top", "/api/crypto/price/{symbol}", "/api/crypto/market-overview"], - "analysis": ["/api/analysis/signals", "/api/analysis/smc", "/api/scoring/snapshot"], - "market": ["/api/market/prices", "/api/market-data/prices"], - "system": ["/api/system/status", "/api/system/config"], - "huggingface": ["/api/hf/health", "/api/hf/refresh", "/api/hf/registry", "/api/hf/run-sentiment"], - }, - "data_sources": ["Binance", "CoinGecko", "CoinPaprika", "CoinCap"], - "providers_loaded": len(PROVIDERS_CONFIG), - "huggingface_space_providers": len(hf_providers), - "features": [ - "Real-time price data", - "OHLCV historical data", - "Trading signals", - "Market analysis", - "Sentiment analysis", - "HuggingFace model integration", - f"{len(PROVIDERS_CONFIG)} providers from providers_config_extended.json", - ], - "ai_registry": registry_status(), - } - - -@app.get("/api/providers") -async def get_providers(): - """Get list of API providers and their health.""" - - try: - statuses = await provider_collector.get_providers_status() - except Exception as exc: # pragma: no cover - network heavy - logger.error("Error getting providers: %s", exc) - raise HTTPException(status_code=503, detail=str(exc)) - - providers_list = [] - for status in statuses: - meta = PROVIDERS_CONFIG.get(status["provider_id"], {}) - providers_list.append( - { - **status, - "base_url": meta.get("base_url"), - "requires_auth": meta.get("requires_auth"), - "priority": meta.get("priority"), - } - ) - - return { - "providers": providers_list, - "total": len(providers_list), - "source": str(PROVIDERS_CONFIG_PATH), - "last_updated": datetime.utcnow().isoformat(), - } - - -@app.get("/api/providers/{provider_id}/health") -async def get_provider_health(provider_id: str): - """Get health status for a specific provider.""" - - # Check if provider exists in config - provider_config = PROVIDERS_CONFIG.get(provider_id) - if not provider_config: - raise HTTPException(status_code=404, detail=f"Provider '{provider_id}' not found") - - try: - # Perform health check using the collector - async with httpx.AsyncClient(timeout=provider_collector.timeout, headers=provider_collector.headers) as client: - health_result = await provider_collector._check_provider(client, provider_id, provider_config) - - # Add metadata from config - health_result.update({ - "base_url": provider_config.get("base_url"), - "requires_auth": provider_config.get("requires_auth"), - "priority": provider_config.get("priority"), - "category": provider_config.get("category"), - "last_checked": datetime.utcnow().isoformat() - }) - - return health_result - except Exception as exc: # pragma: no cover - network heavy - logger.error("Error checking provider health for %s: %s", provider_id, exc) - raise HTTPException(status_code=503, detail=f"Health check failed: {str(exc)}") - - -@app.get("/api/providers/config") -async def get_providers_config(): - """Get providers configuration in format expected by frontend.""" - try: - return { - "success": True, - "providers": PROVIDERS_CONFIG, - "total": len(PROVIDERS_CONFIG), - "source": str(PROVIDERS_CONFIG_PATH), - "last_updated": datetime.utcnow().isoformat() - } - except Exception as exc: - logger.error("Error getting providers config: %s", exc) - raise HTTPException(status_code=500, detail=str(exc)) - - -# ============================================================================ -# OHLCV Data Endpoint -# ============================================================================ - -@app.get("/api/ohlcv") -async def get_ohlcv( - symbol: str = Query("BTCUSDT", description="Trading pair symbol"), - interval: str = Query("1h", description="Time interval (1m, 5m, 15m, 1h, 4h, 1d)"), - limit: int = Query(100, ge=1, le=1000, description="Number of candles") -): - """ - Get OHLCV (candlestick) data for a trading pair - - Supported intervals: 1m, 5m, 15m, 30m, 1h, 4h, 1d - """ - try: - # Check cache - cache_key = f"{symbol}_{interval}_{limit}" - if cache_key in cache["ohlcv"]: - cached_data, cached_time = cache["ohlcv"][cache_key] - if (datetime.now() - cached_time).seconds < 60: # 60s cache - return {"symbol": symbol, "interval": interval, "data": cached_data, "source": "cache"} - - # Fetch from Binance - ohlcv_data = await fetch_binance_ohlcv(symbol, interval, limit) - - if ohlcv_data: - # Update cache - cache["ohlcv"][cache_key] = (ohlcv_data, datetime.now()) - - return { - "symbol": symbol, - "interval": interval, - "count": len(ohlcv_data), - "data": ohlcv_data, - "source": "binance", - "timestamp": datetime.now().isoformat() - } - else: - raise HTTPException(status_code=503, detail="Unable to fetch OHLCV data") - - except HTTPException: - raise - except Exception as e: - logger.error(f"Error in get_ohlcv: {e}") - raise HTTPException(status_code=500, detail=str(e)) - - -# ============================================================================ -# Crypto Prices Endpoints -# ============================================================================ - -@app.get("/api/crypto/prices/top") -async def get_top_prices(limit: int = Query(10, ge=1, le=100, description="Number of top cryptocurrencies")): - """Get top cryptocurrencies by market cap""" - try: - # Check cache - cache_key = f"top_{limit}" - if cache_key in cache["prices"]: - cached_data, cached_time = cache["prices"][cache_key] - if (datetime.now() - cached_time).seconds < 60: - return {"data": cached_data, "source": "cache"} - - # Fetch from CoinGecko - prices, source = await fetch_coingecko_prices(limit=limit) - - if prices: - # Update cache - cache["prices"][cache_key] = (prices, datetime.now()) - - return { - "count": len(prices), - "data": prices, - "source": source, - "timestamp": datetime.now().isoformat() - } - else: - raise HTTPException(status_code=503, detail="Unable to fetch price data") - - except HTTPException: - raise - except Exception as e: - logger.error(f"Error in get_top_prices: {e}") - raise HTTPException(status_code=500, detail=str(e)) - - -@app.get("/api/crypto/price/{symbol}") -async def get_single_price(symbol: str): - """Get price for a single cryptocurrency""" - try: - # Try Binance first for common pairs - binance_symbol = f"{symbol.upper()}USDT" - ticker, ticker_source = await fetch_binance_ticker(binance_symbol) - - if ticker: - return { - "symbol": symbol.upper(), - "price": ticker, - "source": ticker_source, - "timestamp": datetime.now().isoformat() - } - - # Fallback to CoinGecko - prices, source = await fetch_coingecko_prices([symbol]) - if prices: - return { - "symbol": symbol.upper(), - "price": prices[0], - "source": source, - "timestamp": datetime.now().isoformat() - } - - raise HTTPException(status_code=404, detail=f"Price data not found for {symbol}") - - except HTTPException: - raise - except Exception as e: - logger.error(f"Error in get_single_price: {e}") - raise HTTPException(status_code=500, detail=str(e)) - - -@app.get("/api/crypto/market-overview") -async def get_market_overview(): - """Get comprehensive market overview""" - try: - # Fetch top 20 coins - prices, source = await fetch_coingecko_prices(limit=20) - - if not prices: - raise HTTPException(status_code=503, detail="Unable to fetch market data") - - # Calculate market stats - # Try multiple field names for market cap and volume - total_market_cap = 0 - total_volume = 0 - - for p in prices: - # Try different field names for market cap - market_cap = ( - p.get("market_cap") or - p.get("market_cap_usd") or - p.get("market_cap_rank") or # Sometimes this is the value - None - ) - # If market_cap is not found, try calculating from price and supply - if not market_cap: - price = p.get("price") or p.get("current_price") or 0 - supply = p.get("circulating_supply") or p.get("total_supply") or 0 - if price and supply: - market_cap = float(price) * float(supply) - - if market_cap: - try: - total_market_cap += float(market_cap) - except (TypeError, ValueError): - pass - - # Try different field names for volume - volume = ( - p.get("total_volume") or - p.get("volume_24h") or - p.get("volume_24h_usd") or - None - ) - if volume: - try: - total_volume += float(volume) - except (TypeError, ValueError): - pass - - logger.info(f"Market overview: {len(prices)} coins, total_market_cap={total_market_cap:,.0f}, total_volume={total_volume:,.0f}") - - # Sort by 24h change - gainers = sorted( - [p for p in prices if p.get("price_change_percentage_24h")], - key=lambda x: x.get("price_change_percentage_24h", 0), - reverse=True - )[:5] - - losers = sorted( - [p for p in prices if p.get("price_change_percentage_24h")], - key=lambda x: x.get("price_change_percentage_24h", 0) - )[:5] - - return { - "total_market_cap": total_market_cap, - "total_volume_24h": total_volume, - "btc_dominance": (prices[0].get("market_cap", 0) / total_market_cap * 100) if total_market_cap > 0 else 0, - "top_gainers": gainers, - "top_losers": losers, - "top_by_volume": sorted(prices, key=lambda x: x.get("total_volume", 0) or 0, reverse=True)[:5], - "timestamp": datetime.now().isoformat(), - "source": source - } - - except HTTPException: - raise - except Exception as e: - logger.error(f"Error in get_market_overview: {e}") - raise HTTPException(status_code=500, detail=str(e)) - - -@app.get("/api/market") -async def get_market(): - """Get market data in format expected by frontend dashboard""" - try: - overview = await get_market_overview() - prices, source = await fetch_coingecko_prices(limit=50) - - if not prices: - raise HTTPException(status_code=503, detail="Unable to fetch market data") - - return { - "total_market_cap": overview.get("total_market_cap", 0), - "btc_dominance": overview.get("btc_dominance", 0), - "total_volume_24h": overview.get("total_volume_24h", 0), - "cryptocurrencies": prices, - "timestamp": datetime.now().isoformat(), - "source": source - } - except HTTPException: - raise - except Exception as e: - logger.error(f"Error in get_market: {e}") - raise HTTPException(status_code=500, detail=str(e)) - - -@app.get("/api/trending") -async def get_trending(): - """Get trending cryptocurrencies (top gainers by 24h change)""" - try: - prices, source = await fetch_coingecko_prices(limit=100) - - if not prices: - raise HTTPException(status_code=503, detail="Unable to fetch trending data") - - trending = sorted( - [p for p in prices if p.get("price_change_percentage_24h") is not None], - key=lambda x: x.get("price_change_percentage_24h", 0), - reverse=True - )[:10] - - return { - "trending": trending, - "count": len(trending), - "timestamp": datetime.now().isoformat(), - "source": source - } - except HTTPException: - raise - except Exception as e: - logger.error(f"Error in get_trending: {e}") - raise HTTPException(status_code=500, detail=str(e)) - - -@app.get("/api/market/prices") -async def get_multiple_prices(symbols: str = Query("BTC,ETH,SOL", description="Comma-separated symbols")): - """Get prices for multiple cryptocurrencies""" - try: - symbol_list = [s.strip().upper() for s in symbols.split(",")] - - # Fetch prices - prices_data = [] - source = "binance" - for symbol in symbol_list: - try: - ticker, ticker_source = await fetch_binance_ticker(f"{symbol}USDT") - if ticker: - prices_data.append(ticker) - if ticker_source != "binance": - source = ticker_source - except: - continue - if not prices_data: - # Fallback to CoinGecko - prices_data, source = await fetch_coingecko_prices(symbol_list) - - if not prices_data: - fallback_prices = local_resource_service.get_prices_for_symbols(symbol_list) - if fallback_prices: - prices_data = fallback_prices - source = "local-fallback" - - return { - "symbols": symbol_list, - "count": len(prices_data), - "data": prices_data, - "source": source, - "timestamp": datetime.now().isoformat() - } - - except Exception as e: - logger.error(f"Error in get_multiple_prices: {e}") - raise HTTPException(status_code=500, detail=str(e)) - - -@app.get("/api/market-data/prices") -async def get_market_data_prices(symbols: str = Query("BTC,ETH", description="Comma-separated symbols")): - """Alternative endpoint for market data prices""" - return await get_multiple_prices(symbols) - - -# ============================================================================ -# Analysis Endpoints -# ============================================================================ - -@app.get("/api/analysis/signals") -async def get_trading_signals( - symbol: str = Query("BTCUSDT", description="Trading pair"), - timeframe: str = Query("1h", description="Timeframe") -): - """Get trading signals for a symbol""" - try: - # Fetch OHLCV data for analysis - ohlcv = await fetch_binance_ohlcv(symbol, timeframe, 100) - - if not ohlcv: - raise HTTPException(status_code=503, detail="Unable to fetch data for analysis") - - # Simple signal generation (can be enhanced) - latest = ohlcv[-1] - prev = ohlcv[-2] if len(ohlcv) > 1 else latest - - # Calculate simple indicators - close_prices = [c["close"] for c in ohlcv[-20:]] - sma_20 = sum(close_prices) / len(close_prices) - - # Generate signal - trend = "bullish" if latest["close"] > sma_20 else "bearish" - momentum = "strong" if abs(latest["close"] - prev["close"]) / prev["close"] > 0.01 else "weak" - - signal = "buy" if trend == "bullish" and momentum == "strong" else ( - "sell" if trend == "bearish" and momentum == "strong" else "hold" - ) - - ai_summary = analyze_chart_points(symbol, timeframe, ohlcv) - - return { - "symbol": symbol, - "timeframe": timeframe, - "signal": signal, - "trend": trend, - "momentum": momentum, - "indicators": { - "sma_20": sma_20, - "current_price": latest["close"], - "price_change": latest["close"] - prev["close"], - "price_change_percent": ((latest["close"] - prev["close"]) / prev["close"]) * 100 - }, - "analysis": ai_summary, - "timestamp": datetime.now().isoformat() - } - - except HTTPException: - raise - except Exception as e: - logger.error(f"Error in get_trading_signals: {e}") - raise HTTPException(status_code=500, detail=str(e)) - - -@app.get("/api/analysis/smc") -async def get_smc_analysis(symbol: str = Query("BTCUSDT", description="Trading pair")): - """Get Smart Money Concepts (SMC) analysis""" - try: - # Fetch OHLCV data - ohlcv = await fetch_binance_ohlcv(symbol, "1h", 200) - - if not ohlcv: - raise HTTPException(status_code=503, detail="Unable to fetch data") - - # Calculate key levels - highs = [c["high"] for c in ohlcv] - lows = [c["low"] for c in ohlcv] - closes = [c["close"] for c in ohlcv] - - resistance = max(highs[-50:]) - support = min(lows[-50:]) - current_price = closes[-1] - - # Structure analysis - market_structure = "higher_highs" if closes[-1] > closes[-10] > closes[-20] else "lower_lows" - - return { - "symbol": symbol, - "market_structure": market_structure, - "key_levels": { - "resistance": resistance, - "support": support, - "current_price": current_price, - "mid_point": (resistance + support) / 2 - }, - "order_blocks": { - "bullish": support, - "bearish": resistance - }, - "liquidity_zones": { - "above": resistance, - "below": support - }, - "timestamp": datetime.now().isoformat() - } - - except HTTPException: - raise - except Exception as e: - logger.error(f"Error in get_smc_analysis: {e}") - raise HTTPException(status_code=500, detail=str(e)) - - -@app.get("/api/scoring/snapshot") -async def get_scoring_snapshot(symbol: str = Query("BTCUSDT", description="Trading pair")): - """Get comprehensive scoring snapshot""" - try: - # Fetch data - ticker, _ = await fetch_binance_ticker(symbol) - ohlcv = await fetch_binance_ohlcv(symbol, "1h", 100) - - if not ticker or not ohlcv: - raise HTTPException(status_code=503, detail="Unable to fetch data") - - # Calculate scores (0-100) - volatility_score = min(abs(ticker["price_change_percent_24h"]) * 5, 100) - volume_score = min((ticker["volume_24h"] / 1000000) * 10, 100) - trend_score = 50 + (ticker["price_change_percent_24h"] * 2) - - # Overall score - overall_score = (volatility_score + volume_score + trend_score) / 3 - - return { - "symbol": symbol, - "overall_score": round(overall_score, 2), - "scores": { - "volatility": round(volatility_score, 2), - "volume": round(volume_score, 2), - "trend": round(trend_score, 2), - "momentum": round(50 + ticker["price_change_percent_24h"], 2) - }, - "rating": "excellent" if overall_score > 80 else ( - "good" if overall_score > 60 else ( - "average" if overall_score > 40 else "poor" - ) - ), - "timestamp": datetime.now().isoformat() - } - - except HTTPException: - raise - except Exception as e: - logger.error(f"Error in get_scoring_snapshot: {e}") - raise HTTPException(status_code=500, detail=str(e)) - - -@app.get("/api/signals") -async def get_all_signals(): - """Get signals for multiple assets""" - symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "BNBUSDT"] - signals = [] - - for symbol in symbols: - try: - signal_data = await get_trading_signals(symbol, "1h") - signals.append(signal_data) - except: - continue - - return { - "count": len(signals), - "signals": signals, - "timestamp": datetime.now().isoformat() - } - - -@app.get("/api/sentiment") -async def get_sentiment(): - """Get market sentiment data""" - try: - news = await news_collector.get_latest_news(limit=5) - except CollectorError as exc: - logger.warning("Sentiment fallback due to news error: %s", exc) - news = [] - - text = " ".join(item.get("title", "") for item in news).strip() or "Crypto market update" - analysis = analyze_market_text(text) - score = analysis.get("signals", {}).get("crypto", {}).get("score", 0.0) - normalized_value = int((score + 1) * 50) - - if normalized_value < 20: - classification = "extreme_fear" - elif normalized_value < 40: - classification = "fear" - elif normalized_value < 60: - classification = "neutral" - elif normalized_value < 80: - classification = "greed" - else: - classification = "extreme_greed" - - return { - "value": normalized_value, - "classification": classification, - "description": f"Market sentiment is {classification.replace('_', ' ')}", - "analysis": analysis, - "timestamp": datetime.utcnow().isoformat(), - } - - -# ============================================================================ -# System Endpoints -# ============================================================================ - -@app.get("/api/system/status") -async def get_system_status(): - """Get system status""" - providers = await provider_collector.get_providers_status() - online = sum(1 for provider in providers if provider.get("status") == "online") - - cache_items = ( - len(getattr(market_collector.cache, "_store", {})) - + len(getattr(news_collector.cache, "_store", {})) - + len(getattr(provider_collector.cache, "_store", {})) - ) - - return { - "status": "operational" if online else "maintenance", - "uptime_seconds": round(time.time() - START_TIME, 2), - "cache_size": cache_items, - "providers_online": online, - "requests_per_minute": 0, - "timestamp": datetime.utcnow().isoformat(), - } - - -@app.get("/api/system/config") -async def get_system_config(): - """Get system configuration""" - return { - "version": app.version, - "api_version": "v1", - "cache_ttl_seconds": settings.cache_ttl, - "supported_symbols": sorted(set(COIN_SYMBOL_MAPPING.values())), - "supported_intervals": ["1m", "5m", "15m", "30m", "1h", "4h", "1d"], - "max_ohlcv_limit": 1000, - "timestamp": datetime.utcnow().isoformat(), - } - - -@app.get("/api/categories") -async def get_categories(): - """Get data categories""" - return { - "categories": [ - {"name": "market_data", "endpoints": 5, "status": "active"}, - {"name": "analysis", "endpoints": 4, "status": "active"}, - {"name": "signals", "endpoints": 2, "status": "active"}, - {"name": "sentiment", "endpoints": 1, "status": "active"} - ] - } - - -@app.get("/api/rate-limits") -async def get_rate_limits(): - """Get rate limit information""" - return { - "rate_limits": [ - {"endpoint": "/api/ohlcv", "limit": 1200, "window": "per_minute"}, - {"endpoint": "/api/crypto/prices/top", "limit": 600, "window": "per_minute"}, - {"endpoint": "/api/analysis/*", "limit": 300, "window": "per_minute"} - ], - "current_usage": { - "requests_this_minute": 0, - "percentage": 0 - } - } - - -@app.get("/api/logs") -async def get_logs(limit: int = Query(50, ge=1, le=500)): - """Get recent API logs""" - # Mock logs (can be enhanced with real logging) - logs = [] - for i in range(min(limit, 10)): - logs.append({ - "timestamp": (datetime.now() - timedelta(minutes=i)).isoformat(), - "endpoint": "/api/ohlcv", - "status": "success", - "response_time_ms": random.randint(50, 200) - }) - - return {"logs": logs, "count": len(logs)} - - -@app.get("/api/alerts") -async def get_alerts(): - """Get system alerts""" - return { - "alerts": [], - "count": 0, - "timestamp": datetime.now().isoformat() - } - - -# ============================================================================ -# HuggingFace Integration Endpoints -# ============================================================================ - -@app.get("/api/hf/health") -async def hf_health(): - """HuggingFace integration health""" - from ai_models import AI_MODELS_SUMMARY - status = registry_status() - status["models"] = AI_MODELS_SUMMARY - status["timestamp"] = datetime.utcnow().isoformat() - return status - - -@app.post("/api/hf/refresh") -async def hf_refresh(): - """Refresh HuggingFace data""" - from ai_models import initialize_models - result = initialize_models() - return {"status": "ok" if result.get("models_loaded", 0) > 0 else "degraded", **result, "timestamp": datetime.utcnow().isoformat()} - - -@app.get("/api/hf/registry") -async def hf_registry(kind: str = "models"): - """Get HuggingFace registry""" - info = get_model_info() - return {"kind": kind, "items": info.get("model_names", info)} - - -@app.get("/api/resources/unified") -async def get_unified_resources(): - """Get unified API resources from crypto_resources_unified_2025-11-11.json""" - try: - data = local_resource_service.get_registry() - if data: - metadata = data.get("registry", {}).get("metadata", {}) - return { - "success": True, - "data": data, - "metadata": metadata, - "count": metadata.get("total_entries", 0), - "fallback_assets": len(local_resource_service.get_supported_symbols()) - } - return {"success": False, "error": "Resources file not found"} - except Exception as e: - logger.error(f"Error loading unified resources: {e}") - return {"success": False, "error": str(e)} - - -@app.get("/api/resources/ultimate") -async def get_ultimate_resources(): - """Get ultimate API resources from ultimate_crypto_pipeline_2025_NZasinich.json""" - try: - resources_path = WORKSPACE_ROOT / "api-resources" / "ultimate_crypto_pipeline_2025_NZasinich.json" - if resources_path.exists(): - with open(resources_path, 'r', encoding='utf-8') as f: - data = json.load(f) - return { - "success": True, - "data": data, - "total_sources": data.get("total_sources", 0), - "files": len(data.get("files", [])) - } - return {"success": False, "error": "Resources file not found"} - except Exception as e: - logger.error(f"Error loading ultimate resources: {e}") - return {"success": False, "error": str(e)} - - -@app.get("/api/resources/stats") -async def get_resources_stats(): - """Get statistics about available API resources""" - try: - stats = { - "unified": {"available": False, "count": 0}, - "ultimate": {"available": False, "count": 0}, - "total_apis": 0 - } - - # Check unified resources via the centralized loader - registry = local_resource_service.get_registry() - if registry: - stats["unified"] = { - "available": True, - "count": registry.get("registry", {}).get("metadata", {}).get("total_entries", 0), - "fallback_assets": len(local_resource_service.get_supported_symbols()) - } - - # Check ultimate resources - ultimate_path = WORKSPACE_ROOT / "api-resources" / "ultimate_crypto_pipeline_2025_NZasinich.json" - if ultimate_path.exists(): - with open(ultimate_path, 'r', encoding='utf-8') as f: - ultimate_data = json.load(f) - stats["ultimate"] = { - "available": True, - "count": ultimate_data.get("total_sources", 0) - } - - stats["total_apis"] = stats["unified"].get("count", 0) + stats["ultimate"].get("count", 0) - - return {"success": True, "stats": stats} - except Exception as e: - logger.error(f"Error getting resources stats: {e}") - return {"success": False, "error": str(e)} - - -def _resolve_sentiment_payload(payload: Union[List[str], Dict[str, Any]]) -> Dict[str, Any]: - if isinstance(payload, list): - return {"texts": payload, "mode": "auto"} - if isinstance(payload, dict): - texts = payload.get("texts") or payload.get("text") - if isinstance(texts, str): - texts = [texts] - if not isinstance(texts, list): - raise ValueError("texts must be provided") - mode = payload.get("mode") or payload.get("model") or "auto" - return {"texts": texts, "mode": mode} - raise ValueError("Invalid payload") - - -@app.post("/api/hf/run-sentiment") -@app.post("/api/hf/sentiment") -async def hf_sentiment(payload: Union[List[str], Dict[str, Any]] = Body(...)): - """Run sentiment analysis using shared AI helpers.""" - from ai_models import AI_MODELS_SUMMARY - - if AI_MODELS_SUMMARY.get("models_loaded", 0) == 0 or AI_MODELS_SUMMARY.get("mode") == "off": - return { - "ok": False, - "error": "No HF models are currently loaded.", - "mode": AI_MODELS_SUMMARY.get("mode", "off"), - "models_loaded": AI_MODELS_SUMMARY.get("models_loaded", 0) - } - - try: - resolved = _resolve_sentiment_payload(payload) - except ValueError as exc: - raise HTTPException(status_code=400, detail=str(exc)) from exc - - mode = (resolved.get("mode") or "auto").lower() - texts = resolved["texts"] - results: List[Dict[str, Any]] = [] - for text in texts: - if mode == "crypto": - analysis = analyze_crypto_sentiment(text) - elif mode == "financial": - analysis = analyze_market_text(text).get("signals", {}).get("financial", {}) - elif mode == "social": - analysis = analyze_market_text(text).get("signals", {}).get("social", {}) - else: - analysis = analyze_market_text(text) - results.append({"text": text, "result": analysis}) - - return {"mode": mode, "results": results, "timestamp": datetime.utcnow().isoformat()} - - -@app.post("/api/hf/models/sentiment") -async def hf_models_sentiment(payload: Union[List[str], Dict[str, Any]] = Body(...)): - """Compatibility endpoint for HF console sentiment panel.""" - from ai_models import AI_MODELS_SUMMARY - - if AI_MODELS_SUMMARY.get("models_loaded", 0) == 0 or AI_MODELS_SUMMARY.get("mode") == "off": - return { - "ok": False, - "error": "No HF models are currently loaded.", - "mode": AI_MODELS_SUMMARY.get("mode", "off"), - "models_loaded": AI_MODELS_SUMMARY.get("models_loaded", 0) - } - - return await hf_sentiment(payload) - - -@app.post("/api/hf/models/forecast") -async def hf_models_forecast(payload: Dict[str, Any] = Body(...)): - """Generate quick technical forecasts from provided closing prices.""" - series = payload.get("series") or payload.get("values") or payload.get("close") - if not isinstance(series, list) or len(series) < 3: - raise HTTPException(status_code=400, detail="Provide at least 3 closing prices in 'series'.") - - try: - floats = [float(x) for x in series] - except (TypeError, ValueError) as exc: - raise HTTPException(status_code=400, detail="Series must contain numeric values") from exc - - model_name = (payload.get("model") or payload.get("model_name") or "btc_lstm").lower() - steps = int(payload.get("steps") or 3) - - deltas = [floats[i] - floats[i - 1] for i in range(1, len(floats))] - avg_delta = mean(deltas) - volatility = mean(abs(delta - avg_delta) for delta in deltas) if deltas else 0 - - predictions = [] - last = floats[-1] - decay = 0.95 if model_name == "btc_arima" else 1.02 - for _ in range(steps): - last = last + (avg_delta * decay) - predictions.append(round(last, 4)) - - return { - "model": model_name, - "steps": steps, - "input_count": len(floats), - "volatility": round(volatility, 5), - "predictions": predictions, - "source": "local-fallback" if model_name == "btc_arima" else "hybrid", - "timestamp": datetime.utcnow().isoformat() - } - - -@app.get("/api/hf/datasets/market/ohlcv") -async def hf_dataset_market_ohlcv(symbol: str = Query("BTC"), interval: str = Query("1h"), limit: int = Query(120, ge=10, le=500)): - """Expose fallback OHLCV snapshots as a pseudo HF dataset slice.""" - data = local_resource_service.get_ohlcv(symbol.upper(), interval, limit) - source = "local-fallback" - - if not data: - return { - "symbol": symbol.upper(), - "interval": interval, - "count": 0, - "data": [], - "source": source, - "message": "No cached OHLCV available yet" - } - - return { - "symbol": symbol.upper(), - "interval": interval, - "count": len(data), - "data": data, - "source": source, - "timestamp": datetime.utcnow().isoformat() - } - - -@app.get("/api/hf/datasets/market/btc_technical") -async def hf_dataset_market_btc(limit: int = Query(50, ge=10, le=200)): - """Simplified technical metrics derived from fallback OHLCV data.""" - candles = local_resource_service.get_ohlcv("BTC", "1h", limit + 20) - - if not candles: - raise HTTPException(status_code=503, detail="Fallback OHLCV unavailable") - - rows = [] - closes = [c["close"] for c in candles] - for idx, candle in enumerate(candles[-limit:]): - window = closes[max(0, idx): idx + 20] - sma = sum(window) / len(window) if window else candle["close"] - momentum = candle["close"] - candle["open"] - rows.append({ - "timestamp": candle["timestamp"], - "datetime": candle["datetime"], - "close": candle["close"], - "sma_20": round(sma, 4), - "momentum": round(momentum, 4), - "volatility": round((candle["high"] - candle["low"]) / candle["low"], 4) - }) - - return { - "symbol": "BTC", - "interval": "1h", - "count": len(rows), - "items": rows, - "source": "local-fallback" - } - - -@app.get("/api/hf/datasets/news/semantic") -async def hf_dataset_news(limit: int = Query(10, ge=3, le=25)): - """News slice augmented with sentiment tags for HF demos.""" - try: - news = await news_collector.get_latest_news(limit=limit) - source = "providers" - except CollectorError: - news = [] - source = "local-fallback" - - if not news: - items = HF_SAMPLE_NEWS[:limit] - else: - items = [] - for item in news: - items.append({ - "title": item.get("title"), - "source": item.get("source") or item.get("provider"), - "sentiment": item.get("sentiment") or "neutral", - "sentiment_score": item.get("sentiment_confidence", 0.5), - "entities": item.get("symbols") or [], - "summary": item.get("summary") or item.get("description"), - "published_at": item.get("date") or item.get("published_at") - }) - return { - "count": len(items), - "items": items, - "source": source, - "timestamp": datetime.utcnow().isoformat() - } - - -# ============================================================================ -# HTML Routes - Serve UI files -# ============================================================================ - -@app.get("/favicon.ico") -async def favicon(): - """Serve favicon""" - favicon_path = WORKSPACE_ROOT / "static" / "favicon.ico" - if favicon_path.exists(): - return FileResponse(favicon_path) - return JSONResponse({"status": "no favicon"}, status_code=404) - -@app.get("/", response_class=HTMLResponse) -async def root(): - """Serve main HTML UI page (index.html)""" - index_path = WORKSPACE_ROOT / "index.html" - if index_path.exists(): - return FileResponse( - path=str(index_path), - media_type="text/html", - filename="index.html" - ) - return HTMLResponse("

Cryptocurrency Data & Analysis API

See /docs for API documentation

") - -@app.get("/index.html", response_class=HTMLResponse) -async def index(): - """Serve index.html""" - return FileResponse(WORKSPACE_ROOT / "index.html") - -@app.get("/dashboard.html", response_class=HTMLResponse) -async def dashboard(): - """Serve dashboard.html""" - return FileResponse(WORKSPACE_ROOT / "dashboard.html") - -@app.get("/dashboard", response_class=HTMLResponse) -async def dashboard_alt(): - """Alternative route for dashboard""" - return FileResponse(WORKSPACE_ROOT / "dashboard.html") - -@app.get("/admin.html", response_class=HTMLResponse) -async def admin(): - """Serve admin panel""" - admin_path = WORKSPACE_ROOT / "admin.html" - if admin_path.exists(): - return FileResponse( - path=str(admin_path), - media_type="text/html", - filename="admin.html" - ) - return HTMLResponse("

Admin panel not found

") - -@app.get("/admin", response_class=HTMLResponse) -async def admin_alt(): - """Alternative route for admin""" - admin_path = WORKSPACE_ROOT / "admin.html" - if admin_path.exists(): - return FileResponse( - path=str(admin_path), - media_type="text/html", - filename="admin.html" - ) - return HTMLResponse("

Admin panel not found

") - -@app.get("/hf_console.html", response_class=HTMLResponse) -async def hf_console(): - """Serve HuggingFace console""" - return FileResponse(WORKSPACE_ROOT / "hf_console.html") - -@app.get("/console", response_class=HTMLResponse) -async def console_alt(): - """Alternative route for HF console""" - return FileResponse(WORKSPACE_ROOT / "hf_console.html") - -@app.get("/pool_management.html", response_class=HTMLResponse) -async def pool_management(): - """Serve pool management UI""" - return FileResponse(WORKSPACE_ROOT / "pool_management.html") - -@app.get("/unified_dashboard.html", response_class=HTMLResponse) -async def unified_dashboard(): - """Serve unified dashboard""" - return FileResponse(WORKSPACE_ROOT / "unified_dashboard.html") - -@app.get("/simple_overview.html", response_class=HTMLResponse) -async def simple_overview(): - """Serve simple overview""" - return FileResponse(WORKSPACE_ROOT / "simple_overview.html") - -# Generic HTML file handler -@app.get("/{filename}.html", response_class=HTMLResponse) -async def serve_html(filename: str): - """Serve any HTML file from workspace root""" - file_path = WORKSPACE_ROOT / f"{filename}.html" - if file_path.exists(): - return FileResponse(file_path) - return HTMLResponse(f"

File {filename}.html not found

", status_code=404) - - -# ============================================================================ -# Startup Event -# ============================================================================ - - -# ============================================================================ -# ADMIN DASHBOARD ENDPOINTS -# ============================================================================ - -from fastapi import WebSocket, WebSocketDisconnect -import asyncio - -class ConnectionManager: - def __init__(self): - self.active_connections = [] - async def connect(self, websocket: WebSocket): - await websocket.accept() - self.active_connections.append(websocket) - def disconnect(self, websocket: WebSocket): - if websocket in self.active_connections: - self.active_connections.remove(websocket) - async def broadcast(self, message: dict): - disconnected = [] - for conn in list(self.active_connections): - try: - # Check connection state before sending - if conn.client_state == WebSocketState.CONNECTED: - await conn.send_json(message) - else: - disconnected.append(conn) - except Exception as e: - logger.debug(f"Error broadcasting to client: {e}") - disconnected.append(conn) - - # Clean up disconnected clients - for conn in disconnected: - self.disconnect(conn) - -ws_manager = ConnectionManager() - -@app.get("/api/health") -async def api_health(): - h = await health() - return {"status": "healthy" if h.get("status") == "ok" else "degraded", **h} - -# Removed duplicate - using improved version below - -@app.get("/api/coins/{symbol}") -async def get_coin_detail(symbol: str): - coins = await market_collector.get_top_coins(limit=250) - coin = next((c for c in coins if c.get("symbol", "").upper() == symbol.upper()), None) - if not coin: - raise HTTPException(404, f"Coin {symbol} not found") - return {"success": True, "symbol": symbol.upper(), "name": coin.get("name", ""), - "price": coin.get("price") or coin.get("current_price", 0), - "change_24h": coin.get("change_24h") or coin.get("price_change_percentage_24h", 0), - "market_cap": coin.get("market_cap", 0)} - -@app.get("/api/market/stats") -async def get_market_stats(): - """Get global market statistics (duplicate endpoint - keeping for compatibility)""" - try: - overview = await get_market_overview() - - # Calculate ETH dominance from prices if available - eth_dominance = 0 - if overview.get("total_market_cap", 0) > 0: - try: - eth_prices, _ = await fetch_coingecko_prices(symbols=["ETH"], limit=1) - if eth_prices and len(eth_prices) > 0: - eth_market_cap = eth_prices[0].get("market_cap", 0) or 0 - eth_dominance = (eth_market_cap / overview.get("total_market_cap", 1)) * 100 - except: - pass - - return { - "success": True, - "stats": { - "total_market_cap": overview.get("total_market_cap", 0) or 0, - "total_volume_24h": overview.get("total_volume_24h", 0) or 0, - "btc_dominance": overview.get("btc_dominance", 0) or 0, - "eth_dominance": eth_dominance, - "active_cryptocurrencies": 10000, - "markets": 500, - "market_cap_change_24h": 0.0, - "timestamp": datetime.now().isoformat() - } - } - except Exception as e: - logger.error(f"Error in /api/market/stats (duplicate): {e}") - return { - "success": True, - "stats": { - "total_market_cap": 0, - "total_volume_24h": 0, - "btc_dominance": 0, - "eth_dominance": 0, - "active_cryptocurrencies": 0, - "markets": 0, - "market_cap_change_24h": 0.0, - "timestamp": datetime.now().isoformat() - } - } - - -@app.get("/api/stats") -async def get_stats_alias(): - """Alias endpoint for /api/market/stats - backward compatibility""" - return await get_market_stats() - - -@app.get("/api/news/latest") -async def get_latest_news(limit: int = Query(default=40, ge=1, le=100)): - from ai_models import analyze_news_item - news = await news_collector.get_latest_news(limit=limit) - enriched = [] - for item in news[:limit]: - try: - e = analyze_news_item(item) - enriched.append({"title": e.get("title", ""), "source": e.get("source", ""), - "published_at": e.get("published_at") or e.get("date", ""), - "symbols": e.get("symbols", []), "sentiment": e.get("sentiment", "neutral"), - "sentiment_confidence": e.get("sentiment_confidence", 0.5)}) - except: - enriched.append({"title": item.get("title", ""), "source": item.get("source", ""), - "published_at": item.get("date", ""), "symbols": item.get("symbols", []), - "sentiment": "neutral", "sentiment_confidence": 0.5}) - return {"success": True, "news": enriched, "count": len(enriched)} - -@app.post("/api/news/summarize") -async def summarize_news(item: Dict[str, Any] = Body(...)): - from ai_models import analyze_news_item - e = analyze_news_item(item) - return {"success": True, "summary": e.get("title", ""), "sentiment": e.get("sentiment", "neutral")} - -# Duplicate endpoints removed - using the improved versions below in CHARTS ENDPOINTS section - -@app.post("/api/sentiment/analyze") -async def analyze_sentiment(payload: Dict[str, Any] = Body(...)): - from ai_models import ensemble_crypto_sentiment - result = ensemble_crypto_sentiment(payload.get("text", "")) - return {"success": True, "sentiment": result["label"], "confidence": result["confidence"], "details": result} - -@app.post("/api/query") -async def process_query(payload: Dict[str, Any] = Body(...)): - query = payload.get("query", "").lower() - if "price" in query or "btc" in query: - coins = await market_collector.get_top_coins(limit=10) - btc = next((c for c in coins if c.get("symbol", "").upper() == "BTC"), None) - if btc: - return {"success": True, "type": "price", "message": f"Bitcoin is ${btc.get('price', 0):,.2f}", "data": btc} - return {"success": True, "type": "general", "message": "Query processed"} - -@app.get("/api/datasets/list") -async def list_datasets(): - from backend.services.hf_registry import REGISTRY - datasets = REGISTRY.list(kind="datasets") - formatted = [{"name": d.get("id"), "category": d.get("category", "other"), "tags": d.get("tags", [])} for d in datasets] - return {"success": True, "datasets": formatted, "count": len(formatted)} - -@app.get("/api/datasets/sample") -async def get_dataset_sample(name: str = Query(...), limit: int = Query(default=20)): - return {"success": False, "name": name, "sample": [], "message": "Auth required"} - -@app.get("/api/models/list") -async def list_models(): - from ai_models import get_model_info - info = get_model_info() - models = [] - for cat, mlist in info.get("model_catalog", {}).items(): - for mid in mlist: - models.append({"name": mid, "task": "sentiment" if "sentiment" in cat else "analysis", "category": cat}) - return {"success": True, "models": models, "count": len(models)} - -@app.post("/api/models/test") -async def test_model(payload: Dict[str, Any] = Body(...)): - from ai_models import ensemble_crypto_sentiment - result = ensemble_crypto_sentiment(payload.get("text", "")) - return {"success": True, "model": payload.get("model", ""), "result": result} - -@app.websocket("/ws") -async def websocket_endpoint(websocket: WebSocket): - await ws_manager.connect(websocket) - try: - while True: - # Check if connection is still open before sending - if websocket.client_state != WebSocketState.CONNECTED: - logger.info("WebSocket connection closed, breaking loop") - break - - try: - top_coins = await market_collector.get_top_coins(limit=5) - news = await news_collector.get_latest_news(limit=3) - from ai_models import ensemble_crypto_sentiment - sentiment = ensemble_crypto_sentiment(" ".join([n.get("title", "") for n in news])) if news else {"label": "neutral", "confidence": 0.5} - - # Double-check connection state before sending - if websocket.client_state == WebSocketState.CONNECTED: - await websocket.send_json({ - "type": "update", - "payload": { - "market_data": top_coins, - "news": news, - "sentiment": sentiment, - "timestamp": datetime.now().isoformat() - } - }) - else: - logger.info("WebSocket disconnected, breaking loop") - break - - except CollectorError as e: - # Provider errors are already logged by the collector, just continue - logger.debug(f"Provider error in WebSocket update (this is expected with fallbacks): {e}") - # Use cached data if available, or empty data - top_coins = [] - news = [] - sentiment = {"label": "neutral", "confidence": 0.5} - except Exception as e: - # Log other errors with full details - error_msg = str(e) if str(e) else repr(e) - logger.error(f"Error in WebSocket update loop: {type(e).__name__}: {error_msg}") - # Don't break on data errors, just log and continue - # Only break on connection errors - if "send" in str(e).lower() or "close" in str(e).lower(): - break - - await asyncio.sleep(10) - except WebSocketDisconnect: - logger.info("WebSocket disconnect exception caught") - except Exception as e: - logger.error(f"WebSocket endpoint error: {e}") - finally: - try: - ws_manager.disconnect(websocket) - except: - pass - - -@app.on_event("startup") -async def startup_event(): - """Initialize on startup - non-blocking""" - logger.info("=" * 70) - logger.info("Starting Cryptocurrency Data & Analysis API") - logger.info("=" * 70) - logger.info("FastAPI initialized") - logger.info("CORS configured") - logger.info("Cache initialized") - logger.info(f"Providers loaded: {len(PROVIDERS_CONFIG)}") - - # Initialize AI models in background (non-blocking) - async def init_models_background(): - try: - from ai_models import initialize_models - models_init = initialize_models() - logger.info(f"AI Models initialized: {models_init}") - except Exception as e: - logger.warning(f"AI Models initialization failed: {e}") - - # Initialize HF Registry in background (non-blocking) - async def init_registry_background(): - try: - from backend.services.hf_registry import REGISTRY - registry_result = await REGISTRY.refresh() - logger.info(f"HF Registry initialized: {registry_result}") - except Exception as e: - logger.warning(f"HF Registry initialization failed: {e}") - - # Start background tasks - asyncio.create_task(init_models_background()) - asyncio.create_task(init_registry_background()) - logger.info("Background initialization tasks started") - - # Show loaded HuggingFace Space providers - hf_providers = [p for p in PROVIDERS_CONFIG.keys() if 'huggingface_space' in p] - if hf_providers: - logger.info(f"HuggingFace Space providers: {', '.join(hf_providers)}") - - logger.info("Data sources: Binance, CoinGecko, providers_config_extended.json") - - # Check HTML files - html_files = ["index.html", "dashboard.html", "admin.html", "hf_console.html"] - available_html = [f for f in html_files if (WORKSPACE_ROOT / f).exists()] - logger.info(f"UI files: {len(available_html)}/{len(html_files)} available") - logger.info(f"HTML UI available at: http://0.0.0.0:7860/ (index.html)") - - logger.info("=" * 70) - logger.info("API ready at http://0.0.0.0:7860") - logger.info("Docs at http://0.0.0.0:7860/docs") - logger.info("UI at http://0.0.0.0:7860/ (index.html - default HTML page)") - logger.info("=" * 70) - - -# ============================================================================ -# Main Entry Point -# ============================================================================ - -if __name__ == "__main__": - import uvicorn - import sys - import io - - # Fix encoding for Windows console - if sys.platform == "win32": - sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') - sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace') - - try: - print("=" * 70) - print("Starting Cryptocurrency Data & Analysis API") - print("=" * 70) - print("Server: http://localhost:7860") - print("API Docs: http://localhost:7860/docs") - print("Health: http://localhost:7860/health") - print("=" * 70) - except UnicodeEncodeError: - # Fallback if encoding still fails - print("=" * 70) - print("Starting Cryptocurrency Data & Analysis API") - print("=" * 70) - print("Server: http://localhost:7860") - print("API Docs: http://localhost:7860/docs") - print("Health: http://localhost:7860/health") - print("=" * 70) - - uvicorn.run( - app, - host="0.0.0.0", - port=7860, - log_level="info" - ) -# NEW ENDPOINTS FOR ADMIN.HTML - ADD TO hf_unified_server.py - -from fastapi import WebSocket, WebSocketDisconnect -from collections import defaultdict - -# WebSocket Manager -class ConnectionManager: - def __init__(self): - self.active_connections: List[WebSocket] = [] - - async def connect(self, websocket: WebSocket): - await websocket.accept() - self.active_connections.append(websocket) - logger.info(f"WebSocket connected. Total: {len(self.active_connections)}") - - def disconnect(self, websocket: WebSocket): - if websocket in self.active_connections: - self.active_connections.remove(websocket) - logger.info(f"WebSocket disconnected. Total: {len(self.active_connections)}") - - async def broadcast(self, message: dict): - disconnected = [] - for connection in list(self.active_connections): - try: - # Check connection state before sending - if connection.client_state == WebSocketState.CONNECTED: - await connection.send_json(message) - else: - disconnected.append(connection) - except Exception as e: - logger.debug(f"Error broadcasting to client: {e}") - disconnected.append(connection) - - # Clean up disconnected clients - for connection in disconnected: - self.disconnect(connection) - -ws_manager = ConnectionManager() - - -# ===== API HEALTH ===== -@app.get("/api/health") -async def api_health(): - """Health check for admin dashboard""" - health_data = await health() - return { - "status": "healthy" if health_data.get("status") == "ok" else "degraded", - **health_data - } - - -# ===== COINS ENDPOINTS ===== -@app.get("/api/coins/top") -async def get_top_coins(limit: int = Query(default=10, ge=1, le=100)): - """Get top cryptocurrencies by market cap""" - try: - coins = await market_collector.get_top_coins(limit=limit) - - result = [] - for coin in coins: - result.append({ - "id": coin.get("id", coin.get("symbol", "").lower()), - "rank": coin.get("rank", 0), - "symbol": coin.get("symbol", "").upper(), - "name": coin.get("name", ""), - "price": coin.get("price") or coin.get("current_price", 0), - "current_price": coin.get("price") or coin.get("current_price", 0), - "price_change_24h": coin.get("change_24h") or coin.get("price_change_percentage_24h", 0), - "price_change_percentage_24h": coin.get("change_24h") or coin.get("price_change_percentage_24h", 0), - "price_change_percentage_7d_in_currency": coin.get("price_change_percentage_7d", 0), - "volume_24h": coin.get("volume_24h") or coin.get("total_volume", 0), - "total_volume": coin.get("volume_24h") or coin.get("total_volume", 0), - "market_cap": coin.get("market_cap", 0), - "image": coin.get("image", ""), - "sparkline_in_7d": coin.get("sparkline_in_7d") or {"price": []}, - "sparkline_data": coin.get("sparkline_data") or [], - "last_updated": coin.get("last_updated", datetime.now().isoformat()) - }) - - return { - "success": True, - "coins": result, - "count": len(result), - "timestamp": datetime.now().isoformat() - } - except Exception as e: - logger.error(f"Error in /api/coins/top: {e}") - raise HTTPException(status_code=503, detail=str(e)) - - -@app.get("/api/coins/{symbol}") -async def get_coin_detail(symbol: str): - """Get specific coin details""" - try: - coins = await market_collector.get_top_coins(limit=250) - coin = next((c for c in coins if c.get("symbol", "").upper() == symbol.upper()), None) - - if not coin: - raise HTTPException(status_code=404, detail=f"Coin {symbol} not found") - - return { - "success": True, - "symbol": symbol.upper(), - "name": coin.get("name", ""), - "price": coin.get("price") or coin.get("current_price", 0), - "change_24h": coin.get("change_24h") or coin.get("price_change_percentage_24h", 0), - "volume_24h": coin.get("volume_24h") or coin.get("total_volume", 0), - "market_cap": coin.get("market_cap", 0), - "rank": coin.get("rank", 0), - "last_updated": coin.get("last_updated", datetime.now().isoformat()) - } - except HTTPException: - raise - except Exception as e: - logger.error(f"Error in /api/coins/{symbol}: {e}") - raise HTTPException(status_code=503, detail=str(e)) - - -# ===== MARKET STATS ===== -@app.get("/api/market/stats") -async def get_market_stats(): - """Get global market statistics""" - try: - # Use existing endpoint - get_market_overview returns total_market_cap and total_volume_24h - overview = await get_market_overview() - - # Calculate ETH dominance from prices if available - eth_dominance = 0 - if overview.get("total_market_cap", 0) > 0: - # Try to get ETH market cap from top coins - try: - eth_prices, _ = await fetch_coingecko_prices(symbols=["ETH"], limit=1) - if eth_prices and len(eth_prices) > 0: - eth_market_cap = eth_prices[0].get("market_cap", 0) or 0 - eth_dominance = (eth_market_cap / overview.get("total_market_cap", 1)) * 100 - except: - pass - - stats = { - "total_market_cap": overview.get("total_market_cap", 0) or 0, - "total_volume_24h": overview.get("total_volume_24h", 0) or 0, - "btc_dominance": overview.get("btc_dominance", 0) or 0, - "eth_dominance": eth_dominance, - "active_cryptocurrencies": 10000, # Approximate - "markets": 500, # Approximate - "market_cap_change_24h": 0.0, - "timestamp": datetime.now().isoformat() - } - - return {"success": True, "stats": stats} - except Exception as e: - logger.error(f"Error in /api/market/stats: {e}") - raise HTTPException(status_code=503, detail=str(e)) - - -# ===== NEWS ENDPOINTS ===== -@app.get("/api/news/latest") -async def get_latest_news(limit: int = Query(default=40, ge=1, le=100)): - """Get latest crypto news with sentiment""" - try: - news_items = await news_collector.get_latest_news(limit=limit) - - # Attach sentiment to each news item - from ai_models import analyze_news_item - enriched_news = [] - for item in news_items: - try: - enriched = analyze_news_item(item) - enriched_news.append({ - "title": enriched.get("title", ""), - "source": enriched.get("source", ""), - "published_at": enriched.get("published_at") or enriched.get("date", ""), - "symbols": enriched.get("symbols", []), - "sentiment": enriched.get("sentiment", "neutral"), - "sentiment_confidence": enriched.get("sentiment_confidence", 0.5), - "url": enriched.get("url", "") - }) - except: - enriched_news.append({ - "title": item.get("title", ""), - "source": item.get("source", ""), - "published_at": item.get("published_at") or item.get("date", ""), - "symbols": item.get("symbols", []), - "sentiment": "neutral", - "sentiment_confidence": 0.5, - "url": item.get("url", "") - }) - - return { - "success": True, - "news": enriched_news, - "count": len(enriched_news), - "timestamp": datetime.now().isoformat() - } - except Exception as e: - logger.error(f"Error in /api/news/latest: {e}") - return {"success": True, "news": [], "count": 0, "timestamp": datetime.now().isoformat()} - - -@app.get("/api/news") -async def get_news(limit: int = Query(default=40, ge=1, le=100)): - """Alias for /api/news/latest for backward compatibility""" - return await get_latest_news(limit=limit) - - -@app.post("/api/news/summarize") -async def summarize_news(item: Dict[str, Any] = Body(...)): - """Summarize a news article""" - try: - from ai_models import analyze_news_item - enriched = analyze_news_item(item) - - return { - "success": True, - "summary": enriched.get("title", ""), - "sentiment": enriched.get("sentiment", "neutral"), - "sentiment_confidence": enriched.get("sentiment_confidence", 0.5) - } - except Exception as e: - logger.error(f"Error in /api/news/summarize: {e}") - return { - "success": False, - "error": str(e), - "summary": item.get("title", ""), - "sentiment": "neutral" - } - - -# ===== CHARTS ENDPOINTS ===== -@app.get("/api/charts/price/{symbol}") -async def get_price_chart(symbol: str, timeframe: str = Query(default="7d")): - """Get price chart data""" - try: - # Clean and validate symbol - symbol = symbol.strip().upper() - if not symbol: - return JSONResponse( - status_code=400, - content={ - "success": False, - "symbol": "", - "timeframe": timeframe, - "data": [], - "count": 0, - "error": "Symbol cannot be empty" - } - ) - - logger.info(f"Fetching price history for {symbol} with timeframe {timeframe}") - - # market_collector.get_price_history expects timeframe as string, not hours - price_history = await market_collector.get_price_history(symbol, timeframe=timeframe) - - if not price_history or len(price_history) == 0: - logger.warning(f"No price history returned for {symbol}") - return { - "success": True, - "symbol": symbol, - "timeframe": timeframe, - "data": [], - "count": 0, - "message": "No data available" - } - - chart_data = [] - for point in price_history: - # Handle different timestamp formats - timestamp = point.get("timestamp") or point.get("time") or point.get("date") - price = point.get("price") or point.get("close") or point.get("value") or 0 - - # Convert timestamp to ISO format if needed - if timestamp: - try: - # If it's already a string, use it - if isinstance(timestamp, str): - # Try to parse and format - try: - # Try ISO format first - dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00')) - timestamp = dt.isoformat() - except: - try: - # Try other common formats - from dateutil import parser - dt = parser.parse(timestamp) - timestamp = dt.isoformat() - except: - pass - elif isinstance(timestamp, (int, float)): - # Unix timestamp - dt = datetime.fromtimestamp(timestamp) - timestamp = dt.isoformat() - except Exception as e: - logger.warning(f"Error parsing timestamp {timestamp}: {e}") - - chart_data.append({ - "timestamp": timestamp or "", - "time": timestamp or "", - "date": timestamp or "", - "price": float(price) if price else 0, - "close": float(price) if price else 0, - "value": float(price) if price else 0 - }) - - logger.info(f"Returning {len(chart_data)} data points for {symbol}") - - return { - "success": True, - "symbol": symbol, - "timeframe": timeframe, - "data": chart_data, - "count": len(chart_data) - } - except CollectorError as e: - logger.error(f"Collector error in /api/charts/price/{symbol}: {e}", exc_info=True) - return JSONResponse( - status_code=200, - content={ - "success": False, - "symbol": symbol.upper() if symbol else "", - "timeframe": timeframe, - "data": [], - "count": 0, - "error": str(e) - } - ) - except Exception as e: - logger.error(f"Error in /api/charts/price/{symbol}: {e}", exc_info=True) - return JSONResponse( - status_code=200, - content={ - "success": False, - "symbol": symbol.upper() if symbol else "", - "timeframe": timeframe, - "data": [], - "count": 0, - "error": str(e) - } - ) - - -@app.post("/api/charts/analyze") -async def analyze_chart(payload: Dict[str, Any] = Body(...)): - """Analyze chart data""" - try: - symbol = payload.get("symbol") - timeframe = payload.get("timeframe", "7d") - indicators = payload.get("indicators", []) - - if not symbol: - return JSONResponse( - status_code=400, - content={"success": False, "error": "Symbol is required"} - ) - - symbol = symbol.strip().upper() - logger.info(f"Analyzing chart for {symbol} with timeframe {timeframe}") - - # Get price data - use timeframe string, not hours - price_history = await market_collector.get_price_history(symbol, timeframe=timeframe) - - if not price_history or len(price_history) == 0: - return { - "success": False, - "symbol": symbol, - "timeframe": timeframe, - "error": "No price data available for analysis" - } - - # Analyze with AI - from ai_models import analyze_chart_points - try: - analysis = analyze_chart_points(price_history, indicators) - except Exception as ai_error: - logger.error(f"AI analysis error: {ai_error}", exc_info=True) - # Return a basic analysis if AI fails - analysis = { - "direction": "neutral", - "summary": "Analysis unavailable", - "signals": [] - } - - return { - "success": True, - "symbol": symbol, - "timeframe": timeframe, - "analysis": analysis - } - except CollectorError as e: - logger.error(f"Collector error in /api/charts/analyze: {e}", exc_info=True) - return JSONResponse( - status_code=200, - content={"success": False, "error": str(e)} - ) - except Exception as e: - logger.error(f"Error in /api/charts/analyze: {e}", exc_info=True) - return JSONResponse( - status_code=200, - content={"success": False, "error": str(e)} - ) - - -# ===== SENTIMENT ENDPOINTS ===== -@app.post("/api/sentiment/analyze") -async def analyze_sentiment(payload: Dict[str, Any] = Body(...)): - """Analyze sentiment of text""" - try: - text = payload.get("text", "") - - from ai_models import ensemble_crypto_sentiment - result = ensemble_crypto_sentiment(text) - - return { - "success": True, - "sentiment": result["label"], - "confidence": result["confidence"], - "details": result - } - except Exception as e: - logger.error(f"Error in /api/sentiment/analyze: {e}") - return {"success": False, "error": str(e)} - - -# ===== QUERY ENDPOINT ===== -@app.post("/api/query") -async def process_query(payload: Dict[str, Any] = Body(...)): - """Process natural language query""" - try: - query = payload.get("query", "").lower() - - # Simple query processing - if "price" in query or "btc" in query or "bitcoin" in query: - coins = await market_collector.get_top_coins(limit=10) - btc = next((c for c in coins if c.get("symbol", "").upper() == "BTC"), None) - - if btc: - price = btc.get("price") or btc.get("current_price", 0) - return { - "success": True, - "type": "price", - "message": f"Bitcoin (BTC) is currently trading at ${price:,.2f}", - "data": btc - } - - return { - "success": True, - "type": "general", - "message": "Query processed", - "data": None - } - except Exception as e: - logger.error(f"Error in /api/query: {e}") - return {"success": False, "error": str(e), "message": "Query failed"} - - -# ===== DATASETS & MODELS ===== -@app.get("/api/datasets/list") -async def list_datasets(): - """List available datasets""" - try: - from backend.services.hf_registry import REGISTRY - datasets = REGISTRY.list(kind="datasets") - - formatted = [] - for d in datasets: - formatted.append({ - "name": d.get("id"), - "category": d.get("category", "other"), - "records": "N/A", - "updated_at": "", - "tags": d.get("tags", []), - "source": d.get("source", "hub") - }) - - return { - "success": True, - "datasets": formatted, - "count": len(formatted) - } - except Exception as e: - logger.error(f"Error in /api/datasets/list: {e}") - return {"success": True, "datasets": [], "count": 0} - - -@app.get("/api/datasets/sample") -async def get_dataset_sample(name: str = Query(...), limit: int = Query(default=20)): - """Get sample from dataset""" - try: - # Attempt to load dataset - try: - from datasets import load_dataset - from config import get_settings - - # Get HF token for dataset loading - settings = get_settings() - hf_token = settings.hf_token or "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" - - # Set token in environment for datasets library - import os - if hf_token and not os.environ.get("HF_TOKEN"): - os.environ["HF_TOKEN"] = hf_token - - dataset = load_dataset(name, split="train", streaming=True, token=hf_token) - - sample = [] - for i, row in enumerate(dataset): - if i >= limit: - break - sample.append({k: str(v) for k, v in row.items()}) - - return { - "success": True, - "name": name, - "sample": sample, - "count": len(sample) - } - except: - return { - "success": False, - "name": name, - "sample": [], - "count": 0, - "message": "Dataset loading requires authentication or is not available" - } - except Exception as e: - logger.error(f"Error in /api/datasets/sample: {e}") - return {"success": False, "error": str(e)} - - -@app.get("/api/models/list") -async def list_models(): - """List available models""" - try: - from ai_models import get_model_info - info = get_model_info() - - models = [] - catalog = info.get("model_catalog", {}) - - for category, model_list in catalog.items(): - for model_id in model_list: - models.append({ - "name": model_id, - "task": "sentiment" if "sentiment" in category else "decision" if category == "decision" else "analysis", - "status": "available", - "category": category, - "notes": f"{category.replace('_', ' ').title()} model" - }) - - return { - "success": True, - "models": models, - "count": len(models) - } - except Exception as e: - logger.error(f"Error in /api/models/list: {e}") - return {"success": True, "models": [], "count": 0} - - -@app.post("/api/models/test") -async def test_model(payload: Dict[str, Any] = Body(...)): - """Test a specific model""" - try: - model_id = payload.get("model", "") - text = payload.get("text", "") - - from ai_models import ensemble_crypto_sentiment - result = ensemble_crypto_sentiment(text) - - return { - "success": True, - "model": model_id, - "result": result - } - except Exception as e: - logger.error(f"Error in /api/models/test: {e}") - return {"success": False, "error": str(e)} - - -# ===== WEBSOCKET ===== -@app.websocket("/ws") -async def websocket_endpoint(websocket: WebSocket): - """WebSocket endpoint for real-time updates""" - await ws_manager.connect(websocket) - - try: - while True: - # Check if connection is still open before sending - if websocket.client_state != WebSocketState.CONNECTED: - logger.info("WebSocket connection closed, breaking loop") - break - - # Send market updates every 10 seconds - try: - # Get latest data - top_coins = await market_collector.get_top_coins(limit=5) - news_items = await news_collector.get_latest_news(limit=3) - - # Compute global sentiment from news - from ai_models import ensemble_crypto_sentiment - news_texts = " ".join([n.get("title", "") for n in news_items]) - global_sentiment = ensemble_crypto_sentiment(news_texts) if news_texts else {"label": "neutral", "confidence": 0.5} - - payload = { - "market_data": top_coins, - "stats": { - "total_market_cap": sum([c.get("market_cap", 0) for c in top_coins]), - "sentiment": global_sentiment - }, - "news": news_items, - "sentiment": global_sentiment, - "timestamp": datetime.now().isoformat() - } - - # Double-check connection state before sending - if websocket.client_state == WebSocketState.CONNECTED: - await websocket.send_json({ - "type": "update", - "payload": payload - }) - else: - logger.info("WebSocket disconnected, breaking loop") - break - except CollectorError as e: - # Provider errors are already logged by the collector, just continue - logger.debug(f"Provider error in WebSocket update (this is expected with fallbacks): {e}") - # Use empty data on provider errors - payload = { - "market_data": [], - "stats": {"total_market_cap": 0, "sentiment": {"label": "neutral", "confidence": 0.5}}, - "news": [], - "sentiment": {"label": "neutral", "confidence": 0.5}, - "timestamp": datetime.now().isoformat() - } - except Exception as e: - # Log other errors with full details - error_msg = str(e) if str(e) else repr(e) - logger.error(f"Error in WebSocket update: {type(e).__name__}: {error_msg}") - # Don't break on data errors, just log and continue - # Only break on connection errors - if "send" in str(e).lower() or "close" in str(e).lower(): - break - - await asyncio.sleep(10) - except WebSocketDisconnect: - logger.info("WebSocket disconnect exception caught") - except Exception as e: - logger.error(f"WebSocket error: {e}") - finally: - try: - ws_manager.disconnect(websocket) - except: - pass - -@app.get("/api/market/history") -async def get_market_history(symbol: str = "BTC", limit: int = 10): - """ - Get historical prices from the local database if available. - - For this deployment we avoid touching the internal DatabaseManager - and simply report that no history API is wired yet. - """ - symbol = symbol.upper() - # We don't fabricate data here; if you need real history, it should - # be implemented via the shared database models. - return { - "symbol": symbol, - "history": [], - "count": 0, - "message": "History endpoint not wired to DB in this Space", - } - - - -@app.get("/api/status") -async def get_status(): - """ - System status endpoint used by the admin UI. - - This reports real-time information about providers and database, - without fabricating any market data. - """ - providers_cfg = load_providers_config() - providers = providers_cfg or {} - validated_count = sum(1 for p in providers.values() if p.get("validated")) - - db_path = DB_PATH - db_status = "connected" if db_path.exists() else "initializing" - - return { - "system_health": "healthy", - "timestamp": datetime.now().isoformat(), - "total_providers": len(providers), - "validated_providers": validated_count, - "database_status": db_status, - "apl_available": APL_REPORT_PATH.exists(), - "use_mock_data": False, - } - - -@app.get("/api/logs/recent") -async def get_recent_logs(): - """ - Return recent log lines for the admin UI. - - We read from the main server log file if available. - This does not fabricate content; if there are no logs, - an empty list is returned. - """ - log_file = LOG_DIR / "server.log" - lines = tail_log_file(log_file, max_lines=200) - # Wrap plain text lines as structured entries - logs = [{"line": line.rstrip("\n")} for line in lines] - return {"logs": logs, "count": len(logs)} - - -@app.get("/api/logs/errors") -async def get_error_logs(): - """ - Return recent error log lines from the same log file. - - This is a best-effort filter based on typical ERROR prefixes. - """ - log_file = LOG_DIR / "server.log" - lines = tail_log_file(log_file, max_lines=400) - error_lines = [line for line in lines if "ERROR" in line or "WARNING" in line] - logs = [{"line": line.rstrip("\n")} for line in error_lines[-200:]] - return {"errors": logs, "count": len(logs)} - - -def _load_apl_report() -> Optional[Dict[str, Any]]: - """Load the APL (Auto Provider Loader) validation report if available.""" - if not APL_REPORT_PATH.exists(): - return None - try: - with APL_REPORT_PATH.open("r", encoding="utf-8") as f: - return json.load(f) - except Exception as e: - logger.error(f"Error reading APL report: {e}") - return None - - -@app.get("/api/apl/summary") -async def get_apl_summary(): - """ - Summary of the Auto Provider Loader (APL) report. - - If the report is missing, we return a clear not_available status - instead of fabricating metrics. - """ - report = _load_apl_report() - if not report or "stats" not in report: - return { - "status": "not_available", - "message": "APL report not found", - } - - stats = report.get("stats", {}) - return { - "status": "ok", - "http_candidates": stats.get("total_http_candidates", 0), - "http_valid": stats.get("http_valid", 0), - "http_invalid": stats.get("http_invalid", 0), - "http_conditional": stats.get("http_conditional", 0), - "hf_candidates": stats.get("total_hf_candidates", 0), - "hf_valid": stats.get("hf_valid", 0), - "hf_invalid": stats.get("hf_invalid", 0), - "hf_conditional": stats.get("hf_conditional", 0), - "timestamp": datetime.now().isoformat(), - } - - -@app.get("/api/hf/models") -async def get_hf_models_from_apl(): - """ - Return the list of Hugging Face models discovered by the APL report. - - This is used by the admin UI. The data comes from the real - PROVIDER_AUTO_DISCOVERY_REPORT.json file if present. - """ - report = _load_apl_report() - if not report: - return {"models": [], "count": 0, "source": "none"} - - hf_models = report.get("hf_models", {}).get("results", []) - return { - "models": hf_models, - "count": len(hf_models), - "source": "APL report", - } - +"""Unified HuggingFace Space API Server leveraging shared collectors and AI helpers.""" + +import asyncio +import time +import os +import sys +import io + +# Fix encoding for Windows console (must be done before any print/logging) +if sys.platform == "win32": + try: + sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') + sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace') + except Exception: + pass # If already wrapped, ignore + +# Set environment variables to force PyTorch and avoid TensorFlow/Keras issues +os.environ.setdefault('TRANSFORMERS_NO_ADVISORY_WARNINGS', '1') +os.environ.setdefault('TRANSFORMERS_VERBOSITY', 'error') +os.environ.setdefault('TF_CPP_MIN_LOG_LEVEL', '3') # Suppress TensorFlow warnings +# Force PyTorch as default framework +os.environ.setdefault('TRANSFORMERS_FRAMEWORK', 'pt') + +from datetime import datetime, timedelta +from fastapi import Body, FastAPI, HTTPException, Query, WebSocket, WebSocketDisconnect +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import FileResponse, JSONResponse, HTMLResponse +from fastapi.staticfiles import StaticFiles +from starlette.websockets import WebSocketState +from typing import Any, Dict, List, Optional, Union +from statistics import mean +import logging +import random +import json +from pathlib import Path +import httpx + + +from ai_models import ( + analyze_chart_points, + analyze_crypto_sentiment, + analyze_market_text, + get_model_info, + initialize_models, + registry_status, +) +from backend.services.local_resource_service import LocalResourceService +from collectors.aggregator import ( + CollectorError, + MarketDataCollector, + NewsCollector, + ProviderStatusCollector, +) +from config import COIN_SYMBOL_MAPPING, get_settings + +# Setup logging +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +# Create FastAPI app +app = FastAPI( + title="Cryptocurrency Data & Analysis API", + description="Complete API for cryptocurrency data, market analysis, and trading signals", + version="3.0.0" +) + +# CORS +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +# Runtime state +START_TIME = time.time() +cache = {"ohlcv": {}, "prices": {}, "market_data": {}, "providers": [], "last_update": None} +settings = get_settings() +market_collector = MarketDataCollector() +news_collector = NewsCollector() +provider_collector = ProviderStatusCollector() + +# Load providers config +WORKSPACE_ROOT = Path(__file__).parent +PROVIDERS_CONFIG_PATH = settings.providers_config_path +FALLBACK_RESOURCE_PATH = WORKSPACE_ROOT / "crypto_resources_unified_2025-11-11.json" +LOG_DIR = WORKSPACE_ROOT / "logs" +APL_REPORT_PATH = WORKSPACE_ROOT / "PROVIDER_AUTO_DISCOVERY_REPORT.json" + +# Ensure log directory exists +LOG_DIR.mkdir(parents=True, exist_ok=True) + +# Database path (managed by DatabaseManager in the admin API) +DB_PATH = WORKSPACE_ROOT / "data" / "api_monitor.db" + +def tail_log_file(path: Path, max_lines: int = 200) -> List[str]: + """Return the last max_lines from a log file, if it exists.""" + if not path.exists(): + return [] + try: + with path.open("r", encoding="utf-8", errors="ignore") as f: + lines = f.readlines() + return lines[-max_lines:] + except Exception as e: + logger.error(f"Error reading log file {path}: {e}") + return [] + + +def load_providers_config(): + """Load providers from providers_config_extended.json""" + try: + if PROVIDERS_CONFIG_PATH.exists(): + with open(PROVIDERS_CONFIG_PATH, 'r', encoding='utf-8') as f: + config = json.load(f) + providers = config.get('providers', {}) + logger.info(f"Loaded {len(providers)} providers from providers_config_extended.json") + return providers + else: + logger.warning(f"providers_config_extended.json not found at {PROVIDERS_CONFIG_PATH}") + return {} + except Exception as e: + logger.error(f"Error loading providers config: {e}") + return {} + +# Load providers at startup +PROVIDERS_CONFIG = load_providers_config() +local_resource_service = LocalResourceService(FALLBACK_RESOURCE_PATH) + +HF_SAMPLE_NEWS = [ + { + "title": "Bitcoin holds key liquidity zone", + "source": "Fallback Ledger", + "sentiment": "positive", + "sentiment_score": 0.64, + "entities": ["BTC"], + "summary": "BTC consolidates near resistance with steady inflows", + }, + { + "title": "Ethereum staking demand remains resilient", + "source": "Fallback Ledger", + "sentiment": "neutral", + "sentiment_score": 0.12, + "entities": ["ETH"], + "summary": "Validator queue shortens as fees stabilize around L2 adoption", + }, + { + "title": "Solana ecosystem sees TVL uptick", + "source": "Fallback Ledger", + "sentiment": "positive", + "sentiment_score": 0.41, + "entities": ["SOL"], + "summary": "DeFi protocols move to Solana as mempool congestion drops", + }, +] + +# Mount static files (CSS, JS) +try: + static_path = WORKSPACE_ROOT / "static" + if static_path.exists(): + app.mount("/static", StaticFiles(directory=str(static_path)), name="static") + logger.info(f"Static files mounted from {static_path}") + else: + logger.warning(f"Static directory not found: {static_path}") +except Exception as e: + logger.error(f"Error mounting static files: {e}") + +# Mount api-resources for frontend access +try: + api_resources_path = WORKSPACE_ROOT / "api-resources" + if api_resources_path.exists(): + app.mount("/api-resources", StaticFiles(directory=str(api_resources_path)), name="api-resources") + logger.info(f"API resources mounted from {api_resources_path}") + else: + logger.warning(f"API resources directory not found: {api_resources_path}") +except Exception as e: + logger.error(f"Error mounting API resources: {e}") + +# ============================================================================ +# Helper utilities & Data Fetching Functions +# ============================================================================ + +def _normalize_asset_symbol(symbol: str) -> str: + symbol = (symbol or "").upper() + suffixes = ("USDT", "USD", "BTC", "ETH", "BNB") + for suffix in suffixes: + if symbol.endswith(suffix) and len(symbol) > len(suffix): + return symbol[: -len(suffix)] + return symbol + + +def _format_price_record(record: Dict[str, Any]) -> Dict[str, Any]: + price = record.get("price") or record.get("current_price") + change_pct = record.get("change_24h") or record.get("price_change_percentage_24h") + change_abs = None + if price is not None and change_pct is not None: + try: + change_abs = float(price) * float(change_pct) / 100.0 + except (TypeError, ValueError): + change_abs = None + + return { + "id": record.get("id") or record.get("symbol", "").lower(), + "symbol": record.get("symbol", "").upper(), + "name": record.get("name"), + "current_price": price, + "market_cap": record.get("market_cap"), + "market_cap_rank": record.get("rank"), + "total_volume": record.get("volume_24h") or record.get("total_volume"), + "price_change_24h": change_abs, + "price_change_percentage_24h": change_pct, + "high_24h": record.get("high_24h"), + "low_24h": record.get("low_24h"), + "last_updated": record.get("last_updated"), + } + + +async def fetch_binance_ohlcv(symbol: str = "BTCUSDT", interval: str = "1h", limit: int = 100): + """Fetch OHLCV data from Binance via the shared collector.""" + + try: + candles = await market_collector.get_ohlcv(symbol, interval, limit) + return [ + { + **candle, + "timestamp": int(datetime.fromisoformat(candle["timestamp"]).timestamp() * 1000), + "datetime": candle["timestamp"], + } + for candle in candles + ] + except CollectorError as exc: + logger.error("Error fetching OHLCV: %s", exc) + fallback_symbol = _normalize_asset_symbol(symbol) + fallback = local_resource_service.get_ohlcv(fallback_symbol, interval, limit) + if fallback: + return fallback + return [] + + +async def fetch_coingecko_prices(symbols: Optional[List[str]] = None, limit: int = 10): + """Fetch price snapshots using the shared market collector.""" + + source = "coingecko" + try: + if symbols: + tasks = [market_collector.get_coin_details(_normalize_asset_symbol(sym)) for sym in symbols] + results = await asyncio.gather(*tasks, return_exceptions=True) + coins: List[Dict[str, Any]] = [] + for result in results: + if isinstance(result, Exception): + continue + coins.append(_format_price_record(result)) + if coins: + return coins, source + else: + top = await market_collector.get_top_coins(limit=limit) + formatted = [_format_price_record(entry) for entry in top] + if formatted: + return formatted, source + except CollectorError as exc: + logger.error("Error fetching aggregated prices: %s", exc) + + fallback = ( + local_resource_service.get_prices_for_symbols([sym for sym in symbols or []]) + if symbols + else local_resource_service.get_top_prices(limit) + ) + if fallback: + return fallback, "local-fallback" + return [], source + + +async def fetch_binance_ticker(symbol: str): + """Provide ticker-like information sourced from CoinGecko market data.""" + + try: + coin = await market_collector.get_coin_details(_normalize_asset_symbol(symbol)) + except CollectorError as exc: + logger.error("Unable to load ticker for %s: %s", symbol, exc) + coin = None + + if coin: + price = coin.get("price") + change_pct = coin.get("change_24h") or 0.0 + change_abs = price * change_pct / 100 if price is not None and change_pct is not None else None + return { + "symbol": symbol.upper(), + "price": price, + "price_change_24h": change_abs, + "price_change_percent_24h": change_pct, + "high_24h": coin.get("high_24h"), + "low_24h": coin.get("low_24h"), + "volume_24h": coin.get("volume_24h"), + "quote_volume_24h": coin.get("volume_24h"), + }, "binance" + + fallback_symbol = _normalize_asset_symbol(symbol) + fallback = local_resource_service.get_ticker_snapshot(fallback_symbol) + if fallback: + fallback["symbol"] = symbol.upper() + return fallback, "local-fallback" + return None, "binance" + + +# ============================================================================ +# Core Endpoints +# ============================================================================ + +@app.get("/health") +async def health(): + """System health check using shared collectors.""" + + async def _safe_call(coro): + try: + data = await coro + return {"status": "ok", "count": len(data) if hasattr(data, "__len__") else 1} + except Exception as exc: # pragma: no cover - network heavy + return {"status": "error", "detail": str(exc)} + + market_task = asyncio.create_task(_safe_call(market_collector.get_top_coins(limit=3))) + news_task = asyncio.create_task(_safe_call(news_collector.get_latest_news(limit=3))) + providers_task = asyncio.create_task(_safe_call(provider_collector.get_providers_status())) + + market_status, news_status, providers_status = await asyncio.gather( + market_task, news_task, providers_task + ) + + ai_status = registry_status() + service_states = { + "market_data": market_status, + "news": news_status, + "providers": providers_status, + "ai_models": ai_status, + } + + degraded = any(state.get("status") != "ok" for state in (market_status, news_status, providers_status)) + overall = "healthy" if not degraded else "degraded" + + return { + "status": overall, + "service": "cryptocurrency-data-api", + "timestamp": datetime.utcnow().isoformat(), + "version": app.version, + "providers_loaded": market_status.get("count", 0), + "services": service_states, + } + + +@app.get("/info") +async def info(): + """System information""" + hf_providers = [p for p in PROVIDERS_CONFIG.keys() if "huggingface_space" in p] + + return { + "service": "Cryptocurrency Data & Analysis API", + "version": app.version, + "endpoints": { + "core": ["/health", "/info", "/api/providers"], + "data": ["/api/ohlcv", "/api/crypto/prices/top", "/api/crypto/price/{symbol}", "/api/crypto/market-overview"], + "analysis": ["/api/analysis/signals", "/api/analysis/smc", "/api/scoring/snapshot"], + "market": ["/api/market/prices", "/api/market-data/prices"], + "system": ["/api/system/status", "/api/system/config"], + "huggingface": ["/api/hf/health", "/api/hf/refresh", "/api/hf/registry", "/api/hf/run-sentiment"], + }, + "data_sources": ["Binance", "CoinGecko", "CoinPaprika", "CoinCap"], + "providers_loaded": len(PROVIDERS_CONFIG), + "huggingface_space_providers": len(hf_providers), + "features": [ + "Real-time price data", + "OHLCV historical data", + "Trading signals", + "Market analysis", + "Sentiment analysis", + "HuggingFace model integration", + f"{len(PROVIDERS_CONFIG)} providers from providers_config_extended.json", + ], + "ai_registry": registry_status(), + } + + +@app.get("/api/providers") +async def get_providers(): + """Get list of API providers and their health.""" + + try: + statuses = await provider_collector.get_providers_status() + except Exception as exc: # pragma: no cover - network heavy + logger.error("Error getting providers: %s", exc) + raise HTTPException(status_code=503, detail=str(exc)) + + providers_list = [] + for status in statuses: + meta = PROVIDERS_CONFIG.get(status["provider_id"], {}) + providers_list.append( + { + **status, + "base_url": meta.get("base_url"), + "requires_auth": meta.get("requires_auth"), + "priority": meta.get("priority"), + } + ) + + return { + "providers": providers_list, + "total": len(providers_list), + "source": str(PROVIDERS_CONFIG_PATH), + "last_updated": datetime.utcnow().isoformat(), + } + + +@app.get("/api/providers/{provider_id}/health") +async def get_provider_health(provider_id: str): + """Get health status for a specific provider.""" + + # Check if provider exists in config + provider_config = PROVIDERS_CONFIG.get(provider_id) + if not provider_config: + raise HTTPException(status_code=404, detail=f"Provider '{provider_id}' not found") + + try: + # Perform health check using the collector + async with httpx.AsyncClient(timeout=provider_collector.timeout, headers=provider_collector.headers) as client: + health_result = await provider_collector._check_provider(client, provider_id, provider_config) + + # Add metadata from config + health_result.update({ + "base_url": provider_config.get("base_url"), + "requires_auth": provider_config.get("requires_auth"), + "priority": provider_config.get("priority"), + "category": provider_config.get("category"), + "last_checked": datetime.utcnow().isoformat() + }) + + return health_result + except Exception as exc: # pragma: no cover - network heavy + logger.error("Error checking provider health for %s: %s", provider_id, exc) + raise HTTPException(status_code=503, detail=f"Health check failed: {str(exc)}") + + +@app.get("/api/providers/config") +async def get_providers_config(): + """Get providers configuration in format expected by frontend.""" + try: + return { + "success": True, + "providers": PROVIDERS_CONFIG, + "total": len(PROVIDERS_CONFIG), + "source": str(PROVIDERS_CONFIG_PATH), + "last_updated": datetime.utcnow().isoformat() + } + except Exception as exc: + logger.error("Error getting providers config: %s", exc) + raise HTTPException(status_code=500, detail=str(exc)) + + +# ============================================================================ +# OHLCV Data Endpoint +# ============================================================================ + +@app.get("/api/ohlcv") +async def get_ohlcv( + symbol: str = Query("BTCUSDT", description="Trading pair symbol"), + interval: str = Query("1h", description="Time interval (1m, 5m, 15m, 1h, 4h, 1d)"), + limit: int = Query(100, ge=1, le=1000, description="Number of candles") +): + """ + Get OHLCV (candlestick) data for a trading pair + + Supported intervals: 1m, 5m, 15m, 30m, 1h, 4h, 1d + """ + try: + # Check cache + cache_key = f"{symbol}_{interval}_{limit}" + if cache_key in cache["ohlcv"]: + cached_data, cached_time = cache["ohlcv"][cache_key] + if (datetime.now() - cached_time).seconds < 60: # 60s cache + return {"symbol": symbol, "interval": interval, "data": cached_data, "source": "cache"} + + # Fetch from Binance + ohlcv_data = await fetch_binance_ohlcv(symbol, interval, limit) + + if ohlcv_data: + # Update cache + cache["ohlcv"][cache_key] = (ohlcv_data, datetime.now()) + + return { + "symbol": symbol, + "interval": interval, + "count": len(ohlcv_data), + "data": ohlcv_data, + "source": "binance", + "timestamp": datetime.now().isoformat() + } + else: + raise HTTPException(status_code=503, detail="Unable to fetch OHLCV data") + + except HTTPException: + raise + except Exception as e: + logger.error(f"Error in get_ohlcv: {e}") + raise HTTPException(status_code=500, detail=str(e)) + + +# ============================================================================ +# Crypto Prices Endpoints +# ============================================================================ + +@app.get("/api/crypto/prices/top") +async def get_top_prices(limit: int = Query(10, ge=1, le=100, description="Number of top cryptocurrencies")): + """Get top cryptocurrencies by market cap""" + try: + # Check cache + cache_key = f"top_{limit}" + if cache_key in cache["prices"]: + cached_data, cached_time = cache["prices"][cache_key] + if (datetime.now() - cached_time).seconds < 60: + return {"data": cached_data, "source": "cache"} + + # Fetch from CoinGecko + prices, source = await fetch_coingecko_prices(limit=limit) + + if prices: + # Update cache + cache["prices"][cache_key] = (prices, datetime.now()) + + return { + "count": len(prices), + "data": prices, + "source": source, + "timestamp": datetime.now().isoformat() + } + else: + raise HTTPException(status_code=503, detail="Unable to fetch price data") + + except HTTPException: + raise + except Exception as e: + logger.error(f"Error in get_top_prices: {e}") + raise HTTPException(status_code=500, detail=str(e)) + + +@app.get("/api/crypto/price/{symbol}") +async def get_single_price(symbol: str): + """Get price for a single cryptocurrency""" + try: + # Try Binance first for common pairs + binance_symbol = f"{symbol.upper()}USDT" + ticker, ticker_source = await fetch_binance_ticker(binance_symbol) + + if ticker: + return { + "symbol": symbol.upper(), + "price": ticker, + "source": ticker_source, + "timestamp": datetime.now().isoformat() + } + + # Fallback to CoinGecko + prices, source = await fetch_coingecko_prices([symbol]) + if prices: + return { + "symbol": symbol.upper(), + "price": prices[0], + "source": source, + "timestamp": datetime.now().isoformat() + } + + raise HTTPException(status_code=404, detail=f"Price data not found for {symbol}") + + except HTTPException: + raise + except Exception as e: + logger.error(f"Error in get_single_price: {e}") + raise HTTPException(status_code=500, detail=str(e)) + + +@app.get("/api/crypto/market-overview") +async def get_market_overview(): + """Get comprehensive market overview""" + try: + # Fetch top 20 coins + prices, source = await fetch_coingecko_prices(limit=20) + + if not prices: + raise HTTPException(status_code=503, detail="Unable to fetch market data") + + # Calculate market stats + # Try multiple field names for market cap and volume + total_market_cap = 0 + total_volume = 0 + + for p in prices: + # Try different field names for market cap + market_cap = ( + p.get("market_cap") or + p.get("market_cap_usd") or + p.get("market_cap_rank") or # Sometimes this is the value + None + ) + # If market_cap is not found, try calculating from price and supply + if not market_cap: + price = p.get("price") or p.get("current_price") or 0 + supply = p.get("circulating_supply") or p.get("total_supply") or 0 + if price and supply: + market_cap = float(price) * float(supply) + + if market_cap: + try: + total_market_cap += float(market_cap) + except (TypeError, ValueError): + pass + + # Try different field names for volume + volume = ( + p.get("total_volume") or + p.get("volume_24h") or + p.get("volume_24h_usd") or + None + ) + if volume: + try: + total_volume += float(volume) + except (TypeError, ValueError): + pass + + logger.info(f"Market overview: {len(prices)} coins, total_market_cap={total_market_cap:,.0f}, total_volume={total_volume:,.0f}") + + # Sort by 24h change + gainers = sorted( + [p for p in prices if p.get("price_change_percentage_24h")], + key=lambda x: x.get("price_change_percentage_24h", 0), + reverse=True + )[:5] + + losers = sorted( + [p for p in prices if p.get("price_change_percentage_24h")], + key=lambda x: x.get("price_change_percentage_24h", 0) + )[:5] + + return { + "total_market_cap": total_market_cap, + "total_volume_24h": total_volume, + "btc_dominance": (prices[0].get("market_cap", 0) / total_market_cap * 100) if total_market_cap > 0 else 0, + "top_gainers": gainers, + "top_losers": losers, + "top_by_volume": sorted(prices, key=lambda x: x.get("total_volume", 0) or 0, reverse=True)[:5], + "timestamp": datetime.now().isoformat(), + "source": source + } + + except HTTPException: + raise + except Exception as e: + logger.error(f"Error in get_market_overview: {e}") + raise HTTPException(status_code=500, detail=str(e)) + + +@app.get("/api/market") +async def get_market(): + """Get market data in format expected by frontend dashboard""" + try: + overview = await get_market_overview() + prices, source = await fetch_coingecko_prices(limit=50) + + if not prices: + raise HTTPException(status_code=503, detail="Unable to fetch market data") + + return { + "total_market_cap": overview.get("total_market_cap", 0), + "btc_dominance": overview.get("btc_dominance", 0), + "total_volume_24h": overview.get("total_volume_24h", 0), + "cryptocurrencies": prices, + "timestamp": datetime.now().isoformat(), + "source": source + } + except HTTPException: + raise + except Exception as e: + logger.error(f"Error in get_market: {e}") + raise HTTPException(status_code=500, detail=str(e)) + + +@app.get("/api/trending") +async def get_trending(): + """Get trending cryptocurrencies (top gainers by 24h change)""" + try: + prices, source = await fetch_coingecko_prices(limit=100) + + if not prices: + raise HTTPException(status_code=503, detail="Unable to fetch trending data") + + trending = sorted( + [p for p in prices if p.get("price_change_percentage_24h") is not None], + key=lambda x: x.get("price_change_percentage_24h", 0), + reverse=True + )[:10] + + return { + "trending": trending, + "count": len(trending), + "timestamp": datetime.now().isoformat(), + "source": source + } + except HTTPException: + raise + except Exception as e: + logger.error(f"Error in get_trending: {e}") + raise HTTPException(status_code=500, detail=str(e)) + + +@app.get("/api/market/prices") +async def get_multiple_prices(symbols: str = Query("BTC,ETH,SOL", description="Comma-separated symbols")): + """Get prices for multiple cryptocurrencies""" + try: + symbol_list = [s.strip().upper() for s in symbols.split(",")] + + # Fetch prices + prices_data = [] + source = "binance" + for symbol in symbol_list: + try: + ticker, ticker_source = await fetch_binance_ticker(f"{symbol}USDT") + if ticker: + prices_data.append(ticker) + if ticker_source != "binance": + source = ticker_source + except: + continue + if not prices_data: + # Fallback to CoinGecko + prices_data, source = await fetch_coingecko_prices(symbol_list) + + if not prices_data: + fallback_prices = local_resource_service.get_prices_for_symbols(symbol_list) + if fallback_prices: + prices_data = fallback_prices + source = "local-fallback" + + return { + "symbols": symbol_list, + "count": len(prices_data), + "data": prices_data, + "source": source, + "timestamp": datetime.now().isoformat() + } + + except Exception as e: + logger.error(f"Error in get_multiple_prices: {e}") + raise HTTPException(status_code=500, detail=str(e)) + + +@app.get("/api/market-data/prices") +async def get_market_data_prices(symbols: str = Query("BTC,ETH", description="Comma-separated symbols")): + """Alternative endpoint for market data prices""" + return await get_multiple_prices(symbols) + + +# ============================================================================ +# Analysis Endpoints +# ============================================================================ + +@app.get("/api/analysis/signals") +async def get_trading_signals( + symbol: str = Query("BTCUSDT", description="Trading pair"), + timeframe: str = Query("1h", description="Timeframe") +): + """Get trading signals for a symbol""" + try: + # Fetch OHLCV data for analysis + ohlcv = await fetch_binance_ohlcv(symbol, timeframe, 100) + + if not ohlcv: + raise HTTPException(status_code=503, detail="Unable to fetch data for analysis") + + # Simple signal generation (can be enhanced) + latest = ohlcv[-1] + prev = ohlcv[-2] if len(ohlcv) > 1 else latest + + # Calculate simple indicators + close_prices = [c["close"] for c in ohlcv[-20:]] + sma_20 = sum(close_prices) / len(close_prices) + + # Generate signal + trend = "bullish" if latest["close"] > sma_20 else "bearish" + momentum = "strong" if abs(latest["close"] - prev["close"]) / prev["close"] > 0.01 else "weak" + + signal = "buy" if trend == "bullish" and momentum == "strong" else ( + "sell" if trend == "bearish" and momentum == "strong" else "hold" + ) + + ai_summary = analyze_chart_points(symbol, timeframe, ohlcv) + + return { + "symbol": symbol, + "timeframe": timeframe, + "signal": signal, + "trend": trend, + "momentum": momentum, + "indicators": { + "sma_20": sma_20, + "current_price": latest["close"], + "price_change": latest["close"] - prev["close"], + "price_change_percent": ((latest["close"] - prev["close"]) / prev["close"]) * 100 + }, + "analysis": ai_summary, + "timestamp": datetime.now().isoformat() + } + + except HTTPException: + raise + except Exception as e: + logger.error(f"Error in get_trading_signals: {e}") + raise HTTPException(status_code=500, detail=str(e)) + + +@app.get("/api/analysis/smc") +async def get_smc_analysis(symbol: str = Query("BTCUSDT", description="Trading pair")): + """Get Smart Money Concepts (SMC) analysis""" + try: + # Fetch OHLCV data + ohlcv = await fetch_binance_ohlcv(symbol, "1h", 200) + + if not ohlcv: + raise HTTPException(status_code=503, detail="Unable to fetch data") + + # Calculate key levels + highs = [c["high"] for c in ohlcv] + lows = [c["low"] for c in ohlcv] + closes = [c["close"] for c in ohlcv] + + resistance = max(highs[-50:]) + support = min(lows[-50:]) + current_price = closes[-1] + + # Structure analysis + market_structure = "higher_highs" if closes[-1] > closes[-10] > closes[-20] else "lower_lows" + + return { + "symbol": symbol, + "market_structure": market_structure, + "key_levels": { + "resistance": resistance, + "support": support, + "current_price": current_price, + "mid_point": (resistance + support) / 2 + }, + "order_blocks": { + "bullish": support, + "bearish": resistance + }, + "liquidity_zones": { + "above": resistance, + "below": support + }, + "timestamp": datetime.now().isoformat() + } + + except HTTPException: + raise + except Exception as e: + logger.error(f"Error in get_smc_analysis: {e}") + raise HTTPException(status_code=500, detail=str(e)) + + +@app.get("/api/scoring/snapshot") +async def get_scoring_snapshot(symbol: str = Query("BTCUSDT", description="Trading pair")): + """Get comprehensive scoring snapshot""" + try: + # Fetch data + ticker, _ = await fetch_binance_ticker(symbol) + ohlcv = await fetch_binance_ohlcv(symbol, "1h", 100) + + if not ticker or not ohlcv: + raise HTTPException(status_code=503, detail="Unable to fetch data") + + # Calculate scores (0-100) + volatility_score = min(abs(ticker["price_change_percent_24h"]) * 5, 100) + volume_score = min((ticker["volume_24h"] / 1000000) * 10, 100) + trend_score = 50 + (ticker["price_change_percent_24h"] * 2) + + # Overall score + overall_score = (volatility_score + volume_score + trend_score) / 3 + + return { + "symbol": symbol, + "overall_score": round(overall_score, 2), + "scores": { + "volatility": round(volatility_score, 2), + "volume": round(volume_score, 2), + "trend": round(trend_score, 2), + "momentum": round(50 + ticker["price_change_percent_24h"], 2) + }, + "rating": "excellent" if overall_score > 80 else ( + "good" if overall_score > 60 else ( + "average" if overall_score > 40 else "poor" + ) + ), + "timestamp": datetime.now().isoformat() + } + + except HTTPException: + raise + except Exception as e: + logger.error(f"Error in get_scoring_snapshot: {e}") + raise HTTPException(status_code=500, detail=str(e)) + + +@app.get("/api/signals") +async def get_all_signals(): + """Get signals for multiple assets""" + symbols = ["BTCUSDT", "ETHUSDT", "SOLUSDT", "BNBUSDT"] + signals = [] + + for symbol in symbols: + try: + signal_data = await get_trading_signals(symbol, "1h") + signals.append(signal_data) + except: + continue + + return { + "count": len(signals), + "signals": signals, + "timestamp": datetime.now().isoformat() + } + + +@app.get("/api/sentiment") +async def get_sentiment(): + """Get market sentiment data""" + try: + news = await news_collector.get_latest_news(limit=5) + except CollectorError as exc: + logger.warning("Sentiment fallback due to news error: %s", exc) + news = [] + + text = " ".join(item.get("title", "") for item in news).strip() or "Crypto market update" + analysis = analyze_market_text(text) + score = analysis.get("signals", {}).get("crypto", {}).get("score", 0.0) + normalized_value = int((score + 1) * 50) + + if normalized_value < 20: + classification = "extreme_fear" + elif normalized_value < 40: + classification = "fear" + elif normalized_value < 60: + classification = "neutral" + elif normalized_value < 80: + classification = "greed" + else: + classification = "extreme_greed" + + return { + "value": normalized_value, + "classification": classification, + "description": f"Market sentiment is {classification.replace('_', ' ')}", + "analysis": analysis, + "timestamp": datetime.utcnow().isoformat(), + } + + +# ============================================================================ +# System Endpoints +# ============================================================================ + +@app.get("/api/system/status") +async def get_system_status(): + """Get system status""" + providers = await provider_collector.get_providers_status() + online = sum(1 for provider in providers if provider.get("status") == "online") + + cache_items = ( + len(getattr(market_collector.cache, "_store", {})) + + len(getattr(news_collector.cache, "_store", {})) + + len(getattr(provider_collector.cache, "_store", {})) + ) + + return { + "status": "operational" if online else "maintenance", + "uptime_seconds": round(time.time() - START_TIME, 2), + "cache_size": cache_items, + "providers_online": online, + "requests_per_minute": 0, + "timestamp": datetime.utcnow().isoformat(), + } + + +@app.get("/api/system/config") +async def get_system_config(): + """Get system configuration""" + return { + "version": app.version, + "api_version": "v1", + "cache_ttl_seconds": settings.cache_ttl, + "supported_symbols": sorted(set(COIN_SYMBOL_MAPPING.values())), + "supported_intervals": ["1m", "5m", "15m", "30m", "1h", "4h", "1d"], + "max_ohlcv_limit": 1000, + "timestamp": datetime.utcnow().isoformat(), + } + + +@app.get("/api/categories") +async def get_categories(): + """Get data categories""" + return { + "categories": [ + {"name": "market_data", "endpoints": 5, "status": "active"}, + {"name": "analysis", "endpoints": 4, "status": "active"}, + {"name": "signals", "endpoints": 2, "status": "active"}, + {"name": "sentiment", "endpoints": 1, "status": "active"} + ] + } + + +@app.get("/api/rate-limits") +async def get_rate_limits(): + """Get rate limit information""" + return { + "rate_limits": [ + {"endpoint": "/api/ohlcv", "limit": 1200, "window": "per_minute"}, + {"endpoint": "/api/crypto/prices/top", "limit": 600, "window": "per_minute"}, + {"endpoint": "/api/analysis/*", "limit": 300, "window": "per_minute"} + ], + "current_usage": { + "requests_this_minute": 0, + "percentage": 0 + } + } + + +@app.get("/api/logs") +async def get_logs(limit: int = Query(50, ge=1, le=500)): + """Get recent API logs""" + # Mock logs (can be enhanced with real logging) + logs = [] + for i in range(min(limit, 10)): + logs.append({ + "timestamp": (datetime.now() - timedelta(minutes=i)).isoformat(), + "endpoint": "/api/ohlcv", + "status": "success", + "response_time_ms": random.randint(50, 200) + }) + + return {"logs": logs, "count": len(logs)} + + +@app.get("/api/alerts") +async def get_alerts(): + """Get system alerts""" + return { + "alerts": [], + "count": 0, + "timestamp": datetime.now().isoformat() + } + + +# ============================================================================ +# HuggingFace Integration Endpoints +# ============================================================================ + +@app.get("/api/hf/health") +async def hf_health(): + """HuggingFace integration health""" + from ai_models import AI_MODELS_SUMMARY + status = registry_status() + status["models"] = AI_MODELS_SUMMARY + status["timestamp"] = datetime.utcnow().isoformat() + return status + + +@app.post("/api/hf/refresh") +async def hf_refresh(): + """Refresh HuggingFace data""" + from ai_models import initialize_models + result = initialize_models() + return {"status": "ok" if result.get("models_loaded", 0) > 0 else "degraded", **result, "timestamp": datetime.utcnow().isoformat()} + + +@app.get("/api/hf/registry") +async def hf_registry(kind: str = "models"): + """Get HuggingFace registry""" + info = get_model_info() + return {"kind": kind, "items": info.get("model_names", info)} + + +@app.get("/api/resources/unified") +async def get_unified_resources(): + """Get unified API resources from crypto_resources_unified_2025-11-11.json""" + try: + data = local_resource_service.get_registry() + if data: + metadata = data.get("registry", {}).get("metadata", {}) + return { + "success": True, + "data": data, + "metadata": metadata, + "count": metadata.get("total_entries", 0), + "fallback_assets": len(local_resource_service.get_supported_symbols()) + } + return {"success": False, "error": "Resources file not found"} + except Exception as e: + logger.error(f"Error loading unified resources: {e}") + return {"success": False, "error": str(e)} + + +@app.get("/api/resources/ultimate") +async def get_ultimate_resources(): + """Get ultimate API resources from ultimate_crypto_pipeline_2025_NZasinich.json""" + try: + resources_path = WORKSPACE_ROOT / "api-resources" / "ultimate_crypto_pipeline_2025_NZasinich.json" + if resources_path.exists(): + with open(resources_path, 'r', encoding='utf-8') as f: + data = json.load(f) + return { + "success": True, + "data": data, + "total_sources": data.get("total_sources", 0), + "files": len(data.get("files", [])) + } + return {"success": False, "error": "Resources file not found"} + except Exception as e: + logger.error(f"Error loading ultimate resources: {e}") + return {"success": False, "error": str(e)} + + +@app.get("/api/resources/stats") +async def get_resources_stats(): + """Get statistics about available API resources""" + try: + stats = { + "unified": {"available": False, "count": 0}, + "ultimate": {"available": False, "count": 0}, + "total_apis": 0 + } + + # Check unified resources via the centralized loader + registry = local_resource_service.get_registry() + if registry: + stats["unified"] = { + "available": True, + "count": registry.get("registry", {}).get("metadata", {}).get("total_entries", 0), + "fallback_assets": len(local_resource_service.get_supported_symbols()) + } + + # Check ultimate resources + ultimate_path = WORKSPACE_ROOT / "api-resources" / "ultimate_crypto_pipeline_2025_NZasinich.json" + if ultimate_path.exists(): + with open(ultimate_path, 'r', encoding='utf-8') as f: + ultimate_data = json.load(f) + stats["ultimate"] = { + "available": True, + "count": ultimate_data.get("total_sources", 0) + } + + stats["total_apis"] = stats["unified"].get("count", 0) + stats["ultimate"].get("count", 0) + + return {"success": True, "stats": stats} + except Exception as e: + logger.error(f"Error getting resources stats: {e}") + return {"success": False, "error": str(e)} + + +def _resolve_sentiment_payload(payload: Union[List[str], Dict[str, Any]]) -> Dict[str, Any]: + if isinstance(payload, list): + return {"texts": payload, "mode": "auto"} + if isinstance(payload, dict): + texts = payload.get("texts") or payload.get("text") + if isinstance(texts, str): + texts = [texts] + if not isinstance(texts, list): + raise ValueError("texts must be provided") + mode = payload.get("mode") or payload.get("model") or "auto" + return {"texts": texts, "mode": mode} + raise ValueError("Invalid payload") + + +@app.post("/api/hf/run-sentiment") +@app.post("/api/hf/sentiment") +async def hf_sentiment(payload: Union[List[str], Dict[str, Any]] = Body(...)): + """Run sentiment analysis using shared AI helpers.""" + from ai_models import AI_MODELS_SUMMARY + + if AI_MODELS_SUMMARY.get("models_loaded", 0) == 0 or AI_MODELS_SUMMARY.get("mode") == "off": + return { + "ok": False, + "error": "No HF models are currently loaded.", + "mode": AI_MODELS_SUMMARY.get("mode", "off"), + "models_loaded": AI_MODELS_SUMMARY.get("models_loaded", 0) + } + + try: + resolved = _resolve_sentiment_payload(payload) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + mode = (resolved.get("mode") or "auto").lower() + texts = resolved["texts"] + results: List[Dict[str, Any]] = [] + for text in texts: + if mode == "crypto": + analysis = analyze_crypto_sentiment(text) + elif mode == "financial": + analysis = analyze_market_text(text).get("signals", {}).get("financial", {}) + elif mode == "social": + analysis = analyze_market_text(text).get("signals", {}).get("social", {}) + else: + analysis = analyze_market_text(text) + results.append({"text": text, "result": analysis}) + + return {"mode": mode, "results": results, "timestamp": datetime.utcnow().isoformat()} + + +@app.post("/api/hf/models/sentiment") +async def hf_models_sentiment(payload: Union[List[str], Dict[str, Any]] = Body(...)): + """Compatibility endpoint for HF console sentiment panel.""" + from ai_models import AI_MODELS_SUMMARY + + if AI_MODELS_SUMMARY.get("models_loaded", 0) == 0 or AI_MODELS_SUMMARY.get("mode") == "off": + return { + "ok": False, + "error": "No HF models are currently loaded.", + "mode": AI_MODELS_SUMMARY.get("mode", "off"), + "models_loaded": AI_MODELS_SUMMARY.get("models_loaded", 0) + } + + return await hf_sentiment(payload) + + +@app.post("/api/hf/models/forecast") +async def hf_models_forecast(payload: Dict[str, Any] = Body(...)): + """Generate quick technical forecasts from provided closing prices.""" + series = payload.get("series") or payload.get("values") or payload.get("close") + if not isinstance(series, list) or len(series) < 3: + raise HTTPException(status_code=400, detail="Provide at least 3 closing prices in 'series'.") + + try: + floats = [float(x) for x in series] + except (TypeError, ValueError) as exc: + raise HTTPException(status_code=400, detail="Series must contain numeric values") from exc + + model_name = (payload.get("model") or payload.get("model_name") or "btc_lstm").lower() + steps = int(payload.get("steps") or 3) + + deltas = [floats[i] - floats[i - 1] for i in range(1, len(floats))] + avg_delta = mean(deltas) + volatility = mean(abs(delta - avg_delta) for delta in deltas) if deltas else 0 + + predictions = [] + last = floats[-1] + decay = 0.95 if model_name == "btc_arima" else 1.02 + for _ in range(steps): + last = last + (avg_delta * decay) + predictions.append(round(last, 4)) + + return { + "model": model_name, + "steps": steps, + "input_count": len(floats), + "volatility": round(volatility, 5), + "predictions": predictions, + "source": "local-fallback" if model_name == "btc_arima" else "hybrid", + "timestamp": datetime.utcnow().isoformat() + } + + +@app.get("/api/hf/datasets/market/ohlcv") +async def HF_TOKEN_FROM_SPACE_SECRET(symbol: str = Query("BTC"), interval: str = Query("1h"), limit: int = Query(120, ge=10, le=500)): + """Expose fallback OHLCV snapshots as a pseudo HF dataset slice.""" + data = local_resource_service.get_ohlcv(symbol.upper(), interval, limit) + source = "local-fallback" + + if not data: + return { + "symbol": symbol.upper(), + "interval": interval, + "count": 0, + "data": [], + "source": source, + "message": "No cached OHLCV available yet" + } + + return { + "symbol": symbol.upper(), + "interval": interval, + "count": len(data), + "data": data, + "source": source, + "timestamp": datetime.utcnow().isoformat() + } + + +@app.get("/api/hf/datasets/market/btc_technical") +async def hf_dataset_market_btc(limit: int = Query(50, ge=10, le=200)): + """Simplified technical metrics derived from fallback OHLCV data.""" + candles = local_resource_service.get_ohlcv("BTC", "1h", limit + 20) + + if not candles: + raise HTTPException(status_code=503, detail="Fallback OHLCV unavailable") + + rows = [] + closes = [c["close"] for c in candles] + for idx, candle in enumerate(candles[-limit:]): + window = closes[max(0, idx): idx + 20] + sma = sum(window) / len(window) if window else candle["close"] + momentum = candle["close"] - candle["open"] + rows.append({ + "timestamp": candle["timestamp"], + "datetime": candle["datetime"], + "close": candle["close"], + "sma_20": round(sma, 4), + "momentum": round(momentum, 4), + "volatility": round((candle["high"] - candle["low"]) / candle["low"], 4) + }) + + return { + "symbol": "BTC", + "interval": "1h", + "count": len(rows), + "items": rows, + "source": "local-fallback" + } + + +@app.get("/api/hf/datasets/news/semantic") +async def hf_dataset_news(limit: int = Query(10, ge=3, le=25)): + """News slice augmented with sentiment tags for HF demos.""" + try: + news = await news_collector.get_latest_news(limit=limit) + source = "providers" + except CollectorError: + news = [] + source = "local-fallback" + + if not news: + items = HF_SAMPLE_NEWS[:limit] + else: + items = [] + for item in news: + items.append({ + "title": item.get("title"), + "source": item.get("source") or item.get("provider"), + "sentiment": item.get("sentiment") or "neutral", + "sentiment_score": item.get("sentiment_confidence", 0.5), + "entities": item.get("symbols") or [], + "summary": item.get("summary") or item.get("description"), + "published_at": item.get("date") or item.get("published_at") + }) + return { + "count": len(items), + "items": items, + "source": source, + "timestamp": datetime.utcnow().isoformat() + } + + +# ============================================================================ +# HTML Routes - Serve UI files +# ============================================================================ + +@app.get("/favicon.ico") +async def favicon(): + """Serve favicon""" + favicon_path = WORKSPACE_ROOT / "static" / "favicon.ico" + if favicon_path.exists(): + return FileResponse(favicon_path) + return JSONResponse({"status": "no favicon"}, status_code=404) + +@app.get("/", response_class=HTMLResponse) +async def root(): + """Serve main HTML UI page (index.html)""" + index_path = WORKSPACE_ROOT / "index.html" + if index_path.exists(): + return FileResponse( + path=str(index_path), + media_type="text/html", + filename="index.html" + ) + return HTMLResponse("

Cryptocurrency Data & Analysis API

See /docs for API documentation

") + +@app.get("/index.html", response_class=HTMLResponse) +async def index(): + """Serve index.html""" + return FileResponse(WORKSPACE_ROOT / "index.html") + +@app.get("/dashboard.html", response_class=HTMLResponse) +async def dashboard(): + """Serve dashboard.html""" + return FileResponse(WORKSPACE_ROOT / "dashboard.html") + +@app.get("/dashboard", response_class=HTMLResponse) +async def dashboard_alt(): + """Alternative route for dashboard""" + return FileResponse(WORKSPACE_ROOT / "dashboard.html") + +@app.get("/admin.html", response_class=HTMLResponse) +async def admin(): + """Serve admin panel""" + admin_path = WORKSPACE_ROOT / "admin.html" + if admin_path.exists(): + return FileResponse( + path=str(admin_path), + media_type="text/html", + filename="admin.html" + ) + return HTMLResponse("

Admin panel not found

") + +@app.get("/admin", response_class=HTMLResponse) +async def admin_alt(): + """Alternative route for admin""" + admin_path = WORKSPACE_ROOT / "admin.html" + if admin_path.exists(): + return FileResponse( + path=str(admin_path), + media_type="text/html", + filename="admin.html" + ) + return HTMLResponse("

Admin panel not found

") + +@app.get("/hf_console.html", response_class=HTMLResponse) +async def hf_console(): + """Serve HuggingFace console""" + return FileResponse(WORKSPACE_ROOT / "hf_console.html") + +@app.get("/console", response_class=HTMLResponse) +async def console_alt(): + """Alternative route for HF console""" + return FileResponse(WORKSPACE_ROOT / "hf_console.html") + +@app.get("/pool_management.html", response_class=HTMLResponse) +async def pool_management(): + """Serve pool management UI""" + return FileResponse(WORKSPACE_ROOT / "pool_management.html") + +@app.get("/unified_dashboard.html", response_class=HTMLResponse) +async def unified_dashboard(): + """Serve unified dashboard""" + return FileResponse(WORKSPACE_ROOT / "unified_dashboard.html") + +@app.get("/simple_overview.html", response_class=HTMLResponse) +async def simple_overview(): + """Serve simple overview""" + return FileResponse(WORKSPACE_ROOT / "simple_overview.html") + +# Generic HTML file handler +@app.get("/{filename}.html", response_class=HTMLResponse) +async def serve_html(filename: str): + """Serve any HTML file from workspace root""" + file_path = WORKSPACE_ROOT / f"{filename}.html" + if file_path.exists(): + return FileResponse(file_path) + return HTMLResponse(f"

File {filename}.html not found

", status_code=404) + + +# ============================================================================ +# Startup Event +# ============================================================================ + + +# ============================================================================ +# ADMIN DASHBOARD ENDPOINTS +# ============================================================================ + +from fastapi import WebSocket, WebSocketDisconnect +import asyncio + +class ConnectionManager: + def __init__(self): + self.active_connections = [] + async def connect(self, websocket: WebSocket): + await websocket.accept() + self.active_connections.append(websocket) + def disconnect(self, websocket: WebSocket): + if websocket in self.active_connections: + self.active_connections.remove(websocket) + async def broadcast(self, message: dict): + disconnected = [] + for conn in list(self.active_connections): + try: + # Check connection state before sending + if conn.client_state == WebSocketState.CONNECTED: + await conn.send_json(message) + else: + disconnected.append(conn) + except Exception as e: + logger.debug(f"Error broadcasting to client: {e}") + disconnected.append(conn) + + # Clean up disconnected clients + for conn in disconnected: + self.disconnect(conn) + +ws_manager = ConnectionManager() + +@app.get("/api/health") +async def api_health(): + h = await health() + return {"status": "healthy" if h.get("status") == "ok" else "degraded", **h} + +# Removed duplicate - using improved version below + +@app.get("/api/coins/{symbol}") +async def get_coin_detail(symbol: str): + coins = await market_collector.get_top_coins(limit=250) + coin = next((c for c in coins if c.get("symbol", "").upper() == symbol.upper()), None) + if not coin: + raise HTTPException(404, f"Coin {symbol} not found") + return {"success": True, "symbol": symbol.upper(), "name": coin.get("name", ""), + "price": coin.get("price") or coin.get("current_price", 0), + "change_24h": coin.get("change_24h") or coin.get("price_change_percentage_24h", 0), + "market_cap": coin.get("market_cap", 0)} + +@app.get("/api/market/stats") +async def get_market_stats(): + """Get global market statistics (duplicate endpoint - keeping for compatibility)""" + try: + overview = await get_market_overview() + + # Calculate ETH dominance from prices if available + eth_dominance = 0 + if overview.get("total_market_cap", 0) > 0: + try: + eth_prices, _ = await fetch_coingecko_prices(symbols=["ETH"], limit=1) + if eth_prices and len(eth_prices) > 0: + eth_market_cap = eth_prices[0].get("market_cap", 0) or 0 + eth_dominance = (eth_market_cap / overview.get("total_market_cap", 1)) * 100 + except: + pass + + return { + "success": True, + "stats": { + "total_market_cap": overview.get("total_market_cap", 0) or 0, + "total_volume_24h": overview.get("total_volume_24h", 0) or 0, + "btc_dominance": overview.get("btc_dominance", 0) or 0, + "eth_dominance": eth_dominance, + "active_cryptocurrencies": 10000, + "markets": 500, + "market_cap_change_24h": 0.0, + "timestamp": datetime.now().isoformat() + } + } + except Exception as e: + logger.error(f"Error in /api/market/stats (duplicate): {e}") + return { + "success": True, + "stats": { + "total_market_cap": 0, + "total_volume_24h": 0, + "btc_dominance": 0, + "eth_dominance": 0, + "active_cryptocurrencies": 0, + "markets": 0, + "market_cap_change_24h": 0.0, + "timestamp": datetime.now().isoformat() + } + } + + +@app.get("/api/stats") +async def get_stats_alias(): + """Alias endpoint for /api/market/stats - backward compatibility""" + return await get_market_stats() + + +@app.get("/api/news/latest") +async def get_latest_news(limit: int = Query(default=40, ge=1, le=100)): + from ai_models import analyze_news_item + news = await news_collector.get_latest_news(limit=limit) + enriched = [] + for item in news[:limit]: + try: + e = analyze_news_item(item) + enriched.append({"title": e.get("title", ""), "source": e.get("source", ""), + "published_at": e.get("published_at") or e.get("date", ""), + "symbols": e.get("symbols", []), "sentiment": e.get("sentiment", "neutral"), + "sentiment_confidence": e.get("sentiment_confidence", 0.5)}) + except: + enriched.append({"title": item.get("title", ""), "source": item.get("source", ""), + "published_at": item.get("date", ""), "symbols": item.get("symbols", []), + "sentiment": "neutral", "sentiment_confidence": 0.5}) + return {"success": True, "news": enriched, "count": len(enriched)} + +@app.post("/api/news/summarize") +async def summarize_news(item: Dict[str, Any] = Body(...)): + from ai_models import analyze_news_item + e = analyze_news_item(item) + return {"success": True, "summary": e.get("title", ""), "sentiment": e.get("sentiment", "neutral")} + +# Duplicate endpoints removed - using the improved versions below in CHARTS ENDPOINTS section + +@app.post("/api/sentiment/analyze") +async def analyze_sentiment(payload: Dict[str, Any] = Body(...)): + from ai_models import ensemble_crypto_sentiment + result = ensemble_crypto_sentiment(payload.get("text", "")) + return {"success": True, "sentiment": result["label"], "confidence": result["confidence"], "details": result} + +@app.post("/api/query") +async def process_query(payload: Dict[str, Any] = Body(...)): + query = payload.get("query", "").lower() + if "price" in query or "btc" in query: + coins = await market_collector.get_top_coins(limit=10) + btc = next((c for c in coins if c.get("symbol", "").upper() == "BTC"), None) + if btc: + return {"success": True, "type": "price", "message": f"Bitcoin is ${btc.get('price', 0):,.2f}", "data": btc} + return {"success": True, "type": "general", "message": "Query processed"} + +@app.get("/api/datasets/list") +async def list_datasets(): + from backend.services.hf_registry import REGISTRY + datasets = REGISTRY.list(kind="datasets") + formatted = [{"name": d.get("id"), "category": d.get("category", "other"), "tags": d.get("tags", [])} for d in datasets] + return {"success": True, "datasets": formatted, "count": len(formatted)} + +@app.get("/api/datasets/sample") +async def get_dataset_sample(name: str = Query(...), limit: int = Query(default=20)): + return {"success": False, "name": name, "sample": [], "message": "Auth required"} + +@app.get("/api/models/list") +async def list_models(): + from ai_models import get_model_info + info = get_model_info() + models = [] + for cat, mlist in info.get("model_catalog", {}).items(): + for mid in mlist: + models.append({"name": mid, "task": "sentiment" if "sentiment" in cat else "analysis", "category": cat}) + return {"success": True, "models": models, "count": len(models)} + +@app.post("/api/models/test") +async def test_model(payload: Dict[str, Any] = Body(...)): + from ai_models import ensemble_crypto_sentiment + result = ensemble_crypto_sentiment(payload.get("text", "")) + return {"success": True, "model": payload.get("model", ""), "result": result} + +@app.websocket("/ws") +async def websocket_endpoint(websocket: WebSocket): + await ws_manager.connect(websocket) + try: + while True: + # Check if connection is still open before sending + if websocket.client_state != WebSocketState.CONNECTED: + logger.info("WebSocket connection closed, breaking loop") + break + + try: + top_coins = await market_collector.get_top_coins(limit=5) + news = await news_collector.get_latest_news(limit=3) + from ai_models import ensemble_crypto_sentiment + sentiment = ensemble_crypto_sentiment(" ".join([n.get("title", "") for n in news])) if news else {"label": "neutral", "confidence": 0.5} + + # Double-check connection state before sending + if websocket.client_state == WebSocketState.CONNECTED: + await websocket.send_json({ + "type": "update", + "payload": { + "market_data": top_coins, + "news": news, + "sentiment": sentiment, + "timestamp": datetime.now().isoformat() + } + }) + else: + logger.info("WebSocket disconnected, breaking loop") + break + + except CollectorError as e: + # Provider errors are already logged by the collector, just continue + logger.debug(f"Provider error in WebSocket update (this is expected with fallbacks): {e}") + # Use cached data if available, or empty data + top_coins = [] + news = [] + sentiment = {"label": "neutral", "confidence": 0.5} + except Exception as e: + # Log other errors with full details + error_msg = str(e) if str(e) else repr(e) + logger.error(f"Error in WebSocket update loop: {type(e).__name__}: {error_msg}") + # Don't break on data errors, just log and continue + # Only break on connection errors + if "send" in str(e).lower() or "close" in str(e).lower(): + break + + await asyncio.sleep(10) + except WebSocketDisconnect: + logger.info("WebSocket disconnect exception caught") + except Exception as e: + logger.error(f"WebSocket endpoint error: {e}") + finally: + try: + ws_manager.disconnect(websocket) + except: + pass + + +@app.on_event("startup") +async def startup_event(): + """Initialize on startup - non-blocking""" + logger.info("=" * 70) + logger.info("Starting Cryptocurrency Data & Analysis API") + logger.info("=" * 70) + logger.info("FastAPI initialized") + logger.info("CORS configured") + logger.info("Cache initialized") + logger.info(f"Providers loaded: {len(PROVIDERS_CONFIG)}") + + # Initialize AI models in background (non-blocking) + async def init_models_background(): + try: + from ai_models import initialize_models + models_init = initialize_models() + logger.info(f"AI Models initialized: {models_init}") + except Exception as e: + logger.warning(f"AI Models initialization failed: {e}") + + # Initialize HF Registry in background (non-blocking) + async def init_registry_background(): + try: + from backend.services.hf_registry import REGISTRY + registry_result = await REGISTRY.refresh() + logger.info(f"HF Registry initialized: {registry_result}") + except Exception as e: + logger.warning(f"HF Registry initialization failed: {e}") + + # Start background tasks + asyncio.create_task(init_models_background()) + asyncio.create_task(init_registry_background()) + logger.info("Background initialization tasks started") + + # Show loaded HuggingFace Space providers + hf_providers = [p for p in PROVIDERS_CONFIG.keys() if 'huggingface_space' in p] + if hf_providers: + logger.info(f"HuggingFace Space providers: {', '.join(hf_providers)}") + + logger.info("Data sources: Binance, CoinGecko, providers_config_extended.json") + + # Check HTML files + html_files = ["index.html", "dashboard.html", "admin.html", "hf_console.html"] + available_html = [f for f in html_files if (WORKSPACE_ROOT / f).exists()] + logger.info(f"UI files: {len(available_html)}/{len(html_files)} available") + logger.info(f"HTML UI available at: http://0.0.0.0:7860/ (index.html)") + + logger.info("=" * 70) + logger.info("API ready at http://0.0.0.0:7860") + logger.info("Docs at http://0.0.0.0:7860/docs") + logger.info("UI at http://0.0.0.0:7860/ (index.html - default HTML page)") + logger.info("=" * 70) + + +# ============================================================================ +# Main Entry Point +# ============================================================================ + +if __name__ == "__main__": + import uvicorn + import sys + import io + + # Fix encoding for Windows console + if sys.platform == "win32": + sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') + sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace') + + try: + print("=" * 70) + print("Starting Cryptocurrency Data & Analysis API") + print("=" * 70) + print("Server: http://localhost:7860") + print("API Docs: http://localhost:7860/docs") + print("Health: http://localhost:7860/health") + print("=" * 70) + except UnicodeEncodeError: + # Fallback if encoding still fails + print("=" * 70) + print("Starting Cryptocurrency Data & Analysis API") + print("=" * 70) + print("Server: http://localhost:7860") + print("API Docs: http://localhost:7860/docs") + print("Health: http://localhost:7860/health") + print("=" * 70) + + uvicorn.run( + app, + host="0.0.0.0", + port=7860, + log_level="info" + ) +# NEW ENDPOINTS FOR ADMIN.HTML - ADD TO hf_unified_server.py + +from fastapi import WebSocket, WebSocketDisconnect +from collections import defaultdict + +# WebSocket Manager +class ConnectionManager: + def __init__(self): + self.active_connections: List[WebSocket] = [] + + async def connect(self, websocket: WebSocket): + await websocket.accept() + self.active_connections.append(websocket) + logger.info(f"WebSocket connected. Total: {len(self.active_connections)}") + + def disconnect(self, websocket: WebSocket): + if websocket in self.active_connections: + self.active_connections.remove(websocket) + logger.info(f"WebSocket disconnected. Total: {len(self.active_connections)}") + + async def broadcast(self, message: dict): + disconnected = [] + for connection in list(self.active_connections): + try: + # Check connection state before sending + if connection.client_state == WebSocketState.CONNECTED: + await connection.send_json(message) + else: + disconnected.append(connection) + except Exception as e: + logger.debug(f"Error broadcasting to client: {e}") + disconnected.append(connection) + + # Clean up disconnected clients + for connection in disconnected: + self.disconnect(connection) + +ws_manager = ConnectionManager() + + +# ===== API HEALTH ===== +@app.get("/api/health") +async def api_health(): + """Health check for admin dashboard""" + health_data = await health() + return { + "status": "healthy" if health_data.get("status") == "ok" else "degraded", + **health_data + } + + +# ===== COINS ENDPOINTS ===== +@app.get("/api/coins/top") +async def get_top_coins(limit: int = Query(default=10, ge=1, le=100)): + """Get top cryptocurrencies by market cap""" + try: + coins = await market_collector.get_top_coins(limit=limit) + + result = [] + for coin in coins: + result.append({ + "id": coin.get("id", coin.get("symbol", "").lower()), + "rank": coin.get("rank", 0), + "symbol": coin.get("symbol", "").upper(), + "name": coin.get("name", ""), + "price": coin.get("price") or coin.get("current_price", 0), + "current_price": coin.get("price") or coin.get("current_price", 0), + "price_change_24h": coin.get("change_24h") or coin.get("price_change_percentage_24h", 0), + "price_change_percentage_24h": coin.get("change_24h") or coin.get("price_change_percentage_24h", 0), + "price_change_percentage_7d_in_currency": coin.get("price_change_percentage_7d", 0), + "volume_24h": coin.get("volume_24h") or coin.get("total_volume", 0), + "total_volume": coin.get("volume_24h") or coin.get("total_volume", 0), + "market_cap": coin.get("market_cap", 0), + "image": coin.get("image", ""), + "sparkline_in_7d": coin.get("sparkline_in_7d") or {"price": []}, + "sparkline_data": coin.get("sparkline_data") or [], + "last_updated": coin.get("last_updated", datetime.now().isoformat()) + }) + + return { + "success": True, + "coins": result, + "count": len(result), + "timestamp": datetime.now().isoformat() + } + except Exception as e: + logger.error(f"Error in /api/coins/top: {e}") + raise HTTPException(status_code=503, detail=str(e)) + + +@app.get("/api/coins/{symbol}") +async def get_coin_detail(symbol: str): + """Get specific coin details""" + try: + coins = await market_collector.get_top_coins(limit=250) + coin = next((c for c in coins if c.get("symbol", "").upper() == symbol.upper()), None) + + if not coin: + raise HTTPException(status_code=404, detail=f"Coin {symbol} not found") + + return { + "success": True, + "symbol": symbol.upper(), + "name": coin.get("name", ""), + "price": coin.get("price") or coin.get("current_price", 0), + "change_24h": coin.get("change_24h") or coin.get("price_change_percentage_24h", 0), + "volume_24h": coin.get("volume_24h") or coin.get("total_volume", 0), + "market_cap": coin.get("market_cap", 0), + "rank": coin.get("rank", 0), + "last_updated": coin.get("last_updated", datetime.now().isoformat()) + } + except HTTPException: + raise + except Exception as e: + logger.error(f"Error in /api/coins/{symbol}: {e}") + raise HTTPException(status_code=503, detail=str(e)) + + +# ===== MARKET STATS ===== +@app.get("/api/market/stats") +async def get_market_stats(): + """Get global market statistics""" + try: + # Use existing endpoint - get_market_overview returns total_market_cap and total_volume_24h + overview = await get_market_overview() + + # Calculate ETH dominance from prices if available + eth_dominance = 0 + if overview.get("total_market_cap", 0) > 0: + # Try to get ETH market cap from top coins + try: + eth_prices, _ = await fetch_coingecko_prices(symbols=["ETH"], limit=1) + if eth_prices and len(eth_prices) > 0: + eth_market_cap = eth_prices[0].get("market_cap", 0) or 0 + eth_dominance = (eth_market_cap / overview.get("total_market_cap", 1)) * 100 + except: + pass + + stats = { + "total_market_cap": overview.get("total_market_cap", 0) or 0, + "total_volume_24h": overview.get("total_volume_24h", 0) or 0, + "btc_dominance": overview.get("btc_dominance", 0) or 0, + "eth_dominance": eth_dominance, + "active_cryptocurrencies": 10000, # Approximate + "markets": 500, # Approximate + "market_cap_change_24h": 0.0, + "timestamp": datetime.now().isoformat() + } + + return {"success": True, "stats": stats} + except Exception as e: + logger.error(f"Error in /api/market/stats: {e}") + raise HTTPException(status_code=503, detail=str(e)) + + +# ===== NEWS ENDPOINTS ===== +@app.get("/api/news/latest") +async def get_latest_news(limit: int = Query(default=40, ge=1, le=100)): + """Get latest crypto news with sentiment""" + try: + news_items = await news_collector.get_latest_news(limit=limit) + + # Attach sentiment to each news item + from ai_models import analyze_news_item + enriched_news = [] + for item in news_items: + try: + enriched = analyze_news_item(item) + enriched_news.append({ + "title": enriched.get("title", ""), + "source": enriched.get("source", ""), + "published_at": enriched.get("published_at") or enriched.get("date", ""), + "symbols": enriched.get("symbols", []), + "sentiment": enriched.get("sentiment", "neutral"), + "sentiment_confidence": enriched.get("sentiment_confidence", 0.5), + "url": enriched.get("url", "") + }) + except: + enriched_news.append({ + "title": item.get("title", ""), + "source": item.get("source", ""), + "published_at": item.get("published_at") or item.get("date", ""), + "symbols": item.get("symbols", []), + "sentiment": "neutral", + "sentiment_confidence": 0.5, + "url": item.get("url", "") + }) + + return { + "success": True, + "news": enriched_news, + "count": len(enriched_news), + "timestamp": datetime.now().isoformat() + } + except Exception as e: + logger.error(f"Error in /api/news/latest: {e}") + return {"success": True, "news": [], "count": 0, "timestamp": datetime.now().isoformat()} + + +@app.get("/api/news") +async def get_news(limit: int = Query(default=40, ge=1, le=100)): + """Alias for /api/news/latest for backward compatibility""" + return await get_latest_news(limit=limit) + + +@app.post("/api/news/summarize") +async def summarize_news(item: Dict[str, Any] = Body(...)): + """Summarize a news article""" + try: + from ai_models import analyze_news_item + enriched = analyze_news_item(item) + + return { + "success": True, + "summary": enriched.get("title", ""), + "sentiment": enriched.get("sentiment", "neutral"), + "sentiment_confidence": enriched.get("sentiment_confidence", 0.5) + } + except Exception as e: + logger.error(f"Error in /api/news/summarize: {e}") + return { + "success": False, + "error": str(e), + "summary": item.get("title", ""), + "sentiment": "neutral" + } + + +# ===== CHARTS ENDPOINTS ===== +@app.get("/api/charts/price/{symbol}") +async def get_price_chart(symbol: str, timeframe: str = Query(default="7d")): + """Get price chart data""" + try: + # Clean and validate symbol + symbol = symbol.strip().upper() + if not symbol: + return JSONResponse( + status_code=400, + content={ + "success": False, + "symbol": "", + "timeframe": timeframe, + "data": [], + "count": 0, + "error": "Symbol cannot be empty" + } + ) + + logger.info(f"Fetching price history for {symbol} with timeframe {timeframe}") + + # market_collector.get_price_history expects timeframe as string, not hours + price_history = await market_collector.get_price_history(symbol, timeframe=timeframe) + + if not price_history or len(price_history) == 0: + logger.warning(f"No price history returned for {symbol}") + return { + "success": True, + "symbol": symbol, + "timeframe": timeframe, + "data": [], + "count": 0, + "message": "No data available" + } + + chart_data = [] + for point in price_history: + # Handle different timestamp formats + timestamp = point.get("timestamp") or point.get("time") or point.get("date") + price = point.get("price") or point.get("close") or point.get("value") or 0 + + # Convert timestamp to ISO format if needed + if timestamp: + try: + # If it's already a string, use it + if isinstance(timestamp, str): + # Try to parse and format + try: + # Try ISO format first + dt = datetime.fromisoformat(timestamp.replace('Z', '+00:00')) + timestamp = dt.isoformat() + except: + try: + # Try other common formats + from dateutil import parser + dt = parser.parse(timestamp) + timestamp = dt.isoformat() + except: + pass + elif isinstance(timestamp, (int, float)): + # Unix timestamp + dt = datetime.fromtimestamp(timestamp) + timestamp = dt.isoformat() + except Exception as e: + logger.warning(f"Error parsing timestamp {timestamp}: {e}") + + chart_data.append({ + "timestamp": timestamp or "", + "time": timestamp or "", + "date": timestamp or "", + "price": float(price) if price else 0, + "close": float(price) if price else 0, + "value": float(price) if price else 0 + }) + + logger.info(f"Returning {len(chart_data)} data points for {symbol}") + + return { + "success": True, + "symbol": symbol, + "timeframe": timeframe, + "data": chart_data, + "count": len(chart_data) + } + except CollectorError as e: + logger.error(f"Collector error in /api/charts/price/{symbol}: {e}", exc_info=True) + return JSONResponse( + status_code=200, + content={ + "success": False, + "symbol": symbol.upper() if symbol else "", + "timeframe": timeframe, + "data": [], + "count": 0, + "error": str(e) + } + ) + except Exception as e: + logger.error(f"Error in /api/charts/price/{symbol}: {e}", exc_info=True) + return JSONResponse( + status_code=200, + content={ + "success": False, + "symbol": symbol.upper() if symbol else "", + "timeframe": timeframe, + "data": [], + "count": 0, + "error": str(e) + } + ) + + +@app.post("/api/charts/analyze") +async def analyze_chart(payload: Dict[str, Any] = Body(...)): + """Analyze chart data""" + try: + symbol = payload.get("symbol") + timeframe = payload.get("timeframe", "7d") + indicators = payload.get("indicators", []) + + if not symbol: + return JSONResponse( + status_code=400, + content={"success": False, "error": "Symbol is required"} + ) + + symbol = symbol.strip().upper() + logger.info(f"Analyzing chart for {symbol} with timeframe {timeframe}") + + # Get price data - use timeframe string, not hours + price_history = await market_collector.get_price_history(symbol, timeframe=timeframe) + + if not price_history or len(price_history) == 0: + return { + "success": False, + "symbol": symbol, + "timeframe": timeframe, + "error": "No price data available for analysis" + } + + # Analyze with AI + from ai_models import analyze_chart_points + try: + analysis = analyze_chart_points(price_history, indicators) + except Exception as ai_error: + logger.error(f"AI analysis error: {ai_error}", exc_info=True) + # Return a basic analysis if AI fails + analysis = { + "direction": "neutral", + "summary": "Analysis unavailable", + "signals": [] + } + + return { + "success": True, + "symbol": symbol, + "timeframe": timeframe, + "analysis": analysis + } + except CollectorError as e: + logger.error(f"Collector error in /api/charts/analyze: {e}", exc_info=True) + return JSONResponse( + status_code=200, + content={"success": False, "error": str(e)} + ) + except Exception as e: + logger.error(f"Error in /api/charts/analyze: {e}", exc_info=True) + return JSONResponse( + status_code=200, + content={"success": False, "error": str(e)} + ) + + +# ===== SENTIMENT ENDPOINTS ===== +@app.post("/api/sentiment/analyze") +async def analyze_sentiment(payload: Dict[str, Any] = Body(...)): + """Analyze sentiment of text""" + try: + text = payload.get("text", "") + + from ai_models import ensemble_crypto_sentiment + result = ensemble_crypto_sentiment(text) + + return { + "success": True, + "sentiment": result["label"], + "confidence": result["confidence"], + "details": result + } + except Exception as e: + logger.error(f"Error in /api/sentiment/analyze: {e}") + return {"success": False, "error": str(e)} + + +# ===== QUERY ENDPOINT ===== +@app.post("/api/query") +async def process_query(payload: Dict[str, Any] = Body(...)): + """Process natural language query""" + try: + query = payload.get("query", "").lower() + + # Simple query processing + if "price" in query or "btc" in query or "bitcoin" in query: + coins = await market_collector.get_top_coins(limit=10) + btc = next((c for c in coins if c.get("symbol", "").upper() == "BTC"), None) + + if btc: + price = btc.get("price") or btc.get("current_price", 0) + return { + "success": True, + "type": "price", + "message": f"Bitcoin (BTC) is currently trading at ${price:,.2f}", + "data": btc + } + + return { + "success": True, + "type": "general", + "message": "Query processed", + "data": None + } + except Exception as e: + logger.error(f"Error in /api/query: {e}") + return {"success": False, "error": str(e), "message": "Query failed"} + + +# ===== DATASETS & MODELS ===== +@app.get("/api/datasets/list") +async def list_datasets(): + """List available datasets""" + try: + from backend.services.hf_registry import REGISTRY + datasets = REGISTRY.list(kind="datasets") + + formatted = [] + for d in datasets: + formatted.append({ + "name": d.get("id"), + "category": d.get("category", "other"), + "records": "N/A", + "updated_at": "", + "tags": d.get("tags", []), + "source": d.get("source", "hub") + }) + + return { + "success": True, + "datasets": formatted, + "count": len(formatted) + } + except Exception as e: + logger.error(f"Error in /api/datasets/list: {e}") + return {"success": True, "datasets": [], "count": 0} + + +@app.get("/api/datasets/sample") +async def get_dataset_sample(name: str = Query(...), limit: int = Query(default=20)): + """Get sample from dataset""" + try: + # Attempt to load dataset + try: + from datasets import load_dataset + from config import get_settings + + # Get HF token for dataset loading + settings = get_settings() + hf_token = settings.hf_token or "HF_TOKEN_FROM_SPACE_SECRET" + + # Set token in environment for datasets library + import os + if hf_token and not os.environ.get("HF_TOKEN"): + os.environ["HF_TOKEN"] = hf_token + + dataset = load_dataset(name, split="train", streaming=True, token=hf_token) + + sample = [] + for i, row in enumerate(dataset): + if i >= limit: + break + sample.append({k: str(v) for k, v in row.items()}) + + return { + "success": True, + "name": name, + "sample": sample, + "count": len(sample) + } + except: + return { + "success": False, + "name": name, + "sample": [], + "count": 0, + "message": "Dataset loading requires authentication or is not available" + } + except Exception as e: + logger.error(f"Error in /api/datasets/sample: {e}") + return {"success": False, "error": str(e)} + + +@app.get("/api/models/list") +async def list_models(): + """List available models""" + try: + from ai_models import get_model_info + info = get_model_info() + + models = [] + catalog = info.get("model_catalog", {}) + + for category, model_list in catalog.items(): + for model_id in model_list: + models.append({ + "name": model_id, + "task": "sentiment" if "sentiment" in category else "decision" if category == "decision" else "analysis", + "status": "available", + "category": category, + "notes": f"{category.replace('_', ' ').title()} model" + }) + + return { + "success": True, + "models": models, + "count": len(models) + } + except Exception as e: + logger.error(f"Error in /api/models/list: {e}") + return {"success": True, "models": [], "count": 0} + + +@app.post("/api/models/test") +async def test_model(payload: Dict[str, Any] = Body(...)): + """Test a specific model""" + try: + model_id = payload.get("model", "") + text = payload.get("text", "") + + from ai_models import ensemble_crypto_sentiment + result = ensemble_crypto_sentiment(text) + + return { + "success": True, + "model": model_id, + "result": result + } + except Exception as e: + logger.error(f"Error in /api/models/test: {e}") + return {"success": False, "error": str(e)} + + +# ===== WEBSOCKET ===== +@app.websocket("/ws") +async def websocket_endpoint(websocket: WebSocket): + """WebSocket endpoint for real-time updates""" + await ws_manager.connect(websocket) + + try: + while True: + # Check if connection is still open before sending + if websocket.client_state != WebSocketState.CONNECTED: + logger.info("WebSocket connection closed, breaking loop") + break + + # Send market updates every 10 seconds + try: + # Get latest data + top_coins = await market_collector.get_top_coins(limit=5) + news_items = await news_collector.get_latest_news(limit=3) + + # Compute global sentiment from news + from ai_models import ensemble_crypto_sentiment + news_texts = " ".join([n.get("title", "") for n in news_items]) + global_sentiment = ensemble_crypto_sentiment(news_texts) if news_texts else {"label": "neutral", "confidence": 0.5} + + payload = { + "market_data": top_coins, + "stats": { + "total_market_cap": sum([c.get("market_cap", 0) for c in top_coins]), + "sentiment": global_sentiment + }, + "news": news_items, + "sentiment": global_sentiment, + "timestamp": datetime.now().isoformat() + } + + # Double-check connection state before sending + if websocket.client_state == WebSocketState.CONNECTED: + await websocket.send_json({ + "type": "update", + "payload": payload + }) + else: + logger.info("WebSocket disconnected, breaking loop") + break + except CollectorError as e: + # Provider errors are already logged by the collector, just continue + logger.debug(f"Provider error in WebSocket update (this is expected with fallbacks): {e}") + # Use empty data on provider errors + payload = { + "market_data": [], + "stats": {"total_market_cap": 0, "sentiment": {"label": "neutral", "confidence": 0.5}}, + "news": [], + "sentiment": {"label": "neutral", "confidence": 0.5}, + "timestamp": datetime.now().isoformat() + } + except Exception as e: + # Log other errors with full details + error_msg = str(e) if str(e) else repr(e) + logger.error(f"Error in WebSocket update: {type(e).__name__}: {error_msg}") + # Don't break on data errors, just log and continue + # Only break on connection errors + if "send" in str(e).lower() or "close" in str(e).lower(): + break + + await asyncio.sleep(10) + except WebSocketDisconnect: + logger.info("WebSocket disconnect exception caught") + except Exception as e: + logger.error(f"WebSocket error: {e}") + finally: + try: + ws_manager.disconnect(websocket) + except: + pass + +@app.get("/api/market/history") +async def get_market_history(symbol: str = "BTC", limit: int = 10): + """ + Get historical prices from the local database if available. + + For this deployment we avoid touching the internal DatabaseManager + and simply report that no history API is wired yet. + """ + symbol = symbol.upper() + # We don't fabricate data here; if you need real history, it should + # be implemented via the shared database models. + return { + "symbol": symbol, + "history": [], + "count": 0, + "message": "History endpoint not wired to DB in this Space", + } + + + +@app.get("/api/status") +async def get_status(): + """ + System status endpoint used by the admin UI. + + This reports real-time information about providers and database, + without fabricating any market data. + """ + providers_cfg = load_providers_config() + providers = providers_cfg or {} + validated_count = sum(1 for p in providers.values() if p.get("validated")) + + db_path = DB_PATH + db_status = "connected" if db_path.exists() else "initializing" + + return { + "system_health": "healthy", + "timestamp": datetime.now().isoformat(), + "total_providers": len(providers), + "validated_providers": validated_count, + "database_status": db_status, + "apl_available": APL_REPORT_PATH.exists(), + "use_mock_data": False, + } + + +@app.get("/api/logs/recent") +async def get_recent_logs(): + """ + Return recent log lines for the admin UI. + + We read from the main server log file if available. + This does not fabricate content; if there are no logs, + an empty list is returned. + """ + log_file = LOG_DIR / "server.log" + lines = tail_log_file(log_file, max_lines=200) + # Wrap plain text lines as structured entries + logs = [{"line": line.rstrip("\n")} for line in lines] + return {"logs": logs, "count": len(logs)} + + +@app.get("/api/logs/errors") +async def get_error_logs(): + """ + Return recent error log lines from the same log file. + + This is a best-effort filter based on typical ERROR prefixes. + """ + log_file = LOG_DIR / "server.log" + lines = tail_log_file(log_file, max_lines=400) + error_lines = [line for line in lines if "ERROR" in line or "WARNING" in line] + logs = [{"line": line.rstrip("\n")} for line in error_lines[-200:]] + return {"errors": logs, "count": len(logs)} + + +def _load_apl_report() -> Optional[Dict[str, Any]]: + """Load the APL (Auto Provider Loader) validation report if available.""" + if not APL_REPORT_PATH.exists(): + return None + try: + with APL_REPORT_PATH.open("r", encoding="utf-8") as f: + return json.load(f) + except Exception as e: + logger.error(f"Error reading APL report: {e}") + return None + + +@app.get("/api/apl/summary") +async def get_apl_summary(): + """ + Summary of the Auto Provider Loader (APL) report. + + If the report is missing, we return a clear not_available status + instead of fabricating metrics. + """ + report = _load_apl_report() + if not report or "stats" not in report: + return { + "status": "not_available", + "message": "APL report not found", + } + + stats = report.get("stats", {}) + return { + "status": "ok", + "http_candidates": stats.get("total_http_candidates", 0), + "http_valid": stats.get("http_valid", 0), + "http_invalid": stats.get("http_invalid", 0), + "http_conditional": stats.get("http_conditional", 0), + "hf_candidates": stats.get("total_hf_candidates", 0), + "hf_valid": stats.get("hf_valid", 0), + "hf_invalid": stats.get("hf_invalid", 0), + "hf_conditional": stats.get("hf_conditional", 0), + "timestamp": datetime.now().isoformat(), + } + + +@app.get("/api/hf/models") +async def get_hf_models_from_apl(): + """ + Return the list of Hugging Face models discovered by the APL report. + + This is used by the admin UI. The data comes from the real + PROVIDER_AUTO_DISCOVERY_REPORT.json file if present. + """ + report = _load_apl_report() + if not report: + return {"models": [], "count": 0, "source": "none"} + + hf_models = report.get("hf_models", {}).get("results", []) + return { + "models": hf_models, + "count": len(hf_models), + "source": "APL report", + } +