diff --git a/examples/model_rolling/task_manager_rolling.py b/examples/model_rolling/task_manager_rolling.py index ab3a4eee5..175319885 100644 --- a/examples/model_rolling/task_manager_rolling.py +++ b/examples/model_rolling/task_manager_rolling.py @@ -1,24 +1,23 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +""" +This example shows how a TrainerRM work based on TaskManager with rolling tasks. +After training, how to collect the rolling results will be showed in task_collecting. +""" + from pprint import pprint -import time import fire import qlib from qlib.config import REG_CN -from qlib.model.trainer import TrainerR, task_train from qlib.workflow import R from qlib.workflow.task.gen import RollingGen, task_generator -from qlib.workflow.task.manage import TaskManager, run_task +from qlib.workflow.task.manage import TaskManager from qlib.workflow.task.collect import RecorderCollector -from qlib.model.ens.ensemble import RollingEnsemble, ens_workflow -import pandas as pd -from qlib.workflow.task.utils import list_recorders from qlib.model.ens.group import RollingGroup from qlib.model.trainer import TrainerRM -""" -This example shows how a Trainer work based on TaskManager with rolling tasks. -After training, how to collect the rolling results will be showed in task_collecting. -""" data_handler_config = { "start_time": "2008-01-01", @@ -139,11 +138,13 @@ class RollingTaskExample: return True return False - artifact = ens_workflow( - RecorderCollector(experiment=self.experiment_name, rec_key_func=rec_key, rec_filter_func=my_filter), - RollingGroup(), + collector = RecorderCollector( + experiment=self.experiment_name, + process_list=RollingGroup(), + rec_key_func=rec_key, + rec_filter_func=my_filter, ) - print(artifact) + print(collector()) def main(self): self.reset() diff --git a/examples/online_srv/online_management_simulate.py b/examples/online_srv/online_management_simulate.py index 6a1d233ae..16e985ccd 100644 --- a/examples/online_srv/online_management_simulate.py +++ b/examples/online_srv/online_management_simulate.py @@ -1,23 +1,17 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. + """ -This examples is about the OnlineManager and OnlineSimulator based on rolling tasks. -The OnlineManager will focus on the updating of your online models. -The OnlineSimulator will focus on the simulating real updating routine of your online models. +This examples is about how can simulate the OnlineManager based on rolling tasks. """ + import fire import qlib -from qlib.model.ens.ensemble import ens_workflow -from qlib.model.trainer import DelayTrainerR, DelayTrainerRM, TrainerRM -from qlib.workflow import R -from qlib.workflow.online.manager import OnlineM # RollingOnlineManager -from qlib.workflow.online.strategy import OnlineStrategy, RollingAverageStrategy -from qlib.workflow.task.collect import RecorderCollector -from qlib.workflow.task.gen import RollingGen, task_generator +from qlib.model.trainer import DelayTrainerRM +from qlib.workflow.online.manager import OnlineManager +from qlib.workflow.online.strategy import RollingAverageStrategy +from qlib.workflow.task.gen import RollingGen from qlib.workflow.task.manage import TaskManager -from qlib.workflow.task.utils import list_recorders - - data_handler_config = { @@ -89,10 +83,10 @@ class OnlineSimulationExample: rolling_step=80, start_time="2018-09-10", end_time="2018-10-31", - tasks=[task_xgboost_config], # , task_lgb_config] + tasks=[task_xgboost_config, task_lgb_config], ): """ - init OnlineManagerExample. + Init OnlineManagerExample. Args: provider_uri (str, optional): the provider uri. Defaults to "~/.qlib/qlib_data/cn_data". @@ -120,42 +114,28 @@ class OnlineSimulationExample: ) # The rolling tasks generator, modify_end_time is false because we just need simulate to 2018-10-31. self.trainer = DelayTrainerRM(self.exp_name, self.task_pool) self.task_manager = TaskManager(self.task_pool) # A good way to manage all your tasks - self.rolling_online_manager = OnlineM( + self.rolling_online_manager = OnlineManager( RollingAverageStrategy( exp_name, task_template=tasks, rolling_gen=self.rolling_gen, trainer=self.trainer, need_log=False ), begin_time=self.start_time, need_log=False, - ) # The OnlineManager based on Rolling - # self.onlinesimulator = OnlineSimulator( - # start_time=start_time, - # end_time=end_time, - # online_manager=self.rolling_online_manager, - # ) + ) self.tasks = tasks - # Reset all things to the first status, be careful to save important data - def reset(self): - print("========== reset ==========") - self.task_manager.remove() - - exp = R.get_exp(experiment_name=self.exp_name) - for rid in exp.list_recorders(): - exp.delete_recorder(rid) - - for rid in list_recorders("OnlineManagerSignals", lambda x: True if x.info["name"] == self.exp_name else False): - exp.delete_recorder(rid) - - # Run this to run all workflow automaticly + # Run this to run all workflow automatically def main(self): - self.reset() + print("========== reset ==========") + self.rolling_online_manager.reset() print("========== simulate ==========") self.rolling_online_manager.simulate(end_time=self.end_time) + print("========== collect results ==========") print(self.rolling_online_manager.get_collector()()) + print("========== online history ==========") print(self.rolling_online_manager.get_online_history(self.exp_name)) if __name__ == "__main__": - ## to run all workflow automaticly with your own parameters, use the command below + ## to run all workflow automatically with your own parameters, use the command below # python online_management_simulate.py main --experiment_name="your_exp_name" --rolling_step=60 fire.Fire(OnlineSimulationExample) diff --git a/examples/online_srv/rolling_online_management.py b/examples/online_srv/rolling_online_management.py index 7b2f58909..950c9684d 100644 --- a/examples/online_srv/rolling_online_management.py +++ b/examples/online_srv/rolling_online_management.py @@ -1,22 +1,25 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + """ -This example show how RollingOnlineManager works with rolling tasks. +This example show how OnlineManager works with rolling tasks. There are two parts including first train and routine. -Firstly, the RollingOnlineManager will finish the first training and set trained models to `online` models. -Next, the RollingOnlineManager will finish a routine process, including update online prediction -> prepare signals -> prepare tasks -> prepare new models -> reset online models +Firstly, the OnlineManager will finish the first training and set trained models to `online` models. +Next, the OnlineManager will finish a routine process, including update online prediction -> prepare signals -> prepare tasks -> prepare new models -> reset online models """ + import os from pathlib import Path import pickle import fire import qlib from qlib.workflow import R -from qlib.workflow.online.strategy import OnlineStrategy, RollingAverageStrategy +from qlib.workflow.online.strategy import RollingAverageStrategy from qlib.workflow.task.gen import RollingGen from qlib.workflow.task.manage import TaskManager -from qlib.workflow.online.manager import OnlineM +from qlib.workflow.online.manager import OnlineManager from qlib.workflow.task.utils import list_recorders from qlib.model.trainer import TrainerRM -from pprint import pprint data_handler_config = { "start_time": "2013-01-01", @@ -94,7 +97,7 @@ class RollingOnlineExample: self.rolling_step = rolling_step strategy = [] for task in tasks: - name_id = task["model"]["class"] + "_" + str(self.rolling_step) + name_id = task["model"]["class"] # NOTE: Assumption: The model class can specify only one strategy strategy.append( RollingAverageStrategy( name_id, @@ -104,9 +107,12 @@ class RollingOnlineExample: ) ) - self.rolling_online_manager = OnlineM(strategy) + self.rolling_online_manager = OnlineManager(strategy) + self.collector = self.rolling_online_manager.get_collector() - _ROLLING_MANAGER_PATH = ".rolling_manager" # the RollingOnlineManager will dump to this file, for it will be loaded when calling routine. + _ROLLING_MANAGER_PATH = ( + ".RollingOnlineExample" # the OnlineManager will dump to this file, for it can be loaded when calling routine. + ) # Reset all things to the first status, be careful to save important data def reset(self): @@ -125,18 +131,23 @@ class RollingOnlineExample: exp.delete_recorder(rid) def first_run(self): + print("========== reset ==========") + self.rolling_online_manager.reset() print("========== first_run ==========") - self.reset() self.rolling_online_manager.first_train() + print("========== dump ==========") self.rolling_online_manager.to_pickle(self._ROLLING_MANAGER_PATH) - print(self.rolling_online_manager.get_collector()()) + print("========== collect results ==========") + print(self.collector()) def routine(self): - print("========== routine ==========") + print("========== load ==========") with Path(self._ROLLING_MANAGER_PATH).open("rb") as f: self.rolling_online_manager = pickle.load(f) + print("========== routine ==========") self.rolling_online_manager.routine() - print(self.rolling_online_manager.get_collector()()) + print("========== collect results ==========") + print(self.collector()) def main(self): self.first_run() @@ -145,11 +156,11 @@ class RollingOnlineExample: if __name__ == "__main__": ####### to train the first version's models, use the command below - # python task_manager_rolling_with_updating.py first_run + # python rolling_online_management.py first_run ####### to update the models and predictions after the trading time, use the command below - # python task_manager_rolling_with_updating.py after_day + # python rolling_online_management.py after_day ####### to define your own parameters, use `--` - # python task_manager_rolling_with_updating.py first_run --exp_name='your_exp_name' --rolling_step=40 + # python rolling_online_management.py first_run --exp_name='your_exp_name' --rolling_step=40 fire.Fire(RollingOnlineExample) diff --git a/examples/online_srv/update_online_pred.py b/examples/online_srv/update_online_pred.py index a02b209bd..6e2725c7a 100644 --- a/examples/online_srv/update_online_pred.py +++ b/examples/online_srv/update_online_pred.py @@ -1,3 +1,6 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + """ This example show how OnlineTool works when we need update prediction. There are two parts including first_train and update_online_pred. diff --git a/qlib/data/dataset/__init__.py b/qlib/data/dataset/__init__.py index 5485796ef..4457dda5f 100644 --- a/qlib/data/dataset/__init__.py +++ b/qlib/data/dataset/__init__.py @@ -299,7 +299,7 @@ class TSDataSampler: self.start_idx, self.end_idx = self.data_index.slice_locs(start=pd.Timestamp(start), end=pd.Timestamp(end)) self.idx_arr = np.array(self.idx_df.values, dtype=np.float64) # for better performance - + del self.data # save memory @staticmethod @@ -507,17 +507,17 @@ class TSDatasetH(DatasetH): """ dtype = kwargs.pop("dtype") start, end = slc.start, slc.stop - flt_col = kwargs.pop('flt_col', None) + flt_col = kwargs.pop("flt_col", None) # TSDatasetH will retrieve more data for complete data = self._prepare_raw_seg(slc, **kwargs) flt_kwargs = deepcopy(kwargs) if flt_col is not None: - flt_kwargs['col_set'] = flt_col + flt_kwargs["col_set"] = flt_col flt_data = self._prepare_raw_seg(slc, **flt_kwargs) assert len(flt_data.columns) == 1 else: flt_data = None tsds = TSDataSampler(data=data, start=start, end=end, step_len=self.step_len, dtype=dtype, flt_data=flt_data) - return tsds \ No newline at end of file + return tsds diff --git a/qlib/model/ens/ensemble.py b/qlib/model/ens/ensemble.py index 63f6438c2..7ccf98ab2 100644 --- a/qlib/model/ens/ensemble.py +++ b/qlib/model/ens/ensemble.py @@ -1,36 +1,11 @@ -from abc import abstractmethod -from typing import Callable, Union +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +""" +Ensemble can merge the objects in an Ensemble. For example, if there are many submodels predictions, we may need to merge them in an ensemble predictions. +""" import pandas as pd -from qlib.workflow.task.collect import Collector -from qlib.utils.serial import Serializable - - -def ens_workflow(collector: Collector, process_list, *args, **kwargs): - """the ensemble workflow based on collector and different dict processors. - - Args: - collector (Collector): the collector to collect the result into {result_key: things} - process_list (list or Callable): the list of processors or the instance of processor to process dict. - The processor order is same as the list order. - For example: [Group1(..., Ensemble1()), Group2(..., Ensemble2())] - Returns: - dict: the ensemble dict - """ - collect_dict = collector.collect() - if not isinstance(process_list, list): - process_list = [process_list] - - ensemble = {} - for artifact in collect_dict: - value = collect_dict[artifact] - for process in process_list: - if not callable(process): - raise NotImplementedError(f"{type(process)} is not supported in `ens_workflow`.") - value = process(value, *args, **kwargs) - ensemble[artifact] = value - - return ensemble class Ensemble: @@ -53,17 +28,17 @@ class RollingEnsemble(Ensemble): """Merge the rolling objects in an Ensemble""" - def __call__(self, ensemble_dict: dict): + def __call__(self, ensemble_dict: dict) -> pd.DataFrame: """Merge a dict of rolling dataframe like `prediction` or `IC` into an ensemble. - NOTE: The values of dict must be pd.Dataframe, and have the index "datetime" + NOTE: The values of dict must be pd.DataFrame, and have the index "datetime" Args: - ensemble_dict (dict): a dict like {"A": pd.Dataframe, "B": pd.Dataframe}. + ensemble_dict (dict): a dict like {"A": pd.DataFrame, "B": pd.DataFrame}. The key of the dict will be ignored. Returns: - pd.Dataframe: the complete result of rolling. + pd.DataFrame: the complete result of rolling. """ artifact_list = list(ensemble_dict.values()) artifact_list.sort(key=lambda x: x.index.get_level_values("datetime").min()) diff --git a/qlib/model/ens/group.py b/qlib/model/ens/group.py index c80959b0d..d53a55f4c 100644 --- a/qlib/model/ens/group.py +++ b/qlib/model/ens/group.py @@ -1,3 +1,10 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +""" +Group can group a set of object based on `group_func` and change them to a dict. +""" + from qlib.model.ens.ensemble import Ensemble, RollingEnsemble from typing import Callable, Union from joblib import Parallel, delayed @@ -21,20 +28,20 @@ class Group: self._group_func = group_func self._ens_func = ens - def group(self, *args, **kwargs): + def group(self, *args, **kwargs) -> dict: # TODO: such design is weird when `_group_func` is the only configurable part in the class if isinstance(getattr(self, "_group_func", None), Callable): return self._group_func(*args, **kwargs) else: raise NotImplementedError(f"Please specify valid `group_func`.") - def reduce(self, *args, **kwargs): + def reduce(self, *args, **kwargs) -> dict: if isinstance(getattr(self, "_ens_func", None), Callable): return self._ens_func(*args, **kwargs) else: raise NotImplementedError(f"Please specify valid `_ens_func`.") - def __call__(self, ungrouped_dict: dict, n_jobs=1, verbose=0, *args, **kwargs): + def __call__(self, ungrouped_dict: dict, n_jobs=1, verbose=0, *args, **kwargs) -> dict: """Group the ungrouped_dict into different groups. Args: @@ -59,7 +66,7 @@ class Group: class RollingGroup(Group): """group the rolling dict""" - def group(self, rolling_dict: dict): + def group(self, rolling_dict: dict) -> dict: """Given an rolling dict likes {(A,B,R): things}, return the grouped dict likes {(A,B): {R:things}} NOTE: There is a assumption which is the rolling key is at the end of key tuple, because the rolling results always need to be ensemble firstly. diff --git a/qlib/model/task.py b/qlib/model/task.py deleted file mode 100644 index f29f513a4..000000000 --- a/qlib/model/task.py +++ /dev/null @@ -1,27 +0,0 @@ -import abc -import typing - - -class TaskGen(metaclass=abc.ABCMeta): - @abc.abstractmethod - def __call__(self, *args, **kwargs) -> typing.List[dict]: - """ - generate - - Parameters - ---------- - args, kwargs: - The info for generating tasks - Example 1): - input: a specific task template - output: rolling version of the tasks - Example 2): - input: a specific task template - output: a set of tasks with different losses - - Returns - ------- - typing.List[dict]: - A list of tasks - """ - pass diff --git a/qlib/model/trainer.py b/qlib/model/trainer.py index 0dcc1d67a..a0d252ab4 100644 --- a/qlib/model/trainer.py +++ b/qlib/model/trainer.py @@ -1,59 +1,72 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. -import copy +""" +The Trainer will train a list of tasks and return a list of model recorder. +There are two steps in each Trainer including `train`(make model recorder) and `end_train`(modify model recorder). + +This is concept called "DelayTrainer", which can be used in online simulating to parallel training. +In "DelayTrainer", the first step is only to save some necessary info to model recorder, and the second step which will be finished in the end can do some concurrent and time-consuming operations such as model fitting. + +`Qlib` offer two kind of Trainer, TrainerR is simplest and TrainerRM is based on TaskManager to help manager tasks lifecycle automatically. +""" + +import socket import time -from xxlimited import Str -from qlib.utils import init_instance_by_config, flatten_dict, get_cls_kwargs -from qlib.workflow import R -from qlib.workflow.recorder import Recorder -from qlib.workflow.record_temp import SignalRecord -from qlib.workflow.task.manage import TaskManager, run_task +from typing import Callable, List + from qlib.data.dataset import Dataset from qlib.model.base import Model -import socket +from qlib.utils import flatten_dict, get_cls_kwargs, init_instance_by_config +from qlib.workflow import R +from qlib.workflow.record_temp import SignalRecord +from qlib.workflow.recorder import Recorder +from qlib.workflow.task.manage import TaskManager, run_task -def begin_task_train(task_config: dict, experiment_name: str, *args, **kwargs) -> Recorder: +def begin_task_train(task_config: dict, experiment_name: str, recorder_name: str = None) -> Recorder: """ - Begin a task training with starting a recorder and saving the task config. + Begin a task training to start a recorder and save the task config. Args: - task_config (dict) - experiment_name (str) + task_config (dict): the config of a task + experiment_name (str): the name of experiment + recorder_name (str): the given name will be the recorder name. None for using rid. Returns: - Recorder + Recorder: the model recorder """ # FIXME: recorder_id - with R.start(experiment_name=experiment_name, recorder_name=str(time.time())): + if recorder_name is None: + recorder_name = str(time.time()) + with R.start(experiment_name=experiment_name, recorder_name=recorder_name): R.log_params(**flatten_dict(task_config)) R.save_objects(**{"task": task_config}) # keep the original format and datatype - R.set_tags(**{"hostname": socket.gethostname(), "train_status": "begin_task_train"}) + R.set_tags(**{"hostname": socket.gethostname()}) recorder: Recorder = R.get_recorder() return recorder -def end_task_train(rec: Recorder, experiment_name: str, *args, **kwargs): +def end_task_train(rec: Recorder, experiment_name: str) -> Recorder: """ - Finished task training with real model fitting and saving. + Finish task training with real model fitting and saving. Args: - rec (Recorder): This recorder will be resumed - experiment_name (str) + rec (Recorder): the recorder will be resumed + experiment_name (str): the name of experiment Returns: - Recorder + Recorder: the model recorder """ with R.start(experiment_name=experiment_name, recorder_name=rec.info["name"], resume=True): task_config = R.load_object("task") - # model & dataset initiaiton + # model & dataset initiation model: Model = init_instance_by_config(task_config["model"]) dataset: Dataset = init_instance_by_config(task_config["dataset"]) # model training model.fit(dataset) R.save_objects(**{"params.pkl": model}) - # This dataset is saved for online inference. So the concrete data should not be dumped + # this dataset is saved for online inference. So the concrete data should not be dumped dataset.config(dump_all=False, recursive=True) R.save_objects(**{"dataset": dataset}) # generate records: prediction, backtest, and analysis @@ -68,18 +81,18 @@ def end_task_train(rec: Recorder, experiment_name: str, *args, **kwargs): rconf = {"recorder": rec} r = cls(**kwargs, **rconf) r.generate() - R.set_tags(**{"train_status": "end_task_train"}) + return rec def task_train(task_config: dict, experiment_name: str) -> Recorder: """ - task based training + Task based training, will be divided into two steps. Parameters ---------- task_config : dict - A dict describes a task setting. + The config of a task. experiment_name: str The name of experiment @@ -97,42 +110,79 @@ class Trainer: The trainer which can train a list of model """ - def train(self, tasks: list, *args, **kwargs): - """Given a list of model definition, begin a training and return the models. + def __init__(self): + self.delay = False + + def train(self, tasks: list, *args, **kwargs) -> list: + """ + Given a list of model definition, begin a training and return the models. + + Args: + tasks: a list of tasks Returns: list: a list of models """ raise NotImplementedError(f"Please implement the `train` method.") - def end_train(self, models, *args, **kwargs): - """Given a list of models, finished something in the end of training if you need. + def end_train(self, models: list, *args, **kwargs) -> list: + """ + Given a list of models, finished something in the end of training if you need. + The models maybe Recorder, txt file, database and so on. + + Args: + models: a list of models Returns: list: a list of models """ - pass + # do nothing if you finished all work in `train` method + return models - def is_delay(self): - return False + def is_delay(self) -> bool: + """ + If Trainer will delay finishing `end_train`. + + Returns: + bool: if DelayTrainer + """ + return self.delay + + def reset(self): + """ + Reset the Trainer status. + """ + pass class TrainerR(Trainer): - """Trainer based on (R)ecorder. + """ + Trainer based on (R)ecorder. + It will train a list of tasks and return a list of model recorder in a linear way. Assumption: models were defined by `task` and the results will saved to `Recorder` """ - def __init__(self, experiment_name, train_func=task_train): + def __init__(self, experiment_name: str, train_func: Callable = task_train): + """ + Init TrainerR. + + Args: + experiment_name (str): the name of experiment. + train_func (Callable, optional): default training method. Defaults to `task_train`. + """ + super().__init__() self.experiment_name = experiment_name self.train_func = train_func - def train(self, tasks: list, train_func=None, *args, **kwargs): - """Given a list of `task`s and return a list of trained Recorder. The order can be guaranteed. + def train(self, tasks: list, train_func: Callable = None, **kwargs) -> List[Recorder]: + """ + Given a list of `task`s and return a list of trained Recorder. The order can be guaranteed. Args: tasks (list): a list of definition based on `task` dict - train_func (Callable): the train method which need at least `task` and `experiment_name`. None for default. + train_func (Callable): the train method which need at least `task`s and `experiment_name`. None for default training method. + kwargs: the params for train_func. Returns: list: a list of Recorders @@ -141,17 +191,74 @@ class TrainerR(Trainer): train_func = self.train_func recs = [] for task in tasks: - recs.append(train_func(task, self.experiment_name, *args, **kwargs)) + rec = train_func(task, self.experiment_name, **kwargs) + rec.set_tags(**{"train_status": "begin_task_train"}) + recs.append(rec) + return recs + + def end_train(self, recs: list, **kwargs) -> list: + for rec in recs: + rec.set_tags(**{"train_status": "end_task_train"}) + return recs + + +class DelayTrainerR(TrainerR): + """ + A delayed implementation based on TrainerR, which means `train` method may only do some preparation and `end_train` method can do the real model fitting. + """ + + def __init__(self, experiment_name, train_func=begin_task_train, end_train_func=end_task_train): + """ + Init TrainerRM. + + Args: + experiment_name (str): the name of experiment. + train_func (Callable, optional): default train method. Defaults to `begin_task_train`. + end_train_func (Callable, optional): default end_train method. Defaults to `end_task_train`. + """ + super().__init__(experiment_name, train_func) + self.end_train_func = end_train_func + self.delay = True + + def end_train(self, recs, end_train_func=None, **kwargs) -> List[Recorder]: + """ + Given a list of Recorder and return a list of trained Recorder. + This class will finish real data loading and model fitting. + + Args: + recs (list): a list of Recorder, the tasks have been saved to them + end_train_func (Callable, optional): the end_train method which need at least `recorder`s and `experiment_name`. Defaults to None for using self.end_train_func. + kwargs: the params for end_train_func. + + Returns: + list: a list of Recorders + """ + if end_train_func is None: + end_train_func = self.end_train_func + for rec in recs: + end_train_func(rec, **kwargs) + rec.set_tags(**{"train_status": "end_task_train"}) return recs class TrainerRM(Trainer): - """Trainer based on (R)ecorder and Task(M)anager + """ + Trainer based on (R)ecorder and Task(M)anager. + It can train a list of tasks and return a list of model recorder in a multiprocessing way. Assumption: `task` will be saved to TaskManager and `task` will be fetched and trained from TaskManager """ def __init__(self, experiment_name: str, task_pool: str, train_func=task_train): + """ + Init TrainerR. + + Args: + experiment_name (str): the name of experiment. + task_pool (str): task pool name in TaskManager. + train_func (Callable, optional): default training method. Defaults to `task_train`. + """ + super().__init__() self.experiment_name = experiment_name self.task_pool = task_pool self.train_func = train_func @@ -159,20 +266,23 @@ class TrainerRM(Trainer): def train( self, tasks: list, - train_func=None, - before_status=TaskManager.STATUS_WAITING, - after_status=TaskManager.STATUS_DONE, - *args, + train_func: Callable = None, + before_status: str = TaskManager.STATUS_WAITING, + after_status: str = TaskManager.STATUS_DONE, **kwargs, - ): - """Given a list of `task`s and return a list of trained Recorder. The order can be guaranteed. + ) -> List[Recorder]: + """ + Given a list of `task`s and return a list of trained Recorder. The order can be guaranteed. This method defaults to a single process, but TaskManager offered a great way to parallel training. Users can customize their train_func to realize multiple processes or even multiple machines. Args: tasks (list): a list of definition based on `task` dict - train_func (Callable): the train method which need at least `task` and `experiment_name`. None for default. + train_func (Callable): the train method which need at least `task`s and `experiment_name`. None for default training method. + before_status (str): the tasks in before_status will be fetched and trained. Can be STATUS_WAITING, STATUS_PART_DONE. + after_status (str): the tasks after trained will become after_status. Can be STATUS_WAITING, STATUS_PART_DONE. + kwargs: the params for train_func. Returns: list: a list of Recorders @@ -187,65 +297,27 @@ class TrainerRM(Trainer): experiment_name=self.experiment_name, before_status=before_status, after_status=after_status, - *args, **kwargs, ) recs = [] for _id in _id_list: - recs.append(tm.re_query(_id)["res"]) + rec = tm.re_query(_id)["res"] + rec.set_tags(**{"train_status": "begin_task_train"}) + recs.append(rec) return recs - -class DelayTrainerR(TrainerR): - """ - A delayed implementation based on TrainerR, which means `train` method may only do some preparation and `end_train` method can do the real model fitting. - - """ - - def __init__(self, experiment_name, train_func=begin_task_train, end_train_func=end_task_train): - super().__init__(experiment_name, train_func) - self.end_train_func = end_train_func - self.recs = [] - - def train(self, tasks: list, train_func, *args, **kwargs): - """ - Same as `train` of TrainerR, the results will be recorded in self.recs - - Args: - tasks (list): a list of definition based on `task` dict - train_func (Callable): the train method which need at least `task` and `experiment_name`. None for default. - - Returns: - list: a list of Recorders - """ - self.recs = super().train(tasks, train_func=train_func, *args, **kwargs) - return self.recs - - def end_train(self, recs=None, end_train_func=None): - """ - Given a list of Recorder and return a list of trained Recorder. - This class will finished real data loading and model fitting. - - Args: - recs (list, optional): a list of Recorder, the tasks have been saved to them. Defaults to None for using self.recs. - end_train_func (Callable, optional): the end_train method which need at least `rec` and `experiment_name`. Defaults to None for using self.end_train_func. - - Returns: - list: a list of Recorders - """ - if recs is None: - recs = copy.deepcopy(self.recs) - # the models will be only trained once - self.recs = [] - if end_train_func is None: - end_train_func = self.end_train_func + def end_train(self, recs: list, **kwargs) -> list: for rec in recs: - end_train_func(rec) + rec.set_tags(**{"train_status": "end_task_train"}) return recs - def is_delay(self): - return True + def reset(self): + """ + NOTE: this method will delete all task in this task_pool! + """ + tm = TaskManager(task_pool=self.task_pool) + tm.remove() class DelayTrainerRM(TrainerRM): @@ -257,28 +329,28 @@ class DelayTrainerRM(TrainerRM): def __init__(self, experiment_name, task_pool: str, train_func=begin_task_train, end_train_func=end_task_train): super().__init__(experiment_name, task_pool, train_func) self.end_train_func = end_train_func + self.delay = True - def train(self, tasks: list, train_func=None, *args, **kwargs): + def train(self, tasks: list, train_func=None, **kwargs): """ - Same as `train` of TrainerRM, the results will be recorded in self.recs - + Same as `train` of TrainerRM, after_status will be STATUS_PART_DONE. Args: tasks (list): a list of definition based on `task` dict - train_func (Callable): the train method which need at least `task` and `experiment_name`. None for default. - + train_func (Callable): the train method which need at least `task`s and `experiment_name`. Defaults to None for using self.train_func. Returns: list: a list of Recorders """ - return super().train(tasks, train_func=train_func, after_status=TaskManager.STATUS_PART_DONE, *args, **kwargs) + return super().train(tasks, train_func=train_func, after_status=TaskManager.STATUS_PART_DONE, **kwargs) - def end_train(self, recs, end_train_func=None): + def end_train(self, recs, end_train_func=None, **kwargs): """ Given a list of Recorder and return a list of trained Recorder. - This class will finished real data loading and model fitting. + This class will finish real data loading and model fitting. Args: - recs (list, optional): a list of Recorder, the tasks have been saved to them. Defaults to None for using self.recs.. - end_train_func (Callable, optional): the end_train method which need at least `rec` and `experiment_name`. Defaults to None for using self.end_train_func. + recs (list): a list of Recorder, the tasks have been saved to them. + end_train_func (Callable, optional): the end_train method which need at least `recorder`s and `experiment_name`. Defaults to None for using self.end_train_func. + kwargs: the params for end_train_func. Returns: list: a list of Recorders @@ -291,8 +363,8 @@ class DelayTrainerRM(TrainerRM): self.task_pool, experiment_name=self.experiment_name, before_status=TaskManager.STATUS_PART_DONE, + **kwargs, ) + for rec in recs: + rec.set_tags(**{"train_status": "end_task_train"}) return recs - - def is_delay(self): - return True diff --git a/qlib/utils/serial.py b/qlib/utils/serial.py index 1b775d99a..52d326c2a 100644 --- a/qlib/utils/serial.py +++ b/qlib/utils/serial.py @@ -3,11 +3,12 @@ from pathlib import Path import pickle +from typing import Union class Serializable: """ - Serializable will change the behaviours of pickle. + Serializable will change the behaviors of pickle. - It only saves the state whose name **does not** start with `_` It provides a syntactic sugar for distinguish the attributes which user doesn't want. - For examples, a learnable Datahandler just wants to save the parameters without data when dumping to disk @@ -70,7 +71,7 @@ class Serializable: obj.config(**params, recursive=True) del self.__dict__[self.FLAG_KEY] - def to_pickle(self, path: [Path, str], dump_all: bool = None, exclude: list = None): + def to_pickle(self, path: Union[Path, str], dump_all: bool = None, exclude: list = None): self.config(dump_all=dump_all, exclude=exclude) with Path(path).open("wb") as f: pickle.dump(self, f) diff --git a/qlib/workflow/online/manager.py b/qlib/workflow/online/manager.py index f8266577b..4e9290096 100644 --- a/qlib/workflow/online/manager.py +++ b/qlib/workflow/online/manager.py @@ -2,487 +2,40 @@ # Licensed under the MIT License. """ -This class is a component of online serving, it can manage a series of models dynamically. -With the change of time, the decisive models will be also changed. In this module, we called those contributing models as `online` models. +OnlineManager can manage a set of OnlineStrategy and run them dynamically. + +With the change of time, the decisive models will be also changed. In this module, we call those contributing models as `online` models. In every routine(such as everyday or every minutes), the `online` models maybe changed and the prediction of them need to be updated. So this module provide a series methods to control this process. """ -from copy import deepcopy -from pprint import pprint -import pandas as pd -from qlib.model.ens.ensemble import ens_workflow -from qlib.model.ens.group import RollingGroup -from qlib.utils.serial import Serializable + from typing import Dict, List, Union + +import pandas as pd from qlib import get_module_logger from qlib.data.data import D -from qlib.model.trainer import Trainer, TrainerR, task_train -from qlib.workflow import R +from qlib.utils.serial import Serializable from qlib.workflow.online.strategy import OnlineStrategy -from qlib.workflow.online.update import PredUpdater -from qlib.workflow.recorder import Recorder -from qlib.workflow.task.collect import Collector, HyperCollector, RecorderCollector -from qlib.workflow.task.gen import RollingGen, task_generator -from qlib.workflow.task.utils import TimeAdjuster, list_recorders +from qlib.workflow.task.collect import HyperCollector + class OnlineManager(Serializable): - - ONLINE_KEY = "online_status" # the online status key in recorder - ONLINE_TAG = "online" # the 'online' model - # NOTE: The meaning of this tag is that we can not assume the training models can be trained before we need its predition. Whenever finished training, it can be guaranteed that there are some online models. - NEXT_ONLINE_TAG = "next_online" # the 'next online' model, which can be 'online' model when call reset_online_model - OFFLINE_TAG = "offline" # the 'offline' model, not for online serving - - SIGNAL_EXP = "OnlineManagerSignals" # a specific experiment to save signals of different experiment. - - def __init__(self, trainer: Trainer = None, need_log=True): - """ - init OnlineManager. - - Args: - trainer (Trainer, optional): a instance of Trainer. Defaults to None. - need_log (bool, optional): print log or not. Defaults to True. - """ - self.trainer = trainer - self.logger = get_module_logger(self.__class__.__name__) - self.need_log = need_log - self.cur_time = None - - def prepare_signals(self): - """ - After perparing the data of last routine (a box in box-plot) which means the end of the routine, we can prepare trading signals for next routine. - Must use `pass` even though there is nothing to do. - """ - raise NotImplementedError(f"Please implement the `prepare_signals` method.") - - def get_signals(self): - """ - After preparing signals, here is the method to get them. - """ - raise NotImplementedError(f"Please implement the `get_signals` method.") - - def prepare_tasks(self, *args, **kwargs): - """ - After the end of a routine, check whether we need to prepare and train some new tasks. - return the new tasks waiting for training. - """ - raise NotImplementedError(f"Please implement the `prepare_tasks` method.") - - def prepare_new_models(self, tasks, tag=NEXT_ONLINE_TAG, check_func=None, *args, **kwargs): - """ - Use trainer to train a list of tasks and set the trained model to `tag`. - - Args: - tasks (list): a list of tasks. - tag (str): - `ONLINE_TAG` for first train or additional train - `NEXT_ONLINE_TAG` for reset online model when calling `reset_online_tag` - `OFFLINE_TAG` for train but offline those models - check_func: the method to judge if a model can be online. - The parameter is the model record and return True for online. - None for online every models. - *args, **kwargs: will be passed to end_train which means will be passed to customized train method. - - """ - if check_func is None: - check_func = lambda x: True - if len(tasks) > 0: - if self.trainer is not None: - new_models = self.trainer.train(tasks, *args, **kwargs) - if check_func(new_models): - self.set_online_tag(tag, new_models) - if self.need_log: - self.logger.info(f"Finished preparing {len(new_models)} new models and set them to {tag}.") - else: - self.logger.warn("No trainer to train new tasks.") - - def update_online_pred(self): - """ - After the end of a routine, update the predictions of online models to latest. - """ - raise NotImplementedError(f"Please implement the `update_online_pred` method.") - - def set_online_tag(self, tag, recorder): - """ - Set `tag` to the model to sign whether online. - - Args: - tag (str): the tags in `ONLINE_TAG`, `NEXT_ONLINE_TAG`, `OFFLINE_TAG` - """ - raise NotImplementedError(f"Please implement the `set_online_tag` method.") - - def get_online_tag(self): - """ - Given a model and return its online tag. - """ - raise NotImplementedError(f"Please implement the `get_online_tag` method.") - - def reset_online_tag(self, recorders=None): - """offline all models and set the recorders to 'online'. If no parameter and no 'next online' model, then do nothing. - - Args: - recorders (List, optional): - the recorders you want to reset to 'online'. If don't give, set 'next online' model to 'online' model. If there isn't any 'next online' model, then maintain existing 'online' model. - - Returns: - list: new online recorder. [] if there is no update. - """ - raise NotImplementedError(f"Please implement the `reset_online_tag` method.") - - def online_models(self): - """ - Return online models. - """ - raise NotImplementedError(f"Please implement the `online_models` method.") - - def first_train(self): - """ - Train a series of models firstly and set some of them into online models. - """ - raise NotImplementedError(f"Please implement the `first_train` method.") - - def get_collector(self): - """ - Return the collector. - - Returns: - Collector - """ - raise NotImplementedError(f"Please implement the `get_collector` method.") - - def delay_prepare(self, rec_dict, *args, **kwargs): - """ - Prepare all models and signals if there are something waiting for prepare. - NOTE: Assumption: the predictions of online models are between `time_segment`, or this method will work in a wrong way. - - Args: - rec_dict (str): an online models dict likes {(begin_time, end_time):[online models]}. - *args, **kwargs: will be passed to end_train which means will be passed to customized train method. - """ - for time_segment, recs_list in rec_dict.items(): - self.trainer.end_train(recs_list, *args, **kwargs) - self.reset_online_tag(recs_list) - self.prepare_signals() - signal_max = self.get_signals().index.get_level_values("datetime").max() - if time_segment[1] is not None and signal_max > time_segment[1]: - raise ValueError( - f"The max time of signals prepared by online models is {signal_max}, but those models only online in {time_segment}" - ) - - def routine(self, cur_time=None, delay_prepare=False, *args, **kwargs): - """ - The typical update process after a routine, such as day by day or month by month. - update online prediction -> prepare signals -> prepare tasks -> prepare new models -> reset online models - - NOTE: Assumption: if using simulator (delay_prepare is True), the prediction will be prepared well after every training, so there is no need to update predictions. - - Args: - cur_time ([type], optional): [description]. Defaults to None. - delay_prepare (bool, optional): [description]. Defaults to False. - *args, **kwargs: will be passed to `prepare_tasks` and `prepare_new_models`. It can be some hyper parameter or training config. - - Returns: - [type]: [description] - """ - self.cur_time = cur_time # None for latest date - if not delay_prepare: - self.update_online_pred() - self.prepare_signals() - tasks = self.prepare_tasks(*args, **kwargs) - self.prepare_new_models(tasks, *args, **kwargs) - - return self.reset_online_tag() - - -class OnlineManagerR(OnlineManager): - """ - The implementation of OnlineManager based on (R)ecorder. - - """ - - def __init__(self, experiment_name: str, trainer: Trainer = None, need_log=True): - """ - init OnlineManagerR. - - Args: - experiment_name (str): the experiment name. - trainer (Trainer, optional): a instance of Trainer. Defaults to None. - need_log (bool, optional): print log or not. Defaults to True. - """ - if trainer is None: - trainer = TrainerR(experiment_name) - super().__init__(trainer=trainer, need_log=need_log) - self.exp_name = experiment_name - self.signal_rec = None - - def set_online_tag(self, tag, recorder: Union[Recorder, List]): - """ - Set `tag` to the model to sign whether online. - - Args: - tag (str): the tags in `ONLINE_TAG`, `NEXT_ONLINE_TAG`, `OFFLINE_TAG` - recorder (Union[Recorder, List]) - """ - if isinstance(recorder, Recorder): - recorder = [recorder] - for rec in recorder: - rec.set_tags(**{self.ONLINE_KEY: tag}) - if self.need_log: - self.logger.info(f"Set {len(recorder)} models to '{tag}'.") - - def get_online_tag(self, recorder: Recorder): - """ - Given a model and return its online tag. - - Args: - recorder (Recorder): a instance of recorder - - Returns: - str: the tag - """ - tags = recorder.list_tags() - return tags.get(OnlineManager.ONLINE_KEY, OnlineManager.OFFLINE_TAG) - - def reset_online_tag(self, recorder: Union[Recorder, List] = None): - """offline all models and set the recorders to 'online'. If no parameter and no 'next online' model, then do nothing. - - Args: - recorders (Union[Recorder, List], optional): - the recorders you want to reset to 'online'. If don't give, set 'next online' model to 'online' model. If there isn't any 'next online' model, then maintain existing 'online' model. - - Returns: - list: new online recorder. [] if there is no update. - """ - if recorder is None: - recorder = list( - list_recorders( - self.exp_name, lambda rec: self.get_online_tag(rec) == OnlineManager.NEXT_ONLINE_TAG - ).values() - ) - if isinstance(recorder, Recorder): - recorder = [recorder] - if len(recorder) == 0: - if self.need_log: - self.logger.info("No 'next online' model, just use current 'online' models.") - return [] - recs = list_recorders(self.exp_name) - self.set_online_tag(OnlineManager.OFFLINE_TAG, list(recs.values())) - self.set_online_tag(OnlineManager.ONLINE_TAG, recorder) - return recorder - - def get_signals(self): - """ - get signals from the recorder(named self.exp_name) of the experiment(named self.SIGNAL_EXP) - - Returns: - signals - """ - if self.signal_rec is None: - with R.start(experiment_name=self.SIGNAL_EXP, recorder_name=self.exp_name, resume=True): - self.signal_rec = R.get_recorder() - signals = None - try: - signals = self.signal_rec.load_object("signals") - except OSError: - self.logger.warn("Can not find `signals`, have you called `prepare_signals` before?") - return signals - - def online_models(self): - """ - Return online models. - - Returns: - list: the list of online models - """ - return list( - list_recorders(self.exp_name, lambda rec: self.get_online_tag(rec) == OnlineManager.ONLINE_TAG).values() - ) - - def update_online_pred(self): - """ - Update all online model predictions to the latest day in Calendar - """ - online_models = self.online_models() - for rec in online_models: - PredUpdater(rec, to_date=self.cur_time, need_log=self.need_log).update() - - if self.need_log: - self.logger.info(f"Finished updating {len(online_models)} online model predictions of {self.exp_name}.") - - def prepare_signals(self, over_write=False): - """ - Average the predictions of online models and offer a trading signals every routine. - The signals will be saved to `signal` file of a recorder named self.exp_name of a experiment using the name of `SIGNAL_EXP` - Even if the latest signal already exists, the latest calculation result will be overwritten. - NOTE: Given a prediction of a certain time, all signals before this time will be prepared well. - Args: - over_write (bool, optional): If True, the new signals will overwrite the file. If False, the new signals will append to the end of signals. Defaults to False. - """ - if self.signal_rec is None: - with R.start(experiment_name=self.SIGNAL_EXP, recorder_name=self.exp_name, resume=True): - self.signal_rec = R.get_recorder() - - pred = [] - try: - old_signals = self.signal_rec.load_object("signals") - except OSError: - old_signals = None - - for rec in self.online_models(): - pred.append(rec.load_object("pred.pkl")) - - signals = pd.concat(pred, axis=1).mean(axis=1).to_frame("score") - signals = signals.sort_index() - if old_signals is not None and not over_write: - old_max = old_signals.index.get_level_values("datetime").max() - new_signals = signals.loc[old_max:] - signals = pd.concat([old_signals, new_signals], axis=0) - else: - new_signals = signals - if self.need_log: - self.logger.info(f"Finished preparing new {len(new_signals)} signals to {self.SIGNAL_EXP}/{self.exp_name}.") - self.signal_rec.save_objects(**{"signals": signals}) - - -class RollingOnlineManager(OnlineManagerR): - """An implementation of OnlineManager based on Rolling.""" - def __init__( self, - experiment_name: str, - rolling_gen: RollingGen, - trainer: Trainer = None, + strategy: Union[OnlineStrategy, List[OnlineStrategy]], + begin_time: Union[str, pd.Timestamp] = None, + freq="day", need_log=True, ): """ - init RollingOnlineManager. + Init OnlineManager. Args: - experiment_name (str): the experiment name. - rolling_gen (RollingGen): an instance of RollingGen - trainer (Trainer, optional): an instance of Trainer. Defaults to None. - collector (Collector, optional): an instance of Collector. Defaults to None. + strategy (Union[OnlineStrategy, List[OnlineStrategy]]): an instance of OnlineStrategy or a list of OnlineStrategy + begin_time (Union[str,pd.Timestamp], optional): the OnlineManager will begin at this time. Defaults to None. + freq (str, optional): data frequency. Defaults to "day". need_log (bool, optional): print log or not. Defaults to True. """ - if trainer is None: - trainer = TrainerR(experiment_name) - super().__init__(experiment_name=experiment_name, trainer=trainer, need_log=need_log) - self.ta = TimeAdjuster() - self.rg = rolling_gen - self.logger = get_module_logger(self.__class__.__name__) - - def get_collector(self, rec_key_func=None, rec_filter_func=None): - """ - Get the instance of collector to collect results. The returned collector must can distinguish results in different models. - Assumption: the models can be distinguished based on model name and rolling test segments. - If you do not want this assumption, please implement your own method or use another rec_key_func. - - Args: - rec_key_func (Callable): a function to get the key of a recorder. If None, use recorder id. - rec_filter_func (Callable, optional): filter the recorder by return True or False. Defaults to None. - """ - - def rec_key(recorder): - task_config = recorder.load_object("task") - model_key = task_config["model"]["class"] - rolling_key = task_config["dataset"]["kwargs"]["segments"]["test"] - return model_key, rolling_key - - if rec_key_func is None: - rec_key_func = rec_key - - return RecorderCollector(experiment=self.exp_name, rec_key_func=rec_key_func, rec_filter_func=rec_filter_func) - - def collect_artifact(self, rec_key_func=None, rec_filter_func=None): - """ - collecting artifact based on the collector and RollingGroup. - - Args: - rec_key_func (Callable): a function to get the key of a recorder. If None, use recorder id. - rec_filter_func (Callable, optional): filter the recorder by return True or False. Defaults to None. - - Returns: - dict: the artifact dict after rolling ensemble - """ - artifact = ens_workflow( - self.get_collector(rec_key_func=rec_key_func, rec_filter_func=rec_filter_func), RollingGroup() - ) - return artifact - - def first_train(self, task_configs: list): - """ - Use rolling_gen to generate different tasks based on task_configs and trained them. - - Args: - task_configs (list or dict): a list of task configs or a task config - - Returns: - Collector: a instance of a Collector. - """ - tasks = task_generator( - tasks=task_configs, - generators=self.rg, # generate different date segment - ) - self.prepare_new_models(tasks, tag=self.ONLINE_TAG) - return self.get_collector() - - def prepare_tasks(self): - """ - Prepare new tasks based on new date. - - Returns: - list: a list of new tasks. - """ - latest_records, max_test = self.list_latest_recorders( - lambda rec: self.get_online_tag(rec) == OnlineManager.ONLINE_TAG - ) - if max_test is None: - self.logger.warn(f"No latest online recorders, no new tasks.") - return [] - calendar_latest = D.calendar(end_time=self.cur_time)[-1] if self.cur_time is None else self.cur_time - if self.need_log: - self.logger.info( - f"The interval between current time {calendar_latest} and last rolling test begin time {max_test[0]} is {self.ta.cal_interval(calendar_latest, max_test[0])}, the rolling step is {self.rg.step}" - ) - if self.ta.cal_interval(calendar_latest, max_test[0]) >= self.rg.step: - old_tasks = [] - tasks_tmp = [] - for rid, rec in latest_records.items(): - task = rec.load_object("task") - old_tasks.append(deepcopy(task)) - test_begin = task["dataset"]["kwargs"]["segments"]["test"][0] - # modify the test segment to generate new tasks - task["dataset"]["kwargs"]["segments"]["test"] = (test_begin, calendar_latest) - tasks_tmp.append(task) - new_tasks_tmp = task_generator(tasks_tmp, self.rg) - new_tasks = [task for task in new_tasks_tmp if task not in old_tasks] - return new_tasks - return [] - - def list_latest_recorders(self, rec_filter_func=None): - """find latest recorders based on test segments. - - Args: - rec_filter_func (Callable, optional): recorder filter. Defaults to None. - - Returns: - dict, tuple: the latest recorders and the latest date of them - """ - recs_flt = list_recorders(self.exp_name, rec_filter_func) - if len(recs_flt) == 0: - return recs_flt, None - max_test = max(rec.load_object("task")["dataset"]["kwargs"]["segments"]["test"] for rec in recs_flt.values()) - latest_rec = {} - for rid, rec in recs_flt.items(): - if rec.load_object("task")["dataset"]["kwargs"]["segments"]["test"] == max_test: - latest_rec[rid] = rec - return latest_rec, max_test - - -class OnlineM(Serializable): - def __init__( - self, strategy: Union[OnlineStrategy, List[OnlineStrategy]], begin_time=None, freq="day", need_log=True - ): self.logger = get_module_logger(self.__class__.__name__) self.need_log = need_log if not isinstance(strategy, list): @@ -491,38 +44,37 @@ class OnlineM(Serializable): self.freq = freq if begin_time is None: begin_time = D.calendar(freq=self.freq).max() - self.cur_time = pd.Timestamp(begin_time) + self.begin_time = pd.Timestamp(begin_time) + self.cur_time = self.begin_time self.history = {} def first_train(self): """ - Train a series of models firstly and set some of them into online models. + Run every strategy first_train method and record the online history """ for strategy in self.strategy: self.logger.info(f"Strategy `{strategy.name_id}` begins first training...") online_models = strategy.first_train() self.history.setdefault(strategy.name_id, {})[self.cur_time] = online_models - def routine(self, cur_time=None, task_kwargs={}, model_kwargs={}): + def routine(self, cur_time: Union[str, pd.Timestamp] = None, task_kwargs: dict = {}, model_kwargs: dict = {}): """ + Run typical update process for every strategy and record the online history. + The typical update process after a routine, such as day by day or month by month. update online prediction -> prepare signals -> prepare tasks -> prepare new models -> reset online models - NOTE: Assumption: if using simulator (delay_prepare is True), the prediction will be prepared well after every training, so there is no need to update predictions. - Args: - cur_time ([type], optional): [description]. Defaults to None. - delay_prepare (bool, optional): [description]. Defaults to False. - *args, **kwargs: will be passed to `prepare_tasks` and `prepare_new_models`. It can be some hyper parameter or training config. - - Returns: - [type]: [description] + cur_time (Union[str,pd.Timestamp], optional): run routine method in this time. Defaults to None. + task_kwargs (dict): the params for `prepare_tasks` + model_kwargs (dict): the params for `prepare_online_models` """ if cur_time is None: cur_time = D.calendar(freq=self.freq).max() self.cur_time = pd.Timestamp(cur_time) # None for latest date for strategy in self.strategy: - self.logger.info(f"Strategy `{strategy.name_id}` begins routine...") + if self.need_log: + self.logger.info(f"Strategy `{strategy.name_id}` begins routine...") if not strategy.trainer.is_delay(): strategy.prepare_signals() tasks = strategy.prepare_tasks(self.cur_time, **task_kwargs) @@ -530,13 +82,28 @@ class OnlineM(Serializable): if len(online_models) > 0: self.history.setdefault(strategy.name_id, {})[self.cur_time] = online_models - def get_collector(self): + def get_collector(self) -> HyperCollector: + """ + Get the instance of HyperCollector to collect results from every strategy. + + Returns: + HyperCollector: the collector can collect other collectors. + """ collector_dict = {} for strategy in self.strategy: collector_dict[strategy.name_id] = strategy.get_collector() return HyperCollector(collector_dict) - def get_online_history(self, strategy_name_id): + def get_online_history(self, strategy_name_id: str) -> list: + """ + Get the online history based on strategy_name_id. + + Args: + strategy_name_id (str): the name_id of strategy + + Returns: + dict: a list like [(time, [online_models])] + """ history_dict = self.history[strategy_name_id] history = [] for time in sorted(history_dict): @@ -547,22 +114,20 @@ class OnlineM(Serializable): def delay_prepare(self, delay_kwargs={}): """ Prepare all models and signals if there are something waiting for prepare. - NOTE: Assumption: the predictions of online models are between `time_segment`, or this method will work in a wrong way. Args: - rec_dict (str): an online models dict likes {(begin_time, end_time):[online models]}. - *args, **kwargs: will be passed to end_train which means will be passed to customized train method. + delay_kwargs: the params for `delay_prepare` """ for strategy in self.strategy: strategy.delay_prepare(self.get_online_history(strategy.name_id), **delay_kwargs) - def simulate(self, end_time, frequency="day", task_kwargs={}, model_kwargs={}, delay_kwargs={}): + def simulate(self, end_time, frequency="day", task_kwargs={}, model_kwargs={}, delay_kwargs={}) -> HyperCollector: """ - Starting from start time, this method will simulate every routine in OnlineManager. + Starting from cur time, this method will simulate every routine in OnlineManager. NOTE: Considering the parallel training, the models and signals can be perpared after all routine simulating. Returns: - Collector: the OnlineManager's collector + HyperCollector: the OnlineManager's collector """ cal = D.calendar(start_time=self.cur_time, end_time=end_time, freq=frequency) self.first_train() @@ -572,3 +137,12 @@ class OnlineM(Serializable): self.delay_prepare(delay_kwargs=delay_kwargs) self.logger.info(f"Finished preparing signals") return self.get_collector() + + def reset(self): + """ + NOTE: This method will reset all strategy! Be careful to use it. + """ + self.cur_time = self.begin_time + self.history = {} + for strategy in self.strategy: + strategy.reset() diff --git a/qlib/workflow/online/simulator.py b/qlib/workflow/online/simulator.py deleted file mode 100644 index ddaf2471c..000000000 --- a/qlib/workflow/online/simulator.py +++ /dev/null @@ -1,77 +0,0 @@ -from qlib.data import D -from qlib import get_module_logger -from qlib.workflow.online.manager import OnlineM - - -class OnlineSimulator: - """ - To simulate online serving in the past, like a "online serving backtest". - """ - - def __init__( - self, - start_time, - end_time, - online_manager: OnlineManager, - frequency="day", - ): - """ - init OnlineSimulator. - - Args: - start_time (str or pd.Timestamp): the start time of simulating. - end_time (str or pd.Timestamp): the end time of simulating. If None, then end_time is latest. - onlinemanager (OnlineManager): the instance of OnlineManager - frequency (str, optional): the data frequency. Defaults to "day". - """ - self.logger = get_module_logger(self.__class__.__name__) - self.cal = D.calendar(start_time=start_time, end_time=end_time, freq=frequency) - self.start_time = self.cal[0] - self.end_time = self.cal[-1] - self.olm = online_manager - if len(self.cal) == 0: - self.logger.warn(f"There is no need to simulate bacause start_time is larger than end_time.") - - # def simulate(self, *args, **kwargs): - # """ - # Starting from start time, this method will simulate every routine in OnlineManager. - # NOTE: Considering the parallel training, the models and signals can be perpared after all routine simulating. - - # Returns: - # Collector: the OnlineManager's collector - # """ - # self.rec_dict = {} - # tmp_begin = self.start_time - # tmp_end = None - # self.olm.first_train() - # prev_recorders = self.olm.online_models() - # for cur_time in self.cal: - # self.logger.info(f"Simulating at {str(cur_time)}......") - # recorders = self.olm.routine(cur_time, True, *args, **kwargs) - # if len(recorders) == 0: - # tmp_end = cur_time - # else: - # self.rec_dict[(tmp_begin, tmp_end)] = prev_recorders - # tmp_begin = cur_time - # prev_recorders = recorders - # self.rec_dict[(tmp_begin, self.end_time)] = prev_recorders - # # finished perparing models (and pred) and signals - # self.olm.delay_prepare(self.rec_dict) - # self.logger.info(f"Finished preparing signals") - # return self.olm.get_collector() - - def simulate(self, task_kwargs={}, model_kwargs={}): - """ - Starting from start time, this method will simulate every routine in OnlineManager. - NOTE: Considering the parallel training, the models and signals can be perpared after all routine simulating. - - Returns: - Collector: the OnlineManager's collector - """ - self.olm.first_train() - for cur_time in self.cal: - self.logger.info(f"Simulating at {str(cur_time)}......") - self.olm.routine(cur_time, task_kwargs={}, model_kwargs={}) - self.olm.delay_prepare() - self.logger.info(f"Finished preparing signals") - return self.olm.get_collector() diff --git a/qlib/workflow/online/strategy.py b/qlib/workflow/online/strategy.py index 5e4dcc024..3782ee652 100644 --- a/qlib/workflow/online/strategy.py +++ b/qlib/workflow/online/strategy.py @@ -1,11 +1,14 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. + """ -This module is working with OnlineManager, responsing for a set of strategy about how the models are updated and signals are perpared. +OnlineStrategy is a set of strategy of online serving. +It is working with OnlineManager, responsing how the tasks are generated, the models are updated and signals are perpared. """ from copy import deepcopy -from typing import List, Union +from typing import List, Tuple, Union + import pandas as pd from qlib.data.data import D from qlib.log import get_module_logger @@ -13,7 +16,8 @@ from qlib.model.ens.group import RollingGroup from qlib.model.trainer import Trainer, TrainerR from qlib.workflow import R from qlib.workflow.online.utils import OnlineTool, OnlineToolR -from qlib.workflow.task.collect import HyperCollector, RecorderCollector +from qlib.workflow.recorder import Recorder +from qlib.workflow.task.collect import Collector, HyperCollector, RecorderCollector from qlib.workflow.task.gen import RollingGen, task_generator from qlib.workflow.task.utils import TimeAdjuster, list_recorders @@ -21,7 +25,7 @@ from qlib.workflow.task.utils import TimeAdjuster, list_recorders class OnlineStrategy: def __init__(self, name_id: str, trainer: Trainer = None, need_log=True): """ - init OnlineManager. + Init OnlineStrategy. Args: name_id (str): a unique name or id @@ -33,12 +37,15 @@ class OnlineStrategy: self.logger = get_module_logger(self.__class__.__name__) self.need_log = need_log self.tool = OnlineTool() - self.history = {} - def prepare_signals(self, delay=False): + def prepare_signals(self, delay: bool = False): """ After perparing the data of last routine (a box in box-plot) which means the end of the routine, we can prepare trading signals for next routine. - Must use `pass` even though there is nothing to do. + + NOTE: Given a set prediction, all signals before these prediction end time will be prepared well. + Args: + delay: bool + If this method was called by `delay_prepare` """ raise NotImplementedError(f"Please implement the `prepare_signals` method.") @@ -46,6 +53,8 @@ class OnlineStrategy: """ After the end of a routine, check whether we need to prepare and train some new tasks. return the new tasks waiting for training. + + You can find last online models by OnlineTool.online_models. """ raise NotImplementedError(f"Please implement the `prepare_tasks` method.") @@ -53,6 +62,8 @@ class OnlineStrategy: """ Use trainer to train a list of tasks and set the trained model to `online`. + NOTE: This method will first offline all models and online the online models prepared by this method. So you can find last online models by OnlineTool.online_models if you still need them. + Args: tasks (list): a list of tasks. tag (str): @@ -78,33 +89,43 @@ class OnlineStrategy: def first_train(self): """ - Train a series of models firstly and set some of them into online models. + Train a series of models firstly and set some of them as online models. """ raise NotImplementedError(f"Please implement the `first_train` method.") - def get_collector(self): + def get_collector(self) -> Collector: """ - Return the collector. + Get the instance of collector to collect results of online serving. + + For example: + 1) collect predictions in Recorder + 2) collect signals in .txt file Returns: Collector """ raise NotImplementedError(f"Please implement the `get_collector` method.") - def delay_prepare(self, history, **kwargs): + def delay_prepare(self, history: list, **kwargs): """ Prepare all models and signals if there are something waiting for prepare. - NOTE: Assumption: the predictions of online models are between `time_segment`, or this method will work in a wrong way. + NOTE: Assumption: the predictions of online models need less than next begin_time, or this method will work in a wrong way. Args: - rec_dict (str): an online models dict likes {(begin_time, end_time):[online models]}. - *args, **kwargs: will be passed to end_train which means will be passed to customized train method. + history (list): an online models list likes [begin_time:[online models]]. + **kwargs: will be passed to end_train which means will be passed to customized train method. """ - for time_begin, recs_list in history: + for begin_time, recs_list in history: self.trainer.end_train(recs_list, **kwargs) self.tool.reset_online_tag(recs_list) self.prepare_signals(delay=True) + def reset(self): + """ + Delete all things and set them to default status. This method is convenient to explore the strategy for online simulation. + """ + pass + class RollingAverageStrategy(OnlineStrategy): @@ -122,7 +143,7 @@ class RollingAverageStrategy(OnlineStrategy): signal_exp_name="OnlineManagerSignals", ): """ - init OnlineManagerR. + Init RollingAverageStrategy. Assumption: the str of name_id, the experiment name and the trainer's experiment name are same one. @@ -139,11 +160,11 @@ class RollingAverageStrategy(OnlineStrategy): if not isinstance(task_template, list): task_template = [task_template] self.task_template = task_template - self.signal_rec = None self.signal_exp_name = signal_exp_name - self.ta = TimeAdjuster() self.rg = rolling_gen self.tool = OnlineToolR(self.exp_name) + self.ta = TimeAdjuster() + self.signal_rec = None # the recorder to record signals def get_collector(self, rec_key_func=None, rec_filter_func=None): """ @@ -180,12 +201,12 @@ class RollingAverageStrategy(OnlineStrategy): ) return HyperCollector({"artifacts": artifacts_collector, "signals": signals_collector}) - def first_train(self): + def first_train(self) -> List[Recorder]: """ Use rolling_gen to generate different tasks based on task_template and trained them. Returns: - Collector: a instance of a Collector. + List[Recorder]: a list of Recorder. """ tasks = task_generator( tasks=self.task_template, @@ -193,12 +214,14 @@ class RollingAverageStrategy(OnlineStrategy): ) return self.prepare_online_models(tasks) - def prepare_tasks(self, cur_time): + def prepare_tasks(self, cur_time) -> List[dict]: """ Prepare new tasks based on cur_time (None for latest). + You can find last online models by OnlineToolR.online_models. + Returns: - list: a list of new tasks. + List[dict]: a list of new tasks. """ latest_records, max_test = self._list_latest(self.tool.online_models()) if max_test is None: @@ -224,7 +247,7 @@ class RollingAverageStrategy(OnlineStrategy): return new_tasks return [] - def prepare_signals(self, delay=False, over_write=False): + def prepare_signals(self, delay=False, over_write=False) -> pd.DataFrame: """ Average the predictions of online models and offer a trading signals every routine. The signals will be saved to `signal` file of a recorder named self.exp_name of a experiment using the name of `SIGNAL_EXP` @@ -233,7 +256,7 @@ class RollingAverageStrategy(OnlineStrategy): Args: over_write (bool, optional): If True, the new signals will overwrite the file. If False, the new signals will append to the end of signals. Defaults to False. Returns: - object: the signals. + pd.DataFrame: the signals. """ if not delay: self.tool.update_online_pred() @@ -250,7 +273,7 @@ class RollingAverageStrategy(OnlineStrategy): for rec in self.tool.online_models(): pred.append(rec.load_object("pred.pkl")) - signals = pd.concat(pred, axis=1).mean(axis=1).to_frame("score") + signals: pd.DataFrame = pd.concat(pred, axis=1).mean(axis=1).to_frame("score") signals = signals.sort_index() if old_signals is not None and not over_write: old_max = old_signals.index.get_level_values("datetime").max() @@ -275,14 +298,19 @@ class RollingAverageStrategy(OnlineStrategy): # if self.signal_rec is None: # with R.start(experiment_name=self.signal_exp_name, recorder_name=self.exp_name, resume=True): # self.signal_rec = R.get_recorder() - # signals = None - # try: - # signals = self.signal_rec.load_object("signals") - # except OSError: - # self.logger.warn("Can not find `signals`, have you called `prepare_signals` before?") + # signals = self.signal_rec.load_object("signals") # return signals - def _list_latest(self, rec_list): + def _list_latest(self, rec_list: List[Recorder]): + """ + List latest recorder form rec_list + + Args: + rec_list (List[Recorder]): a list of Recorder + + Returns: + List[Recorder], pd.Timestamp: the latest recorders and its test end time + """ if len(rec_list) == 0: return rec_list, None max_test = max(rec.load_object("task")["dataset"]["kwargs"]["segments"]["test"] for rec in rec_list) @@ -291,3 +319,16 @@ class RollingAverageStrategy(OnlineStrategy): if rec.load_object("task")["dataset"]["kwargs"]["segments"]["test"] == max_test: latest_rec.append(rec) return latest_rec, max_test + + def reset(self): + """ + NOTE: This method will delete all recorder in Experiment and reset the Trainer! + """ + self.trainer.reset() + # delete models + exp = R.get_exp(experiment_name=self.exp_name) + for rid in exp.list_recorders(): + exp.delete_recorder(rid) + # delete signals + for rid in list_recorders(self.signal_exp_name, lambda x: True if x.info["name"] == self.exp_name else False): + exp.delete_recorder(rid) diff --git a/qlib/workflow/online/update.py b/qlib/workflow/online/update.py index 5b58360d8..69ad55324 100644 --- a/qlib/workflow/online/update.py +++ b/qlib/workflow/online/update.py @@ -1,18 +1,20 @@ -from typing import Union, List -from qlib.data.dataset import DatasetH -from qlib.workflow import R -from qlib.data import D +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +""" +Update is a module to update artifacts such as predictions, when the stock data updating. +""" + +from abc import ABCMeta, abstractmethod + import pandas as pd from qlib import get_module_logger -from qlib.workflow import R -from qlib.model import Model -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 +from qlib.data import D from qlib.data.dataset import DatasetH -from abc import ABCMeta, abstractmethod +from qlib.data.dataset.handler import DataHandlerLP +from qlib.model import Model from qlib.utils import get_date_by_shift +from qlib.workflow.recorder import Recorder class RMDLoader: @@ -25,19 +27,22 @@ class RMDLoader: def get_dataset(self, start_time, end_time, segments=None) -> DatasetH: """ - load, config and setup dataset. + Load, config and setup dataset. - This dataset is for inference + This dataset is for inference. + + Args: + start_time : + the start_time of underlying data + end_time : + the end_time of underlying data + segments : dict + the segments config for dataset + Due to the time series dataset (TSDatasetH), the test segments maybe different from start_time and end_time + + Returns: + DatasetH: the instance of DatasetH - Parameters - ---------- - start_time : - the start_time of underlying data - end_time : - the end_time of underlying data - segments : dict - the segments config for dataset - Due to the time series dataset (TSDatasetH), the test segments maybe different from start_time and end_time """ if segments is None: segments = {"test": (start_time, end_time)} @@ -52,7 +57,7 @@ class RMDLoader: class RecordUpdater(metaclass=ABCMeta): """ - Updata a specific recorders + Update a specific recorders """ def __init__(self, record: Recorder, need_log=True, *args, **kwargs): @@ -75,16 +80,17 @@ class PredUpdater(RecordUpdater): def __init__(self, record: Recorder, to_date=None, hist_ref: int = 0, freq="day", need_log=True): """ - Parameters - ---------- - record : Recorder - to_date : - update to prediction to the `to_date` - hist_ref : int - Sometimes, the dataset will have historical depends. - Leave the problem to user to set the length of historical dependancy - NOTE: the start_time is not included in the hist_ref - # TODO: automate this step in the future. + Init PredUpdater. + + Args: + record : Recorder + to_date : + update to prediction to the `to_date` + hist_ref : int + Sometimes, the dataset will have historical depends. + Leave the problem to user to set the length of historical dependency + NOTE: the start_time is not included in the hist_ref + # TODO: automate this step in the future. """ super().__init__(record=record, need_log=need_log) @@ -101,9 +107,12 @@ class PredUpdater(RecordUpdater): def prepare_data(self) -> DatasetH: """ - # Load dataset + Load dataset Seperating this function will make it easier to reuse the dataset + + Returns: + DatasetH: the instance of DatasetH """ start_time_buffer = get_date_by_shift(self.last_end, -self.hist_ref + 1, clip_shift=False, freq=self.freq) start_time = get_date_by_shift(self.last_end, 1, freq=self.freq) @@ -113,9 +122,12 @@ class PredUpdater(RecordUpdater): def update(self, dataset: DatasetH = None): """ - update the precition in a recorder + Update the precition in a recorder + + Args: + DatasetH: the instance of DatasetH. None for reprepare. """ - # FIXME: the problme below is not solved + # FIXME: the problem below is not solved # The model dumped on GPU instances can not be loaded on CPU instance. Follow exception will raised # RuntimeError: Attempting to deserialize object on a CUDA device but torch.cuda.is_available() is False. If you are running on a CPU-only machine, please use torch.load with map_location=torch.device('cpu') to map your storages to the CPU. # https://github.com/pytorch/pytorch/issues/16797 diff --git a/qlib/workflow/online/utils.py b/qlib/workflow/online/utils.py index 1cd89d668..4d630a665 100644 --- a/qlib/workflow/online/utils.py +++ b/qlib/workflow/online/utils.py @@ -1,7 +1,14 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + """ -This module is like a online backend, deciding which models are `online` models and how can change them +OnlineTool is a module to set and unset a series of `online` models. +The `online` models are some decisive models in some time point, which can be changed with the change of time. +This allows us to use efficient submodels as the market style changing. """ + from typing import List, Union + from qlib.log import get_module_logger from qlib.workflow.online.update import PredUpdater from qlib.workflow.recorder import Recorder @@ -12,60 +19,66 @@ class OnlineTool: ONLINE_KEY = "online_status" # the online status key in recorder ONLINE_TAG = "online" # the 'online' model - # NOTE: The meaning of this tag is that we can not assume the training models can be trained before we need its predition. Whenever finished training, it can be guaranteed that there are some online models. - NEXT_ONLINE_TAG = "next_online" # the 'next online' model, which can be 'online' model when call reset_online_model OFFLINE_TAG = "offline" # the 'offline' model, not for online serving def __init__(self, need_log=True): """ - init OnlineTool. + Init OnlineTool. Args: need_log (bool, optional): print log or not. Defaults to True. """ self.logger = get_module_logger(self.__class__.__name__) self.need_log = need_log - self.cur_time = None - def set_online_tag(self, tag, recorder): + def set_online_tag(self, tag, recorder: Union[list, object]): """ Set `tag` to the model to sign whether online. Args: - tag (str): the tags in `ONLINE_TAG`, `NEXT_ONLINE_TAG`, `OFFLINE_TAG` + tag (str): the tags in `ONLINE_TAG`, `OFFLINE_TAG` + recorder (Union[list,object]): the model's recorder """ raise NotImplementedError(f"Please implement the `set_online_tag` method.") - def get_online_tag(self): + def get_online_tag(self, recorder: object) -> str: """ - Given a model and return its online tag. + Given a model recorder and return its online tag. + + Args: + recorder (Object): the model's recorder + + Returns: + str: the online tag """ raise NotImplementedError(f"Please implement the `get_online_tag` method.") - def reset_online_tag(self, recorders=None): - """offline all models and set the recorders to 'online'. If no parameter and no 'next online' model, then do nothing. + def reset_online_tag(self, recorder: Union[list, object]): + """ + Offline all models and set the recorders to 'online'. Args: - recorders (List, optional): - the recorders you want to reset to 'online'. If don't give, set 'next online' model to 'online' model. If there isn't any 'next online' model, then maintain existing 'online' model. + recorder (Union[list,object]): + the recorder you want to reset to 'online'. - Returns: - list: new online recorder. [] if there is no update. """ raise NotImplementedError(f"Please implement the `reset_online_tag` method.") - def online_models(self): + def online_models(self) -> list: """ - Return `online` models. + Get current `online` models + + Returns: + list: a list of `online` models. """ raise NotImplementedError(f"Please implement the `online_models` method.") def update_online_pred(self, to_date=None): """ - Update the predictions of online models to a date. + Update the predictions of `online` models to a date. Args: - to_date (pd.Timestamp): the pred before this date will be updated. None for latest. + to_date (pd.Timestamp): the pred before this date will be updated. None for update to latest. """ raise NotImplementedError(f"Please implement the `update_online_pred` method.") @@ -74,12 +87,11 @@ class OnlineTool: class OnlineToolR(OnlineTool): """ The implementation of OnlineTool based on (R)ecorder. - """ def __init__(self, experiment_name: str, need_log=True): """ - init OnlineToolR. + Init OnlineToolR. Args: experiment_name (str): the experiment name. @@ -90,11 +102,11 @@ class OnlineToolR(OnlineTool): def set_online_tag(self, tag, recorder: Union[Recorder, List]): """ - Set `tag` to the model to sign whether online. + Set `tag` to the model's recorder to sign whether online. Args: tag (str): the tags in `ONLINE_TAG`, `NEXT_ONLINE_TAG`, `OFFLINE_TAG` - recorder (Union[Recorder, List]) + recorder (Union[Recorder, List]): a list of Recorder or an instance of Recorder """ if isinstance(recorder, Recorder): recorder = [recorder] @@ -103,50 +115,40 @@ class OnlineToolR(OnlineTool): if self.need_log: self.logger.info(f"Set {len(recorder)} models to '{tag}'.") - def get_online_tag(self, recorder: Recorder): + def get_online_tag(self, recorder: Recorder) -> str: """ - Given a model and return its online tag. + Given a model recorder and return its online tag. Args: - recorder (Recorder): a instance of recorder + recorder (Recorder): an instance of recorder Returns: - str: the tag + str: the online tag """ tags = recorder.list_tags() return tags.get(self.ONLINE_KEY, self.OFFLINE_TAG) - def reset_online_tag(self, recorder: Union[Recorder, List] = None): - """offline all models and set the recorders to 'online'. If no parameter and no 'next online' model, then do nothing. + def reset_online_tag(self, recorder: Union[Recorder, List]): + """ + Offline all models and set the recorders to 'online'. Args: - recorders (Union[Recorder, List], optional): - the recorders you want to reset to 'online'. If don't give, set 'next online' model to 'online' model. If there isn't any 'next online' model, then maintain existing 'online' model. + recorder (Union[Recorder, List]): + the recorder you want to reset to 'online'. - Returns: - list: new online recorder. [] if there is no update. """ - if recorder is None: - recorder = list( - list_recorders(self.exp_name, lambda rec: self.get_online_tag(rec) == self.NEXT_ONLINE_TAG).values() - ) if isinstance(recorder, Recorder): recorder = [recorder] - if len(recorder) == 0: - if self.need_log: - self.logger.info("No 'next online' model, just use current 'online' models.") - return [] recs = list_recorders(self.exp_name) self.set_online_tag(self.OFFLINE_TAG, list(recs.values())) self.set_online_tag(self.ONLINE_TAG, recorder) - return recorder - def online_models(self): + def online_models(self) -> list: """ - Return online models. + Get current `online` models Returns: - list: the list of online models + list: a list of `online` models. """ return list(list_recorders(self.exp_name, lambda rec: self.get_online_tag(rec) == self.ONLINE_TAG).values()) @@ -155,7 +157,7 @@ class OnlineToolR(OnlineTool): Update the predictions of online models to a date. Args: - to_date (pd.Timestamp): the pred before this date will be updated. None for latest in Calendar. + to_date (pd.Timestamp): the pred before this date will be updated. None for update to latest time in Calendar. """ online_models = self.online_models() for rec in online_models: diff --git a/qlib/workflow/task/collect.py b/qlib/workflow/task/collect.py index eb0a20029..d74d08184 100644 --- a/qlib/workflow/task/collect.py +++ b/qlib/workflow/task/collect.py @@ -1,9 +1,11 @@ -from abc import abstractmethod -from typing import Callable, Union -from qlib import init +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +""" +Collector can collect object from everywhere and process them such as merging, grouping, averaging and so on. +""" + from qlib.workflow import R -from qlib.workflow.task.utils import list_recorders -from qlib.utils.serial import Serializable import dill as pickle @@ -19,7 +21,7 @@ class Collector: process_list = [process_list] self.process_list = process_list - def collect(self): + def collect(self) -> dict: """Collect the results and return a dict like {key: things} Returns: @@ -36,7 +38,7 @@ class Collector: raise NotImplementedError(f"Please implement the `collect` method.") @staticmethod - def process_collect(collected_dict, process_list=[], *args, **kwargs): + def process_collect(collected_dict, process_list=[], *args, **kwargs) -> dict: """do a series of processing to the dict returned by collect and return a dict like {key: things} For example: you can group and ensemble. @@ -61,7 +63,7 @@ class Collector: result[artifact] = value return result - def __call__(self, *args, **kwargs): + def __call__(self, *args, **kwargs) -> dict: """ do the workflow including collect and process_collect @@ -124,7 +126,7 @@ class HyperCollector(Collector): super().__init__(process_list=process_list) self.collector_dict = collector_dict - def collect(self): + def collect(self) -> dict: collect_dict = {} for key, collector in self.collector_dict.items(): collect_dict[key] = collector() @@ -153,10 +155,10 @@ class RecorderCollector(Collector): artifacts_path (dict, optional): The artifacts name and its path in Recorder. Defaults to {"pred": "pred.pkl", "IC": "sig_analysis/ic.pkl"}. artifacts_key (str or List, optional): the artifacts key you want to get. If None, get all artifacts. """ + super().__init__(process_list=process_list) if isinstance(experiment, str): experiment = R.get_exp(experiment_name=experiment) self.experiment = experiment - self.process_list = process_list self.artifacts_path = artifacts_path if rec_key_func is None: rec_key_func = lambda rec: rec.info["id"] @@ -166,7 +168,7 @@ class RecorderCollector(Collector): self.artifacts_key = artifacts_key self._rec_filter_func = rec_filter_func - def collect(self, artifacts_key=None, rec_filter_func=None): + def collect(self, artifacts_key=None, rec_filter_func=None) -> dict: """Collect different artifacts based on recorder after filtering. Args: @@ -203,5 +205,11 @@ class RecorderCollector(Collector): return collect_dict - def get_exp_name(self): + def get_exp_name(self) -> str: + """ + Get experiment name + + Returns: + str: experiment name + """ return self.experiment.name diff --git a/qlib/workflow/task/gen.py b/qlib/workflow/task/gen.py index 158bc9916..c4c6bab7f 100644 --- a/qlib/workflow/task/gen.py +++ b/qlib/workflow/task/gen.py @@ -1,7 +1,7 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. """ -this is a task generator +Task generator can generate many tasks based on TaskGen and some task templates. """ import abc import copy @@ -113,7 +113,7 @@ class RollingGen(TaskGen): self.test_key = "test" self.train_key = "train" - def generate(self, task: dict): + def generate(self, task: dict) -> typing.List[dict]: """ Converting the task into a rolling task. @@ -158,6 +158,10 @@ class RollingGen(TaskGen): }, ] } + + Returns + ---------- + typing.List[dict]: a list of tasks """ res = [] @@ -196,16 +200,18 @@ class RollingGen(TaskGen): # update segments of this task t["dataset"]["kwargs"]["segments"] = copy.deepcopy(segments) - # if end_time < the end of test_segments, then change end_time to allow load more data - if ( - self.modify_end_time - and self.ta.cal_interval( + + try: + interval = self.ta.cal_interval( t["dataset"]["kwargs"]["handler"]["kwargs"]["end_time"], t["dataset"]["kwargs"]["segments"][self.test_key][1], ) - < 0 - ): - t["dataset"]["kwargs"]["handler"]["kwargs"]["end_time"] = copy.deepcopy(segments[self.test_key][1]) + # if end_time < the end of test_segments, then change end_time to allow load more data + if self.modify_end_time and interval < 0: + t["dataset"]["kwargs"]["handler"]["kwargs"]["end_time"] = copy.deepcopy(segments[self.test_key][1]) + except KeyError: + # Maybe the user dataset has no handler or end_time + pass prev_seg = segments res.append(t) return res diff --git a/qlib/workflow/task/manage.py b/qlib/workflow/task/manage.py index 9d50d8563..3c3144fe8 100644 --- a/qlib/workflow/task/manage.py +++ b/qlib/workflow/task/manage.py @@ -1,31 +1,39 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. + """ -A task consists of 3 parts +TaskManager can fetch unused tasks automatically and manager the lifecycle of a set of tasks with error handling. +These features can run tasks concurrently and ensure every task will be used only once. +Task Manager will store all tasks in `MongoDB `_. +Users **MUST** finished the configuration of `MongoDB `_ when using this module. + +A task in TaskManager consists of 3 parts - tasks description: the desc will define the task - tasks status: the status of the task - tasks result information : A user can get the task with the task description and task result. - """ -from bson.binary import Binary -import pickle -from pymongo.errors import InvalidDocument -from bson.objectid import ObjectId -from contextlib import contextmanager -import qlib -from tqdm.cli import tqdm -import time import concurrent -import pymongo -from qlib.config import C -from .utils import get_mongodb -from qlib import get_module_logger, auto_init +import pickle +import time +from contextlib import contextmanager +from typing import Callable, List + import fire +import pymongo +from bson.binary import Binary +from bson.objectid import ObjectId +from pymongo.errors import InvalidDocument +from qlib import auto_init, get_module_logger +from tqdm.cli import tqdm + +from .utils import get_mongodb class TaskManager: - """TaskManager - here is what will a task looks like when it created by TaskManager + """ + TaskManager + + Here is what will a task looks like when it created by TaskManager .. code-block:: python @@ -42,6 +50,12 @@ class TaskManager: .. note:: Assumption: the data in MongoDB was encoded and the data out of MongoDB was decoded + + Here are four status which are: + STATUS_WAITING: waiting for train + STATUS_RUNNING: training + STATUS_PART_DONE: finished some step and waiting for next step. + STATUS_DONE: all work done """ STATUS_WAITING = "waiting" @@ -53,7 +67,7 @@ class TaskManager: def __init__(self, task_pool: str = None): """ - init Task Manager, remember to make the statement of MongoDB url and database name firstly. + Init Task Manager, remember to make the statement of MongoDB url and database name firstly. Parameters ---------- @@ -65,7 +79,7 @@ class TaskManager: self.task_pool = getattr(self.mdb, task_pool) self.logger = get_module_logger(self.__class__.__name__) - def list(self): + def list(self) -> list: """ list the all collection(task_pool) of the db @@ -92,7 +106,9 @@ class TaskManager: return {k: str(v) for k, v in flt.items()} def replace_task(self, task, new_task): - # assume that the data out of interface was decoded and the data in interface was encoded + """ + Use a new task to replace a old one + """ new_task = self._encode_task(new_task) query = {"_id": ObjectId(task["_id"])} try: @@ -121,7 +137,7 @@ class TaskManager: Returns ------- - + pymongo.results.InsertOneResult """ task = self._encode_task( { @@ -133,9 +149,9 @@ class TaskManager: insert_result = self.insert_task(task) return insert_result - def create_task(self, task_def_l, dry_run=False, print_nt=False): + def create_task(self, task_def_l, dry_run=False, print_nt=False) -> List[str]: """ - if the tasks in task_def_l is new, then insert new tasks into the task_pool + If the tasks in task_def_l is new, then insert new tasks into the task_pool Parameters ---------- @@ -145,6 +161,7 @@ class TaskManager: if insert those new tasks to task pool print_nt: bool if print new task + Returns ------- list @@ -165,7 +182,7 @@ class TaskManager: print(t) if dry_run: - return + return [] _id_list = [] for t in new_tasks: @@ -174,7 +191,17 @@ class TaskManager: return _id_list - def fetch_task(self, query={}, status=STATUS_WAITING): + def fetch_task(self, query={}, status=STATUS_WAITING) -> dict: + """ + Use query to fetch tasks + + Args: + query (dict, optional): query dict. Defaults to {}. + status (str, optional): [description]. Defaults to STATUS_WAITING. + + Returns: + dict: a task(document in collection) after decoding + """ query = query.copy() if "_id" in query: query["_id"] = ObjectId(query["_id"]) @@ -191,7 +218,7 @@ class TaskManager: @contextmanager def safe_fetch_task(self, query={}, status=STATUS_WAITING): """ - fetch task from task_pool using query with contextmanager + Fetch task from task_pool using query with contextmanager Parameters ---------- @@ -200,7 +227,7 @@ class TaskManager: Returns ------- - + dict: a task(document in collection) after decoding """ task = self.fetch_task(query=query, status=status) try: @@ -231,7 +258,7 @@ class TaskManager: Returns ------- - + dict: a task(document in collection) after decoding """ query = query.copy() if "_id" in query: @@ -240,16 +267,40 @@ class TaskManager: yield self._decode_task(t) def re_query(self, _id): + """ + Use _id to query task. + + Args: + _id (str): _id of a document + + Returns: + dict: a task(document in collection) after decoding + """ t = self.task_pool.find_one({"_id": ObjectId(_id)}) return self._decode_task(t) - def commit_task_res(self, task, res, status=None): + def commit_task_res(self, task, res, status=STATUS_DONE): + """ + Commit the result to task['res']. + + Args: + task ([type]): [description] + res (object): the result you want to save + status (str, optional): STATUS_WAITING, STATUS_RUNNING, STATUS_DONE, STATUS_PART_DONE. Defaults to STATUS_DONE. + """ # A workaround to use the class attribute. if status is None: status = TaskManager.STATUS_DONE self.task_pool.update_one({"_id": task["_id"]}, {"$set": {"status": status, "res": Binary(pickle.dumps(res))}}) - def return_task(self, task, status=None): + def return_task(self, task, status=STATUS_WAITING): + """ + Return a task to status. Alway using in error handling. + + Args: + task ([type]): [description] + status (str, optional): STATUS_WAITING, STATUS_RUNNING, STATUS_DONE, STATUS_PART_DONE. Defaults to STATUS_WAITING. + """ if status is None: status = TaskManager.STATUS_WAITING update_dict = {"$set": {"status": status}} @@ -257,7 +308,7 @@ class TaskManager: def remove(self, query={}): """ - remove the task using query + Remove the task using query Parameters ---------- @@ -295,7 +346,7 @@ class TaskManager: def prioritize(self, task, priority: int): """ - set priority for task + Set priority for task Parameters ---------- @@ -331,29 +382,37 @@ class TaskManager: def run_task( - task_func, - task_pool, - force_release=False, - before_status=TaskManager.STATUS_WAITING, - after_status=TaskManager.STATUS_DONE, - *args, + task_func: Callable, + task_pool: str, + force_release: bool = False, + before_status: str = TaskManager.STATUS_WAITING, + after_status: str = TaskManager.STATUS_DONE, **kwargs, ): """ While task pool is not empty (has WAITING tasks), use task_func to fetch and run tasks in task_pool + After running this method, here are 4 situations (before_status -> after_status): + STATUS_WAITING -> STATUS_DONE: use task["def"] as `task_func` param + STATUS_WAITING -> STATUS_PART_DONE: use task["def"] as `task_func` param + STATUS_PART_DONE -> STATUS_PART_DONE: use task["res"] as `task_func` param + STATUS_PART_DONE -> STATUS_DONE: use task["res"] as `task_func` param + Parameters ---------- - task_func : def (task_def, *args, **kwargs) -> - the function to run the task + task_func : Callable + def (task_def, **kwargs) -> + the function to run the task task_pool : str the name of the task pool (Collection in MongoDB) - force_release : + force_release : bool will the program force to release the resource - args : - args - kwargs : - kwargs + before_status : str: + the tasks in before_status will be fetched and trained. Can be STATUS_WAITING, STATUS_PART_DONE. + after_status : str: + the tasks after trained will become after_status. Can be STATUS_WAITING, STATUS_PART_DONE. + kwargs + the params for `task_func` """ tm = TaskManager(task_pool) @@ -364,19 +423,19 @@ def run_task( if task is None: break get_module_logger("run_task").info(task["def"]) - # when fetching `WAITING` task, use task_def to train + # when fetching `WAITING` task, use task["def"] to train if before_status == TaskManager.STATUS_WAITING: param = task["def"] - # when fetching `PART_DONE` task, use task_res to train for the result has been saved + # when fetching `PART_DONE` task, use task["res"] to train because the middle result has been saved to task["res"] elif before_status == TaskManager.STATUS_PART_DONE: param = task["res"] else: raise ValueError("The fetched task must be `STATUS_WAITING` or `STATUS_PART_DONE`!") if force_release: with concurrent.futures.ProcessPoolExecutor(max_workers=1) as executor: - res = executor.submit(task_func, param, *args, **kwargs).result() + res = executor.submit(task_func, param, **kwargs).result() else: - res = task_func(param, *args, **kwargs) + res = task_func(param, **kwargs) tm.commit_task_res(task, res, status=after_status) ever_run = True diff --git a/qlib/workflow/task/utils.py b/qlib/workflow/task/utils.py index ce8e0dfa3..ed5e1a235 100644 --- a/qlib/workflow/task/utils.py +++ b/qlib/workflow/task/utils.py @@ -1,5 +1,10 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT License. + +""" +Some tools for task management. +""" + import bisect import pandas as pd from qlib.data import D @@ -7,13 +12,14 @@ from qlib.workflow import R from qlib.config import C from qlib.log import get_module_logger from pymongo import MongoClient +from pymongo.database import Database from typing import Union -def get_mongodb(): - """ +def get_mongodb() -> Database: - get database in MongoDB, which means you need to declare the address and the name of database. + """ + Get database in MongoDB, which means you need to declare the address and the name of database. for example: Using qlib.init(): @@ -31,6 +37,8 @@ def get_mongodb(): "task_db_name" : "rolling_db" } + Returns: + Database: the Database instance """ try: cfg = C["mongo"] @@ -43,7 +51,8 @@ def get_mongodb(): def list_recorders(experiment, rec_filter_func=None): - """list all recorders which can pass the filter in a experiment. + """ + List all recorders which can pass the filter in a experiment. Args: experiment (str or Experiment): the name of a Experiment or a instance @@ -65,7 +74,7 @@ def list_recorders(experiment, rec_filter_func=None): class TimeAdjuster: """ - find appropriate date and adjust date. + Find appropriate date and adjust date. """ def __init__(self, future=True, end_time=None): @@ -88,15 +97,15 @@ class TimeAdjuster: return None return self.cals[idx] - def max(self): + def max(self) -> pd.Timestamp: """ Return the max calendar datetime """ return max(self.cals) - def align_idx(self, time_point, tp_type="start"): + def align_idx(self, time_point, tp_type="start") -> int: """ - align the index of time_point in the calendar + Align the index of time_point in the calendar Parameters ---------- @@ -116,9 +125,9 @@ class TimeAdjuster: raise NotImplementedError(f"This type of input is not supported") return idx - def cal_interval(self, time_point_A, time_point_B): + def cal_interval(self, time_point_A, time_point_B) -> int: """ - calculate the trading day interval + Calculate the trading day interval (time_point_A - time_point_B) Args: time_point_A : time_point_A @@ -129,20 +138,22 @@ class TimeAdjuster: """ return self.align_idx(time_point_A) - self.align_idx(time_point_B) - def align_time(self, time_point, tp_type="start"): + def align_time(self, time_point, tp_type="start") -> pd.Timestamp: """ Align time_point to trade date of calendar - Parameters - ---------- - time_point - Time point - tp_type : str - time point type (`"start"`, `"end"`) + Args: + time_point + Time point + tp_type : str + time point type (`"start"`, `"end"`) + + Returns: + pd.Timestamp """ return self.cals[self.align_idx(time_point, tp_type=tp_type)] - def align_seg(self, segment: Union[dict, tuple]): + def align_seg(self, segment: Union[dict, tuple]) -> Union[dict, tuple]: """ align the given date to trade date @@ -162,7 +173,7 @@ class TimeAdjuster: Returns ------- - the start and end trade date (pd.Timestamp) between the given start and end date. + Union[dict, tuple]: the start and end trade date (pd.Timestamp) between the given start and end date. """ if isinstance(segment, dict): return {k: self.align_seg(seg) for k, seg in segment.items()} @@ -171,7 +182,7 @@ class TimeAdjuster: else: raise NotImplementedError(f"This type of input is not supported") - def truncate(self, segment: tuple, test_start, days: int): + def truncate(self, segment: tuple, test_start, days: int) -> tuple: """ truncate the segment based on the test_start date @@ -183,6 +194,10 @@ class TimeAdjuster: days : int The trading days to be truncated the data in this segment may need 'days' data + + Returns + --------- + tuple: new segment """ test_idx = self.align_idx(test_start) if isinstance(segment, tuple): @@ -198,7 +213,7 @@ class TimeAdjuster: SHIFT_SD = "sliding" SHIFT_EX = "expanding" - def shift(self, seg: tuple, step: int, rtype=SHIFT_SD): + def shift(self, seg: tuple, step: int, rtype=SHIFT_SD) -> tuple: """ shift the datatime of segment @@ -211,6 +226,10 @@ class TimeAdjuster: rtype : str rolling type ("sliding" or "expanding") + Returns + -------- + tuple: new segment + Raises ------ KeyError: