Untitled

mail@pastecode.io avatar
unknown
python
a year ago
3.6 kB
5
Indexable
Never
def HWC(df: pd.DataFrame, na: int, nb: int, nc: int, nd: int, m: int, vol: int, th: int) -> pd.DataFrame:
    def calculate_tis(df: pd.DataFrame, na: int, nb: int, nc: int, nd: int, m: int) -> 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'] = np.maximum(df['High'], df['Open'], df['Close']) 
        # df['haLow'] = np.minimum(df['Low'], df['Open'], df['Close'])

        # df['grad'] = ((m*df['haClose'] + df['haHigh'] + df['haLow'])/(m+2)).diff()

        # df['norClose'] = (m*df['haClose'] + df['haHigh'] + df['haLow'])/(m+2)
        # # Volatility
        # df['volatility'] = df['High'] - df['Low']

        # df['MACD'] = ta.trend.MACD(df['haClose'], windonb=nb, windona=na, windonc=nc).macd()
        # df['MA_signal'] = ta.trend.MACD(df['haClose'], windonb=nb, windona=na, windonc=nc).HWCsignal()
        # df['HWChist'] = ta.trend.MACD(df['haClose'], windonb=nb, windona=na, windonc=nc).HWCdiff()

        df['hwm'] = pandas_ta.hwc(df['Close'], na=na/10, nb=nb/10, nc=nc/10, nd=nd/10)['HWM']
        df['hwu'] = pandas_ta.hwc(df['Close'], na=na/10, nb=nb/10, nc=nc/10, nd=nd/10)['HWU']
        df['hwl'] = pandas_ta.hwc(df['Close'], na=na/10, nb=nb/10, nc=nc/10, nd=nd/10)['HWL']

        return df.dropna()
    
    def calculate_signals(df: pd.DataFrame) -> pd.DataFrame:
        # Enter Long Rule: Price breaks through the high channel
        df['enter_long'] = np.where((df['Close'].shift(1) > df['hwl'].shift(1)) & (df['Close'].shift(2) < df['hwl'].shift(2)), 1, 0)

        #  Exit Long Rule: Price drops below the mid channel
        df['exit_long'] = np.where((df['Close'].shift(1) < df['hwu'].shift(1)) & (df['Close'].shift(2) > df['hwu'].shift(2)), 1, 0)

        #  Enter Short Rule: Price drops below the low channel
        df['enter_short'] = np.where((df['Close'].shift(1) < df['hwu'].shift(1)) & (df['Close'].shift(2) > df['hwu'].shift(2)), 1, 0)
                                    
        #  Enter Short Rule: Price breaks through the mid channel
        df['exit_short'] = np.where((df['Close'].shift(1) > df['hwl'].shift(1)) & (df['Close'].shift(2) < df['hwl'].shift(2)), 1, 0)

        return df
    
    
    df = df.copy(deep=True)
    tradings = calculate_tis(df, na=na, nb=nb, nc=nc, nd=nd, m=0)
    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, -1)

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

    # indices = np.where(tradings.loc[:, 'Position'] != tradings.loc[:, 'Position'].shift())[0]
    # for i in range(len(indices)-1):
    #     start = indices[i]
    #     end = indices[i+1] - 1

    #     if tradings.loc[start, 'Position'] == 1:
    #         for j in range(end-start+1):
    #             if tradings.loc[start+j, 'Close'] < tradings.loc[start, 'Close'] - th:
    #                 tradings.loc[start+j+1:end, 'Position'] = 0
    #     elif tradings.loc[start, 'Position'] == -1:
    #         for j in range(end-start+1):
    #             if tradings.loc[start+j, 'Close'] > tradings.loc[start, 'Close'] + th:
    #                 tradings.loc[start+j+1:end, 'Position'] = 0 

    return tradings