brighter-trading/Signals.py

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 get_signal_by_name(self, name):
        for signal in self.signals:
            if signal.name == name:
                return signal
        return None

    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