brighter-trading/Signals.py

import json
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, loaded_signals=None):

        # list of Signal objects.
        self.signals = []

        # Initialize signals with loaded data.
        if loaded_signals is not None:
            self.create_signal_from_dic(loaded_signals)

    @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 create_signal_from_dic(self, signals_list=None):
        """
            :param signals_list: list of dict
            :return True: on success.
            Create and store signal objects from list of dictionaries.
        """

        # If no signals were provided used a default list.
        if signals_list is None:
            signals_list = self.get_signals_defaults()
        # Loop through the provided list, unpack the dictionaries, create and store the signal objects.
        for sig in signals_list:
            self.new_signal(sig)
        return True

    def get_signals(self, form):
        # Return a python object of all the signals stored in this instance.
        if form == 'obj':
            return self.signals
        # Return a JSON object of all the signals stored in this instance.
        elif form == 'json':
            sigs = self.signals
            json_str = []
            for sig in sigs:
                # TODO: note - Explore why I had to treat signals and strategies different here.
                json_str.append(json.dumps(sig.__dict__))
            return json_str
        # Return a dictionary object of all the signals stored in this instance.
        elif form == 'dict':
            sigs = self.signals
            s_list = []
            for sig in sigs:
                dic = sig.__dict__
                s_list.append(dic)
            return s_list

    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