from fastapi import FastAPI, HTTPException import pandas as pd import numpy as np import xgboost as xgb import requests import warnings import time from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry warnings.filterwarnings('ignore') app = FastAPI(title="Swing Quant Engine API") # --- YOUR WINNING OOS PARAMETERS --- MIN_PROB = 0.36 EDGE = 0.010 MODEL_PATH = 'swing_quant_engine_v1.json' # Load model globally on startup model = xgb.XGBClassifier() try: model.load_model(MODEL_PATH) print("✅ Quant Engine Loaded Successfully") except Exception as e: print(f"⚠️ Failed to load model: {e}") def get_top_liquid_coins(limit=50): """Fetches all tradable USDT-M Futures symbols and sorts them by 24h volume.""" info_url = "https://fapi.binance.com/fapi/v1/exchangeInfo" ticker_url = "https://fapi.binance.com/fapi/v1/ticker/24hr" # 1. Add headers to prevent basic bot-blocking headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'} # 2. Set up a requests session with automatic retries session = requests.Session() retry = Retry(total=3, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504]) adapter = HTTPAdapter(max_retries=retry) session.mount('http://', adapter) session.mount('https://', adapter) # 3. Updated fallback list (MATIC swapped to POL) fallback_list = [ 'BTCUSDT', 'ETHUSDT', 'SOLUSDT', 'BNBUSDT', 'ADAUSDT', 'XRPUSDT', 'DOTUSDT', 'LINKUSDT', 'AVAXUSDT', 'POLUSDT', 'LTCUSDT', 'BCHUSDT', 'SHIBUSDT', 'TRXUSDT', 'NEARUSDT', 'FILUSDT', 'ATOMUSDT', 'ETCUSDT' ] try: print("Fetching Binance exchange info...") # Get Exchange Info to find valid USDT pairs info_res = session.get(info_url, headers=headers, timeout=10) info_res.raise_for_status() # Raises an error if the status code isn't 200 OK info_data = info_res.json() # Use a set for faster lookups valid_symbols = { s['symbol'] for s in info_data['symbols'] if s['quoteAsset'] == 'USDT' and s['status'] == 'TRADING' } print("Fetching Binance 24hr ticker data...") # Get 24hr Ticker data to sort by volume (increased timeout slightly for large payload) ticker_res = session.get(ticker_url, headers=headers, timeout=15) ticker_res.raise_for_status() ticker_data = ticker_res.json() # Filter ticker data for only our valid USDT trading pairs if isinstance(ticker_data, dict): ticker_data = [ticker_data] volume_map = [] for item in ticker_data: symbol = item['symbol'] if symbol in valid_symbols: volume_map.append({ 'symbol': symbol, 'volume': float(item['quoteVolume']) }) # Sort by Volume Descending volume_map.sort(key=lambda x: x['volume'], reverse=True) top_symbols = [d['symbol'] for d in volume_map[:limit]] if len(top_symbols) >= 10: print(f"✅ Successfully fetched {len(top_symbols)} liquid symbols.") return top_symbols else: print("⚠️ Warning: Too few symbols found, dropping to fallback.") return fallback_list[:limit] except requests.exceptions.RequestException as e: # This catches connection errors, timeouts, and bad HTTP statuses print(f"🚨 Network/API Error fetching universe: {e}") return fallback_list[:limit] except Exception as e: # Catches JSON parsing errors or unexpected bugs print(f"🚨 Unexpected Universe Fetch Error: {e}") return fallback_list[:limit] def fetch_and_engineer(symbol: str, limit: int = 250): # 1. Switched from Spot API (api.binance.com) to Futures API (fapi.binance.com) url = f"https://fapi.binance.com/fapi/v1/klines?symbol={symbol}&interval=1h&limit={limit}" # 2. Added headers and a timeout to bypass Binance bot-blocking headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'} try: res = requests.get(url, headers=headers, timeout=10) if res.status_code != 200: raise ValueError(f"Binance API error for {symbol} - Status Code {res.status_code}") data = res.json() except Exception as e: raise ValueError(f"Network error fetching klines for {symbol}: {e}") if len(data) < 200: raise ValueError(f"Not enough data for 200 EMA. Found {len(data)} candles.") cols = ['open_time', 'open', 'high', 'low', 'close', 'volume', 'close_time', 'quote_asset_volume', 'number_of_trades', 'taker_buy_base', 'taker_buy_quote', 'ignore'] df = pd.DataFrame(data, columns=cols) for col in ['open', 'high', 'low', 'close', 'volume']: df[col] = pd.to_numeric(df[col], errors='coerce') # --- Feature Engineering Logic --- df['ema_200'] = df['close'].ewm(span=200, adjust=False).mean() df['above_200ema'] = np.where(df['close'] > df['ema_200'], 1, 0) df['dist_to_ema200'] = (df['close'] - df['ema_200']) / df['ema_200'] df['ema_200_slope'] = df['ema_200'].diff(10) / df['ema_200'] df['ema_20'] = df['close'].ewm(span=20, adjust=False).mean() df['ema_50'] = df['close'].ewm(span=50, adjust=False).mean() df['trend_alignment'] = (df['ema_20'] - df['ema_50']) / df['ema_50'] df['dist_to_ema20'] = (df['close'] - df['ema_20']) / df['ema_20'] df['tr'] = np.maximum(df['high'] - df['low'], np.maximum(abs(df['high'] - df['close'].shift()), abs(df['low'] - df['close'].shift()))) df['atr_14'] = df['tr'].rolling(14).mean() df['atr_pct'] = df['atr_14'] / df['close'] df['std_20'] = df['close'].rolling(20).std() df['bb_pct'] = (df['close'] - (df['ema_20'] - (df['std_20'] * 2))) / ((df['ema_20'] + (df['std_20'] * 2)) - (df['ema_20'] - (df['std_20'] * 2))).replace(0, 0.001) ema_12 = df['close'].ewm(span=12, adjust=False).mean() ema_26 = df['close'].ewm(span=26, adjust=False).mean() df['macd_hist'] = (ema_12 - ema_26) - (ema_12 - ema_26).ewm(span=9, adjust=False).mean() delta = df['close'].diff() gain = (delta.where(delta > 0, 0)).rolling(window=14).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean() df['rsi_14'] = 100 - (100 / (1 + (gain / loss.replace(0, 1)))) df['vol_ema_24'] = df['volume'].ewm(span=24, adjust=False).mean() df['vol_surge'] = df['volume'] / df['vol_ema_24'] return df.dropna() @app.get("/") def read_root(): return {"status": "Swing Quant Engine Online"} @app.get("/health") def health_check(): return {"status": "alive"} @app.get("/scan/{symbol}") def scan_symbol(symbol: str): try: df = fetch_and_engineer(symbol.upper()) current_state = df.iloc[-2] # Last fully closed 1H candle features = ['dist_to_ema20', 'trend_alignment', 'macd_hist', 'rsi_14', 'bb_pct', 'vol_surge', 'atr_pct', 'above_200ema', 'dist_to_ema200', 'ema_200_slope'] current_features = df[features].iloc[-2: -1] probs = model.predict_proba(current_features)[0] p_neutral, p_long, p_short = float(probs[0]), float(probs[1]), float(probs[2]) signal = "WAIT" tp_price, sl_price = 0.0, 0.0 if p_long > p_neutral and p_long > p_short: if (p_long - p_short) > EDGE and p_long >= MIN_PROB and current_state['above_200ema'] == 1: if 0.005 < current_state['atr_pct'] < 0.035: signal = "LONG" atr_val = current_state['atr_pct'] * current_state['close'] tp_price = current_state['close'] + (atr_val * 3.0) sl_price = current_state['close'] - (atr_val * 1.5) return { "symbol": symbol.upper(), "price": float(current_state['close']), "signal": signal, "probabilities": {"neutral": p_neutral, "long": p_long, "short": p_short}, "targets": {"take_profit": tp_price, "stop_loss": sl_price} if signal == "LONG" else None } except Exception as e: raise HTTPException(status_code=400, detail=str(e)) @app.get("/scan_universe") def scan_universe(limit: int = 50): """Deep space scanner: Loops through liquid coins, returning LONG signals and ALL probabilities.""" universe = get_top_liquid_coins(limit) hits = [] all_results = [] for symbol in universe: try: result = scan_symbol(symbol) # Record the probability regardless of the signal all_results.append({ "symbol": result["symbol"], "signal": result["signal"], "probabilities": result["probabilities"] }) # If it's a hit, add it to the execution targets if result["signal"] == "LONG": hits.append(result) time.sleep(0.1) # Rate limit protection except Exception as e: print(f"⚠️ Skipping {symbol} due to scan error: {e}") continue return { "timestamp": time.time(), "coins_scanned": len(all_results), "signals_found": len(hits), "long_targets": hits, "all_results": all_results # New key added here } @app.get("/debug_universe") def debug_universe(): symbols = get_top_liquid_coins(50) return { "count": len(symbols), "symbols": symbols }