diff --git a/src/pages/backtesting/backtesting_page.py b/src/pages/backtesting/backtesting_page.py index 10ba0fa..ada8e1b 100644 --- a/src/pages/backtesting/backtesting_page.py +++ b/src/pages/backtesting/backtesting_page.py @@ -18,25 +18,55 @@ class DynamicStrategy(Strategy): # Initialize all selected indicators for ind_name, ind_config in self.indicator_configs.items(): if ind_config['type'] == 'SMA': - self.indicators[ind_name] = self.I(lambda x: ta.sma(x, length=int(ind_config['params']['length'])).fillna(0).to_numpy(), self.data.Close) + def sma_calc(x): + result = ta.sma(x, length=int(ind_config['params']['length'])) + return result.fillna(method='ffill').fillna(0).values + self.indicators[ind_name] = self.I(sma_calc, self.data.Close) + elif ind_config['type'] == 'EMA': - self.indicators[ind_name] = self.I(lambda x: ta.ema(x, length=int(ind_config['params']['length'])).fillna(0).to_numpy(), self.data.Close) + def ema_calc(x): + result = ta.ema(x, length=int(ind_config['params']['length'])) + return result.fillna(method='ffill').fillna(0).values + self.indicators[ind_name] = self.I(ema_calc, self.data.Close) + elif ind_config['type'] == 'RSI': - self.indicators[ind_name] = self.I(lambda x: ta.rsi(x, length=int(ind_config['params']['length'])).fillna(0).to_numpy(), self.data.Close) + def rsi_calc(x): + result = ta.rsi(x, length=int(ind_config['params']['length'])) + return result.fillna(method='ffill').fillna(0).values + self.indicators[ind_name] = self.I(rsi_calc, self.data.Close) + elif ind_config['type'] == 'MACD': fast = int(ind_config['params']['fast']) slow = int(ind_config['params']['slow']) signal = int(ind_config['params']['signal']) - macd = self.I(lambda x: ta.macd(x, fast=fast, slow=slow, signal=signal), self.data.Close) - self.indicators[f"{ind_name}_macd"] = self.I(lambda x: x.iloc[:,0].fillna(0).to_numpy(), macd) - self.indicators[f"{ind_name}_signal"] = self.I(lambda x: x.iloc[:,2].fillna(0).to_numpy(), macd) + + def macd_calc(x): + result = ta.macd(x, fast=fast, slow=slow, signal=signal) + if result is None or result.empty: + return np.zeros(len(x)), np.zeros(len(x)) + return (result.iloc[:,0].fillna(method='ffill').fillna(0).values, + result.iloc[:,2].fillna(method='ffill').fillna(0).values) + + macd_vals = self.I(macd_calc, self.data.Close) + self.indicators[f"{ind_name}_macd"] = macd_vals[0] + self.indicators[f"{ind_name}_signal"] = macd_vals[1] + elif ind_config['type'] == 'BB': length = int(ind_config['params']['length']) std = float(ind_config['params']['std']) - bbands = self.I(lambda x: ta.bbands(x, length=length, std=std), self.data.Close) - self.indicators[f"{ind_name}_upper"] = self.I(lambda x: x.iloc[:,0].fillna(0).to_numpy(), bbands) - self.indicators[f"{ind_name}_middle"] = self.I(lambda x: x.iloc[:,1].fillna(0).to_numpy(), bbands) - self.indicators[f"{ind_name}_lower"] = self.I(lambda x: x.iloc[:,2].fillna(0).to_numpy(), bbands) + + def bb_calc(x): + result = ta.bbands(x, length=length, std=std) + if result is None or result.empty: + return np.zeros(len(x)), np.zeros(len(x)), np.zeros(len(x)) + return (result.iloc[:,0].fillna(method='ffill').fillna(0).values, + result.iloc[:,1].fillna(method='ffill').fillna(0).values, + result.iloc[:,2].fillna(method='ffill').fillna(0).values) + + bb_vals = self.I(bb_calc, self.data.Close) + self.indicators[f"{ind_name}_upper"] = bb_vals[0] + self.indicators[f"{ind_name}_middle"] = bb_vals[1] + self.indicators[f"{ind_name}_lower"] = bb_vals[2] def next(self): # Simple example strategy - should be customized based on user input