diff --git a/src/StockBench/algorithm/algorithm.py b/src/StockBench/algorithm/algorithm.py index b5a6e7e..1aedf48 100644 --- a/src/StockBench/algorithm/algorithm.py +++ b/src/StockBench/algorithm/algorithm.py @@ -79,7 +79,7 @@ def get_additional_days(self) -> int: key = keys[i] value = values[i] for trigger in triggers: - if trigger.strategy_symbol in key: + if trigger.indicator_symbol in key: num = trigger.additional_days(key, value) if additional_days < num: additional_days = num @@ -95,11 +95,11 @@ def add_indicator_data(self, data_manager) -> None: # find all buy algorithm and add their indicator to the data for key in self.strategy[BUY_SIDE].keys(): for trigger in triggers: - if trigger.strategy_symbol in key: + if trigger.indicator_symbol in key: trigger.add_to_data(key, self.strategy[BUY_SIDE][key], BUY_SIDE, data_manager) elif AND_KEY in key: for inner_key in self.strategy[BUY_SIDE][key].keys(): - if trigger.strategy_symbol in inner_key: + if trigger.indicator_symbol in inner_key: trigger.add_to_data(inner_key, self.strategy[BUY_SIDE][key][inner_key], BUY_SIDE, data_manager) @@ -109,11 +109,11 @@ def add_indicator_data(self, data_manager) -> None: # find all sell algorithm and add their indicator to the data for key in self.strategy[SELL_SIDE].keys(): for trigger in triggers: - if trigger.strategy_symbol in key: + if trigger.indicator_symbol in key: trigger.add_to_data(key, self.strategy[SELL_SIDE][key], SELL_SIDE, data_manager) elif AND_KEY in key: for inner_key in self.strategy[SELL_SIDE][key].keys(): - if trigger.strategy_symbol in inner_key: + if trigger.indicator_symbol in inner_key: trigger.add_to_data(inner_key, self.strategy[SELL_SIDE][key][inner_key], SELL_SIDE, data_manager) @@ -244,7 +244,7 @@ def __handle_and_triggers(self, triggers: List[Trigger], data_manager: DataManag key_matched_with_trigger = False # check all algorithm for trigger in triggers: - if trigger.strategy_symbol in inner_key: + if trigger.indicator_symbol in inner_key: key_matched_with_trigger = True trigger_hit = trigger.check_trigger( inner_key, @@ -280,7 +280,7 @@ def __handle_or_triggers(self, triggers: List[Trigger], data_manager: DataManage key_matched_with_trigger = False # check all algorithm for trigger in triggers: - if trigger.strategy_symbol in key: + if trigger.indicator_symbol in key: key_matched_with_trigger = True trigger_hit = trigger.check_trigger( key, diff --git a/src/StockBench/indicator/trigger.py b/src/StockBench/indicator/trigger.py index 353f0ad..2a55c80 100644 --- a/src/StockBench/indicator/trigger.py +++ b/src/StockBench/indicator/trigger.py @@ -18,8 +18,8 @@ class Trigger: SELL = 1 AGNOSTIC = 2 - def __init__(self, strategy_symbol: str, side: str): - self.strategy_symbol = strategy_symbol + def __init__(self, indicator_symbol: str, side: str): + self.indicator_symbol = indicator_symbol self.__side = side def get_side(self): @@ -63,6 +63,123 @@ def _parse_rule_value(self, rule_value: str, data_manager: DataManager, return operator, trigger_value + @staticmethod + def _parse_rule_key(rule_key: str, indicator_symbol: str, data_manager: DataManager, + current_day_index: int) -> float: + """Translates a complex rule key for an indicator value where the indicator has a default value. + Can have 0, 1, or 2 number groupings. + """ + rule_key_number_groups = Trigger.find_all_nums_in_str(rule_key) + if len(rule_key_number_groups) == 0: + # rule key does not define an indicator length (use default) + if SLOPE_SYMBOL in rule_key: + raise StrategyIndicatorError(f'{indicator_symbol} rule key: {rule_key} does not contain ' + f'enough number groupings!') + indicator_value = float(data_manager.get_data_point(indicator_symbol, current_day_index)) + elif len(rule_key_number_groups) == 1: + if SLOPE_SYMBOL in rule_key: + # make sure the number is after the slope emblem and not the RSI emblem + if rule_key.split(str(rule_key_number_groups))[0] == indicator_symbol + SLOPE_SYMBOL: + raise StrategyIndicatorError(f'{indicator_symbol} rule key: {rule_key} does not contain ' + f'a slope value!') + # rule key defines an indicator length (not using default) + column_title = f'{indicator_symbol}{int(rule_key_number_groups[0])}' + indicator_value = float(data_manager.get_data_point(column_title, current_day_index)) + elif len(rule_key_number_groups) == 2: + column_title = f'{indicator_symbol}{int(rule_key_number_groups[0])}' + # 2 number groupings suggests the $slope indicator is being used + if SLOPE_SYMBOL in rule_key: + slope_window_length = int(rule_key_number_groups[1]) + + # data request length is window - 1 to account for the current day index being a part of the window + slope_data_request_length = slope_window_length - 1 + + indicator_value = Trigger.calculate_slope( + float(data_manager.get_data_point(column_title, current_day_index)), + float(data_manager.get_data_point(column_title, current_day_index - slope_data_request_length)), + slope_window_length + ) + else: + raise StrategyIndicatorError(f'{indicator_symbol} rule key: {rule_key} contains too many number ' + f'groupings! Are you missing a $slope emblem?') + else: + raise StrategyIndicatorError(f'{indicator_symbol} rule key: {rule_key} contains invalid number ' + f'groupings!') + + return indicator_value + + @staticmethod + def _parse_rule_key_no_default_indicator_length(rule_key: str, indicator_symbol: str, data_manager: DataManager, + current_day_index: int) -> float: + """Translates a complex rule key for an indicator value where the indicator DOES NOT have a default value. + Can have 1, or 2 number groupings. + """ + key_number_groupings = Trigger.find_all_nums_in_str(rule_key) + + if len(key_number_groupings) == 1: + if SLOPE_SYMBOL in rule_key: + raise StrategyIndicatorError(f'{indicator_symbol} rule key: {rule_key} does not contain ' + f'enough number groupings!') + column_title = f'{indicator_symbol}{int(key_number_groupings[0])}' + indicator_value = float(data_manager.get_data_point(column_title, current_day_index)) + elif len(key_number_groupings) == 2: + column_title = f'{indicator_symbol}{int(key_number_groupings[0])}' + # 2 number groupings suggests the $slope indicator is being used + if SLOPE_SYMBOL in rule_key: + slope_window_length = int(key_number_groupings[1]) + + # data request length is window - 1 to account for the current day index being a part of the window + slope_data_request_length = slope_window_length - 1 + + indicator_value = Trigger.calculate_slope( + float(data_manager.get_data_point(column_title, current_day_index)), + float(data_manager.get_data_point(column_title, current_day_index - slope_data_request_length)), + slope_window_length + ) + else: + raise StrategyIndicatorError(f'{indicator_symbol} rule key: {rule_key} contains too many number ' + f'groupings! Are you missing a $slope emblem?') + else: + raise StrategyIndicatorError(f'{indicator_symbol} rule key: {rule_key} contains invalid number ' + f'groupings!') + + return indicator_value + + @staticmethod + def _parse_rule_key_no_indicator_length(rule_key: str, indicator_symbol: str, data_manager: DataManager, + current_day_index: int) -> float: + """Parser for parsing the key into the indicator value.""" + key_number_groupings = Trigger.find_all_nums_in_str(rule_key) + + # MACD can only have slope emblem therefore 1 or 0 number groupings are acceptable + if len(key_number_groupings) == 0: + if SLOPE_SYMBOL in rule_key: + raise StrategyIndicatorError(f'{indicator_symbol} rule key: {rule_key} does not contain' + f' enough number groupings!') + indicator_value = float(data_manager.get_data_point(indicator_symbol, current_day_index)) + elif len(key_number_groupings) == 1: + # 1 number grouping suggests the $slope indicator is being used + if SLOPE_SYMBOL in rule_key: + slope_window_length = int(key_number_groupings[0]) + + # data request length is window - 1 to account for the current day index being a part of the window + slope_data_request_length = slope_window_length - 1 + + indicator_value = Trigger.calculate_slope( + float(data_manager.get_data_point(indicator_symbol, current_day_index)), + float(data_manager.get_data_point(indicator_symbol, current_day_index - + slope_data_request_length)), + slope_window_length + ) + else: + raise StrategyIndicatorError(f'{indicator_symbol} rule key: {rule_key} contains too many number ' + f'groupings! Are you missing a $slope emblem?') + else: + raise StrategyIndicatorError(f'{indicator_symbol} rule key: {rule_key} contains ' + f'invalid number groupings!') + + return indicator_value + @staticmethod def _add_trigger_column(column_name: str, trigger_value: float, data_manager: DataManager): """Add a trigger value to the df.""" diff --git a/src/StockBench/indicators/candlestick_color/trigger.py b/src/StockBench/indicators/candlestick_color/trigger.py index 03cff5b..3022ede 100644 --- a/src/StockBench/indicators/candlestick_color/trigger.py +++ b/src/StockBench/indicators/candlestick_color/trigger.py @@ -8,8 +8,8 @@ class CandlestickColorTrigger(Trigger): - def __init__(self, strategy_symbol): - super().__init__(strategy_symbol, side=Trigger.AGNOSTIC) + def __init__(self, indicator_symbol): + super().__init__(indicator_symbol, side=Trigger.AGNOSTIC) def additional_days(self, rule_key, value_value) -> int: """Calculate the additional days required. @@ -19,7 +19,7 @@ def additional_days(self, rule_key, value_value) -> int: value_value (any): The value from the strategy. """ if len(value_value.keys()) == 0: - raise StrategyIndicatorError(f'{self.strategy_symbol} key: {rule_key} must have at least one color child ' + raise StrategyIndicatorError(f'{self.indicator_symbol} key: {rule_key} must have at least one color child ' f'key') additional_days = 0 @@ -58,7 +58,7 @@ def check_trigger(self, rule_key, rule_value, data_manager, position, current_da key_count = len(rule_value) if key_count == 0: - raise StrategyIndicatorError(f'{self.strategy_symbol} key: {rule_key} must have at least one color child ' + raise StrategyIndicatorError(f'{self.indicator_symbol} key: {rule_key} must have at least one color child ' f'key') trigger_colors = [rule_value[value_key] for value_key in sorted(rule_value.keys())] diff --git a/src/StockBench/indicators/ema/trigger.py b/src/StockBench/indicators/ema/trigger.py index a7cdbdd..bc696ad 100644 --- a/src/StockBench/indicators/ema/trigger.py +++ b/src/StockBench/indicators/ema/trigger.py @@ -1,6 +1,5 @@ import logging import statistics -from StockBench.constants import * from StockBench.indicator.trigger import Trigger from StockBench.indicator.exceptions import StrategyIndicatorError from StockBench.simulation_data.data_manager import DataManager @@ -9,8 +8,8 @@ class EMATrigger(Trigger): - def __init__(self, strategy_symbol): - super().__init__(strategy_symbol, side=Trigger.AGNOSTIC) + def __init__(self, indicator_symbol): + super().__init__(indicator_symbol, side=Trigger.AGNOSTIC) def additional_days(self, rule_key, value_value) -> int: """Calculate the additional days required. @@ -23,7 +22,7 @@ def additional_days(self, rule_key, value_value) -> int: nums = list(map(int, self.find_all_nums_in_str(rule_key))) if nums: return max(nums) - raise StrategyIndicatorError(f'{self.strategy_symbol} key: {rule_key} must have an indicator length!') + raise StrategyIndicatorError(f'{self.indicator_symbol} key: {rule_key} must have an indicator length!') def add_to_data(self, rule_key, rule_value, side, data_manager): """Add data to the dataframe. @@ -39,7 +38,7 @@ def add_to_data(self, rule_key, rule_value, side, data_manager): indicator_length = int(nums[0]) self.__add_ema(indicator_length, data_manager) else: - raise StrategyIndicatorError(f'{self.strategy_symbol} key: {rule_key} must have an indicator length!') + raise StrategyIndicatorError(f'{self.indicator_symbol} key: {rule_key} must have an indicator length!') def check_trigger(self, rule_key, rule_value, data_manager, position, current_day_index) -> bool: """Trigger logic for EMA. @@ -54,52 +53,20 @@ def check_trigger(self, rule_key, rule_value, data_manager, position, current_da return: bool: True if a trigger was hit. """ - log.debug(f'Checking {self.strategy_symbol} algorithm: {rule_key}...') + log.debug(f'Checking {self.indicator_symbol} algorithm: {rule_key}...') - indicator_value = self.__parse_key(rule_key, data_manager, current_day_index) + indicator_value = Trigger._parse_rule_key_no_default_indicator_length(rule_key, self.indicator_symbol, data_manager, + current_day_index) operator, trigger_value = self._parse_rule_value(rule_value, data_manager, current_day_index) - log.debug(f'{self.strategy_symbol} algorithm: {rule_key} checked successfully') + log.debug(f'{self.indicator_symbol} algorithm: {rule_key} checked successfully') return Trigger.basic_trigger_check(indicator_value, operator, trigger_value) - def __parse_key(self, rule_key: str, data_manager: DataManager, current_day_index: int) -> float: - """Parser for parsing the key into the indicator value.""" - key_number_groupings = self.find_all_nums_in_str(rule_key) - - if len(key_number_groupings) == 1: - if SLOPE_SYMBOL in rule_key: - raise StrategyIndicatorError(f'{self.strategy_symbol} rule key: {rule_key} does not contain ' - f'enough number groupings!') - column_title = f'{self.strategy_symbol}{int(key_number_groupings[0])}' - indicator_value = float(data_manager.get_data_point(column_title, current_day_index)) - elif len(key_number_groupings) == 2: - column_title = f'{self.strategy_symbol}{int(key_number_groupings[0])}' - # 2 number groupings suggests the $slope indicator is being used - if SLOPE_SYMBOL in rule_key: - slope_window_length = int(key_number_groupings[1]) - - # data request length is window - 1 to account for the current day index being a part of the window - slope_data_request_length = slope_window_length - 1 - - indicator_value = self.calculate_slope( - float(data_manager.get_data_point(column_title, current_day_index)), - float(data_manager.get_data_point(column_title, current_day_index - slope_data_request_length)), - slope_window_length - ) - else: - raise StrategyIndicatorError(f'{self.strategy_symbol} rule key: {rule_key} contains too many number ' - f'groupings! Are you missing a $slope emblem?') - else: - raise StrategyIndicatorError(f'{self.strategy_symbol} rule key: {rule_key} contains invalid number ' - f'groupings!') - - return indicator_value - def __add_ema(self, length: int, data_manager: DataManager): """Pre-calculate the EMA values and add them to the df.""" - column_title = f'{self.strategy_symbol}{length}' + column_title = f'{self.indicator_symbol}{length}' # if we already have EMA values in the df, we don't need to add them again for col_name in data_manager.get_column_names(): diff --git a/src/StockBench/indicators/macd/trigger.py b/src/StockBench/indicators/macd/trigger.py index f222e52..b0f3b44 100644 --- a/src/StockBench/indicators/macd/trigger.py +++ b/src/StockBench/indicators/macd/trigger.py @@ -15,10 +15,8 @@ class MACDTrigger(Trigger): SMALL_EMA_LENGTH = 12 - DATA_COLUMN_TITLE = 'MACD' - - def __init__(self, strategy_symbol): - super().__init__(strategy_symbol, side=Trigger.AGNOSTIC) + def __init__(self, indicator_symbol): + super().__init__(indicator_symbol, side=Trigger.AGNOSTIC) def additional_days(self, rule_key, value_value) -> int: """Calculate the additional days required. @@ -40,12 +38,12 @@ def add_to_data(self, rule_key, rule_value, side, data_manager): """ # if we already have MACD values in the df, we don't need to add them again for col_name in data_manager.get_column_names(): - if self.DATA_COLUMN_TITLE == col_name: + if self.indicator_symbol == col_name: return price_data = data_manager.get_column_data(data_manager.CLOSE) - data_manager.add_column(self.DATA_COLUMN_TITLE, self.__calculate_macd(price_data)) + data_manager.add_column(self.indicator_symbol, self.__calculate_macd(price_data)) def check_trigger(self, rule_key, rule_value, data_manager, position, current_day_index) -> bool: """Trigger logic for EMA. @@ -60,49 +58,17 @@ def check_trigger(self, rule_key, rule_value, data_manager, position, current_da return: bool: True if a trigger was hit. """ - log.debug(f'Checking {self.strategy_symbol} algorithm: {rule_key}...') + log.debug(f'Checking {self.indicator_symbol} algorithm: {rule_key}...') - indicator_value = self.__parse_key(rule_key, data_manager, current_day_index) + indicator_value = Trigger._parse_rule_key_no_indicator_length(rule_key, self.indicator_symbol, data_manager, + current_day_index) operator, trigger_value = self._parse_rule_value(rule_value, data_manager, current_day_index) - log.debug(f'{self.strategy_symbol} algorithm: {rule_key} checked successfully') + log.debug(f'{self.indicator_symbol} algorithm: {rule_key} checked successfully') return Trigger.basic_trigger_check(indicator_value, operator, trigger_value) - def __parse_key(self, rule_key: str, data_manager: DataManager, current_day_index: int) -> float: - """Parser for parsing the key into the indicator value.""" - key_number_groupings = self.find_all_nums_in_str(rule_key) - - # MACD can only have slope emblem therefore 1 or 0 number groupings are acceptable - if len(key_number_groupings) == 0: - if SLOPE_SYMBOL in rule_key: - raise StrategyIndicatorError(f'{self.strategy_symbol} rule key: {rule_key} does not contain' - f' enough number groupings!') - indicator_value = float(data_manager.get_data_point(self.DATA_COLUMN_TITLE, current_day_index)) - elif len(key_number_groupings) == 1: - # 1 number grouping suggests the $slope indicator is being used - if SLOPE_SYMBOL in rule_key: - slope_window_length = int(key_number_groupings[0]) - - # data request length is window - 1 to account for the current day index being a part of the window - slope_data_request_length = slope_window_length - 1 - - indicator_value = self.calculate_slope( - float(data_manager.get_data_point(self.DATA_COLUMN_TITLE, current_day_index)), - float(data_manager.get_data_point(self.DATA_COLUMN_TITLE, current_day_index - - slope_data_request_length)), - slope_window_length - ) - else: - raise StrategyIndicatorError(f'{self.strategy_symbol} rule key: {rule_key} contains too many number ' - f'groupings! Are you missing a $slope emblem?') - else: - raise StrategyIndicatorError(f'{self.strategy_symbol} rule key: {rule_key} contains ' - f'invalid number groupings!') - - return indicator_value - def __calculate_macd(self, price_data: list) -> list: """Calculate MACD values for a list of price values""" large_ema_length_values = MACDTrigger.__calculate_ema(self.LARGE_EMA_LENGTH, price_data) @@ -110,7 +76,7 @@ def __calculate_macd(self, price_data: list) -> list: small_ema_length_values = MACDTrigger.__calculate_ema(self.SMALL_EMA_LENGTH, price_data) if len(large_ema_length_values) != len(small_ema_length_values): - raise StrategyIndicatorError(f'{self.strategy_symbol} value lists for {self.strategy_symbol} must be the ' + raise StrategyIndicatorError(f'{self.indicator_symbol} value lists for {self.indicator_symbol} must be the ' f'same length!') macd_values = [] diff --git a/src/StockBench/indicators/price/trigger.py b/src/StockBench/indicators/price/trigger.py index 0e66d16..dac1502 100644 --- a/src/StockBench/indicators/price/trigger.py +++ b/src/StockBench/indicators/price/trigger.py @@ -10,8 +10,8 @@ class PriceTrigger(Trigger): # cannot use strategy symbol because its "price" DISPLAY_NAME = 'Price' - def __init__(self, strategy_symbol): - super().__init__(strategy_symbol, side=Trigger.AGNOSTIC) + def __init__(self, indicator_symbol): + super().__init__(indicator_symbol, side=Trigger.AGNOSTIC) def additional_days(self, rule_key, value_value) -> int: """Calculate the additional days required. @@ -50,6 +50,8 @@ def check_trigger(self, rule_key, rule_value, data_manager, position, current_da """ log.debug(f'Checking price algorithm: {rule_key}...') + # price uses special key parses because the indicator called 'price', but in the data it is 'close', to make it + # more clear we are using a dedicate parser indicator_value = self.__parse_key(rule_key, data_manager, current_day_index) operator, trigger_value = self._parse_rule_value(rule_value, data_manager, current_day_index) @@ -77,9 +79,8 @@ def __parse_key(self, rule_key, data_manager, current_day_index) -> float: indicator_value = self.calculate_slope( float(data_manager.get_data_point(data_manager.CLOSE, current_day_index)), - float(data_manager.get_data_point(data_manager.CLOSE, current_day_index - slope_data_request_length)), - slope_window_length - ) + float(data_manager.get_data_point(data_manager.CLOSE, current_day_index - + slope_data_request_length)), slope_window_length) else: raise StrategyIndicatorError(f'{self.DISPLAY_NAME} rule key: {rule_key} contains too many number ' f'groupings! Are you missing a $slope emblem?') diff --git a/src/StockBench/indicators/rsi/trigger.py b/src/StockBench/indicators/rsi/trigger.py index 1136579..b796b4e 100644 --- a/src/StockBench/indicators/rsi/trigger.py +++ b/src/StockBench/indicators/rsi/trigger.py @@ -4,14 +4,13 @@ from StockBench.indicator.trigger import Trigger from StockBench.simulation_data.data_manager import DataManager from StockBench.position.position import Position -from StockBench.indicator.exceptions import StrategyIndicatorError log = logging.getLogger() class RSITrigger(Trigger): - def __init__(self, strategy_symbol): - super().__init__(strategy_symbol, side=Trigger.AGNOSTIC) + def __init__(self, indicator_symbol): + super().__init__(indicator_symbol, side=Trigger.AGNOSTIC) def additional_days(self, rule_key, value_value) -> int: """Calculate the additional days required. @@ -48,7 +47,7 @@ def add_to_data(self, rule_key, rule_value, side, data_manager): nums = self.find_all_nums_in_str(rule_value) if len(nums) > 0: trigger_value = float(nums[0]) - Trigger._add_trigger_column(f'{self.strategy_symbol}_{trigger_value}', trigger_value, + Trigger._add_trigger_column(f'{self.indicator_symbol}_{trigger_value}', trigger_value, data_manager) def check_trigger(self, rule_key, rule_value, data_manager, position, current_day_index) -> bool: @@ -64,70 +63,28 @@ def check_trigger(self, rule_key, rule_value, data_manager, position, current_da return: bool: True if a trigger was hit. """ - log.debug(f'Checking {self.strategy_symbol} algorithm: {rule_key}...') + log.debug(f'Checking {self.indicator_symbol} algorithm: {rule_key}...') - indicator_value = self.__parse_key(rule_key, data_manager, current_day_index) + indicator_value = Trigger._parse_rule_key(rule_key, self.indicator_symbol, data_manager, current_day_index) operator, trigger_value = self._parse_rule_value(rule_value, data_manager, current_day_index) - log.debug(f'{self.strategy_symbol} algorithm: {rule_key} checked successfully') + log.debug(f'{self.indicator_symbol} algorithm: {rule_key} checked successfully') return Trigger.basic_trigger_check(indicator_value, operator, trigger_value) - def __parse_key(self, rule_key: any, data_manager: DataManager, current_day_index: int) -> float: - """Parser for parsing the key into the indicator value.""" - rule_key_number_groups = self.find_all_nums_in_str(rule_key) - - if len(rule_key_number_groups) == 0: - # RSI is default length (14) - if SLOPE_SYMBOL in rule_key: - raise StrategyIndicatorError(f'{self.strategy_symbol} rule key: {rule_key} does not contain ' - f'enough number groupings!') - indicator_value = float(data_manager.get_data_point(self.strategy_symbol, current_day_index)) - elif len(rule_key_number_groups) == 1: - if SLOPE_SYMBOL in rule_key: - # make sure the number is after the slope emblem and not the RSI emblem - if rule_key.split(str(rule_key_number_groups))[0] == self.strategy_symbol + SLOPE_SYMBOL: - raise StrategyIndicatorError(f'{self.strategy_symbol} rule key: {rule_key} does not contain ' - f'a slope value!') - # RSI is custom length (not 14) - column_title = f'{self.strategy_symbol}{int(rule_key_number_groups[0])}' - indicator_value = float(data_manager.get_data_point(column_title, current_day_index)) - elif len(rule_key_number_groups) == 2: - column_title = f'{self.strategy_symbol}{int(rule_key_number_groups[0])}' - # 2 number groupings suggests the $slope indicator is being used - if SLOPE_SYMBOL in rule_key: - slope_window_length = int(rule_key_number_groups[1]) - - # data request length is window - 1 to account for the current day index being a part of the window - slope_data_request_length = slope_window_length - 1 - - indicator_value = self.calculate_slope( - float(data_manager.get_data_point(column_title, current_day_index)), - float(data_manager.get_data_point(column_title, current_day_index - slope_data_request_length)), - slope_window_length - ) - else: - raise StrategyIndicatorError(f'{self.strategy_symbol} rule key: {rule_key} contains too many number ' - f'groupings! Are you missing a $slope emblem?') - else: - raise StrategyIndicatorError(f'{self.strategy_symbol} rule key: {rule_key} contains invalid number ' - f'groupings!') - - return indicator_value - def __add_rsi_column(self, length: int, data_manager: DataManager): """Calculate the RSI values and add them to the df.""" # if we already have RSI upper values in the df, we don't need to add them again for col_name in data_manager.get_column_names(): - if self.strategy_symbol in col_name: + if self.indicator_symbol in col_name: return price_data = data_manager.get_column_data(data_manager.CLOSE) rsi_values = RSITrigger.__calculate_rsi(length, price_data) - data_manager.add_column(self.strategy_symbol, rsi_values) + data_manager.add_column(self.indicator_symbol, rsi_values) @staticmethod def __calculate_rsi(length: int, price_data: list) -> list: diff --git a/src/StockBench/indicators/sma/trigger.py b/src/StockBench/indicators/sma/trigger.py index edb503b..81c805a 100644 --- a/src/StockBench/indicators/sma/trigger.py +++ b/src/StockBench/indicators/sma/trigger.py @@ -10,8 +10,8 @@ class SMATrigger(Trigger): - def __init__(self, strategy_symbol): - super().__init__(strategy_symbol, side=Trigger.AGNOSTIC) + def __init__(self, indicator_symbol): + super().__init__(indicator_symbol, side=Trigger.AGNOSTIC) def additional_days(self, rule_key, value_value) -> int: """Calculate the additional days required. @@ -24,7 +24,7 @@ def additional_days(self, rule_key, value_value) -> int: rule_key_number_groups = list(map(int, self.find_all_nums_in_str(rule_key))) if rule_key_number_groups: return max(rule_key_number_groups) - raise StrategyIndicatorError(f'{self.strategy_symbol} rule key: {rule_key} must have an indicator length!') + raise StrategyIndicatorError(f'{self.indicator_symbol} rule key: {rule_key} must have an indicator length!') def add_to_data(self, rule_key, rule_value, side, data_manager): """Add data to the dataframe. @@ -40,7 +40,7 @@ def add_to_data(self, rule_key, rule_value, side, data_manager): indicator_length = int(rule_key_number_groups[0]) self.__add_sma(indicator_length, data_manager) else: - raise StrategyIndicatorError(f'{self.strategy_symbol} rule key: {rule_key} must have an indicator length!') + raise StrategyIndicatorError(f'{self.indicator_symbol} rule key: {rule_key} must have an indicator length!') def check_trigger(self, rule_key, rule_value, data_manager, position, current_day_index) -> bool: """Trigger logic for SMA. @@ -55,55 +55,20 @@ def check_trigger(self, rule_key, rule_value, data_manager, position, current_da return: bool: True if a trigger was hit. """ - log.debug(f'Checking {self.strategy_symbol} algorithm: {rule_key}...') + log.debug(f'Checking {self.indicator_symbol} algorithm: {rule_key}...') - indicator_value = self.__parse_key(rule_key, data_manager, current_day_index) + indicator_value = Trigger._parse_rule_key_no_default_indicator_length(rule_key, self.indicator_symbol, data_manager, + current_day_index) operator, trigger_value = self._parse_rule_value(rule_value, data_manager, current_day_index) - log.debug(f'{self.strategy_symbol} algorithm: {rule_key} checked successfully') + log.debug(f'{self.indicator_symbol} algorithm: {rule_key} checked successfully') return Trigger.basic_trigger_check(indicator_value, operator, trigger_value) - def __parse_key(self, rule_key: any, data_manager: DataManager, current_day_index: int) -> float: - """Parser for parsing the key into the indicator value.""" - # find the indicator value (left hand side of the comparison) - rule_key_number_groups = self.find_all_nums_in_str(rule_key) - - # do not build title outside of conditional as nums could be [] which would result in index error - - if len(rule_key_number_groups) == 1: - if SLOPE_SYMBOL in rule_key: - raise StrategyIndicatorError(f'{self.strategy_symbol} rule key: {rule_key} does not contain ' - f'enough number groupings!') - title = f'{self.strategy_symbol}{int(rule_key_number_groups[0])}' - indicator_value = float(data_manager.get_data_point(title, current_day_index)) - elif len(rule_key_number_groups) == 2: - title = f'{self.strategy_symbol}{int(rule_key_number_groups[0])}' - # likely that the $slope indicator is being used - if SLOPE_SYMBOL in rule_key: - slope_window_length = int(rule_key_number_groups[1]) - - # data request length is window - 1 to account for the current day index being a part of the window - slope_data_request_length = slope_window_length - 1 - - indicator_value = self.calculate_slope( - float(data_manager.get_data_point(title, current_day_index)), - float(data_manager.get_data_point(title, current_day_index - slope_data_request_length)), - slope_window_length - ) - else: - raise StrategyIndicatorError(f'{self.strategy_symbol} rule key: {rule_key} contains ' - f'too many number groupings! Are you missing a $slope emblem?') - else: - raise StrategyIndicatorError(f'{self.strategy_symbol} rule key: {rule_key} contains ' - f'invalid number groupings!') - - return indicator_value - def __add_sma(self, length: int, data_manager: DataManager): """Pre-calculate the SMA values and add them to the df.""" - column_title = f'{self.strategy_symbol}{length}' + column_title = f'{self.indicator_symbol}{length}' # if SMA values ar already in the df, we don't need to add them again for col_name in data_manager.get_column_names(): diff --git a/src/StockBench/indicators/stochastic/trigger.py b/src/StockBench/indicators/stochastic/trigger.py index 97796e3..d9c39cb 100644 --- a/src/StockBench/indicators/stochastic/trigger.py +++ b/src/StockBench/indicators/stochastic/trigger.py @@ -12,8 +12,8 @@ class StochasticTrigger(Trigger): # cannot use strategy symbol because it is "stochastic" DISPLAY_NAME = 'Stochastic' - def __init__(self, strategy_symbol): - super().__init__(strategy_symbol, side=Trigger.AGNOSTIC) + def __init__(self, indicator_symbol): + super().__init__(indicator_symbol, side=Trigger.AGNOSTIC) def additional_days(self, rule_key, value_value) -> int: """Calculate the additional days required. @@ -48,7 +48,7 @@ def add_to_data(self, rule_key, rule_value, side, data_manager): rule_key_number_groups = self.find_all_nums_in_str(rule_value) if len(rule_key_number_groups) > 0: trigger_value = float(rule_key_number_groups[0]) - Trigger._add_trigger_column(f'{self.strategy_symbol}_{trigger_value}', trigger_value, + Trigger._add_trigger_column(f'{self.indicator_symbol}_{trigger_value}', trigger_value, data_manager) def check_trigger(self, rule_key, rule_value, data_manager, position, current_day_index) -> bool: @@ -66,7 +66,7 @@ def check_trigger(self, rule_key, rule_value, data_manager, position, current_da """ log.debug(f'Checking stochastic algorithm: {rule_key}...') - indicator_value = self.__parse_key(rule_key, data_manager, current_day_index) + indicator_value = Trigger._parse_rule_key(rule_key, self.indicator_symbol, data_manager, current_day_index) operator, trigger_value = self._parse_rule_value(rule_value, data_manager, current_day_index) @@ -74,52 +74,11 @@ def check_trigger(self, rule_key, rule_value, data_manager, position, current_da return Trigger.basic_trigger_check(indicator_value, operator, trigger_value) - def __parse_key(self, rule_key, data_manager, current_day_index) -> float: - """Parser for parsing the key into the indicator value.""" - # find the indicator value (left hand side of the comparison) - rule_key_number_groups = self.find_all_nums_in_str(rule_key) - - if len(rule_key_number_groups) == 0: - if SLOPE_SYMBOL in rule_key: - raise StrategyIndicatorError(f'{self.DISPLAY_NAME} rule key: {rule_key} does not contain enough number ' - f'groupings!') - indicator_value = float(data_manager.get_data_point(self.strategy_symbol, current_day_index)) - elif len(rule_key_number_groups) == 1: - if SLOPE_SYMBOL in rule_key: - # make sure the number is after the slope emblem and not the stochastic emblem - if rule_key.split(str(rule_key_number_groups))[0] == self.strategy_symbol + SLOPE_SYMBOL: - raise StrategyIndicatorError( - f'{self.DISPLAY_NAME} rule key: {rule_key} contains too many number groupings! ' - f'Are you missing a $slope emblem?') - indicator_value = float(data_manager.get_data_point(self.strategy_symbol, current_day_index)) - elif len(rule_key_number_groups) == 2: - title = f'{self.strategy_symbol}{int(rule_key_number_groups[0])}' - # 2 number groups suggests $slope indicator is being used - if SLOPE_SYMBOL in rule_key: - slope_window_length = int(rule_key_number_groups[1]) - - # data request length is window - 1 to account for the current day index being a part of the window - slope_data_request_length = slope_window_length - 1 - - indicator_value = self.calculate_slope( - float(data_manager.get_data_point(title, current_day_index)), - float(data_manager.get_data_point(title, current_day_index - slope_data_request_length)), - slope_window_length - ) - else: - raise StrategyIndicatorError(f'{self.DISPLAY_NAME} rule key: {rule_key} contains ' - f'too many number groupings! Are you missing a $slope emblem?') - else: - raise StrategyIndicatorError(f'{self.DISPLAY_NAME} rule key: {rule_key} contains ' - f'too many number groupings!') - - return indicator_value - def __add_stochastic_column(self, length: int, data_manager: DataManager): """Calculate the stochastic values and add them to the df.""" # if we already have values in the df, we don't need to add them again for col_name in data_manager.get_column_names(): - if self.strategy_symbol in col_name: + if self.indicator_symbol in col_name: return high_data = data_manager.get_column_data(data_manager.HIGH) @@ -128,7 +87,7 @@ def __add_stochastic_column(self, length: int, data_manager: DataManager): stochastic_values = StochasticTrigger.__stochastic_oscillator(length, high_data, low_data, close_data) - data_manager.add_column(self.strategy_symbol, stochastic_values) + data_manager.add_column(self.indicator_symbol, stochastic_values) @staticmethod def __stochastic_oscillator(length: int, high_data: list, low_data: list, close_data: list) -> list: diff --git a/src/StockBench/indicators/stop_loss/trigger.py b/src/StockBench/indicators/stop_loss/trigger.py index 8e58ddc..93e084b 100644 --- a/src/StockBench/indicators/stop_loss/trigger.py +++ b/src/StockBench/indicators/stop_loss/trigger.py @@ -7,8 +7,8 @@ class StopLossTrigger(Trigger): - def __init__(self, strategy_symbol): - super().__init__(strategy_symbol, side=Trigger.SELL) + def __init__(self, indicator_symbol): + super().__init__(indicator_symbol, side=Trigger.SELL) def additional_days(self, rule_key, value_value) -> int: """Calculate the additional days required. diff --git a/src/StockBench/indicators/stop_profit/trigger.py b/src/StockBench/indicators/stop_profit/trigger.py index ae0581e..4cc9f3e 100644 --- a/src/StockBench/indicators/stop_profit/trigger.py +++ b/src/StockBench/indicators/stop_profit/trigger.py @@ -7,8 +7,8 @@ class StopProfitTrigger(Trigger): - def __init__(self, strategy_symbol): - super().__init__(strategy_symbol, side=Trigger.SELL) + def __init__(self, indicator_symbol): + super().__init__(indicator_symbol, side=Trigger.SELL) def additional_days(self, rule_key: str, value_value: str) -> int: """Calculate the additional days required. diff --git a/src/StockBench/indicators/volume/trigger.py b/src/StockBench/indicators/volume/trigger.py index f5dc3b4..1895e7f 100644 --- a/src/StockBench/indicators/volume/trigger.py +++ b/src/StockBench/indicators/volume/trigger.py @@ -8,8 +8,8 @@ class VolumeTrigger(Trigger): - def __init__(self, strategy_symbol): - super().__init__(strategy_symbol, side=Trigger.AGNOSTIC) + def __init__(self, indicator_symbol): + super().__init__(indicator_symbol, side=Trigger.AGNOSTIC) def additional_days(self, rule_key, value_value) -> int: """Calculate the additional days required. @@ -57,6 +57,6 @@ def check_trigger(self, rule_key, rule_value, data_manager, position, current_da result = Trigger.basic_trigger_check(volume, operator, trigger_value) - log.debug(f'{self.strategy_symbol} algorithm: {rule_key} checked successfully') + log.debug(f'{self.indicator_symbol} algorithm: {rule_key} checked successfully') return result