init version of online serving and rolling

examples/model_rolling/task_manager_rolling.py

@@ -1,24 +1,23 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""
+This example shows how a TrainerRM work based on TaskManager with rolling tasks.
+After training, how to collect the rolling results will be showed in task_collecting.
+"""
+
 from pprint import pprint
-import time
 
 import fire
 import qlib
 from qlib.config import REG_CN
-from qlib.model.trainer import TrainerR, task_train
 from qlib.workflow import R
 from qlib.workflow.task.gen import RollingGen, task_generator
-from qlib.workflow.task.manage import TaskManager, run_task
+from qlib.workflow.task.manage import TaskManager
 from qlib.workflow.task.collect import RecorderCollector
-from qlib.model.ens.ensemble import RollingEnsemble, ens_workflow
-import pandas as pd
-from qlib.workflow.task.utils import list_recorders
 from qlib.model.ens.group import RollingGroup
 from qlib.model.trainer import TrainerRM
 
-"""
-This example shows how a Trainer work based on TaskManager with rolling tasks.
-After training, how to collect the rolling results will be showed in task_collecting.
-"""
 
 data_handler_config = {
     "start_time": "2008-01-01",
@@ -139,11 +138,13 @@ def my_filter(recorder):
                 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()

examples/online_srv/online_management_simulate.py

@@ -1,21 +1,18 @@
-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 RollingOnlineManager
-from qlib.workflow.online.simulator import OnlineSimulator
-from qlib.workflow.task.collect import RecorderCollector
-from qlib.workflow.task.gen import RollingGen, task_generator
-from qlib.workflow.task.manage import TaskManager
-from qlib.workflow.task.utils import list_recorders
+# 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.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
+
 
 data_handler_config = {
     "start_time": "2018-01-01",
@@ -86,10 +83,10 @@ def __init__(
         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".
@@ -105,6 +102,8 @@ def __init__(
         """
         self.exp_name = exp_name
         self.task_pool = task_pool
+        self.start_time = start_time
+        self.end_time = end_time
         mongo_conf = {
             "task_url": task_url,
             "task_db_name": task_db_name,
@@ -115,62 +114,28 @@ def __init__(
         )  # 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 = RollingOnlineManager(
-            experiment_name=exp_name,
-            rolling_gen=self.rolling_gen,
-            trainer=self.trainer,
+        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):
+    # Run this to run all workflow automatically
+    def main(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(
-            RollingOnlineManager.SIGNAL_EXP, lambda x: True if x.info["name"] == self.exp_name else False
-        ):
-            exp.delete_recorder(rid)
-
-    # Run this firstly to see the workflow in OnlineManager
-    def first_train(self):
-        print("========== first train ==========")
-        self.reset()
-        self.rolling_online_manager.first_train(self.tasks)
-
-    # Run this secondly to see the simulating in OnlineSimulator
-    def simulate(self):
+        self.rolling_online_manager.reset()
         print("========== simulate ==========")
-        self.onlinesimulator.simulate()
-        print(self.rolling_online_manager.collect_artifact())
-
-        print("========== online models ==========")
-        recs_dict = self.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"])
-
-        print("========== online signals ==========")
-        print(self.rolling_online_manager.get_signals())
-
-    # Run this to run all workflow automaticly
-    def main(self):
-        self.first_train()
-        self.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)

examples/online_srv/rolling_online_management.py

@@ -1,22 +1,26 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""
+This example show how OnlineManager works with rolling tasks.
+There are two parts including first train and routine.
+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 RollingAverageStrategy
 from qlib.workflow.task.gen import RollingGen
 from qlib.workflow.task.manage import TaskManager
-from qlib.workflow.online.manager import RollingOnlineManager
+from qlib.workflow.online.manager import OnlineManager
 from qlib.workflow.task.utils import list_recorders
 from qlib.model.trainer import TrainerRM
 
-"""
-This example show how RollingOnlineManager 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
-"""
-
 data_handler_config = {
     "start_time": "2013-01-01",
     "end_time": "2020-09-25",
@@ -77,58 +81,73 @@
 class RollingOnlineExample:
     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=550,
+        tasks=[task_xgboost_config, task_lgb_config],
     ):
