Convert to Pathlib
This commit is contained in:
parent
cc7b820978
commit
20de8c82e4
@ -4,7 +4,7 @@
|
|||||||
This module load custom hyperopts
|
This module load custom hyperopts
|
||||||
"""
|
"""
|
||||||
import logging
|
import logging
|
||||||
from os import path
|
from pathlib import Path
|
||||||
from typing import Optional, Dict
|
from typing import Optional, Dict
|
||||||
|
|
||||||
from freqtrade.constants import DEFAULT_HYPEROPT
|
from freqtrade.constants import DEFAULT_HYPEROPT
|
||||||
@ -40,16 +40,16 @@ class HyperOptResolver(IResolver):
|
|||||||
:param extra_dir: additional directory to search for the given hyperopt
|
:param extra_dir: additional directory to search for the given hyperopt
|
||||||
:return: HyperOpt instance or None
|
:return: HyperOpt instance or None
|
||||||
"""
|
"""
|
||||||
current_path = path.join(path.dirname(path.dirname(path.realpath(__file__))), 'optimize')
|
current_path = Path(__file__).parent.parent.joinpath('optimize').resolve()
|
||||||
|
|
||||||
abs_paths = [
|
abs_paths = [
|
||||||
path.join(current_path, '..', '..', 'user_data', 'hyperopts'),
|
current_path.parent.parent.joinpath('user_data/hyperopts'),
|
||||||
current_path,
|
current_path,
|
||||||
]
|
]
|
||||||
|
|
||||||
if extra_dir:
|
if extra_dir:
|
||||||
# Add extra hyperopt directory on top of search paths
|
# Add extra hyperopt directory on top of search paths
|
||||||
abs_paths.insert(0, extra_dir)
|
abs_paths.insert(0, Path(extra_dir))
|
||||||
|
|
||||||
for _path in abs_paths:
|
for _path in abs_paths:
|
||||||
hyperopt = self._search_object(directory=_path, object_type=IHyperOpt,
|
hyperopt = self._search_object(directory=_path, object_type=IHyperOpt,
|
||||||
|
@ -6,7 +6,7 @@ This module load custom objects
|
|||||||
import importlib.util
|
import importlib.util
|
||||||
import inspect
|
import inspect
|
||||||
import logging
|
import logging
|
||||||
import os
|
from pathlib import Path
|
||||||
from typing import Optional, Dict, Type, Any
|
from typing import Optional, Dict, Type, Any
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
@ -25,7 +25,7 @@ class IResolver(object):
|
|||||||
config = config or {}
|
config = config or {}
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _get_valid_objects(object_type, module_path: str,
|
def _get_valid_objects(object_type, module_path: Path,
|
||||||
object_name: str) -> Optional[Type[Any]]:
|
object_name: str) -> Optional[Type[Any]]:
|
||||||
"""
|
"""
|
||||||
Returns a list of all possible objects for the given module_path of type oject_type
|
Returns a list of all possible objects for the given module_path of type oject_type
|
||||||
@ -36,7 +36,7 @@ class IResolver(object):
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
# Generate spec based on absolute path
|
# Generate spec based on absolute path
|
||||||
spec = importlib.util.spec_from_file_location('unknown', module_path)
|
spec = importlib.util.spec_from_file_location('unknown', str(module_path))
|
||||||
module = importlib.util.module_from_spec(spec)
|
module = importlib.util.module_from_spec(spec)
|
||||||
spec.loader.exec_module(module) # type: ignore # importlib does not use typehints
|
spec.loader.exec_module(module) # type: ignore # importlib does not use typehints
|
||||||
|
|
||||||
@ -47,7 +47,7 @@ class IResolver(object):
|
|||||||
return next(valid_objects_gen, None)
|
return next(valid_objects_gen, None)
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _search_object(directory: str, object_type, object_name: str,
|
def _search_object(directory: Path, object_type, object_name: str,
|
||||||
kwargs: dict = {}) -> Optional[Any]:
|
kwargs: dict = {}) -> Optional[Any]:
|
||||||
"""
|
"""
|
||||||
Search for the objectname in the given directory
|
Search for the objectname in the given directory
|
||||||
@ -55,13 +55,13 @@ class IResolver(object):
|
|||||||
:return: object instance
|
:return: object instance
|
||||||
"""
|
"""
|
||||||
logger.debug('Searching for %s %s in \'%s\'', object_type.__name__, object_name, directory)
|
logger.debug('Searching for %s %s in \'%s\'', object_type.__name__, object_name, directory)
|
||||||
for entry in os.listdir(directory):
|
for entry in directory.iterdir():
|
||||||
# Only consider python files
|
# Only consider python files
|
||||||
if not entry.endswith('.py'):
|
if not str(entry).endswith('.py'):
|
||||||
logger.debug('Ignoring %s', entry)
|
logger.debug('Ignoring %s', entry)
|
||||||
continue
|
continue
|
||||||
obj = IResolver._get_valid_objects(
|
obj = IResolver._get_valid_objects(
|
||||||
object_type, os.path.abspath(os.path.join(directory, entry)), object_name
|
object_type, Path.resolve(directory.joinpath(entry)), object_name
|
||||||
)
|
)
|
||||||
if obj:
|
if obj:
|
||||||
return obj(**kwargs)
|
return obj(**kwargs)
|
||||||
|
@ -5,7 +5,6 @@ This module load custom strategies
|
|||||||
"""
|
"""
|
||||||
import inspect
|
import inspect
|
||||||
import logging
|
import logging
|
||||||
from os import getcwd, path
|
|
||||||
import tempfile
|
import tempfile
|
||||||
from base64 import urlsafe_b64decode
|
from base64 import urlsafe_b64decode
|
||||||
from collections import OrderedDict
|
from collections import OrderedDict
|
||||||
@ -103,16 +102,16 @@ class StrategyResolver(IResolver):
|
|||||||
:param extra_dir: additional directory to search for the given strategy
|
:param extra_dir: additional directory to search for the given strategy
|
||||||
:return: Strategy instance or None
|
:return: Strategy instance or None
|
||||||
"""
|
"""
|
||||||
current_path = path.join(path.dirname(path.dirname(path.realpath(__file__))), 'strategy')
|
current_path = Path(__file__).parent.parent.joinpath('strategy').resolve()
|
||||||
|
|
||||||
abs_paths = [
|
abs_paths = [
|
||||||
path.join(getcwd(), 'user_data', 'strategies'),
|
Path.cwd().joinpath('user_data/strategies'),
|
||||||
current_path,
|
current_path,
|
||||||
]
|
]
|
||||||
|
|
||||||
if extra_dir:
|
if extra_dir:
|
||||||
# Add extra strategy directory on top of search paths
|
# Add extra strategy directory on top of search paths
|
||||||
abs_paths.insert(0, extra_dir)
|
abs_paths.insert(0, Path(extra_dir).resolve())
|
||||||
|
|
||||||
if ":" in strategy_name:
|
if ":" in strategy_name:
|
||||||
logger.info("loading base64 endocded strategy")
|
logger.info("loading base64 endocded strategy")
|
||||||
@ -125,17 +124,17 @@ class StrategyResolver(IResolver):
|
|||||||
temp.joinpath(name).write_text(urlsafe_b64decode(strat[1]).decode('utf-8'))
|
temp.joinpath(name).write_text(urlsafe_b64decode(strat[1]).decode('utf-8'))
|
||||||
temp.joinpath("__init__.py").touch()
|
temp.joinpath("__init__.py").touch()
|
||||||
|
|
||||||
strategy_name = path.splitext(name)[0]
|
strategy_name = strat[0]
|
||||||
|
|
||||||
# register temp path with the bot
|
# register temp path with the bot
|
||||||
abs_paths.insert(0, str(temp.resolve()))
|
abs_paths.insert(0, temp.resolve())
|
||||||
|
|
||||||
for _path in abs_paths:
|
for _path in abs_paths:
|
||||||
try:
|
try:
|
||||||
strategy = self._search_object(directory=_path, object_type=IStrategy,
|
strategy = self._search_object(directory=_path, object_type=IStrategy,
|
||||||
object_name=strategy_name, kwargs={'config': config})
|
object_name=strategy_name, kwargs={'config': config})
|
||||||
if strategy:
|
if strategy:
|
||||||
logger.info('Using resolved strategy %s from \'%s\'', strategy_name, path)
|
logger.info('Using resolved strategy %s from \'%s\'', strategy_name, _path)
|
||||||
strategy._populate_fun_len = len(
|
strategy._populate_fun_len = len(
|
||||||
inspect.getfullargspec(strategy.populate_indicators).args)
|
inspect.getfullargspec(strategy.populate_indicators).args)
|
||||||
strategy._buy_fun_len = len(
|
strategy._buy_fun_len = len(
|
||||||
@ -145,7 +144,7 @@ class StrategyResolver(IResolver):
|
|||||||
|
|
||||||
return import_strategy(strategy, config=config)
|
return import_strategy(strategy, config=config)
|
||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
logger.warning('Path "%s" does not exist', _path)
|
logger.warning('Path "%s" does not exist', _path.relative_to(Path.cwd()))
|
||||||
|
|
||||||
raise ImportError(
|
raise ImportError(
|
||||||
"Impossible to load Strategy '{}'. This class does not exist"
|
"Impossible to load Strategy '{}'. This class does not exist"
|
||||||
|
@ -2,6 +2,7 @@
|
|||||||
import logging
|
import logging
|
||||||
from base64 import urlsafe_b64encode
|
from base64 import urlsafe_b64encode
|
||||||
from os import path
|
from os import path
|
||||||
|
from pathlib import Path
|
||||||
import warnings
|
import warnings
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
@ -40,9 +41,7 @@ def test_import_strategy(caplog):
|
|||||||
|
|
||||||
def test_search_strategy():
|
def test_search_strategy():
|
||||||
default_config = {}
|
default_config = {}
|
||||||
default_location = path.join(path.dirname(
|
default_location = Path(__file__).parent.parent.joinpath('strategy').resolve()
|
||||||
path.realpath(__file__)), '..', '..', 'strategy'
|
|
||||||
)
|
|
||||||
assert isinstance(
|
assert isinstance(
|
||||||
StrategyResolver._search_object(
|
StrategyResolver._search_object(
|
||||||
directory=default_location,
|
directory=default_location,
|
||||||
|
Loading…
Reference in New Issue
Block a user