liuyu/qlib
1
0
Fork 0
mirror of https://github.com/microsoft/qlib.git synced 2026-07-06 20:41:09 +08:00
Code Activity

ensemble & get_exp & dataset_pickle

Browse Source
This commit is contained in:
lzh222333
2021-03-31 02:39:14 +00:00
parent eae94d1ee8
commit 544365f3a9
7 changed files with 249 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
examples/model_rolling/task_manager_rolling.py
View File

@@ -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(

19
examples/online_svr/task_manager_rolling_with_updating.py
View File

@@ -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)

3
qlib/model/trainer.py
View File

@@ -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", [])

179
qlib/workflow/task/collect.py
View File

@@ -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": <VALUE>, "IC": <VALUE>}
}
"""
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

180
qlib/workflow/task/ensemble.py Normal file
View File

@@ -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]

9
qlib/workflow/task/update.py
View File

@@ -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'):

2
qlib/workflow/task/utils.py
View File

@@ -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():