CB

from pandas import DataFrame
import talib.abstract as ta, numpy as np
from freqtrade.strategy import IStrategy

def fe(d):
    df=d.copy()
    for p in[1,3,6,12]: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']+1e-6)
    for p in[9,21]:e=ta.EMA(df,p);df['e'+str(p)+'d']=(df['close']-e)/e*100
    md=ta.MACD(df);df['md']=md['macd'];df['ms']=md['macdsignal']
    df['rsi']=ta.RSI(df,14);df['adx']=ta.ADX(df,14);df['vr']=df['volume']/ta.SMA(df['volume'],20);df['atr']=ta.ATR(df,14)/df['close']*100
    return df
FC=['r1','r3','r6','r12','hlr','cp','e9d','e21d','md','ms','rsi','adx','vr','atr']

class CB(IStrategy):
    """CatBoost per-pair"""
    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.06,"480":0.04,"1440":0.025,"4320":0};max_open_trades=4;startup_candle_count=2000;process_only_new_candles=False;use_exit_signal=False
    def __init__(self,config):super().__init__(config);self.m={};self.sm={};self.ss={}
    def populate_indicators(self,d,meta):
        p=meta['pair'];n=len(d)
        if p not in self.m and n>=2000:self._t(d.iloc[:1900].copy(),p)
        if p in self.m:
            d2=fe(d);X=np.nan_to_num(d2[FC].values,0,0,0);Xs=(X-self.sm[p])/self.ss[p];pp=self.m[p].predict_proba(Xs);cl=self.m[p].classes_;li=list(cl).index(2)if 2 in cl else 0;si=list(cl).index(0)if 0 in cl else 0;d['xs']=pp[:,li]-pp[:,si];d['xc']=np.maximum(pp[:,li],pp[:,si])
        return d
    def _t(self,df,p):
        from catboost import CatBoostClassifier as CCB
        d2=fe(df);X=np.nan_to_num(d2[FC].values,0,0,0);y=np.where(d2['close'].pct_change(12).shift(-12).fillna(0)>0.005,2,np.where(d2['close'].pct_change(12).shift(-12).fillna(0)<-0.005,0,1))
        v=np.arange(len(X)-12);v2=np.array([i for i in v if not np.isnan(y[i])and not np.isnan(X[i]).any()])
        if len(v2)<400:return False
        Xt,yt=X[v2],y[v2];self.sm[p]=Xt.mean(0);self.ss[p]=Xt.std(0)+1e-6;self.m[p]=CCB(iterations=300,depth=5,learning_rate=0.04,l2_leaf_reg=3,random_seed=42,verbose=0);self.m[p].fit((Xt-self.sm[p])/self.ss[p],yt);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),['enter_long','enter_tag']]=(1,'L');d.loc[(d['xc']>0.55)&(d['xs']<0),['enter_short','enter_tag']]=(1,'S');return d
    def populate_exit_trend(self,d,m):return d

class NN(IStrategy):
    """MLP Neural Network per-pair"""
    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.06,"480":0.04,"1440":0.025,"4320":0};max_open_trades=4;startup_candle_count=2000;process_only_new_candles=False;use_exit_signal=False
    def __init__(self,config):super().__init__(config);self.m={};self.s={};self.b={}
    def populate_indicators(self,d,meta):
        p=meta['pair'];n=len(d)
        if p not in self.m and n>=2000:self._t(d.iloc[:1900].copy(),p)
        if p in self.m:
            d2=fe(d);X=np.nan_to_num(d2[FC].values,0,0,0);Xs=(X-self.s[p])/self.b[p];pp=self.m[p].predict_proba(Xs);cl=self.m[p].classes_;li=list(cl).index(2)if 2 in cl else 0;si=list(cl).index(0)if 0 in cl else 0;d['xs']=pp[:,li]-pp[:,si];d['xc']=np.maximum(pp[:,li],pp[:,si])
        return d
    def _t(self,df,p):
        from sklearn.neural_network import MLPClassifier
        d2=fe(df);X=np.nan_to_num(d2[FC].values,0,0,0);y=np.where(d2['close'].pct_change(12).shift(-12).fillna(0)>0.005,2,np.where(d2['close'].pct_change(12).shift(-12).fillna(0)<-0.005,0,1))
        v=np.arange(len(X)-12);v2=np.array([i for i in v if not np.isnan(y[i])and not np.isnan(X[i]).any()])
        if len(v2)<400:return False
        Xt,yt=X[v2],y[v2];self.s[p]=Xt.mean(0);self.b[p]=Xt.std(0)+1e-6;self.m[p]=MLPClassifier(hidden_layer_sizes=(32,16),activation='relu',max_iter=500,random_state=42,early_stopping=True,validation_fraction=0.15);self.m[p].fit((Xt-self.s[p])/self.b[p],yt);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),['enter_long','enter_tag']]=(1,'L');d.loc[(d['xc']>0.55)&(d['xs']<0),['enter_short','enter_tag']]=(1,'S');return d
    def populate_exit_trend(self,d,m):return d

