4131164979

Build Type

push

github-actions

Committed by Matthias

Commit Message

filled-date shouldn't update again

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

17024 of 17946 relevant lines covered (94.86%)

0.95 hits per line

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

97.54

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

freqtrade / freqtrade / 4131164979

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