9394559170

Committed 26 Apr 2024 06:36AM UTC coverage: 94.656% (-0.02%) from 94.674%

Build # 9394559170

Build Type

push

github

Committed by

xmatthias

Commit Message

Loader should be passed as kwarg for clarity

Run Details

20280 of 21425 relevant lines covered (94.66%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.56

/freqtrade/resolvers/iresolver.py

# pragma pylint: disable=attribute-defined-outside-init

"""
This module load custom objects
"""
import importlib.util
import inspect
import logging
import sys
from pathlib import Path
from typing import Any, Dict, Iterator, List, Optional, Tuple, Type, Union

from freqtrade.constants import Config
from freqtrade.exceptions import OperationalException


logger = logging.getLogger(__name__)


class PathModifier:
    def __init__(self, path: Path):
        self.path = path

    def __enter__(self):
        """Inject path to allow importing with relative imports."""
        sys.path.insert(0, str(self.path))
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Undo insertion of local path."""
        str_path = str(self.path)
        if str_path in sys.path:
            sys.path.remove(str_path)


class IResolver:
    """
    This class contains all the logic to load custom classes
    """
    # Childclasses need to override this
    object_type: Type[Any]
    object_type_str: str
    user_subdir: Optional[str] = None
    initial_search_path: Optional[Path] = None
    # Optional config setting containing a path (strategy_path, freqaimodel_path)
    extra_path: Optional[str] = None

    @classmethod
    def build_search_paths(cls, config: Config, user_subdir: Optional[str] = None,
                           extra_dirs: Optional[List[str]] = None) -> List[Path]:

        abs_paths: List[Path] = []
        if cls.initial_search_path:
            abs_paths.append(cls.initial_search_path)

        if user_subdir:
            abs_paths.insert(0, config['user_data_dir'].joinpath(user_subdir))

        # Add extra directory to the top of the search paths
        if extra_dirs:
            for dir in extra_dirs:
                abs_paths.insert(0, Path(dir).resolve())

        if cls.extra_path and (extra := config.get(cls.extra_path)):
            abs_paths.insert(0, Path(extra).resolve())

        return abs_paths

    @classmethod
    def _get_valid_object(cls, module_path: Path, object_name: Optional[str],
                          enum_failed: bool = False) -> Iterator[Any]:
        """
        Generator returning objects with matching object_type and object_name in the path given.
        :param module_path: absolute path to the module
        :param object_name: Class name of the object
        :param enum_failed: If True, will return None for modules which fail.
            Otherwise, failing modules are skipped.
        :return: generator containing tuple of matching objects
             Tuple format: [Object, source]
        """

        # Generate spec based on absolute path
        # Pass object_name as first argument to have logging print a reasonable name.
        with PathModifier(module_path.parent):
            module_name = module_path.stem or ""
            spec = importlib.util.spec_from_file_location(module_name, str(module_path))
            if not spec:
                return iter([None])

            module = importlib.util.module_from_spec(spec)
            try:
                spec.loader.exec_module(module)  # type: ignore # importlib does not use typehints
            except (AttributeError, ModuleNotFoundError, SyntaxError,
                    ImportError, NameError) as err:
                # Catch errors in case a specific module is not installed
                logger.warning(f"Could not import {module_path} due to '{err}'")
                if enum_failed:
                    return iter([None])

            valid_objects_gen = (
                (obj, inspect.getsource(module)) for
                name, obj in inspect.getmembers(
                    module, inspect.isclass) if ((object_name is None or object_name == name)
                                                 and issubclass(obj, cls.object_type)
                                                 and obj is not cls.object_type
                                                 and obj.__module__ == module_name
                                                 )
            )
            # The __module__ check ensures we only use strategies that are defined in this folder.
            return valid_objects_gen

    @classmethod
    def _search_object(cls, directory: Path, *, object_name: str, add_source: bool = False
                       ) -> Union[Tuple[Any, Path], Tuple[None, None]]:
        """
        Search for the objectname in the given directory
        :param directory: relative or absolute directory path
        :param object_name: ClassName of the object to load
        :return: object class
        """
        logger.debug(f"Searching for {cls.object_type.__name__} {object_name} in '{directory}'")
        for entry in directory.iterdir():
            # Only consider python files
            if entry.suffix != '.py':
                logger.debug('Ignoring %s', entry)
                continue
            if entry.is_symlink() and not entry.is_file():
                logger.debug('Ignoring broken symlink %s', entry)
                continue
            module_path = entry.resolve()

            obj = next(cls._get_valid_object(module_path, object_name), None)

            if obj:
                obj[0].__file__ = str(entry)
                if add_source:
                    obj[0].__source__ = obj[1]
                return (obj[0], module_path)
        return (None, None)

    @classmethod
    def _load_object(cls, paths: List[Path], *, object_name: str, add_source: bool = False,
                     kwargs: Dict) -> Optional[Any]:
        """
        Try to load object from path list.
        """

        for _path in paths:
            try:
                (module, module_path) = cls._search_object(directory=_path,
                                                           object_name=object_name,
                                                           add_source=add_source)
                if module:
                    logger.info(
                        f"Using resolved {cls.object_type.__name__.lower()[1:]} {object_name} "
                        f"from '{module_path}'...")
                    return module(**kwargs)
            except FileNotFoundError:
                logger.warning('Path "%s" does not exist.', _path.resolve())

        return None

    @classmethod
    def load_object(cls, object_name: str, config: Config, *, kwargs: dict,
                    extra_dir: Optional[str] = None) -> Any:
        """
        Search and loads the specified object as configured in the child class.
        :param object_name: name of the module to import
        :param config: configuration dictionary
        :param extra_dir: additional directory to search for the given pairlist
        :raises: OperationalException if the class is invalid or does not exist.
        :return: Object instance or None
        """

        extra_dirs: List[str] = []
        if extra_dir:
            extra_dirs.append(extra_dir)

        abs_paths = cls.build_search_paths(config,
                                           user_subdir=cls.user_subdir,
                                           extra_dirs=extra_dirs)

        found_object = cls._load_object(paths=abs_paths, object_name=object_name,
                                        kwargs=kwargs)
        if found_object:
            return found_object
        raise OperationalException(
            f"Impossible to load {cls.object_type_str} '{object_name}'. This class does not exist "
            "or contains Python code errors."
        )

    @classmethod
    def search_all_objects(cls, config: Config, enum_failed: bool,
                           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)
        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 str(
            entry.relative_to(directory))

    @classmethod
    def _search_all_objects(
            cls, directory: Path, enum_failed: bool, recursive: bool = False,
            basedir: Optional[Path] = None) -> List[Dict[str, Any]]:
        """
        Searches a directory for valid objects
        :param directory: Path to search
        :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
        """
        logger.debug(f"Searching for {cls.object_type.__name__} '{directory}'")
        objects: List[Dict[str, Any]] = []
        if not directory.is_dir():
            logger.info(f"'{directory}' is not a directory, skipping.")
            return objects
        for entry in directory.iterdir():
            if (
                recursive and entry.is_dir()
                and not entry.name.startswith('__')
                and not entry.name.startswith('.')
            ):
                objects.extend(cls._search_all_objects(
                    entry, enum_failed, recursive, basedir or directory))
            # Only consider python files
            if entry.suffix != '.py':
                logger.debug('Ignoring %s', entry)
                continue
            module_path = entry.resolve()
            logger.debug(f"Path {module_path}")
            for obj in cls._get_valid_object(module_path, object_name=None,
                                             enum_failed=enum_failed):
                objects.append(
                    {'name': obj[0].__name__ if obj is not None else '',
                     'class': obj[0] if obj is not None else None,
                     'location': entry,
                     'location_rel': cls._build_rel_location(basedir or directory, entry),
                     })
        return objects

1	# pragma pylint: disable=attribute-defined-outside-init
2
3	"""	1✔
4	This module load custom objects
5	"""
6	import importlib.util	1✔
7	import inspect	1✔
8	import logging	1✔
9	import sys	1✔
10	from pathlib import Path	1✔
11	from typing import Any, Dict, Iterator, List, Optional, Tuple, Type, Union	1✔
12
13	from freqtrade.constants import Config	1✔
14	from freqtrade.exceptions import OperationalException	1✔
15
16
17	logger = logging.getLogger(__name__)	1✔
18
19
20	class PathModifier:	1✔
21	def __init__(self, path: Path):	1✔
22	self.path = path	1✔
23
24	def __enter__(self):	1✔
25	"""Inject path to allow importing with relative imports."""
26	sys.path.insert(0, str(self.path))	1✔
27	return self	1✔
28
29	def __exit__(self, exc_type, exc_val, exc_tb):	1✔
30	"""Undo insertion of local path."""
31	str_path = str(self.path)	1✔
32	if str_path in sys.path:	1✔
33	sys.path.remove(str_path)	1✔
34
35
36	class IResolver:	1✔
37	"""
38	This class contains all the logic to load custom classes
39	"""
40	# Childclasses need to override this
41	object_type: Type[Any]	1✔
42	object_type_str: str	1✔
43	user_subdir: Optional[str] = None	1✔
44	initial_search_path: Optional[Path] = None	1✔
45	# Optional config setting containing a path (strategy_path, freqaimodel_path)
46	extra_path: Optional[str] = None	1✔
47
48	@classmethod	1✔
49	def build_search_paths(cls, config: Config, user_subdir: Optional[str] = None,	1✔
50	extra_dirs: Optional[List[str]] = None) -> List[Path]:
51
52	abs_paths: List[Path] = []	1✔
53	if cls.initial_search_path:	1✔
54	abs_paths.append(cls.initial_search_path)	1✔
55
56	if user_subdir:	1✔
57	abs_paths.insert(0, config['user_data_dir'].joinpath(user_subdir))	1✔
58
59	# Add extra directory to the top of the search paths
60	if extra_dirs:	1✔
61	for dir in extra_dirs:	1✔
62	abs_paths.insert(0, Path(dir).resolve())	1✔
63
64	if cls.extra_path and (extra := config.get(cls.extra_path)):	1✔
65	abs_paths.insert(0, Path(extra).resolve())	1✔
66
67	return abs_paths	1✔
68
69	@classmethod	1✔
70	def _get_valid_object(cls, module_path: Path, object_name: Optional[str],	1✔
71	enum_failed: bool = False) -> Iterator[Any]:
72	"""
73	Generator returning objects with matching object_type and object_name in the path given.
74	:param module_path: absolute path to the module
75	:param object_name: Class name of the object
76	:param enum_failed: If True, will return None for modules which fail.
77	Otherwise, failing modules are skipped.
78	:return: generator containing tuple of matching objects
79	Tuple format: [Object, source]
80	"""
81
82	# Generate spec based on absolute path
83	# Pass object_name as first argument to have logging print a reasonable name.
84	with PathModifier(module_path.parent):	1✔
85	module_name = module_path.stem or ""	1✔
86	spec = importlib.util.spec_from_file_location(module_name, str(module_path))	1✔
87	if not spec:	1✔
88	return iter([None])	×
89
90	module = importlib.util.module_from_spec(spec)	1✔
91	try:	1✔
92	spec.loader.exec_module(module) # type: ignore # importlib does not use typehints	1✔
93	except (AttributeError, ModuleNotFoundError, SyntaxError,	1✔
94	ImportError, NameError) as err:
95	# Catch errors in case a specific module is not installed
96	logger.warning(f"Could not import {module_path} due to '{err}'")	1✔
97	if enum_failed:	1✔
98	return iter([None])	1✔
99
100	valid_objects_gen = (	1✔
101	(obj, inspect.getsource(module)) for
102	name, obj in inspect.getmembers(
103	module, inspect.isclass) if ((object_name is None or object_name == name)
104	and issubclass(obj, cls.object_type)
105	and obj is not cls.object_type
106	and obj.__module__ == module_name
107	)
108	)
109	# The __module__ check ensures we only use strategies that are defined in this folder.
110	return valid_objects_gen	1✔
111
112	@classmethod	1✔
113	def _search_object(cls, directory: Path, *, object_name: str, add_source: bool = False	1✔
114	) -> Union[Tuple[Any, Path], Tuple[None, None]]:
115	"""
116	Search for the objectname in the given directory
117	:param directory: relative or absolute directory path
118	:param object_name: ClassName of the object to load
119	:return: object class
120	"""
121	logger.debug(f"Searching for {cls.object_type.__name__} {object_name} in '{directory}'")	1✔
122	for entry in directory.iterdir():	1✔
123	# Only consider python files
124	if entry.suffix != '.py':	1✔
125	logger.debug('Ignoring %s', entry)	1✔
126	continue	1✔
127	if entry.is_symlink() and not entry.is_file():	1✔
128	logger.debug('Ignoring broken symlink %s', entry)	×
129	continue	×
130	module_path = entry.resolve()	1✔
131
132	obj = next(cls._get_valid_object(module_path, object_name), None)	1✔
133
134	if obj:	1✔
135	obj[0].__file__ = str(entry)	1✔
136	if add_source:	1✔
137	obj[0].__source__ = obj[1]	1✔
138	return (obj[0], module_path)	1✔
139	return (None, None)	1✔
140
141	@classmethod	1✔
142	def _load_object(cls, paths: List[Path], *, object_name: str, add_source: bool = False,	1✔
143	kwargs: Dict) -> Optional[Any]:
144	"""
145	Try to load object from path list.
146	"""
147
148	for _path in paths:	1✔
149	try:	1✔
150	(module, module_path) = cls._search_object(directory=_path,	1✔
151	object_name=object_name,
152	add_source=add_source)
153	if module:	1✔
154	logger.info(	1✔
155	f"Using resolved {cls.object_type.__name__.lower()[1:]} {object_name} "
156	f"from '{module_path}'...")
157	return module(**kwargs)	1✔
158	except FileNotFoundError:	1✔
159	logger.warning('Path "%s" does not exist.', _path.resolve())	1✔
160
161	return None	1✔
162
163	@classmethod	1✔
164	def load_object(cls, object_name: str, config: Config, *, kwargs: dict,	1✔
165	extra_dir: Optional[str] = None) -> Any:
166	"""
167	Search and loads the specified object as configured in the child class.
168	:param object_name: name of the module to import
169	:param config: configuration dictionary
170	:param extra_dir: additional directory to search for the given pairlist
171	:raises: OperationalException if the class is invalid or does not exist.
172	:return: Object instance or None
173	"""
174
175	extra_dirs: List[str] = []	1✔
176	if extra_dir:	1✔
177	extra_dirs.append(extra_dir)	1✔
178
179	abs_paths = cls.build_search_paths(config,	1✔
180	user_subdir=cls.user_subdir,
181	extra_dirs=extra_dirs)
182
183	found_object = cls._load_object(paths=abs_paths, object_name=object_name,	1✔
184	kwargs=kwargs)
185	if found_object:	1✔
186	return found_object	1✔
187	raise OperationalException(	1✔
188	f"Impossible to load {cls.object_type_str} '{object_name}'. This class does not exist "
189	"or contains Python code errors."
190	)
191
192	@classmethod	1✔
193	def search_all_objects(cls, config: Config, enum_failed: bool,	1✔
194	recursive: bool = False) -> List[Dict[str, Any]]:
195	"""
196	Searches for valid objects
197	:param config: Config object
198	:param enum_failed: If True, will return None for modules which fail.
199	Otherwise, failing modules are skipped.
200	:param recursive: Recursively walk directory tree searching for strategies
201	:return: List of dicts containing 'name', 'class' and 'location' entries
202	"""
203	result = []	1✔
204
205	abs_paths = cls.build_search_paths(config, user_subdir=cls.user_subdir)	1✔
206	for path in abs_paths:	1✔
207	result.extend(cls._search_all_objects(path, enum_failed, recursive))	1✔
208	return result	1✔
209
210	@classmethod	1✔
211	def _build_rel_location(cls, directory: Path, entry: Path) -> str:	1✔
212
213	builtin = cls.initial_search_path == directory	1✔
214	return f"<builtin>/{entry.relative_to(directory)}" if builtin else str(	1✔
215	entry.relative_to(directory))
216
217	@classmethod	1✔
218	def _search_all_objects(	1✔
219	cls, directory: Path, enum_failed: bool, recursive: bool = False,
220	basedir: Optional[Path] = None) -> List[Dict[str, Any]]:
221	"""
222	Searches a directory for valid objects
223	:param directory: Path to search
224	:param enum_failed: If True, will return None for modules which fail.
225	Otherwise, failing modules are skipped.
226	:param recursive: Recursively walk directory tree searching for strategies
227	:return: List of dicts containing 'name', 'class' and 'location' entries
228	"""
229	logger.debug(f"Searching for {cls.object_type.__name__} '{directory}'")	1✔
230	objects: List[Dict[str, Any]] = []	1✔
231	if not directory.is_dir():	1✔
232	logger.info(f"'{directory}' is not a directory, skipping.")	1✔
233	return objects	1✔
234	for entry in directory.iterdir():	1✔
235	if (	1✔
236	recursive and entry.is_dir()
237	and not entry.name.startswith('__')
238	and not entry.name.startswith('.')
239	):
240	objects.extend(cls._search_all_objects(	1✔
241	entry, enum_failed, recursive, basedir or directory))
242	# Only consider python files
243	if entry.suffix != '.py':	1✔
244	logger.debug('Ignoring %s', entry)	1✔
245	continue	1✔
246	module_path = entry.resolve()	1✔
247	logger.debug(f"Path {module_path}")	1✔
248	for obj in cls._get_valid_object(module_path, object_name=None,	1✔
249	enum_failed=enum_failed):
250	objects.append(	1✔
251	{'name': obj[0].__name__ if obj is not None else '',
252	'class': obj[0] if obj is not None else None,
253	'location': entry,
254	'location_rel': cls._build_rel_location(basedir or directory, entry),
255	})
256	return objects	1✔

freqtrade / freqtrade / 9394559170

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous