From 544365f3a9ac13860ed1075d0211abeb5fbf8e42 Mon Sep 17 00:00:00 2001 From: lzh222333 Date: Wed, 31 Mar 2021 02:39:14 +0000 Subject: [PATCH] ensemble & get_exp & dataset_pickle --- .../model_rolling/task_manager_rolling.py | 21 +- .../task_manager_rolling_with_updating.py | 19 +- qlib/model/trainer.py | 3 - qlib/workflow/task/collect.py | 179 ++++------------- qlib/workflow/task/ensemble.py | 180 ++++++++++++++++++ qlib/workflow/task/update.py | 9 +- qlib/workflow/task/utils.py | 2 +- 7 files changed, 249 insertions(+), 164 deletions(-) create mode 100644 qlib/workflow/task/ensemble.py diff --git a/examples/model_rolling/task_manager_rolling.py b/examples/model_rolling/task_manager_rolling.py index 70a4f7d7e..e5de1ef60 100644 --- a/examples/model_rolling/task_manager_rolling.py +++ b/examples/model_rolling/task_manager_rolling.py @@ -5,9 +5,12 @@ import qlib from qlib.config import REG_CN from qlib.model.trainer import task_train from qlib.workflow import R -from qlib.workflow.task.collect import RollingCollector from qlib.workflow.task.gen import RollingGen, task_generator from qlib.workflow.task.manage import TaskManager, run_task +from qlib.workflow.task.collect import RecorderCollector +from qlib.workflow.task.ensemble import RollingEnsemble +import pandas as pd +from qlib.workflow.task.utils import list_recorders data_handler_config = { "start_time": "2008-01-01", @@ -70,7 +73,7 @@ def reset(task_pool, exp_name): print("========== reset ==========") TaskManager(task_pool=task_pool).remove() - exp, _ = R.exp_manager._get_or_create_exp(experiment_name=exp_name) + exp, _ = R.get_exp(experiment_name=exp_name) for rid in exp.list_recorders(): exp.delete_recorder(rid) @@ -110,19 +113,21 @@ def task_collecting(task_pool, exp_name): def get_group_key_func(recorder): task_config = recorder.load_object("task") - return task_config["model"]["class"] + model_key = task_config["model"]["class"] + rolling_key = task_config["dataset"]["kwargs"]["segments"]["test"] + return model_key, model_key, rolling_key def my_filter(recorder): # only choose the results of "LGBModel" - task_key = get_group_key_func(recorder) - if task_key == "LGBModel": + model_key, rolling_key = get_group_key_func(recorder) + if model_key == "LGBModel": return True return False - rolling_collector = RollingCollector(exp_name) + collector = RecorderCollector(exp_name) # group tasks by "get_task_key" and filter tasks by "my_filter" - pred_rolling = rolling_collector.collect(get_group_key_func, my_filter) - print(pred_rolling) + artifact = collector.collect(RollingEnsemble(), get_group_key_func, rec_filter_func=my_filter) + print(artifact) def main( diff --git a/examples/online_svr/task_manager_rolling_with_updating.py b/examples/online_svr/task_manager_rolling_with_updating.py index 24bc38a02..4e9fdd336 100644 --- a/examples/online_svr/task_manager_rolling_with_updating.py +++ b/examples/online_svr/task_manager_rolling_with_updating.py @@ -5,7 +5,8 @@ import qlib from qlib.config import REG_CN from qlib.model.trainer import task_train from qlib.workflow import R -from qlib.workflow.task.collect import RollingCollector +from qlib.workflow.task.collect import RecorderCollector +from qlib.workflow.task.ensemble import RollingEnsemble from qlib.workflow.task.gen import RollingGen, task_generator from qlib.workflow.task.manage import TaskManager, run_task from qlib.workflow.task.online import RollingOnlineManager @@ -114,26 +115,28 @@ def task_collecting(): def get_group_key_func(recorder): task_config = recorder.load_object("task") - return task_config["model"]["class"] + model_key = task_config["model"]["class"] + rolling_key = task_config["dataset"]["kwargs"]["segments"]["test"] + return model_key, model_key, rolling_key def my_filter(recorder): # only choose the results of "LGBModel" - task_key = get_group_key_func(recorder) - if task_key == "LGBModel": + model_key, rolling_key = get_group_key_func(recorder) + if model_key == "LGBModel": return True return False - rolling_collector = RollingCollector(exp_name) + collector = RecorderCollector(exp_name) # group tasks by "get_task_key" and filter tasks by "my_filter" - pred_rolling = rolling_collector.collect(get_group_key_func, my_filter) - print(pred_rolling) + artifact = collector.collect(RollingEnsemble(), get_group_key_func, rec_filter_func=my_filter) + print(artifact) # Reset all things to the first status, be careful to save important data def reset(): print("========== reset ==========") task_manager.remove() - exp, _ = R.exp_manager._get_or_create_exp(experiment_name=exp_name) + exp, _ = R.get_exp(experiment_name=exp_name) for rid in exp.list_recorders(): exp.delete_recorder(rid) diff --git a/qlib/model/trainer.py b/qlib/model/trainer.py index 60f56609f..45650c0c7 100644 --- a/qlib/model/trainer.py +++ b/qlib/model/trainer.py @@ -26,8 +26,6 @@ def task_train(task_config: dict, experiment_name: str) -> str: # model initiaiton model = init_instance_by_config(task_config["model"]) dataset = init_instance_by_config(task_config["dataset"]) - datahandler = dataset.handler - dataset.config(exclude=["handler"]) # start exp with R.start(experiment_name=experiment_name): @@ -39,7 +37,6 @@ def task_train(task_config: dict, experiment_name: str) -> str: R.save_objects(**{"params.pkl": model}) R.save_objects(**{"task": task_config}) # keep the original format and datatype R.save_objects(**{"dataset": dataset}) - R.save_objects(**{"datahandler": datahandler}) # generate records: prediction, backtest, and analysis records = task_config.get("record", []) diff --git a/qlib/workflow/task/collect.py b/qlib/workflow/task/collect.py index 0a007cc5c..a7a6ce4bb 100644 --- a/qlib/workflow/task/collect.py +++ b/qlib/workflow/task/collect.py @@ -7,166 +7,69 @@ from qlib.workflow.task.utils import list_recorders class Collector: - """ - This class will divide disorderly records or anything worth collecting into different groups based on the group_key. - After grouping, we can reduce the useful information from different groups. + """The collector to collect different results based on experiment backend and ensemble method """ - def group(self, *args, **kwargs): - """ - According to the get_group_key_func, divide disorderly things into different groups. - - For example: - - .. code-block:: python - - input: - [thing1, thing2, thing3, thing4, thing5] - - output: - { - "group_name1": [thing3, thing5, thing1] - "group_name2": [thing2, thing4] - } + def collect(self, ensemble, get_group_key_func, *args, **kwargs): + """To collect the results, we need to get the experiment record firstly and divided them into + different groups. Then use ensemble methods to merge the group. Args: - get_group_key_func (Callable): get a group key based on a thing - things_list (list): a list of things - - Returns: - dict: a dict including the group key and members of the group. + ensemble (Ensemble): an instance of Ensemble + get_group_key_func (Callable): a function to get the group of a experiment record """ - raise NotImplementedError(f"Please implement the `group` method.") - - def reduce(self, things_group: dict): - """ - Using the dict from `group`, reduce useful information. - - Args: - things_group (dict): a dict after grouping - - Returns: - dict: a dict including the group key, the information key and the information value - - """ - raise NotImplementedError(f"Please implement the `reduce` method.") - - def collect(self, *args, **kwargs): - """group and reduce - - Returns: - dict: a dict including the group key, the information key and the information value - """ - grouped = self.group(*args, **kwargs) - return self.reduce(grouped) + raise NotImplementedError(f"Please implement the `collect` method.") class RecorderCollector(Collector): - """ - The Recorder's Collector. This class is a implementation of Collector, collecting some artifacts saved by Recorder. - """ - - def __init__(self, experiment_name: str) -> None: - self.exp_name = experiment_name - self.logger = get_module_logger(self.__class__.__name__) - - _artifacts_key_path = {"pred": "pred.pkl", "IC": "sig_analysis/ic.pkl"} - _artifacts_key_merge_method = {} - - def default_merge(self, artifact_list): - """Merge disorderly artifacts in artifact list. + def __init__(self, exp_name, artifacts_path = {"pred": "pred.pkl", "IC": "sig_analysis/ic.pkl"}) -> None: + """init RecorderCollector Args: - artifact_list (list): A artifact list. - - Raises: - NotImplementedError: [description] + exp_name (str): the name of Experiment + artifacts_path (dict, optional): The artifacts name and its path in Recorder. Defaults to {"pred": "pred.pkl", "IC": "sig_analysis/ic.pkl"}. """ - raise NotImplementedError(f"Please implement the `default_merge` method.") + self.exp_name = exp_name + self.artifacts_path = artifacts_path - def group(self, get_group_key_func, rec_filter_func=None): - """ - Filter recorders and group recorders by group key. + def collect(self, ensemble, get_group_key_func, artifacts_key=None, rec_filter_func=None): + """Collect different artifacts based on recorder after filtering and ensemble method. + Group recorder by get_group_key_func. Args: - get_group_key_func (Callable): get a group key based on a recorder - rec_filter_func (Callable, optional): filter the recorders in this experiment. Defaults to None. + ensemble (Ensemble): an instance of Ensemble + get_group_key_func (Callable): a function to get the group of a experiment record + artifacts_key (str or List, optional): the artifacts key you want to get. Defaults to None. + rec_filter_func (Callable, optional): filter the recorder by return True or False. Defaults to None. Returns: - dict: a dict including the group key and recorders of the group + dict: the dict after collected. """ + if artifacts_key is None: + artifacts_key = self.artifacts_path.keys() + + if isinstance(artifacts_key, str): + artifacts_key = [artifacts_key] + + # prepare_ensemble + ensemble_dict = {} + for key in artifacts_key: + ensemble_dict.setdefault(key,{}) # filter records recs_flt = list_recorders(self.exp_name, rec_filter_func) - - # group - recs_group = {} for _, rec in recs_flt.items(): group_key = get_group_key_func(rec) - recs_group.setdefault(group_key, []).append(rec) - - return recs_group - - def reduce(self, recs_group: dict, artifact_keys_list: list = None): - """ - Reduce artifacts based on the dict of grouped recorder. - The artifacts need be declared by artifact_keys_list. - The artifacts path in recorder need be declared by _artifacts_key_path. - If there is no declartion in _artifacts_key_merge_method, then use default_merge method to merge it. - - Args: - recs_group (dict): The dict grouped by `group` - artifact_keys_list (list): The list of artifact keys. If it is None, then use all artifacts in _artifacts_key_path. - - Returns: - a dict including the group key, the artifact key and the artifact value. - - For example: - - .. code-block:: python - - { - group_key: {"pred": , "IC": } - } - """ - if artifact_keys_list == None: - artifact_keys_list = self._artifacts_key_path.keys() - reduce_group = {} - for group_key, recorder_list in recs_group.items(): - reduced_artifacts = {} - for artifact_key in artifact_keys_list: - artifact_list = [] - for recorder in recorder_list: - artifact_list.append(recorder.load_object(self._artifacts_key_path[artifact_key])) - merge_method = self._artifacts_key_merge_method.get(artifact_key, self.default_merge) - artifact = merge_method(artifact_list) - reduced_artifacts[artifact_key] = artifact - reduce_group[group_key] = reduced_artifacts - return reduce_group + for key in artifacts_key: + artifact = rec.load_object(self.artifacts_path[key]) + ensemble_dict[key][group_key] = artifact -class RollingCollector(RecorderCollector): - """ - Collect the record results of the rolling tasks - """ + if isinstance(artifacts_key, str): + return ensemble(ensemble_dict[artifacts_key]) - def __init__(self, experiment_name: str): - super().__init__(experiment_name) - self.logger = get_module_logger(self.__class__.__name__) - - def default_merge(self, artifact_list): - """merge disorderly artifacts based on the datetime. - - Args: - artifact_list (list): a list of artifacts from different recorders - - Returns: - merged artifact - """ - # Make sure the pred are sorted according to the rolling start time - artifact_list.sort(key=lambda x: x.index.get_level_values("datetime").min()) - artifact = pd.concat(artifact_list) - # If there are duplicated predition, we use the latest perdiction - artifact = artifact[~artifact.index.duplicated(keep="last")] - artifact = artifact.sort_index() - return artifact + collect_dict = {} + for key in artifacts_key: + collect_dict[key] = ensemble(ensemble_dict[key]) + return collect_dict + \ No newline at end of file diff --git a/qlib/workflow/task/ensemble.py b/qlib/workflow/task/ensemble.py new file mode 100644 index 000000000..649ce9415 --- /dev/null +++ b/qlib/workflow/task/ensemble.py @@ -0,0 +1,180 @@ +from abc import abstractmethod +from typing import Callable, Union + +import pandas as pd +from qlib import get_module_logger +from qlib.workflow.task.utils import list_recorders +from typing import Dict + + + +class Ensemble: + """Merge the objects in an Ensemble. + """ + + def __init__(self, merge_func = None, get_grouped_key_func = None) -> None: + """init Ensemble + + Args: + merge_func (Callable, optional): The specific merge function. Defaults to None. + get_grouped_key_func (Callable, optional): Get group_inner_key and group_outer_key by group_key. Defaults to None. + """ + self.logger = get_module_logger(self.__class__.__name__) + if merge_func is not None: + self.merge_func = merge_func + if get_grouped_key_func is not None: + self.get_grouped_key_func = get_grouped_key_func + + def merge_func(self, group_inner_dict): + """Given a group_inner_dict such as {Rollinga_b: object, Rollingb_c: object}, + merge it to object + + Args: + group_inner_dict (dict): the inner group dict + + """ + raise NotImplementedError(f"Please implement the `merge_func` method.") + + def get_grouped_key_func(self, group_key): + """Given a group_key and return the group_outer_key, group_inner_key. + + For example: + (A,B,Rolling) -> (A,B):Rolling + (A,B) -> C:(A,B) + + Args: + group_key (tuple or str): the group key + """ + raise NotImplementedError(f"Please implement the `get_grouped_key_func` method.") + + def group(self, group_dict: Dict[tuple or str, object]) -> Dict[tuple or str, Dict[tuple or str, object]]: + """In a group of dict, further divide them into outgroups and innergroup. + + For example: + + .. code-block:: python + + RollingEnsemble: + input: + { + (ModelA,Horizon5,Rollinga_b): object + (ModelA,Horizon5,Rollingb_c): object + (ModelA,Horizon10,Rollinga_b): object + (ModelA,Horizon10,Rollingb_c): object + (ModelB,Horizon5,Rollinga_b): object + (ModelB,Horizon5,Rollingb_c): object + (ModelB,Horizon10,Rollinga_b): object + (ModelB,Horizon10,Rollingb_c): object + } + + output: + { + (ModelA,Horizon5): {Rollinga_b: object, Rollingb_c: object} + (ModelA,Horizon10): {Rollinga_b: object, Rollingb_c: object} + (ModelB,Horizon5): {Rollinga_b: object, Rollingb_c: object} + (ModelB,Horizon10): {Rollinga_b: object, Rollingb_c: object} + } + + Args: + group_dict (Dict[tuple or str, object]): a group of dict + + Returns: + Dict[tuple or str, Dict[tuple or str, object]]: the dict after `group` + """ + grouped_dict = {} + for group_key, artifact in group_dict.items(): + group_outer_key, group_inner_key = self.get_grouped_key_func(group_key) # (A,B,Rolling) -> (A,B):Rolling + grouped_dict.setdefault(group_outer_key, {})[group_inner_key] = artifact + return grouped_dict + + def reduce(self, grouped_dict: dict): + """After grouping, reduce the innergroup. + + For example: + + .. code-block:: python + + RollingEnsemble: + input: + { + (ModelA,Horizon5): {Rollinga_b: object, Rollingb_c: object} + (ModelA,Horizon10): {Rollinga_b: object, Rollingb_c: object} + (ModelB,Horizon5): {Rollinga_b: object, Rollingb_c: object} + (ModelB,Horizon10): {Rollinga_b: object, Rollingb_c: object} + } + + output: + { + (ModelA,Horizon5): object + (ModelA,Horizon10): object + (ModelB,Horizon5): object + (ModelB,Horizon10): object + } + + Args: + grouped_dict (dict): the dict after `group` + + Returns: + dict: the dict after `reduce` + """ + reduce_group = {} + for group_outer_key, group_inner_dict in grouped_dict.items(): + artifact = self.merge_func(group_inner_dict) + reduce_group[group_outer_key] = artifact + return reduce_group + + def __call__(self, group_dict): + """The process of Ensemble is group it firstly and then reduce it, like MapReduce. + + Args: + group_dict (Dict[tuple or str, object]): a group of dict + + Returns: + dict: the dict after `reduce` + """ + grouped_dict = self.group(group_dict) + return self.reduce(grouped_dict) + +class RollingEnsemble(Ensemble): + """A specific implementation of Ensemble for Rolling. + + """ + def merge_func(self, group_inner_dict): + """merge group_inner_dict by datetime. + + Args: + group_inner_dict (dict): the inner group dict + + Returns: + object: the artifact after merging + """ + artifact_list = list(group_inner_dict.values()) + artifact_list.sort(key=lambda x: x.index.get_level_values("datetime").min()) + artifact = pd.concat(artifact_list) + # If there are duplicated predition, use the latest perdiction + artifact = artifact[~artifact.index.duplicated(keep="last")] + artifact = artifact.sort_index() + return artifact + + def get_grouped_key_func(self, group_key): + """The final axis of group_key must be the Rolling key. + When `collect`, get_group_key_func can add the statement below. + + .. code-block:: python + + def get_group_key_func(recorder): + task_config = recorder.load_object("task") + ...... + rolling_key = task_config["dataset"]["kwargs"]["segments"]["test"] + return ......, rolling_key + + Args: + group_key (tuple or str): the group key + + Returns: + tuple or str, tuple or str: group_outer_key, group_inner_key + """ + assert len(group_key)>=2 + return group_key[:-1], group_key[-1] + + diff --git a/qlib/workflow/task/update.py b/qlib/workflow/task/update.py index bab7df3c8..43c304239 100644 --- a/qlib/workflow/task/update.py +++ b/qlib/workflow/task/update.py @@ -7,6 +7,7 @@ from qlib.workflow import R from qlib.model.trainer import task_train from qlib.workflow.recorder import Recorder from qlib.workflow.task.utils import list_recorders +from qlib.data.dataset.handler import DataHandlerLP class ModelUpdater: """ @@ -42,13 +43,9 @@ class ModelUpdater: the instance of Dataset """ segments = {"test": (start_time, end_time)} - dataset = recorder.load_object("dataset") - datahandler = recorder.load_object("datahandler") - - datahandler.conf_data(**{"start_time": start_time, "end_time": end_time}) - dataset.setup_data(handler=datahandler, segments=segments) - datahandler.init(datahandler.IT_LS) + dataset.config(handler_kwargs={"start_time": start_time, "end_time": end_time}) + dataset.setup_data(handler_kwargs={"init_type": DataHandlerLP.IT_LS}, segments=segments) return dataset def update_pred(self, recorder: Recorder, frequency='day'): diff --git a/qlib/workflow/task/utils.py b/qlib/workflow/task/utils.py index 29d7a495c..b34b75306 100644 --- a/qlib/workflow/task/utils.py +++ b/qlib/workflow/task/utils.py @@ -42,7 +42,7 @@ def list_recorders(experiment, rec_filter_func=None): dict: a dict {rid: recorder} after filtering. """ if isinstance(experiment, str): - experiment, _ = R.exp_manager._get_or_create_exp(experiment_name=experiment) + experiment, _ = R.get_exp(experiment_name=experiment) recs = experiment.list_recorders() recs_flt = {} for rid, rec in recs.items():