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