```python import xgboost as xgb import pandas as pd import numpy as np from fastapi import APIRouter from ta.momentum import RSIIndicator from ta.trend import MACD, EMAIndicator from ta.volatility import BollingerBands, AverageTrueRange from ta.volume import OnBalanceVolumeIndicator router = APIRouter() model = xgb.XGBClassifier() model.load_model("models/model_xgb_v1.json") def compute_rsi(close_prices, window=14): rsi_indicator = RSIIndicator(close=close_prices, window=window) return rsi_indicator.rsi() def compute_technical_indicators(df): # Trend indicators ema8 = EMAIndicator(close=df['close'], window=8) ema21 = EMAIndicator(close=df['close'], window=21) df['EMA_8'] = ema8.ema_indicator() df['EMA_21'] = ema21.ema_indicator() df['EMA_cross'] = (df['EMA_8'] > df['EMA_21']).astype(int) # Momentum indicators df['RSI_14'] = compute_rsi(df['close']) macd = MACD(close=df['close']) df['MACD'] = macd.macd() df['MACD_signal'] = macd.macd_signal() df['MACD_hist'] = macd.macd_diff() # Volatility indicators atr = AverageTrueRange(high=df['high'], low=df['low'], close=df['close'], window=14) df['ATR_14'] = atr.average_true_range() bb = BollingerBands(close=df['close']) df['Bollinger_width'] = (bb.bollinger_hband() - bb.bollinger_lband()) / df['close'] # Volume indicators df['Volume_change'] = df['volume'].pct_change() obv = OnBalanceVolumeIndicator(close=df['close'], volume=df['volume']) df['OBV'] = obv.on_balance_volume() # Price shape features df['candle_body'] = df['close'] - df['open'] df['upper_shadow'] = df['high'] - df[['open', 'close']].max(axis=1) df['lower_shadow'] = df[['open', 'close']].min(axis=1) - df['low'] # Lag returns for n in [1, 3, 5]: df[f'return_{n}'] = df['close'].pct_change(periods=n) # Time features df['minute_sin'] = np.sin(2 * np.pi * df.index.minute / 60) df['minute_cos'] = np.cos(2 * np.pi * df.index.minute / 60) return df @router.post("/api/signal/{symbol}") def get_signal(symbol: str, candles: list[dict]): df = pd.DataFrame(candles) df = compute_technical_indicators(df) latest = df.tail(1) features = ['EMA_8', 'EMA_21', 'EMA_cross', 'RSI_14', 'MACD', 'MACD_signal', 'MACD_hist', 'ATR_14', 'Bollinger_width', 'Volume_change', 'OBV', 'candle_body', 'upper_shadow', 'lower_shadow', 'return_1', 'return_3', 'return_5', 'minute_sin', 'minute_cos'] X = latest[features] prob_up = float(model.predict_proba(X)[0,1]) prob_down = 1 - prob_up confidence = abs(prob_up - prob_down) advice = "long_bias" if prob_up > 0.6 and confidence > 0.15 else "short_bias" if prob_down > 0.6 else "neutral" return { "symbol": symbol, "prob_up": round(prob_up, 3), "prob_down": round(prob_down, 3), "confidence": round(confidence, 3), "advice": advice } ```