|
|
```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): |
|
|
|
|
|
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) |
|
|
|
|
|
|
|
|
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() |
|
|
|
|
|
|
|
|
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'] |
|
|
|
|
|
|
|
|
df['Volume_change'] = df['volume'].pct_change() |
|
|
obv = OnBalanceVolumeIndicator(close=df['close'], volume=df['volume']) |
|
|
df['OBV'] = obv.on_balance_volume() |
|
|
|
|
|
|
|
|
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'] |
|
|
|
|
|
|
|
|
for n in [1, 3, 5]: |
|
|
df[f'return_{n}'] = df['close'].pct_change(periods=n) |
|
|
|
|
|
|
|
|
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 |
|
|
} |
|
|
``` |