XGBStandalone

"""XGBoost standalone — classification, 1h, BTC+ETH"""
from pandas import DataFrame
import talib.abstract as ta, numpy as np
from freqtrade.strategy import IStrategy

class XGBStandalone(IStrategy):
    INTERFACE_VERSION = 3; timeframe = '1h'; can_short = True
    stoploss = -0.07; minimal_roi = {"0": 0.1, "240": -1}
    max_open_trades = 3; startup_candle_count = 1500
    process_only_new_candles = False; use_exit_signal = True

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

    def _feat(self, d):
        df = d.copy()
        for p in [10,20,50]:
            up = df['close'].diff().clip(0); dn = (-df['close'].diff()).clip(0)
            df['rsi_'+str(p)] = 100-100/(1+up.rolling(p).mean()/(dn.rolling(p).mean()+0.0001))
            df['z_'+str(p)] = (df['close']-df['close'].rolling(p).mean())/(df['close'].rolling(p).std()+0.0001)
            df['ret_'+str(p)] = df['close'].pct_change(p)
        df['vol'] = np.log1p(df['volume']); df['ema20'] = ta.EMA(df,20); df['ema50'] = ta.EMA(df,50)
        df['rsi14'] = ta.RSI(df,14); md = ta.MACD(df); df['md'] = md['macd']; df['ms'] = md['macdsignal']
        df['adx'] = ta.ADX(df,14); df['atr'] = ta.ATR(df,14)/df['close']
        df['vr'] = df['volume']/ta.SMA(df['volume'],20)
        bb = ta.BBANDS(df); df['bb'] = (df['close']-bb['lowerband'])/(bb['upperband']-bb['lowerband']+0.0001)
        return df

    def populate_indicators(self, dataframe, metadata):
        n = len(dataframe)
        if not self._ok and n >= 1200:
            self._train(dataframe.iloc[200:1000].copy())
            if not self._ok and n >= 1500:
                self._train(dataframe.iloc[200:1200].copy())
        if self._ok:
            d = self._feat(dataframe)
            X = np.nan_to_num(d[self._cols].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['&t'] = ap[:,li]-ap[:,si]
            dataframe['dp'] = 1
        return dataframe

    def _train(self, df):
        import xgboost as xgb
        d = self._feat(df)
        ret = d['close'].pct_change(24).shift(-24).fillna(0)
        y = np.where(ret>0.005, 2, np.where(ret<-0.005, 0, 1))
        skip = set(['open','high','low','close','volume','date'])
        self._cols = [c for c in d.columns if c not in skip]
        X = np.nan_to_num(d[self._cols].values, 0, 0, 0)
        v = np.arange(len(d)-24)
        valid = np.ones(len(v), dtype=bool)
        for i in range(len(v)):
            if v[i] >= len(X) or np.isnan(y[v[i]]):
                valid[i] = 0; continue
            for j in range(len(self._cols)):
                if np.isnan(X[v[i],j]) or np.isinf(abs(X[v[i],j])):
                    valid[i] = 0; break
        v2 = v[valid==1]
        Xt = X[v2]; yt = y[v2]
        if len(Xt) < 300: return False
        self._sm = Xt.mean(0); self._ss = Xt.std(0) + 0.0001
        Xs = (Xt-self._sm)/self._ss
        u,cnt = np.unique(yt, return_counts=True)
        w = {u[i]:len(yt)/(len(u)*cnt[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=400, max_depth=5, learning_rate=0.025,
                subsample=0.7, colsample_bytree=0.7, min_child_weight=5,
                gamma=0.2, reg_alpha=0.3, reg_lambda=2.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 'dp' not in d.columns: return d
        d.loc[(d['dp']==1)&(d['&t']>0.1), ['enter_long','enter_tag']] = (1,'L')
        d.loc[(d['dp']==1)&(d['&t']<-0.1), ['enter_short','enter_tag']] = (1,'S')
        return d

    def populate_exit_trend(self, d, m):
        if 'dp' not in d.columns: return d
        d.loc[d['&t'].abs()<0.03, ['exit_long','exit_short','exit_tag']] = (1,1,'E')
        return d