Execute the indicator analysis on a particular exchange and pair. Args: exchange (str): The exchange to get the indicator results for. market_pair (str): The pair to get the market pair results for. Returns: list: A list of dictinaries containing
(self, exchange, market_pair)
| 113 | |
| 114 | |
| 115 | def _get_indicator_results(self, exchange, market_pair): |
| 116 | """Execute the indicator analysis on a particular exchange and pair. |
| 117 | |
| 118 | Args: |
| 119 | exchange (str): The exchange to get the indicator results for. |
| 120 | market_pair (str): The pair to get the market pair results for. |
| 121 | |
| 122 | Returns: |
| 123 | list: A list of dictinaries containing the results of the analysis. |
| 124 | """ |
| 125 | |
| 126 | indicator_dispatcher = self.strategy_analyzer.indicator_dispatcher() |
| 127 | results = { indicator: list() for indicator in self.indicator_conf.keys() } |
| 128 | historical_data_cache = dict() |
| 129 | |
| 130 | for indicator in self.indicator_conf: |
| 131 | if indicator not in indicator_dispatcher: |
| 132 | self.logger.warn("No such indicator %s, skipping.", indicator) |
| 133 | continue |
| 134 | |
| 135 | for indicator_conf in self.indicator_conf[indicator]: |
| 136 | if indicator_conf['enabled']: |
| 137 | candle_period = indicator_conf['candle_period'] |
| 138 | else: |
| 139 | self.logger.debug("%s is disabled, skipping.", indicator) |
| 140 | continue |
| 141 | |
| 142 | if candle_period not in historical_data_cache: |
| 143 | historical_data_cache[candle_period] = self._get_historical_data( |
| 144 | market_pair, |
| 145 | exchange, |
| 146 | candle_period |
| 147 | ) |
| 148 | |
| 149 | if historical_data_cache[candle_period]: |
| 150 | analysis_args = { |
| 151 | 'historical_data': historical_data_cache[candle_period], |
| 152 | 'signal': indicator_conf['signal'], |
| 153 | 'hot_thresh': indicator_conf['hot'], |
| 154 | 'cold_thresh': indicator_conf['cold'] |
| 155 | } |
| 156 | |
| 157 | if 'period_count' in indicator_conf: |
| 158 | analysis_args['period_count'] = indicator_conf['period_count'] |
| 159 | |
| 160 | results[indicator].append({ |
| 161 | 'result': self._get_analysis_result( |
| 162 | indicator_dispatcher, |
| 163 | indicator, |
| 164 | analysis_args, |
| 165 | market_pair |
| 166 | ), |
| 167 | 'config': indicator_conf |
| 168 | }) |
| 169 | return results |
| 170 | |
| 171 | |
| 172 | def _get_informant_results(self, exchange, market_pair): |
no test coverage detected