X1h

"""XGBoost multi-TF — instance-level state, step-by-step"""
from pandas import DataFrame
import talib.abstract as ta
import numpy as np
from freqtrade.strategy import IStrategy

def build_features(d):
    df = d.copy()
    for p in [1,3,6,12,24,48]: df['r'+str(p)] = df['close'].pct_change(p)
    df['hlr'] = (df['high']-df['low'])/df['close']
    df['cp'] = (df['close']-df['low'])/(df['high']-df['low']+0.0001)
    for p in [9,21,55]:
        e = ta.EMA(df, timeperiod=p)
        df['e'+str(p)+'d'] = (df['close']-e)/e*100
    macd = ta.MACD(df); df['md'] = macd['macd']; df['ms'] = macd['macdsignal']
    df['rsi'] = ta.RSI(df, 14)
    bb = ta.BBANDS(df, 20)
    df['bw'] = (bb['upperband']-bb['lowerband'])/bb['middleband']
    df['atr'] = ta.ATR(df,14)/df['close']*100
    df['vr'] = df['volume']/ta.SMA(df['volume'],20)
    df['adx'] = ta.ADX(df,14)
    st = ta.STOCH(df); df['sk'] = st['slowk']; df['sd'] = st['slowd']
    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+0.0001))
    return df

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

def get_X(df):
    return np.nan_to_num(df[[c for c in FEATURE_COLS if c in df.columns]].values, 0, 0, 0)

class X1h(IStrategy):
    INTERFACE_VERSION = 3; timeframe = '1h'; can_short = True
    stoploss = -0.03; trailing_stop = True
    trailing_stop_positive = 0.01; trailing_stop_positive_offset = 0.02
    trailing_only_offset_is_reached = True
    minimal_roi = {"0":0.06,"480":0.04,"1440":0.025,"2880":0}
    max_open_trades = 4; startup_candle_count = 2000
    process_only_new_candles = False; use_exit_signal = True

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

    def populate_indicators(self, dataframe, metadata):
        n = len(dataframe)
        if not self._trained and n >= 2100:
            self._train(dataframe.iloc[:2000].copy())
        if self._models is not None:
            dataframe = self._predict(dataframe)
        return dataframe

    def _train(self, df):
        import xgboost as xgb
        df = build_features(df)
        y = np.zeros(len(df))
        for i in range(len(df)-12):
            r = df['close'].iloc[i+12]/df['close'].iloc[i] - 1
            y[i] = 2 if r > 0.005 else (0 if r < -0.005 else 1)
        X = get_X(df); Xt = X[:len(df)-12]; yt = y[:len(df)-12]
        if len(Xt) < 200: return
        self._sm = np.mean(Xt,0); self._ss = np.std(Xt,0)+0.0001
        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[v] for v in yt])
        models = []
        for s in [42,73,99,17,55]:
            m = xgb.XGBClassifier(n_estimators=200, max_depth=4, learning_rate=0.05,
                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)
            models.append(m)
        self._models = models; self._trained = True

    def _predict(self, df):
        df2 = build_features(df); X = get_X(df2)
        Xs = (X-self._sm)/self._ss
        ap = np.mean([m.predict_proba(Xs) for m in self._models], axis=0)
        cl = self._models[0].classes_
        li, si = list(cl).index(2) if 2 in cl else -1, list(cl).index(0) if 0 in cl else -1
        df['xgL'] = ap[:,li] if li>=0 else 0
        df['xgS'] = ap[:,si] if si>=0 else 0
        df['xgC'] = np.maximum(df['xgL'].values, df['xgS'].values)
        return df

    def populate_entry_trend(self, d, m):
        if 'xgC' not in d.columns: return d
        d.loc[(d['xgC']>0.55)&(d['xgL']>d['xgS']), ['enter_long','enter_tag']] = (1, 'L')
        d.loc[(d['xgC']>0.55)&(d['xgS']>d['xgL']), ['enter_short','enter_tag']] = (1, 'S')
        return d

    def populate_exit_trend(self, d, m):
        if 'xgC' not in d.columns: return d
        d.loc[d['xgC']<0.45, ['exit_long','exit_short','exit_tag']] = (1, 1, 'E')
        return d


