diff --git a/examples/online_srv/online_simulate.py b/examples/online_srv/online_simulate.py new file mode 100644 index 000000000..007085c73 --- /dev/null +++ b/examples/online_srv/online_simulate.py @@ -0,0 +1,163 @@ +from abc import abstractmethod +import copy +from pprint import pprint + +import fire +import qlib +from qlib.config import REG_CN +from qlib.model.trainer import task_train +from qlib.workflow import R +from qlib.workflow.task.gen import TaskGen +from qlib.workflow.online.simulator import OnlineSimulator +from qlib.workflow.task.collect import RecorderCollector +from qlib.model.ens.ensemble import RollingEnsemble, ens_workflow +from qlib.workflow.task.gen import RollingGen, task_generator +from qlib.workflow.task.manage import TaskManager, run_task +from qlib.workflow.online.manager import RollingOnlineManager +from qlib.workflow.task.utils import TimeAdjuster, list_recorders +from qlib.model.trainer import TrainerRM +from qlib.model.ens.group import RollingGroup + +data_handler_config = { + "start_time": "2018-01-01", + "end_time": "2018-10-31", + "fit_start_time": "2018-01-01", + "fit_end_time": "2018-03-31", + "instruments": "csi100", +} + +dataset_config = { + "class": "DatasetH", + "module_path": "qlib.data.dataset", + "kwargs": { + "handler": { + "class": "Alpha158", + "module_path": "qlib.contrib.data.handler", + "kwargs": data_handler_config, + }, + "segments": { + "train": ("2018-01-01", "2018-03-31"), + "valid": ("2018-04-01", "2018-05-31"), + "test": ("2018-06-01", "2018-09-10"), + }, + }, +} + +record_config = [ + { + "class": "SignalRecord", + "module_path": "qlib.workflow.record_temp", + }, + { + "class": "SigAnaRecord", + "module_path": "qlib.workflow.record_temp", + }, +] + +# use lgb model +task_lgb_config = { + "model": { + "class": "LGBModel", + "module_path": "qlib.contrib.model.gbdt", + }, + "dataset": dataset_config, + "record": record_config, +} + +# use xgboost model +task_xgboost_config = { + "model": { + "class": "XGBModel", + "module_path": "qlib.contrib.model.xgboost", + }, + "dataset": dataset_config, + "record": record_config, +} + + +class OnlineSimulatorExample: + def __init__( + self, + exp_name="rolling_exp", + task_pool="rolling_task", + provider_uri="~/.qlib/qlib_data/cn_data", + region="cn", + task_url="mongodb://10.0.0.4:27017/", + task_db_name="rolling_db", + rolling_step=80, + ): + self.exp_name = exp_name + self.task_pool = task_pool + mongo_conf = { + "task_url": task_url, # your MongoDB url + "task_db_name": task_db_name, # database name + } + qlib.init(provider_uri=provider_uri, region=region, mongo=mongo_conf) + + self.rolling_gen = RollingGen(step=rolling_step, rtype=RollingGen.ROLL_SD) + self.trainer = TrainerRM(self.exp_name, self.task_pool) + self.task_manager = TaskManager(self.task_pool) + self.rolling_online_manager = RollingOnlineManager( + experiment_name=exp_name, rolling_gen=self.rolling_gen, trainer=self.trainer, need_log=False + ) + + # 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) + + @staticmethod + 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 + + # Run this firstly to see the workflow in Task Management + def first_run(self): + print("========== first_run ==========") + self.reset() + + tasks = task_generator( + tasks=task_xgboost_config, + generators=[self.rolling_gen], # generate different date segment + ) + + pprint(tasks) + + self.trainer.train(tasks) + + print("========== task collecting ==========") + + artifact = ens_workflow(RecorderCollector(exp_name=self.exp_name, rec_key_func=self.rec_key), RollingGroup()) + print(artifact) + + latest_rec, _ = self.rolling_online_manager.list_latest_recorders() + self.rolling_online_manager.set_online_tag(RollingOnlineManager.ONLINE_TAG, list(latest_rec.values())) + + def simulate(self): + + print("========== simulate ==========") + onlinesimulator = OnlineSimulator( + start_time="2018-09-10", + end_time="2018-10-31", + onlinemanager=self.rolling_online_manager, + collector=RecorderCollector(exp_name=self.exp_name, rec_key_func=self.rec_key), + process_list=[RollingGroup()], + ) + results = onlinesimulator.simulate() + print(results) + recs_dict = onlinesimulator.online_models() + for time, recs in recs_dict.items(): + print(f"{str(time[0])} to {str(time[1])}:") + for rec in recs: + print(rec.info["id"]) + + +if __name__ == "__main__": + ose = OnlineSimulatorExample() + ose.first_run() + ose.simulate() diff --git a/examples/online_srv/task_manager_rolling_with_updating.py b/examples/online_srv/task_manager_rolling_with_updating.py index bfdc5f3c0..9195a1de6 100644 --- a/examples/online_srv/task_manager_rolling_with_updating.py +++ b/examples/online_srv/task_manager_rolling_with_updating.py @@ -100,9 +100,9 @@ class RollingOnlineExample: def print_online_model(self): print("========== print_online_model ==========") print("Current 'online' model:") - for rid, rec in list_recorders(self.exp_name).items(): - if self.rolling_online_manager.get_online_tag(rec) == self.rolling_online_manager.ONLINE_TAG: - print(rid) + + for rec in self.rolling_online_manager.online_models(): + print(rec.info["id"]) print("Current 'next online' model:") for rid, rec in list_recorders(self.exp_name).items(): if self.rolling_online_manager.get_online_tag(rec) == self.rolling_online_manager.NEXT_ONLINE_TAG: @@ -161,12 +161,15 @@ class RollingOnlineExample: self.reset() tasks = self.task_generating() + pprint(tasks) self.task_training(tasks) self.task_collecting() latest_rec, _ = self.rolling_online_manager.list_latest_recorders() self.rolling_online_manager.reset_online_tag(list(latest_rec.values())) + self.routine() + def routine(self): print("========== routine ==========") self.print_online_model() diff --git a/qlib/workflow/online/manager.py b/qlib/workflow/online/manager.py index 66df160cd..a40512cf3 100644 --- a/qlib/workflow/online/manager.py +++ b/qlib/workflow/online/manager.py @@ -3,7 +3,7 @@ from qlib import get_module_logger from qlib.workflow import R from qlib.model.trainer import task_train from qlib.workflow.recorder import MLflowRecorder, Recorder -from qlib.workflow.online.update import ModelUpdater +from qlib.workflow.online.update import PredUpdater, RecordUpdater from qlib.workflow.task.utils import TimeAdjuster from qlib.workflow.task.gen import RollingGen, task_generator from qlib.workflow.task.manage import TaskManager @@ -11,6 +11,7 @@ from qlib.workflow.task.manage import run_task from qlib.workflow.task.utils import list_recorders from qlib.utils.serial import Serializable from qlib.model.trainer import Trainer, TrainerR +from copy import deepcopy class OnlineManager(Serializable): @@ -20,9 +21,11 @@ class OnlineManager(Serializable): 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, trainer: Trainer = None): + def __init__(self, trainer: Trainer = None, need_log=True): self._trainer = trainer self.logger = get_module_logger(self.__class__.__name__) + self.need_log = need_log + self.delay_signals = {} def prepare_signals(self, *args, **kwargs): raise NotImplementedError(f"Please implement the `prepare_signals` method.") @@ -31,7 +34,7 @@ class OnlineManager(Serializable): """return the new tasks waiting for training.""" raise NotImplementedError(f"Please implement the `prepare_tasks` method.") - def prepare_new_models(self, tasks, *args, **kwargs): + def prepare_new_models(self, tasks): """Use trainer to train a list of tasks and set the trained model to next_online. Args: @@ -39,7 +42,7 @@ class OnlineManager(Serializable): """ if not (tasks is None or len(tasks) == 0): if self._trainer is not None: - new_models = self._trainer.train(tasks, *args, **kwargs) + new_models = self._trainer.train(tasks) self.set_online_tag(self.NEXT_ONLINE_TAG, new_models) self.logger.info( f"Finished prepare {len(new_models)} new models and set them to `{self.NEXT_ONLINE_TAG}`." @@ -66,15 +69,27 @@ class OnlineManager(Serializable): """offline all models and set the recorders to 'online'. If no parameter and no 'next online' model, then do nothing.""" raise NotImplementedError(f"Please implement the `reset_online_tag` method.") - def routine(self, *args, **kwargs): - """The typical update process in a routine such as day by day or month by month""" - self.prepare_signals(*args, **kwargs) - tasks = self.prepare_tasks(*args, **kwargs) - self.prepare_new_models(tasks, *args, **kwargs) - self.update_online_pred(*args, **kwargs) - self.reset_online_tag(*args, **kwargs) + def online_models(self): + """return online models""" + raise NotImplementedError(f"Please implement the `online_models` method.") - # TODO: first_train? + def run_delay_signals(self): + for cur_time, params in self.delay_signals.items(): + self.cur_time = cur_time + self.prepare_signals(*params[0], **params[1]) + self.delay_signals = {} + + def routine(self, cur_time=None, delay_prepare=False, *args, **kwargs): + """The typical update process in a routine such as day by day or month by month""" + self.cur_time = cur_time # None for latest date + if not delay_prepare: + self.prepare_signals(*args, **kwargs) + else: + self.delay_signals[cur_time] = (args, kwargs) + tasks = self.prepare_tasks(*args, **kwargs) + self.prepare_new_models(tasks) + self.update_online_pred() + return self.reset_online_tag() class OnlineManagerR(OnlineManager): @@ -83,10 +98,9 @@ class OnlineManagerR(OnlineManager): """ - def __init__(self, experiment_name: str, trainer: Trainer = None): + def __init__(self, experiment_name: str, trainer: Trainer = None, need_log=True): trainer = TrainerR(experiment_name) - super().__init__(trainer) - self.logger = get_module_logger(self.__class__.__name__) + super().__init__(trainer, need_log) self.exp_name = experiment_name def set_online_tag(self, tag, recorder: Union[Recorder, List]): @@ -94,7 +108,8 @@ class OnlineManagerR(OnlineManager): recorder = [recorder] for rec in recorder: rec.set_tags(**{self.ONLINE_KEY: tag}) - self.logger.info(f"Set {len(recorder)} models to '{tag}'.") + if self.need_log: + self.logger.info(f"Set {len(recorder)} models to '{tag}'.") def get_online_tag(self, recorder: Recorder): tags = recorder.list_tags() @@ -106,6 +121,9 @@ class OnlineManagerR(OnlineManager): Args: recorders (Union[List, Dict], 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( @@ -116,31 +134,35 @@ class OnlineManagerR(OnlineManager): if isinstance(recorder, Recorder): recorder = [recorder] if len(recorder) == 0: - self.logger.info("No 'next online' model, just use current 'online' models.") - return + 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) - self.logger.info(f"Reset {len(recorder)} models to 'online'.") + return recorder - def update_online_pred(self, *args, **kwargs): + def online_models(self): + 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""" - mu = ModelUpdater(self.exp_name) - cnt = mu.update_all_pred(lambda rec: self.get_online_tag(rec) == OnlineManager.ONLINE_TAG) - self.logger.info(f"Finish updating {cnt} online model predictions of {self.exp_name}.") + 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"Finish updating {len(online_models)} online model predictions of {self.exp_name}.") class RollingOnlineManager(OnlineManagerR): """An implementation of OnlineManager based on Rolling.""" - def __init__( - self, - experiment_name: str, - rolling_gen: RollingGen, - trainer: Trainer = None, - ): + def __init__(self, experiment_name: str, rolling_gen: RollingGen, trainer: Trainer = None, need_log=True): trainer = TrainerR(experiment_name) - super().__init__(experiment_name, trainer) + super().__init__(experiment_name, trainer, need_log=need_log) self.ta = TimeAdjuster() self.rg = rolling_gen self.logger = get_module_logger(self.__class__.__name__) @@ -154,22 +176,25 @@ class RollingOnlineManager(OnlineManagerR): Returns: list: a list of new tasks. """ + self.ta.set_end_time(self.cur_time) 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 = self.ta.last_date() + calendar_latest = self.ta.last_date() if self.cur_time is None else self.cur_time 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) - old_tasks.append(task) - new_tasks_tmp = task_generator(old_tasks, self.rg) + 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 [] diff --git a/qlib/workflow/online/simulator.py b/qlib/workflow/online/simulator.py new file mode 100644 index 000000000..7f08549f0 --- /dev/null +++ b/qlib/workflow/online/simulator.py @@ -0,0 +1,80 @@ +from typing import Callable +import pandas as pd +from qlib.config import C +from qlib.data import D +from qlib import get_module_logger +from qlib.log import set_log_with_config +from qlib.model.ens.ensemble import ens_workflow +from qlib.workflow.online.manager import OnlineManager +from qlib.workflow.task.collect import Collector + + +class OnlineSimulator: + """ + To simulate online serving in the past, like a "online serving backtest". + """ + + def __init__( + self, + start_time, + end_time, + onlinemanager: OnlineManager, + frequency="day", + time_delta="20 hours", + collector: Collector = None, + process_list: list = None, + ): + 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 = onlinemanager + self.time_delta = time_delta + + if len(self.cal) == 0: + self.logger.warn(f"There is no need to simulate bacause start_time is larger than end_time.") + self.collector = collector + self.process_list = process_list + + def simulate(self, *args, **kwargs): + """ + Starting from start time, this method will simulate every routine in OnlineManager. + NOTE: Considering the parallel training, the signals will be perpared after all routine simulating. + + Returns: + dict: the simulated results collected by collector + """ + self.rec_dict = {} + tmp_begin = self.start_time + tmp_end = None + prev_recorders = self.olm.online_models() + for cur_time in self.cal: + cur_time = cur_time + pd.Timedelta(self.time_delta) + 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 + # prepare signals again incase there is no trained model when call it + self.olm.run_delay_signals() + self.logger.info(f"Finished preparing signals") + + if self.collector is not None: + return ens_workflow(self.collector, self.process_list) + + def online_models(self): + """ + Return a online models dict likes {(begin_time, end_time):[online models]}. + + Returns: + dict + """ + if hasattr(self, "rec_dict"): + return self.rec_dict + self.logger.warn(f"Please call `simulate` firstly when calling `online_models`") + return {} diff --git a/qlib/workflow/online/update.py b/qlib/workflow/online/update.py index 8835fdae2..8aa32ff29 100644 --- a/qlib/workflow/online/update.py +++ b/qlib/workflow/online/update.py @@ -27,7 +27,7 @@ class RMDLoader: """ load, config and setup dataset. - This dataset is for inferene + This dataset is for inference Parameters ---------- @@ -55,8 +55,10 @@ class RecordUpdater(metaclass=ABCMeta): Updata a specific recorders """ - def __init__(self, record: Recorder, *args, **kwargs): + def __init__(self, record: Recorder, need_log=True, *args, **kwargs): self.record = record + self.logger = get_module_logger(self.__class__.__name__) + self.need_log = need_log @abstractmethod def update(self, *args, **kwargs): @@ -73,7 +75,7 @@ class PredUpdater(RecordUpdater): LATEST = "__latest" - def __init__(self, record: Recorder, to_date=LATEST, hist_ref: int = 0, freq="day"): + def __init__(self, record: Recorder, to_date=LATEST, hist_ref: int = 0, freq="day", need_log=True): """ Parameters ---------- @@ -86,14 +88,15 @@ class PredUpdater(RecordUpdater): NOTE: the start_time is not included in the hist_ref # TODO: automate this step in the future. """ - super().__init__(record=record) + super().__init__(record=record, need_log=need_log) self.to_date = to_date self.hist_ref = hist_ref self.freq = freq self.rmdl = RMDLoader(rec=record) - if to_date == self.LATEST: + # FIXME: why we need LATEST? can we use to_date=None instead? + if to_date == self.LATEST or to_date == None: to_date = D.calendar(freq=freq)[-1] self.to_date = pd.Timestamp(to_date) self.old_pred = record.load_object("pred.pkl") @@ -119,6 +122,12 @@ class PredUpdater(RecordUpdater): # 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. + start_time = get_date_by_shift(self.last_end, 1, freq=self.freq) + if start_time >= self.to_date: + if self.need_log: + self.logger.info(f"The prediction in {self.record.info['id']} are latest. No need to update.") + return + # load dataset if dataset is None: # For reusing the dataset @@ -134,114 +143,5 @@ class PredUpdater(RecordUpdater): self.record.save_objects(**{"pred.pkl": cb_pred}) - get_module_logger(self.__class__.__name__).info( - f"Finish updating new {new_pred.shape[0]} predictions in {self.record.info['id']}." - ) - - -class ModelUpdater: - """ - The model updater to update model results in new data. - """ - - def __init__(self, experiment_name: str) -> None: - """ModelUpdater needs experiment name to find the records - - Parameters - ---------- - experiment_name : str - experiment name string - """ - self.exp_name = experiment_name - self.logger = get_module_logger(self.__class__.__name__) - - def _reload_dataset(self, recorder, start_time, end_time): - """reload dataset from pickle file - - Parameters - ---------- - recorder : Recorder - the instance of the Recorder - start_time : Timestamp - the start time you want to load - end_time : Timestamp - the end time you want to load - - Returns - ------- - Dataset - the instance of Dataset - """ - segments = {"test": (start_time, end_time)} - dataset = recorder.load_object("dataset") - dataset.config(handler_kwargs={"start_time": start_time, "end_time": end_time}, segments=segments) - dataset.setup_data(handler_kwargs={"init_type": DataHandlerLP.IT_LS}) - return dataset - - def update_pred(self, recorder: Recorder, frequency="day"): - """update predictions to the latest day in Calendar based on rid - - Parameters - ---------- - recorder: Union[str,Recorder] - the id of a Recorder or the Recorder instance - """ - old_pred = recorder.load_object("pred.pkl") - last_end = old_pred.index.get_level_values("datetime").max() - - # updated to the latest trading day - if frequency == "day": - cal = D.calendar(start_time=last_end + pd.Timedelta(days=1), end_time=None) - else: - raise NotImplementedError("Now `ModelUpdater` only support update daily frequency prediction") - - if len(cal) == 0: - self.logger.info( - f"The prediction in {recorder.info['id']} of {self.exp_name} are latest. No need to update." - ) - return - - start_time, end_time = cal[0], cal[-1] - - dataset = self._reload_dataset(recorder, start_time, end_time) - - model = recorder.load_object("params.pkl") - new_pred = model.predict(dataset) - - cb_pred = pd.concat([old_pred, new_pred.to_frame("score")], axis=0) - cb_pred = cb_pred.sort_index() - - recorder.save_objects(**{"pred.pkl": cb_pred}) - - self.logger.info( - f"Finish updating new {new_pred.shape[0]} predictions in {recorder.info['id']} of {self.exp_name}." - ) - - def update_all_pred(self, rec_filter_func=None): - """update all predictions in this experiment after filter. - - An example of filter function: - - .. code-block:: python - - def record_filter(record): - task_config = record.load_object("task") - if task_config["model"]["class"]=="LGBModel": - return True - return False - - Parameters - ---------- - rec_filter_func : Callable[[Recorder], bool], optional - the filter function to decide whether this record will be updated, by default None - - Returns - ---------- - cnt: int - the count of updated record - - """ - recs = list_recorders(self.exp_name, rec_filter_func=rec_filter_func) - for rid, rec in recs.items(): - self.update_pred(rec) - return len(recs) + if self.need_log: + self.logger.info(f"Finish updating new {new_pred.shape[0]} predictions in {self.record.info['id']}.") diff --git a/qlib/workflow/task/utils.py b/qlib/workflow/task/utils.py index b6287abc2..87a3a41f3 100644 --- a/qlib/workflow/task/utils.py +++ b/qlib/workflow/task/utils.py @@ -57,8 +57,12 @@ class TimeAdjuster: find appropriate date and adjust date. """ - def __init__(self, future=False): - self.cals = D.calendar(future=future) + def __init__(self, future=True, end_time=None): + self._future = future + self.cals = D.calendar(future=future, end_time=end_time) + + def set_end_time(self, end_time=None): + self.cals = D.calendar(future=self._future, end_time=end_time) def get(self, idx: int): """