jhub123's picture
🧠 Predictive Algorithm — “Short-Term Direction Classifier”
ac725f6 verified
```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
}
```