update _search_all_objects functioning

This commit is contained in:
Matthias 2022-10-14 14:32:30 +00:00
parent c6d2eed4fc
commit 1d8d360a12
2 changed files with 33 additions and 9 deletions

View File

@ -1,7 +1,6 @@
import csv import csv
import logging import logging
import sys import sys
from pathlib import Path
from typing import Any, Dict, List from typing import Any, Dict, List
import rapidjson import rapidjson
@ -10,7 +9,6 @@ from colorama import init as colorama_init
from tabulate import tabulate from tabulate import tabulate
from freqtrade.configuration import setup_utils_configuration from freqtrade.configuration import setup_utils_configuration
from freqtrade.constants import USERPATH_STRATEGIES
from freqtrade.enums import RunMode from freqtrade.enums import RunMode
from freqtrade.exceptions import OperationalException from freqtrade.exceptions import OperationalException
from freqtrade.exchange import market_is_active, validate_exchanges from freqtrade.exchange import market_is_active, validate_exchanges
@ -41,7 +39,7 @@ def start_list_exchanges(args: Dict[str, Any]) -> None:
print(tabulate(exchanges, headers=['Exchange name', 'Valid', 'reason'])) print(tabulate(exchanges, headers=['Exchange name', 'Valid', 'reason']))
def _print_objs_tabular(objs: List, print_colorized: bool, base_dir: Path) -> None: def _print_objs_tabular(objs: List, print_colorized: bool) -> None:
if print_colorized: if print_colorized:
colorama_init(autoreset=True) colorama_init(autoreset=True)
red = Fore.RED red = Fore.RED
@ -55,7 +53,7 @@ def _print_objs_tabular(objs: List, print_colorized: bool, base_dir: Path) -> No
names = [s['name'] for s in objs] names = [s['name'] for s in objs]
objs_to_print = [{ objs_to_print = [{
'name': s['name'] if s['name'] else "--", 'name': s['name'] if s['name'] else "--",
'location': s['location'].relative_to(base_dir), 'location': s['location_rel'],
'status': (red + "LOAD FAILED" + reset if s['class'] is None 'status': (red + "LOAD FAILED" + reset if s['class'] is None
else "OK" if names.count(s['name']) == 1 else "OK" if names.count(s['name']) == 1
else yellow + "DUPLICATE NAME" + reset) else yellow + "DUPLICATE NAME" + reset)
@ -76,9 +74,8 @@ def start_list_strategies(args: Dict[str, Any]) -> None:
""" """
config = setup_utils_configuration(args, RunMode.UTIL_NO_EXCHANGE) config = setup_utils_configuration(args, RunMode.UTIL_NO_EXCHANGE)
directory = Path(config.get('strategy_path', config['user_data_dir'] / USERPATH_STRATEGIES))
strategy_objs = StrategyResolver.search_all_objects( strategy_objs = StrategyResolver.search_all_objects(
directory, not args['print_one_column'], config.get('recursive_strategy_search', False)) config, not args['print_one_column'], config.get('recursive_strategy_search', False))
# Sort alphabetically # Sort alphabetically
strategy_objs = sorted(strategy_objs, key=lambda x: x['name']) strategy_objs = sorted(strategy_objs, key=lambda x: x['name'])
for obj in strategy_objs: for obj in strategy_objs:
@ -90,7 +87,7 @@ def start_list_strategies(args: Dict[str, Any]) -> None:
if args['print_one_column']: if args['print_one_column']:
print('\n'.join([s['name'] for s in strategy_objs])) print('\n'.join([s['name'] for s in strategy_objs]))
else: else:
_print_objs_tabular(strategy_objs, config.get('print_colorized', False), directory) _print_objs_tabular(strategy_objs, config.get('print_colorized', False))
def start_list_timeframes(args: Dict[str, Any]) -> None: def start_list_timeframes(args: Dict[str, Any]) -> None:

View File

@ -183,9 +183,35 @@ class IResolver:
) )
@classmethod @classmethod
def search_all_objects(cls, directory: Path, enum_failed: bool, def search_all_objects(cls, config: Config, enum_failed: bool,
recursive: bool = False) -> List[Dict[str, Any]]: recursive: bool = False) -> List[Dict[str, Any]]:
""" """
Searches for valid objects
:param config: Config object
:param enum_failed: If True, will return None for modules which fail.
Otherwise, failing modules are skipped.
:param recursive: Recursively walk directory tree searching for strategies
:return: List of dicts containing 'name', 'class' and 'location' entries
"""
result = []
abs_paths = cls.build_search_paths(config, user_subdir=cls.user_subdir)
print(abs_paths)
for path in abs_paths:
result.extend(cls._search_all_objects(path, enum_failed, recursive))
return result
@classmethod
def _build_rel_location(cls, directory: Path, entry: Path) -> str:
builtin = cls.initial_search_path == directory
return f"<builtin>/{entry.relative_to(directory)}" if builtin else entry.relative_to(
directory)
@classmethod
def _search_all_objects(cls, directory: Path, enum_failed: bool,
recursive: bool = False) -> List[Dict[str, Any]]:
"""
Searches a directory for valid objects Searches a directory for valid objects
:param directory: Path to search :param directory: Path to search
:param enum_failed: If True, will return None for modules which fail. :param enum_failed: If True, will return None for modules which fail.
@ -204,7 +230,7 @@ class IResolver:
and not entry.name.startswith('__') and not entry.name.startswith('__')
and not entry.name.startswith('.') and not entry.name.startswith('.')
): ):
objects.extend(cls.search_all_objects(entry, enum_failed, recursive=recursive)) objects.extend(cls._search_all_objects(entry, enum_failed, recursive=recursive))
# Only consider python files # Only consider python files
if entry.suffix != '.py': if entry.suffix != '.py':
logger.debug('Ignoring %s', entry) logger.debug('Ignoring %s', entry)
@ -217,5 +243,6 @@ class IResolver:
{'name': obj[0].__name__ if obj is not None else '', {'name': obj[0].__name__ if obj is not None else '',
'class': obj[0] if obj is not None else None, 'class': obj[0] if obj is not None else None,
'location': entry, 'location': entry,
'location_rel': cls._build_rel_location(directory, entry),
}) })
return objects return objects