class VT(IStrategy):
    """Voting: Winner15m + RF both must agree"""
    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.06,"480":0.04,"1440":0.025,"4320":0};max_open_trades=4;startup_candle_count=2000;process_only_new_candles=False;use_exit_signal=False
    def __init__(self,config):super().__init__(config);self.m={};self.s={};self.b={}
    def populate_indicators(self,d,meta):
        d['e10']=ta.EMA(d,10);d['e30']=ta.EMA(d,30);md=ta.MACD(d);d['md']=md['macd'];d['ms']=md['macdsignal'];d['mom']=ta.ROC(d,3);d['adx']=ta.ADX(d,14);d['vr']=d['volume']/ta.SMA(d['volume'],20)
        d['wL']=(d['e10']>d['e30'])&(d['md']>d['ms'])&(d['mom']>0.1)&(d['adx']>18)&(d['vr']>1.0)
        d['wS']=(d['e10']<d['e30'])&(d['md']<d['ms'])&(d['mom']<-0.1)&(d['adx']>18)&(d['vr']>1.0)
        p=meta['pair'];n=len(d)
        if p not in self.m and n>=2000:self._t(d.iloc[:1900].copy(),p)
        if p in self.m:
            d2=fe(d);X=np.nan_to_num(d2[FC].values,0,0,0);Xs=(X-self.s[p])/self.b[p];pp=self.m[p].predict_proba(Xs);cl=self.m[p].classes_;li=list(cl).index(2)if 2 in cl else 0;si=list(cl).index(0)if 0 in cl else 0;d['xs']=pp[:,li]-pp[:,si];d['xc']=np.maximum(pp[:,li],pp[:,si])
        return d
    def _t(self,df,p):
        from sklearn.ensemble import RandomForestClassifier as RF
        d2=fe(df);X=np.nan_to_num(d2[FC].values,0,0,0);y=np.where(d2['close'].pct_change(12).shift(-12).fillna(0)>0.005,2,np.where(d2['close'].pct_change(12).shift(-12).fillna(0)<-0.005,0,1))
        v=np.arange(len(X)-12);v2=np.array([i for i in v if not np.isnan(y[i])and not np.isnan(X[i]).any()])
        if len(v2)<400:return False
        Xt,yt=X[v2],y[v2];self.s[p]=Xt.mean(0);self.b[p]=Xt.std(0)+1e-6;self.m[p]=RF(n_estimators=200,max_depth=10,min_samples_leaf=15,class_weight='balanced',random_state=42,n_jobs=-1);self.m[p].fit((Xt-self.s[p])/self.b[p],yt);return True
    def populate_entry_trend(self,d,m):
        if'xs'not in d.columns:return d;d.loc[(d['wL'])&(d['xc']>0.5)&(d['xs']>0),['enter_long','enter_tag']]=(1,'VL');d.loc[(d['wS'])&(d['xc']>0.5)&(d['xs']<0),['enter_short','enter_tag']]=(1,'VS');return d
    def populate_exit_trend(self,d,m):return d