config_parser.py

from technical import indicators
import talib.abstract as ta
import rtmtb.ta.rtmtb_indicators
from rtmtb.plot.autoplot.config_provider import PlotConfigProvider
from rtmtb.log import LoggerMixin
from rtmtb.plot.autoplot.indicator_parser import IndicatorParser
import copy
from itertools import chain
import collections
from finta import TA
import numpy as np


class ConfigParser(LoggerMixin):
    def __init__(self, main_config, plot_file):

        super().__init__()
        self.plot_config = PlotConfigProvider(
            main_config, plot_file).read_configuration()
        self.technical = indicators
        self.talib = ta
        self.rtmtbta = rtmtb.ta.rtmtb_indicators
        self.finta = TA
        self.indicator_parser = IndicatorParser(main_config, plot_file)
        self.indicator_list = []
        self.module_indicators = []
        self.indicator_dicts = []

    def parse_global_options(self):
        for k,v in self.plot_config[0].items():
            for o in v:
                if 'grid' in o.keys():
                    self.indicator_list.append(copy.deepcopy(o))

    def assign_colors(self):

        for q in self.indicator_list:

            if isinstance(q['outputs'], list) and len(q['outputs']) > 1:
                for r in range(len(q['outputs'])):
                    col_2 = self.rgb_generator()
                    q['outputs'][r] = [q['outputs'][r], col_2]

            elif isinstance(q['outputs'], list) and len(q['outputs']) == 1:
                col_3 = self.rgb_generator()
                q['outputs'] = [[q['outputs'][0], col_3]]

    def rgb_generator(self):
        color = np.random.randint(256, size=3).tolist()
        return color

    def parse_groups_section(self):
        indicator_groups = []
        for k, v in (self.plot_config[2].items()):
            for indicator in v:
                for key, value in indicator.items():
                    for m, mv in value.items():
                        if mv == 'talib':
                            for indicator_units in self.indicator_parser.gather_talib_params():
                                if key == indicator_units['indicator']:
                                    indicator_groups.append(indicator_units)

                        elif mv == 'technical':
                            for indicator_units in self.indicator_parser.gather_indicator_params(self.technical):
                                if key == indicator_units['indicator']:
                                    indicator_groups.append(indicator_units)

                        elif mv == 'rtmtbta':
                            for indicator_units in self.indicator_parser.gather_indicator_params(self.rtmtbta):
                                if key == indicator_units['indicator']:
                                    indicator_groups.append(indicator_units)

                        elif mv == 'finta':
                            for indicator_units in self.indicator_parser.gather_finta_params():
                                if key == indicator_units['indicator']:
                                    indicator_groups.append(indicator_units)

        for k, v in (self.plot_config[2].items()):
            for module_indicator in indicator_groups:
                for indicator in v:
                    for key, value in indicator.items():
                        if key == module_indicator['indicator']:
                            for n in range(value['amount']):
                                module_indicator.update(
                                    dict(index=(n + 1), group_tag=value['group'], in_group=True, clone_tag=None))
                                self.indicator_list.append(
                                    copy.deepcopy(module_indicator))

        return self.indicator_list

    def parse_indicator_options(self):

        for ind in self.indicator_dicts:
            continue

        for i in self.module_indicators:
            tag_list = []
            for o in ind:
                if o == i['indicator']:
                    col_1 = self.rgb_generator()
                    i.update(dict(clone_tag='single'))
                    i.update(dict(in_group=False))

                    self.indicator_list.append(copy.deepcopy(i))

                elif type(o) is dict:
                    for k, v in o.items():

                        if k == 'VOLUME':
                            print(k)
                            i.update(dict(clone_tag='single'))
                            i.update(dict(in_group=False))
                            i.update(dict(outputs=v))

                            self.indicator_list.append(copy.deepcopy(i))

                        try:

                            if v['group']:
                                tag_list.append(v['group'])
                                data = list(chain.from_iterable(
                                    [i.split() for i in tag_list]))
                                if k == i['indicator']:
                                    i.update(dict(index=data.count(
                                        v['group']), group_tag=v['group'], in_group=True, clone_tag=None))

                                    self.indicator_list.append(
                                        copy.deepcopy(i))

                        except KeyError as group_not_present_error:
                            pass

                        try:
                            if 'params' in v.keys():
                                for n in self.indicator_list:
                                    if k == n['indicator']:
                                        for kv in n['params'].keys():
                                            if kv in v['params'].keys():
                                                n['params'] = v['params']
                                            else:
                                                raise SyntaxError(
                                                    'check your syntax, param ' + str(v['params']) + ' at ' + str(n['indicator']) + ' is probably not correctly typed')

                        except Exception:
                            pass

        clone_tag_list = []
        counter_list = []
        for x in self.indicator_list:
            if x['clone_tag'] == 'single':
                x['index'] = 0
                x['group_tag'] = None
                x['clone_tag'] = x['indicator']
                clone_tag_list.append(x['indicator'])
                counter_list.append(collections.Counter(clone_tag_list))
                for c in counter_list:
                    for k, v in c.items():
                        if v > 0:
                            for r in range(v):
                                x['clone_tag'] = [x['indicator'], r]

    def module_route(self, indicator_dict, module):
        # gather indicator data, send list of indicators to generate indicator configuration

        if type(module) is dict:
            self.module_indicators.append(module)
            self.indicator_dicts.append(indicator_dict)

        elif module is not dict and module.__name__ == 'talib.abstract':
            for indicator_units in self.indicator_parser.gather_talib_params():
                self.module_indicators.append(indicator_units)
            self.indicator_dicts.append(indicator_dict)

        elif module is not dict and module.__name__ == 'technical.indicators' or module.__name__ == 'rtmtb.ta.rtmtb_indicators':
            for indicator_units in self.indicator_parser.gather_indicator_params(module):
                self.module_indicators.append(indicator_units)
            self.indicator_dicts.append(indicator_dict)

        elif module is not dict and module.__name__ == 'TA':
            for indicator_units in self.indicator_parser.gather_finta_params():
                self.module_indicators.append(indicator_units)
            self.indicator_dicts.append(indicator_dict)

        self.parse_indicator_options()

    def parse_indicator_section(self):
        # check if modules contain indicators in main config

        for modules in self.plot_config[1]['indicators']['modules']:
            try:
                for key, values in modules.items():
                    if key == 'talib' and len(list(modules['talib'])):
                        self.module_route(modules['talib'], self.talib)

                    elif key == 'price' and len(list(modules['price'])):
                        price_units = self.indicator_parser.define_price_axe()
                        self.module_route(modules['price'], price_units)

                    elif key == 'technical' and len(list(modules['technical'])):
                        self.module_route(modules['technical'], self.technical)

                    elif key == 'rtmtbta' and len(list(modules['rtmtbta'])):
                        self.module_route(modules['rtmtbta'], self.rtmtbta)

                    elif key == 'finta' and len(list(modules['finta'])):
                        self.module_route(modules['finta'], self.finta)

                    elif key == 'volume' and len(list(modules['volume'])):
                        volume_units = self.indicator_parser.define_volume_axe()
                        self.module_route(modules['volume'], volume_units)

            except TypeError as none_type:
                pass

        return self.indicator_list
    
    
    

    def create_configuration(self):
        try:
            self.parse_indicator_section()
            self.parse_groups_section()
            self.assign_colors()

        except TypeError as none_type:
            print(none_type)
            self.assign_colors()
            pass
        self.parse_global_options()
        return self.indicator_list

def generate_indicator_configuration():
    main_config = 'config/autoplot/main_config.yml'
    plot_config = 'config/autoplot/simple.yml'
    plots = ConfigParser(main_config, plot_config).create_configuration()
    write = PlotConfigProvider(
        main_config, plot_config).write_plot_configuration(plots)


generate_indicator_configuration()