Untitled

def EMA_RVI(df: pd.DataFrame, km, adx, rvi, swma) -> pd.DataFrame: #5min test >2021
    def calculate_tis(df: pd.DataFrame) -> pd.DataFrame:
        # # Heikin ashi
        df['haClose'] = (df['Open'] + df['High'] + df['Low'] + df['Close']) / 4
        df['haOpen'] = (df['Open'].shift() + df['Close'].shift()) / 2
        df['haHigh'] = df[['haClose', 'haOpen', 'High']].max(axis=1)
        df['haLow'] = df[['haClose', 'haOpen', 'Low']].max(axis=1)
        df['haGreen'] = (df['haClose'] > df['haOpen']).apply(lambda x: bool(x))   
        df['log_return'] = np.log(df['haClose'] / df['haClose'].shift(1))   
        
        df['adx'] = pandas_ta.adx(df['High'], df['Low'], df['Close'], length=adx).iloc[:, 0]
        df['ema'] = pandas_ta.ema(df['Close'], length=km)
        df['ema_m'] = pandas_ta.ema(df['Close'], length=50)
        df['ema_s'] = pandas_ta.ema(df['Close'], length=100)
        rvgi = pandas_ta.rvgi(df['Open'], df['High'], df['Low'], df['Close'], rvi, swma)
        df['rvgi'] = rvgi.iloc[:, 0]
        df['rvgi_s'] = rvgi.iloc[:, 1]

        return df.dropna()
    
    def calculate_signals(df: pd.DataFrame) -> pd.DataFrame:
        df['enter_long'] = np.where((df['adx'] > df['adx'].rolling(adx).median())
                                    & (df['haGreen']==True)
                                    & (df['Close'] > df['ema'])
                                    & (df['rvgi'] > df['rvgi_s'])
                                    , 1, 0)
        df['exit_long'] = np.where((df['adx'] < df['adx'].rolling(adx).median())
                                    & (df['rvgi'] < df['rvgi_s'])
                                    & (df['haGreen']==False)
                                    & (df['ema'] < df['ema_m'])
                                    , 1, 0)
        df['enter_short'] = np.where((df['adx'] > df['adx'].rolling(adx).median())
                                    & (df['haGreen']==False)
                                    & (df['Close'] < df['ema'])
                                    & (df['rvgi'] < df['rvgi_s'])
                                    , 1, 0)
        df['exit_short'] = np.where((df['adx'] < df['adx'].rolling(adx).median())
                                    & (df['rvgi'] > df['rvgi_s'])
                                    & (df['haGreen']==True)
                                    & (df['ema'] > df['ema_m'])
                                    , 1, 0)
        

        return df
    
    df = df.copy(deep=True)
    tradings = calculate_tis(df)
    tradings = calculate_signals(tradings)

    tradings = tradings.loc[~((tradings['enter_long']==0)&(tradings['exit_long']==0)&(tradings['enter_short']==0)&(tradings['exit_short']==0))]
    tradings['Position'] = np.where(tradings['enter_long'] == 1, 1, 0)
    tradings['Position'] = np.where((tradings['enter_short'] == 1), -1, tradings['Position'])
    tradings['Position'] = np.where((tradings['Position'].shift() == -1) & (tradings['exit_long'] == 1), -1, tradings['Position'])
    tradings['Position'] = np.where((tradings['Position'].shift() == 1) & (tradings['exit_short'] == 1), 1, tradings['Position'])

    tradings = df.merge(tradings.loc[:, :], how='left')
    tradings = tradings.fillna(method='ffill').dropna()
    tradings = tradings.reset_index(drop=True)   

    return tradings