Convert StrategyLoader to static loader
This commit is contained in:
@@ -20,12 +20,11 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
class StrategyResolver(IResolver):
|
||||
"""
|
||||
This class contains all the logic to load custom strategy class
|
||||
This class contains the logic to load custom strategy class
|
||||
"""
|
||||
|
||||
__slots__ = ['strategy']
|
||||
|
||||
def __init__(self, config: Optional[Dict] = None) -> None:
|
||||
@staticmethod
|
||||
def load_strategy(config: Optional[Dict] = None) -> IStrategy:
|
||||
"""
|
||||
Load the custom class from config parameter
|
||||
:param config: configuration dictionary or None
|
||||
@@ -37,9 +36,9 @@ class StrategyResolver(IResolver):
|
||||
"the strategy class to use.")
|
||||
|
||||
strategy_name = config['strategy']
|
||||
self.strategy: IStrategy = self._load_strategy(strategy_name,
|
||||
config=config,
|
||||
extra_dir=config.get('strategy_path'))
|
||||
strategy: IStrategy = StrategyResolver._load_strategy(
|
||||
strategy_name, config=config,
|
||||
extra_dir=config.get('strategy_path'))
|
||||
|
||||
# make sure ask_strategy dict is available
|
||||
if 'ask_strategy' not in config:
|
||||
@@ -68,9 +67,11 @@ class StrategyResolver(IResolver):
|
||||
]
|
||||
for attribute, default, ask_strategy in attributes:
|
||||
if ask_strategy:
|
||||
self._override_attribute_helper(config['ask_strategy'], attribute, default)
|
||||
StrategyResolver._override_attribute_helper(strategy, config['ask_strategy'],
|
||||
attribute, default)
|
||||
else:
|
||||
self._override_attribute_helper(config, attribute, default)
|
||||
StrategyResolver._override_attribute_helper(strategy, config,
|
||||
attribute, default)
|
||||
|
||||
# Loop this list again to have output combined
|
||||
for attribute, _, exp in attributes:
|
||||
@@ -80,14 +81,16 @@ class StrategyResolver(IResolver):
|
||||
logger.info("Strategy using %s: %s", attribute, config[attribute])
|
||||
|
||||
# Sort and apply type conversions
|
||||
self.strategy.minimal_roi = OrderedDict(sorted(
|
||||
{int(key): value for (key, value) in self.strategy.minimal_roi.items()}.items(),
|
||||
strategy.minimal_roi = OrderedDict(sorted(
|
||||
{int(key): value for (key, value) in strategy.minimal_roi.items()}.items(),
|
||||
key=lambda t: t[0]))
|
||||
self.strategy.stoploss = float(self.strategy.stoploss)
|
||||
strategy.stoploss = float(strategy.stoploss)
|
||||
|
||||
self._strategy_sanity_validations()
|
||||
StrategyResolver._strategy_sanity_validations(strategy)
|
||||
return strategy
|
||||
|
||||
def _override_attribute_helper(self, config, attribute: str, default):
|
||||
@staticmethod
|
||||
def _override_attribute_helper(strategy, config, attribute: str, default):
|
||||
"""
|
||||
Override attributes in the strategy.
|
||||
Prevalence:
|
||||
@@ -96,30 +99,32 @@ class StrategyResolver(IResolver):
|
||||
- default (if not None)
|
||||
"""
|
||||
if attribute in config:
|
||||
setattr(self.strategy, attribute, config[attribute])
|
||||
setattr(strategy, attribute, config[attribute])
|
||||
logger.info("Override strategy '%s' with value in config file: %s.",
|
||||
attribute, config[attribute])
|
||||
elif hasattr(self.strategy, attribute):
|
||||
val = getattr(self.strategy, attribute)
|
||||
elif hasattr(strategy, attribute):
|
||||
val = getattr(strategy, attribute)
|
||||
# None's cannot exist in the config, so do not copy them
|
||||
if val is not None:
|
||||
config[attribute] = val
|
||||
# Explicitly check for None here as other "falsy" values are possible
|
||||
elif default is not None:
|
||||
setattr(self.strategy, attribute, default)
|
||||
setattr(strategy, attribute, default)
|
||||
config[attribute] = default
|
||||
|
||||
def _strategy_sanity_validations(self):
|
||||
if not all(k in self.strategy.order_types for k in constants.REQUIRED_ORDERTYPES):
|
||||
raise ImportError(f"Impossible to load Strategy '{self.strategy.__class__.__name__}'. "
|
||||
@staticmethod
|
||||
def _strategy_sanity_validations(strategy):
|
||||
if not all(k in strategy.order_types for k in constants.REQUIRED_ORDERTYPES):
|
||||
raise ImportError(f"Impossible to load Strategy '{strategy.__class__.__name__}'. "
|
||||
f"Order-types mapping is incomplete.")
|
||||
|
||||
if not all(k in self.strategy.order_time_in_force for k in constants.REQUIRED_ORDERTIF):
|
||||
raise ImportError(f"Impossible to load Strategy '{self.strategy.__class__.__name__}'. "
|
||||
if not all(k in strategy.order_time_in_force for k in constants.REQUIRED_ORDERTIF):
|
||||
raise ImportError(f"Impossible to load Strategy '{strategy.__class__.__name__}'. "
|
||||
f"Order-time-in-force mapping is incomplete.")
|
||||
|
||||
def _load_strategy(
|
||||
self, strategy_name: str, config: dict, extra_dir: Optional[str] = None) -> IStrategy:
|
||||
@staticmethod
|
||||
def _load_strategy(strategy_name: str,
|
||||
config: dict, extra_dir: Optional[str] = None) -> IStrategy:
|
||||
"""
|
||||
Search and loads the specified strategy.
|
||||
:param strategy_name: name of the module to import
|
||||
@@ -129,9 +134,9 @@ class StrategyResolver(IResolver):
|
||||
"""
|
||||
current_path = Path(__file__).parent.parent.joinpath('strategy').resolve()
|
||||
|
||||
abs_paths = self.build_search_paths(config, current_path=current_path,
|
||||
user_subdir=constants.USERPATH_STRATEGY,
|
||||
extra_dir=extra_dir)
|
||||
abs_paths = IResolver.build_search_paths(config, current_path=current_path,
|
||||
user_subdir=constants.USERPATH_STRATEGY,
|
||||
extra_dir=extra_dir)
|
||||
|
||||
if ":" in strategy_name:
|
||||
logger.info("loading base64 encoded strategy")
|
||||
@@ -149,8 +154,8 @@ class StrategyResolver(IResolver):
|
||||
# register temp path with the bot
|
||||
abs_paths.insert(0, temp.resolve())
|
||||
|
||||
strategy = self._load_object(paths=abs_paths, object_type=IStrategy,
|
||||
object_name=strategy_name, kwargs={'config': config})
|
||||
strategy = IResolver._load_object(paths=abs_paths, object_type=IStrategy,
|
||||
object_name=strategy_name, kwargs={'config': config})
|
||||
if strategy:
|
||||
strategy._populate_fun_len = len(getfullargspec(strategy.populate_indicators).args)
|
||||
strategy._buy_fun_len = len(getfullargspec(strategy.populate_buy_trend).args)
|
||||
|
Reference in New Issue
Block a user