simulator & examples

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

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:
+        for rec in self.rolling_online_manager.online_models():
-                print(rid)
+            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()

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):
+    def online_models(self):
-        """The typical update process in a routine such as day by day or month by month"""
+        """return online models"""
-        self.prepare_signals(*args, **kwargs)
+        raise NotImplementedError(f"Please implement the `online_models` method.")
-        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)
 
-    # 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)
+        super().__init__(trainer, need_log)
-        self.logger = get_module_logger(self.__class__.__name__)
         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.")
+            if self.need_log:
-            return
+                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)
+        online_models = self.online_models()
-        cnt = mu.update_all_pred(lambda rec: self.get_online_tag(rec) == OnlineManager.ONLINE_TAG)
+        for rec in online_models:
-        self.logger.info(f"Finish updating {cnt} online model predictions of {self.exp_name}.")
+            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__(
+    def __init__(self, experiment_name: str, rolling_gen: RollingGen, trainer: Trainer = None, need_log=True):
-        self,
-        experiment_name: str,
-        rolling_gen: RollingGen,
-        trainer: Trainer = None,
-    ):
         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)
+                tasks_tmp.append(task)
-            new_tasks_tmp = task_generator(old_tasks, self.rg)
+            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 []

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 {}

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(
+        if self.need_log:
-            f"Finish updating new {new_pred.shape[0]} predictions in {self.record.info['id']}."
+            self.logger.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)

qlib/workflow/task/utils.py

@@ -57,8 +57,12 @@ class TimeAdjuster:
     find appropriate date and adjust date.
     """
 
-    def __init__(self, future=False):
+    def __init__(self, future=True, end_time=None):
-        self.cals = D.calendar(future=future)
+        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):
         """