class X4h(IStrategy):
    INTERFACE_VERSION = 3; timeframe = '4h'; can_short = True
    stoploss = -0.04; trailing_stop = True
    trailing_stop_positive = 0.015; trailing_stop_positive_offset = 0.03
    trailing_only_offset_is_reached = True
    minimal_roi = {"0":0.08,"480":0.05,"1440":0.03,"2880":0}
    max_open_trades = 3; startup_candle_count = 1000
    process_only_new_candles = False; use_exit_signal = True

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

    def populate_indicators(self, dataframe, metadata):
        n = len(dataframe)
        if not self._trained and n >= 900:
            self._train(dataframe.iloc[:800].copy())
        if self._models is not None:
            dataframe = self._predict(dataframe)
        return dataframe

    def _train(self, df):
        import xgboost as xgb
        df = build_features(df); y = np.zeros(len(df))
        for i in range(len(df)-12):
            r = df['close'].iloc[i+12]/df['close'].iloc[i] - 1
            y[i] = 2 if r > 0.005 else (0 if r < -0.005 else 1)
        X = get_X(df); Xt = X[:len(df)-12]; yt = y[:len(df)-12]
        if len(Xt) < 200: return
        self._sm = np.mean(Xt,0); self._ss = np.std(Xt,0)+0.0001
        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[v] for v in yt])
        models = []
        for s in [42,73,99,17,55]:
            m = xgb.XGBClassifier(n_estimators=200, max_depth=4, learning_rate=0.05,
                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); models.append(m)
        self._models = models; self._trained = True

    def _predict(self, df):
        df2 = build_features(df); X = get_X(df2)
        Xs = (X-self._sm)/self._ss
        ap = np.mean([m.predict_proba(Xs) for m in self._models], axis=0)
        cl = self._models[0].classes_
        li, si = list(cl).index(2) if 2 in cl else -1, list(cl).index(0) if 0 in cl else -1
        df['xgL'] = ap[:,li] if li>=0 else 0
        df['xgS'] = ap[:,si] if si>=0 else 0
        df['xgC'] = np.maximum(df['xgL'].values, df['xgS'].values)
        return df

    def populate_entry_trend(self, d, m):
        if 'xgC' not in d.columns: return d
        d.loc[(d['xgC']>0.6)&(d['xgL']>d['xgS']), ['enter_long','enter_tag']] = (1, 'L')
        d.loc[(d['xgC']>0.6)&(d['xgS']>d['xgL']), ['enter_short','enter_tag']] = (1, 'S')
        return d

    def populate_exit_trend(self, d, m):
        if 'xgC' not in d.columns: return d
        d.loc[d['xgC']<0.5, ['exit_long','exit_short','exit_tag']] = (1, 1, 'E')
        return d


class X1d(IStrategy):
    INTERFACE_VERSION = 3; timeframe = '1d'; can_short = True
    stoploss = -0.05; trailing_stop = True
    trailing_stop_positive = 0.02; trailing_stop_positive_offset = 0.04
    trailing_only_offset_is_reached = True
    minimal_roi = {"0":0.10,"720":0.06,"2160":0.04,"4320":0}
    max_open_trades = 2; startup_candle_count = 500
    process_only_new_candles = False; use_exit_signal = True

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

    def populate_indicators(self, dataframe, metadata):
        n = len(dataframe)
        if not self._trained and n >= 500:
            self._train(dataframe.iloc[:400].copy())
        if self._models is not None:
            dataframe = self._predict(dataframe)
        return dataframe

    def _train(self, df):
        import xgboost as xgb
        df = build_features(df); y = np.zeros(len(df))
        for i in range(len(df)-12):
            r = df['close'].iloc[i+12]/df['close'].iloc[i] - 1
            y[i] = 2 if r > 0.005 else (0 if r < -0.005 else 1)
        X = get_X(df); Xt = X[:len(df)-12]; yt = y[:len(df)-12]
        if len(Xt) < 200: return
        self._sm = np.mean(Xt,0); self._ss = np.std(Xt,0)+0.0001
        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[v] for v in yt])
        models = []
        for s in [42,73,99,17,55]:
            m = xgb.XGBClassifier(n_estimators=200, max_depth=4, learning_rate=0.05,
                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); models.append(m)
        self._models = models; self._trained = True

    def _predict(self, df):
        df2 = build_features(df); X = get_X(df2)
        Xs = (X-self._sm)/self._ss
        ap = np.mean([m.predict_proba(Xs) for m in self._models], axis=0)
        cl = self._models[0].classes_
        li, si = list(cl).index(2) if 2 in cl else -1, list(cl).index(0) if 0 in cl else -1
        df['xgL'] = ap[:,li] if li>=0 else 0
        df['xgS'] = ap[:,si] if si>=0 else 0
        df['xgC'] = np.maximum(df['xgL'].values, df['xgS'].values)
        return df

    def populate_entry_trend(self, d, m):
        if 'xgC' not in d.columns: return d
        d.loc[(d['xgC']>0.6)&(d['xgL']>d['xgS']), ['enter_long','enter_tag']] = (1, 'L')
        d.loc[(d['xgC']>0.6)&(d['xgS']>d['xgL']), ['enter_short','enter_tag']] = (1, 'S')
        return d

    def populate_exit_trend(self, d, m):
        if 'xgC' not in d.columns: return d
        d.loc[d['xgC']<0.5, ['exit_long','exit_short','exit_tag']] = (1, 1, 'E')
        return d