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

OnlineServing V9

Browse Source
This commit is contained in:
lzh222333
2021-04-29 04:30:09 +00:00
parent 6f669348a8
commit 67c5740c83
19 changed files with 677 additions and 1010 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
examples/model_rolling/task_manager_rolling.py
View File

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

54
examples/online_srv/online_management_simulate.py
View File

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

43
examples/online_srv/rolling_online_management.py
View File

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

3
examples/online_srv/update_online_pred.py
View File

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

8
qlib/data/dataset/__init__.py
View File

@@ -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
return tsds

45
qlib/model/ens/ensemble.py
View File

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

15
qlib/model/ens/group.py
View File

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

27
qlib/model/task.py
View File

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

288
qlib/model/trainer.py
View File

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

5
qlib/utils/serial.py
View File

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

540
qlib/workflow/online/manager.py
View File

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

77
qlib/workflow/online/simulator.py
View File

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

103
qlib/workflow/online/strategy.py
View File

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

84
qlib/workflow/online/update.py
View File

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

94
qlib/workflow/online/utils.py
View File

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

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

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

24
qlib/workflow/task/gen.py
View File

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

155
qlib/workflow/task/manage.py
View File

@@ -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 <https://www.mongodb.com/>`_.
Users **MUST** finished the configuration of `MongoDB <https://www.mongodb.com/>`_ 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) -> <res which will be committed>
the function to run the task
task_func : Callable
def (task_def, **kwargs) -> <res which will be committed>
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

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

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