-        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_online_manager = RollingOnlineManager(
-            experiment_name=exp_name,
-            rolling_gen=RollingGen(step=rolling_step, rtype=RollingGen.ROLL_SD),
-            trainer=TrainerRM(self.exp_name, self.task_pool),
-        )
-
-    _ROLLING_MANAGER_PATH = ".rolling_manager"  # the RollingOnlineManager will dump to this file, for it will be loaded when calling routine.
+        self.tasks = tasks
+        self.rolling_step = rolling_step
+        strategy = []
+        for task in tasks:
+            name_id = task["model"]["class"]  # NOTE: Assumption: The model class can specify only one strategy
+            strategy.append(
+                RollingAverageStrategy(
+                    name_id,
+                    task,
+                    RollingGen(step=rolling_step, rtype=RollingGen.ROLL_SD),
+                    TrainerRM(experiment_name=name_id, task_pool=name_id),
+                )
+            )
+
+        self.rolling_online_manager = OnlineManager(strategy)
+        self.collector = self.rolling_online_manager.get_collector()
+
+    _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):
         print("========== reset ==========")
-        TaskManager(self.task_pool).remove()
-        exp = R.get_exp(experiment_name=self.exp_name)
-        for rid in exp.list_recorders():
-            exp.delete_recorder(rid)
+        for task in self.tasks:
+            name_id = task["model"]["class"] + "_" + str(self.rolling_step)
+            TaskManager(name_id).remove()
+            exp = R.get_exp(experiment_name=name_id)
+            for rid in exp.list_recorders():
+                exp.delete_recorder(rid)
 
-        if os.path.exists(self._ROLLING_MANAGER_PATH):
-            os.remove(self._ROLLING_MANAGER_PATH)
+            if os.path.exists(self._ROLLING_MANAGER_PATH):
+                os.remove(self._ROLLING_MANAGER_PATH)
 
-        for rid in list_recorders(
-            RollingOnlineManager.SIGNAL_EXP, lambda x: True if x.info["name"] == self.exp_name else False
-        ):
-            exp.delete_recorder(rid)
+            for rid in list_recorders("OnlineManagerSignals", lambda x: True if x.info["name"] == name_id else False):
+                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([task_xgboost_config, task_lgb_config])
+        self.rolling_online_manager.first_train()
+        print("========== dump ==========")
         self.rolling_online_manager.to_pickle(self._ROLLING_MANAGER_PATH)
-        print(self.rolling_online_manager.collect_artifact())
+        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.collect_artifact())
+        print("========== collect results ==========")
+        print(self.collector())
 
     def main(self):
         self.first_run()
@@ -137,11 +156,11 @@ def main(self):
 
 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)

examples/online_srv/update_online_pred.py

@@ -1,16 +1,17 @@
-import fire
-import qlib
-from qlib.config import REG_CN
-from qlib.model.trainer import task_train
-from qlib.workflow.online.manager import OnlineManagerR
-from qlib.workflow.task.utils import list_recorders
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
 
 """
-This example show how OnlineManager works when we need update prediction.
+This example show how OnlineTool works when we need update prediction.
 There are two parts including first_train and update_online_pred.
-Firstly, the RollingOnlineManager will finish the first training and set the trained model to `online` model.
-Next, the RollingOnlineManager will finish updating online prediction
+Firstly, we will finish the training and set the trained model to `online` model.
+Next, we will finish updating online prediction.
 """
+import fire
+import qlib
+from qlib.config import REG_CN
+from qlib.model.trainer import task_train
+from qlib.workflow.online.utils import OnlineToolR
 
 data_handler_config = {
     "start_time": "2008-01-01",
@@ -65,15 +66,15 @@ def __init__(
     ):
         qlib.init(provider_uri=provider_uri, region=region)
         self.experiment_name = experiment_name
-        self.online_manager = OnlineManagerR(self.experiment_name)
+        self.online_tool = OnlineToolR(self.experiment_name)
         self.task_config = task_config
 
     def first_train(self):
         rec = task_train(self.task_config, experiment_name=self.experiment_name)
-        self.online_manager.reset_online_tag(rec)  # set to online model
+        self.online_tool.reset_online_tag(rec)  # set to online model
 
     def update_online_pred(self):
-        self.online_manager.update_online_pred()
+        self.online_tool.update_online_pred()
 
     def main(self):
         self.first_train()