import copy
import yaml
import importlib
from pathlib import Path
from typing import Union
import onetick.py as otp
[docs]class ExperimentBuilder():
"""Builder class used to restore Experiment from YAML/dict config.
Parameters
----------
config : Union[dict, str]
YAML config or dict with config.
globals_dict : dict
Dict with globals, used to restore classes from global scope.
"""
def __init__(self, config: Union[dict, str] = {}, globals_dict: dict = {}):
self.config = config
self.globals_dict = globals_dict
def _build_instance(self, params, force_use=False):
"""Builds class instance from params dict with class definition.
Parameters
----------
params : dict
Dict with params for class instance.
force_use : bool
If True, will try to build instance even if `use=False` in params.
Returns
-------
instance
Instance of class defined in params.
"""
if not params:
return None
params = copy.deepcopy(params)
if not params.pop('use', False) and not force_use:
return None
class_name = params.pop('class')
_class = self.get_class_def(class_name)
return _class.restore_instance(params)
def _build_instances_list(self, params_list):
"""Builds list of instances from list of params dicts.
Parameters
----------
params_list : list
List of params dicts.
Returns
-------
list
List of instances.
"""
result = []
params_list = [] if params_list is None else params_list
for params in params_list:
if isinstance(params, str):
result.append(params)
continue
instance = self._build_instance(params)
if instance:
result.append(instance)
return result
[docs] @staticmethod
def get_class_name(class_obj):
"""Returns class name with module name.
Parameters
----------
class_obj : class
Class object.
Returns
-------
str
Class name with module name.
"""
if class_obj.__module__.startswith("onetick.ml."):
return "onetick.ml." + class_obj.__class__.__name__
else:
return class_obj.__class__.__qualname__
[docs] def get_class_def(self, class_qualname):
"""Returns class object (not instance) by class name with module name.
Parameters
----------
class_qualname : str
Class name with module name.
Returns
-------
class
Class object (not instance).
"""
if "." not in class_qualname:
return None
module_name, class_name = class_qualname.rsplit(".", 1)
if "<locals>" in class_qualname:
class_def = self.globals_dict.get(class_name, None)
if not class_def:
raise Exception(
f"{class_name} wasn't supplied to ExperimentBuilder"
" in global_dict argument, cannot restore")
else:
module = importlib.import_module(module_name)
class_def = getattr(module, class_name)
return class_def
[docs] def build_experiment_class(self):
"""Builds on-fly Experiment class with config loaded from YAML.
Returns
-------
class
Experiment class built from YAML config supplied to builder.
"""
config = self.config
if isinstance(config, (str, Path)):
with open(config, encoding='utf8') as fp:
config = yaml.load(fp, Loader=yaml.Loader)
config = copy.deepcopy(config)
# prevent cyclic import
from onetick.ml.impl.experiments.experiment import Experiment
class _Class(Experiment):
# Please, keep this order exactly as in Experiment
general = config.get('general', {})
datafeeds = self._build_instances_list(config.get("datafeeds", []))
splitters = self._build_instances_list(config.get("splitters", []))
pipeline = self._build_instances_list(config.get("pipeline", []))
models = self._build_instances_list(config.get("models", []))
evaluators = self._build_instances_list(config.get("evaluators", []))
features_columns = config.get('features_columns', [])
target_columns = config.get('target_columns', [])
# train params
train_params = config.get('train_params', {})
# prediction params
prediction_params = config.get('prediction', {})
return _Class
