from dataclasses import dataclass @dataclass() class Signal: """Class for individual signal properties and state.""" name: str source1: str prop1: str source2: str prop2: str operator: str state: bool = False value1: int = None value2: int = None range: int = None def set_value1(self, value): self.value1 = value def set_value2(self, value): self.value2 = value def compare(self): if self.value1 is None: raise ValueError('Signal: Cannot compare: value1 not set') if self.value2 is None: raise ValueError('Signal: Cannot compare: value2 not set') previous_state = self.state if self.operator == '+/-': if self.range is None: raise ValueError('Signal: Cannot compare: range not set') if abs(self.value1 - self.value2) < self.range: self.state = True else: self.state = False else: string = str(self.value1) + self.operator + str(self.value2) print(string) if eval(string): self.state = True else: self.state = False state_change = False if self.state != previous_state: state_change = True return state_change class Signals: def __init__(self): self.signals = [] # self.set_signals_defaults() def set_signals_defaults(self): """These defaults are loaded if the config file is not found.""" sigs = self.get_signals_defaults() for sig in sigs: self.signals.append(Signal(name=sig['name'], source1=sig['source1'], prop1=sig['prop1'], operator=sig['operator'], source2=sig['source2'], prop2=sig['prop2'], state=sig['state'])) return @staticmethod def get_signals_defaults(): """These defaults are loaded if the config file is not found.""" s1 = {"name": "20x50_GC", "source1": "EMA 20", "prop1": "value", "source2": "EMA 50", "prop2": "value", "operator": ">", "state": False, "value1": None, "value2": None, "range": None} s2 = {"name": "50x200_GC", "source1": "EMA 50", "prop1": "value", "source2": "EMA 200", "prop2": "value", "operator": ">", "state": False, "value1": None, "value2": None, "range": None} s3 = {"name": "5x15_GC", "source1": "EMA 5", "prop1": "value", "source2": "EMA 15", "prop2": "value", "operator": ">", "state": False, "value1": None, "value2": None, "range": None} return [s1, s2, s3] def init_loaded_signals(self, signals_list): for sig in signals_list: self.signals.append(Signal(name=sig['name'], source1=sig['source1'], prop1=sig['prop1'], operator=sig['operator'], source2=sig['source2'], prop2=sig['prop2'], state=sig['state'])) def get_signals(self): return self.signals def new_signal(self, data): self.signals.append(Signal(**data)) def delete_signal(self, signal_name): print(f'removing {signal_name}') for sig in self.signals: if sig.name == signal_name: self.signals.remove(sig) break def update_signals(self, candles, indicators): # TODO: This function is not used. but may be used later if candles need to be specified. for signal in self.signals: self.process_signal(signal, candles, indicators) def process_all_signals(self, indicators): """Loop through all the signals and process them based on the last indicator results.""" state_changes = {} for signal in self.signals: change_in_state = self.process_signal(signal, indicators) if change_in_state: state_changes.update({signal.name: signal.state}) return state_changes def process_signal(self, signal, indicators, candles=None): """Receives a signal, makes the comparison with the last value calculated by the indicators and returns the result. If candles are provided it will ask the indicators to calculate new values based on those candles.""" if candles is None: # Get the source of the first signal source_1 = signal.source1 # Ask the indicator for the last result. if source_1 in indicators.indicators: signal.value1 = indicators.indicators[source_1].properties[signal.prop1] else: print('Could not calculate signal source indicator not found.') return False # Get the source of the second signal source_2 = signal.source2 # If the source is a set value it will be stored in prop2 if source_2 == 'value': signal.value2 = signal.prop2 else: # Ask the indicator for the last result. if source_2 in indicators.indicators: signal.value2 = indicators.indicators[source_2].properties[signal.prop2] else: print('Could not calculate signal source2 indicator not found.') return False # Compare the retrieved values. state_change = signal.compare() return state_change