XGBScore

"""XGBoost scoring - 20 factor classification, 15m, BTC+ETH"""
from pandas import DataFrame
import talib.abstract as ta
import numpy as np
from freqtrade.strategy import IStrategy

FEATURES = None  # computed in _train

def build_features(d):
    df = d.copy()
    # Price returns
    for p in [1,3,6,12,24,48]:
        df['r'+str(p)] = df['close'].pct_change(p)
    # OHLC ratios
    df['hlr'] = (df['high']-df['low'])/df['close']
    df['cp'] = (df['close']-df['low'])/(df['high']-df['low']+1e-6)
    # EMA
    for p in [9,21,55,200]:
        e = ta.EMA(df, p)
        df['e'+str(p)+'d'] = (df['close']-e)/e*100
    # MACD
    macd = ta.MACD(df); df['md']=macd['macd']; df['ms']=macd['macdsignal']; df['mh']=macd['macdhist']
    # Oscillators
    df['rsi'] = ta.RSI(df,14); df['adx'] = ta.ADX(df,14)
    # Bollinger
    bb = ta.BBANDS(df,20); df['bw']=(bb['upperband']-bb['lowerband'])/bb['middleband']
    df['bbp'] = (df['close']-bb['lowerband'])/(bb['upperband']-bb['lowerband']+1e-6)
    # Volatility & Volume
    df['atr'] = ta.ATR(df,14)/df['close']*100
    df['vr'] = df['volume']/ta.SMA(df['volume'],20)
    # Stochastic
    st = ta.STOCH(df); df['sk'] = st['slowk']; df['sd'] = st['slowd']
    # MFI
    tp = (df['high']+df['low']+df['close'])/3; mf = tp*df['volume']
    pf = mf.where(tp>tp.shift(1),0).rolling(14).sum()
    nf = mf.where(tp<tp.shift(1),0).rolling(14).sum()
    df['mfi'] = 100-100/(1+pf/(nf+1e-6))
    return df

ALL_FEATS = ['r1','r3','r6','r12','r24','r48','hlr','cp','e9d','e21d','e55d','e200d',
             'md','ms','mh','rsi','adx','bw','bbp','atr','vr','sk','sd','mfi']


class XGBScore(IStrategy):
    INTERFACE_VERSION = 3; timeframe = '15m'; can_short = True
    stoploss = -0.025; trailing_stop = True
    trailing_stop_positive = 0.005; trailing_stop_positive_offset = 0.018
    trailing_only_offset_is_reached = True
    minimal_roi = {"0":0.08,"480":0.05,"1440":0.03,"4320":0}
    max_open_trades = 6; startup_candle_count = 2500
    process_only_new_candles = False; use_exit_signal = False

    def __init__(self, config):
        super().__init__(config)
        self._m = None; self._sm = None; self._ss = None; self._ok = False

    def populate_indicators(self, dataframe, metadata):
        n = len(dataframe)
        if not self._ok and n >= 2400:
            self._train(dataframe.iloc[:2000].copy())
        if self._ok:
            df = build_features(dataframe)
            X = np.nan_to_num(df[ALL_FEATS].values,0,0,0)
            Xs = (X-self._sm)/self._ss
            ap = np.mean([m.predict_proba(Xs) for m in self._m], axis=0)
            cl = self._m[0].classes_
            li = list(cl).index(2) if 2 in cl else 0
            si = list(cl).index(0) if 0 in cl else 0
            dataframe['xs'] = ap[:,li]-ap[:,si]  # long - short score
            dataframe['xc'] = np.maximum(ap[:,li], ap[:,si])  # confidence
        return dataframe

    def _train(self, df):
        import xgboost as xgb
        d = build_features(df)
        ret = d['close'].pct_change(12).shift(-12).fillna(0)
        y = np.where(ret>0.005,2,np.where(ret<-0.005,0,1))
        X = np.nan_to_num(d[ALL_FEATS].values,0,0,0)
        v = np.arange(len(d)-12); v = v[v<len(X)]
        valid = np.ones(len(v), dtype=bool)
        for i in range(len(v)):
            if np.isnan(y[v[i]]): valid[i]=0; continue
            for j in range(X.shape[1]):
                if np.isnan(X[v[i],j]): valid[i]=0; break
        v2 = v[valid==1]; Xt=X[v2]; yt=y[v2]
        if len(Xt)<500: return False
        self._sm = Xt.mean(0); self._ss = Xt.std(0)+1e-6; Xs=(Xt-self._sm)/self._ss
        u,c = np.unique(yt,return_counts=True)
        w = {u[i]:len(yt)/(len(u)*c[i]) for i in range(len(u))}
        sw = np.array([w[z] for z in yt])
        self._m = []
        for s in [42,73,99,17,55]:
            m = xgb.XGBClassifier(n_estimators=300, max_depth=5, learning_rate=0.03,
                subsample=0.8, colsample_bytree=0.7, min_child_weight=3,
                reg_alpha=0.2, reg_lambda=1.0, random_state=s, verbosity=0)
            m.fit(Xs,yt,sample_weight=sw); self._m.append(m)
        self._ok = True; return True

    def populate_entry_trend(self, d, m):
        if 'xs' not in d.columns: return d
        d.loc[(d['xc']>0.55)&(d['xs']>0.05),['enter_long','enter_tag']]=(1,'L')
        d.loc[(d['xc']>0.55)&(d['xs']<-0.05),['enter_short','enter_tag']]=(1,'S')
        return d

    def populate_exit_trend(self, d, m): return d