From b84156fde8bab93c220c8df25fa973f75bd2ccb0 Mon Sep 17 00:00:00 2001
From: lzh222333 <lzh222333@163.com>
Date: Wed, 3 Mar 2021 11:25:37 +0800
Subject: [PATCH 01/10] Consider more situations about task_config. Save the
 "param" file which is collect.py need.

---
 qlib/model/trainer.py | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/qlib/model/trainer.py b/qlib/model/trainer.py
index f0bc0b780..71cf9061f 100644
--- a/qlib/model/trainer.py
+++ b/qlib/model/trainer.py
@@ -27,16 +27,22 @@ def task_train(task_config: dict, experiment_name):
         model.fit(dataset)
         recorder = R.get_recorder()
         R.save_objects(**{"params.pkl": model})
+        R.save_objects(param=task_config)  # keep the original format and datatype
 
         # generate records: prediction, backtest, and analysis
-        for record in task_config["record"]:
+        records = task_config.get('record', [])
+        if isinstance(records, dict):  # prevent only one dict
+            records = [records]
+        for record in records:
             if record["class"] == SignalRecord.__name__:
                 srconf = {"model": model, "dataset": dataset, "recorder": recorder}
+                record.setdefault("kwargs", {})
                 record["kwargs"].update(srconf)
                 sr = init_instance_by_config(record)
                 sr.generate()
             else:
                 rconf = {"recorder": recorder}
+                record.setdefault("kwargs", {})
                 record["kwargs"].update(rconf)
                 ar = init_instance_by_config(record)
                 ar.generate()

From 05cf0e1edcdbe0b696b7e8c1cde538e3a5168dfa Mon Sep 17 00:00:00 2001
From: lzh222333 <lzh222333@163.com>
Date: Wed, 3 Mar 2021 15:42:39 +0800
Subject: [PATCH 02/10] add task_generator method and update some hint

---
 qlib/workflow/task/gen.py | 66 ++++++++++++++++++++++++++++++++++-----
 1 file changed, 59 insertions(+), 7 deletions(-)

diff --git a/qlib/workflow/task/gen.py b/qlib/workflow/task/gen.py
index 9b031435e..efbfe94a6 100644
--- a/qlib/workflow/task/gen.py
+++ b/qlib/workflow/task/gen.py
@@ -9,11 +9,64 @@ import typing
 from .utils import TimeAdjuster
 
 
+def task_generator(*args, **kwargs) -> list:
+    """
+    Accept the dict of task config and the TaskGen to generate different tasks.
+    There is no limit to the number and position of input.
+    The key of input will add to task config.
+
+    for example:
+        There are 3 task_config(a,b,c) and 2 TaskGen(A,B). A will double the task_config and B will triple.
+        task_generator(a=a, b=b, c=c, A=A, B=B) will finally generate 18 task_config.
+
+    Parameters
+    ----------
+    args : dict or TaskGen
+    kwargs : dict or TaskGen
+
+    Returns
+    -------
+    gen_task_list : list
+        a list of task config after generating
+    """
+    tasks_list = []
+    gen_list = []
+
+    tmp_id = 1
+    for task in args:
+        if isinstance(task, dict):
+            task["task_key"] = tmp_id
+            tmp_id += 1
+            tasks_list.append(task)
+        elif isinstance(task, TaskGen):
+            gen_list.append(task)
+        else:
+            raise NotImplementedError(f"{type(task)} is not supported in task_generator")
+
+    for key, task in kwargs.items():
+        if isinstance(task, dict):
+            task["task_key"] = key
+            tasks_list.append(task)
+        elif isinstance(task, TaskGen):
+            gen_list.append(task)
+        else:
+            raise NotImplementedError(f"{type(task)} is not supported in task_generator")
+
+    # generate gen_task_list
+    gen_task_list = []
+    for gen in gen_list:
+        new_task_list = []
+        for task in tasks_list:
+            new_task_list.extend(gen(task))
+        gen_task_list = new_task_list
+    return gen_task_list
+
+
 class TaskGen(metaclass=abc.ABCMeta):
     @abc.abstractmethod
     def __call__(self, *args, **kwargs) -> typing.List[dict]:
         """
-        generate
+        the base class for generate different tasks
 
         Parameters
         ----------
@@ -35,9 +88,8 @@ class TaskGen(metaclass=abc.ABCMeta):
 
 
 class RollingGen(TaskGen):
-
-    ROLL_EX = TimeAdjuster.SHIFT_EX
-    ROLL_SD = TimeAdjuster.SHIFT_SD
+    ROLL_EX = TimeAdjuster.SHIFT_EX  # fixed start date, expanding end date
+    ROLL_SD = TimeAdjuster.SHIFT_SD  # fixed window size, slide it from start date
 
     def __init__(self, step: int = 40, rtype: str = ROLL_EX):
         """
@@ -48,7 +100,7 @@ class RollingGen(TaskGen):
         step : int
             step to rolling
         rtype : str
-            rolling type (expanding, rolling)
+            rolling type (expanding, sliding)
         """
         self.step = step
         self.rtype = rtype
@@ -111,12 +163,12 @@ class RollingGen(TaskGen):
                 segments = {}
                 try:
                     for k, seg in prev_seg.items():
-                        # 决定怎么shift
+                        # decide how to shift
                         if k == self.train_key and self.rtype == self.ROLL_EX:
                             rtype = self.ta.SHIFT_EX
                         else:
                             rtype = self.ta.SHIFT_SD
-                        # 整段数据做shift
+                        # shift the segments data
                         segments[k] = self.ta.shift(seg, step=self.step, rtype=rtype)
                     if segments[self.test_key][0] > test_end:
                         break

From fd2c1ba1ed1c3919b6ddd418f0b3f82239f0baf5 Mon Sep 17 00:00:00 2001
From: lzh222333 <lzh222333@163.com>
Date: Wed, 3 Mar 2021 16:36:15 +0800
Subject: [PATCH 03/10] Update some hint

---
 qlib/workflow/task/collect.py |  9 ++++-----
 qlib/workflow/task/manage.py  |  8 ++------
 qlib/workflow/task/utils.py   | 29 +++++++++++++++++++++--------
 3 files changed, 27 insertions(+), 19 deletions(-)

diff --git a/qlib/workflow/task/collect.py b/qlib/workflow/task/collect.py
index 7cdca30fa..9a67d8e06 100644
--- a/qlib/workflow/task/collect.py
+++ b/qlib/workflow/task/collect.py
@@ -4,17 +4,17 @@ from typing import Union
 from tqdm.auto import tqdm
 
 
-class RollingEnsemble:
+class RollingCollector:
     """
     Rolling Models Ensemble based on (R)ecord
 
     This shares nothing with Ensemble
     """
 
-    # TODO: 这边还可以加加速
+    # TODO: speed up this class
     def __init__(self, get_key_func, flt_func=None):
-        self.get_key_func = get_key_func
-        self.flt_func = flt_func
+        self.get_key_func = get_key_func  # user need to implement this method to get the key of a task based on task config
+        self.flt_func = flt_func  # determine whether a task can be retained based on task config
 
     def __call__(self, exp_name) -> Union[pd.Series, dict]:
         # TODO;
@@ -26,7 +26,6 @@ class RollingEnsemble:
 
         recs_flt = {}
         for rid, rec in tqdm(recs.items(), desc="Loading data"):
-            # rec = exp.get_recorder(recorder_id=rid)
             params = rec.load_object("param")
             if rec.status == rec.STATUS_FI:
                 if self.flt_func is None or self.flt_func(params):
diff --git a/qlib/workflow/task/manage.py b/qlib/workflow/task/manage.py
index 3bcac8360..1a4c341de 100644
--- a/qlib/workflow/task/manage.py
+++ b/qlib/workflow/task/manage.py
@@ -36,12 +36,8 @@ class TaskManager:
     The tasks manager assume that you will only update the tasks you fetched.
     The mongo fetch one and update will make it date updating secure.
 
-    Usage Examples from the CLI.
-    python -m blocks.tasks.__init__ task_stat --task_pool meta_task_rule
-
-
     NOTE:
-    - 假设： 存储在db里面的都是encode过的， 拿出来的都是decode过的
+    - assumption: the data in MongoDB was encoded and the data out of MongoDB was decoded
     """
 
     STATUS_WAITING = "waiting"
@@ -85,7 +81,7 @@ class TaskManager:
         return {k: str(v) for k, v in flt.items()}
 
     def replace_task(self, task, new_task, task_pool=None):
-        # 这里的假设是从接口拿出来的都是decode过的，在接口内部的都是 encode过的
+        # assume that the data out of interface was decoded and the data in interface was encoded
         new_task = self._encode_task(new_task)
         task_pool = self._get_task_pool(task_pool)
         query = {"_id": ObjectId(task["_id"])}
diff --git a/qlib/workflow/task/utils.py b/qlib/workflow/task/utils.py
index d6089ff66..719359d5b 100644
--- a/qlib/workflow/task/utils.py
+++ b/qlib/workflow/task/utils.py
@@ -6,9 +6,19 @@ from qlib.data import D
 from qlib.config import C
 from qlib.log import get_module_logger
 from pymongo import MongoClient
-
+from typing import Union
 
 def get_mongodb():
+    """
+
+    get database in MongoDB, which means you need to declare the address and the name of database.
+    for example:
+        C["mongo"] = {
+            "task_url" : "mongodb://localhost:27017/",
+            "task_db_name" : "rolling_db"
+        }
+
+    """
     try:
         cfg = C["mongo"]
     except KeyError:
@@ -20,7 +30,9 @@ def get_mongodb():
 
 
 class TimeAdjuster:
-    """找到合适的日期，然后adjust date"""
+    """
+    find appropriate date and adjust date.
+    """
 
     def __init__(self, future=False):
         self.cals = D.calendar(future=future)
@@ -40,7 +52,7 @@ class TimeAdjuster:
 
     def max(self):
         """
-        Return return the max calendar date
+        Return the max calendar date
         """
         return max(self.cals)
 
@@ -56,7 +68,7 @@ class TimeAdjuster:
 
     def align_time(self, time_point, tp_type="start"):
         """
-        Align a timepoint to calendar  weekdays
+        Align a timepoint to calendar weekdays
 
         Parameters
         ----------
@@ -67,7 +79,7 @@ class TimeAdjuster:
         """
         return self.cals[self.align_idx(time_point, tp_type=tp_type)]
 
-    def align_seg(self, segment):
+    def align_seg(self, segment: Union[dict, tuple]):
         if isinstance(segment, dict):
             return {k: self.align_seg(seg) for k, seg in segment.items()}
         elif isinstance(segment, tuple):
@@ -75,14 +87,15 @@ class TimeAdjuster:
         else:
             raise NotImplementedError(f"This type of input is not supported")
 
-    def truncate(self, segment, test_start, days: int):
+    def truncate(self, segment: tuple, test_start, days: int):
         """
         truncate the segment based on the test_start date
 
         Parameters
         ----------
-        segment :
+        segment : tuple
             time segment
+        test_start
         days : int
             The trading days to be truncated
             大部分情况是因为这个时间段的数据(一般是特征)会用到 `days` 天的数据
@@ -101,7 +114,7 @@ class TimeAdjuster:
     SHIFT_SD = "sliding"
     SHIFT_EX = "expanding"
 
-    def shift(self, seg, step: int, rtype=SHIFT_SD):
+    def shift(self, seg: tuple, step: int, rtype=SHIFT_SD):
         """
         shift the datatiem of segment
 

From 2882929c5d91e2f655265036fd26ca6c50cbdd6a Mon Sep 17 00:00:00 2001
From: lzh222333 <lzh222333@163.com>
Date: Wed, 3 Mar 2021 16:58:05 +0800
Subject: [PATCH 04/10] Add an example about workflow using RollingGen.

---
 examples/workflow_task_rolling.ipynb | 177 +++++++++++++++++++++++++++
 1 file changed, 177 insertions(+)
 create mode 100644 examples/workflow_task_rolling.ipynb

diff --git a/examples/workflow_task_rolling.ipynb b/examples/workflow_task_rolling.ipynb
new file mode 100644
index 000000000..c2d399be0
--- /dev/null
+++ b/examples/workflow_task_rolling.ipynb
@@ -0,0 +1,177 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {
+    "collapsed": true
+   },
+   "outputs": [],
+   "source": [
+    "import qlib\n",
+    "from qlib.config import REG_CN\n",
+    "from qlib.workflow.task.gen import RollingGen, task_generator\n",
+    "from qlib.workflow.task.manage import TaskManager\n",
+    "from qlib.config import C\n",
+    "\n",
+    "data_handler_config = {\n",
+    "    \"start_time\": \"2008-01-01\",\n",
+    "    \"end_time\": \"2020-08-01\",\n",
+    "    \"fit_start_time\": \"2008-01-01\",\n",
+    "    \"fit_end_time\": \"2014-12-31\",\n",
+    "    \"instruments\": 'csi100',\n",
+    "}\n",
+    "\n",
+    "dataset_config = {\n",
+    "        \"class\": \"DatasetH\",\n",
+    "        \"module_path\": \"qlib.data.dataset\",\n",
+    "        \"kwargs\": {\n",
+    "            \"handler\": {\n",
+    "                \"class\": \"Alpha158\",\n",
+    "                \"module_path\": \"qlib.contrib.data.handler\",\n",
+    "                \"kwargs\": data_handler_config,\n",
+    "            },\n",
+    "            \"segments\": {\n",
+    "                \"train\": (\"2008-01-01\", \"2014-12-31\"),\n",
+    "                \"valid\": (\"2015-01-01\", \"2016-12-31\"),\n",
+    "                \"test\": (\"2017-01-01\", \"2020-08-01\"),\n",
+    "            },\n",
+    "        },\n",
+    "    }\n",
+    "\n",
+    "record_config = [\n",
+    "    {\n",
+    "        \"class\": \"SignalRecord\",\n",
+    "        \"module_path\": \"qlib.workflow.record_temp\",\n",
+    "    },\n",
+    "    {\n",
+    "        \"class\": \"SigAnaRecord\",\n",
+    "        \"module_path\": \"qlib.workflow.record_temp\",\n",
+    "    }\n",
+    "]\n",
+    "\n",
+    "# use lgb\n",
+    "task_lgb_config = {\n",
+    "    \"model\": {\n",
+    "        \"class\": \"LGBModel\",\n",
+    "        \"module_path\": \"qlib.contrib.model.gbdt\",\n",
+    "    },\n",
+    "    \"dataset\": dataset_config,\n",
+    "    \"record\": record_config,\n",
+    "}\n",
+    "\n",
+    "# use xgboost\n",
+    "task_xgboost_config = {\n",
+    "    \"model\": {\n",
+    "        \"class\": \"XGBModel\",\n",
+    "        \"module_path\": \"qlib.contrib.model.xgboost\",\n",
+    "    },\n",
+    "    \"dataset\": dataset_config,\n",
+    "    \"record\": record_config,\n",
+    "}\n",
+    "provider_uri = r\"../qlib-main/qlib_data/cn_data\"\n",
+    "#provider_uri = \"~/.qlib/qlib_data/cn_data\"  # target_dir\n",
+    "qlib.init(provider_uri=provider_uri, region=REG_CN)\n",
+    "\n",
+    "C[\"mongo\"] = {\n",
+    "    \"task_url\" : \"mongodb://localhost:27017/\", # maybe you need to change it to your url\n",
+    "    \"task_db_name\" : \"rolling_db\"\n",
+    "}\n",
+    "\n",
+    "exp_name = 'rolling_exp' # experiment name, will be used as the experiment in MLflow\n",
+    "task_pool = 'rolling_task' # task pool name, will be used as the document in MongoDB"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "outputs": [],
+   "source": [
+    "tasks = task_generator(\n",
+    "    task_xgboost_config, # default task name\n",
+    "    RollingGen(step=550,rtype=RollingGen.ROLL_SD), # generate different date segment\n",
+    "    task_lgb=task_lgb_config # use \"task_lgb\" as the task name\n",
+    ")\n",
+    "# Uncomment next two lines to see the generated tasks\n",
+    "# from pprint import pprint\n",
+    "# pprint(tasks)\n",
+    "tm = TaskManager(task_pool=task_pool)\n",
+    "tm.create_task(tasks) # all tasks will be saved to MongoDB"
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%%\n"
+    }
+   }
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "outputs": [],
+   "source": [
+    "from qlib.workflow.task.manage import run_task\n",
+    "from qlib.workflow.task.collect import RollingCollector\n",
+    "from qlib.model.trainer import task_train\n",
+    "\n",
+    "run_task(task_train, task_pool, experiment_name=exp_name) # all tasks will be trained using \"task_train\" method"
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%%\n"
+    }
+   }
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "outputs": [],
+   "source": [
+    "def get_task_key(task_config):\n",
+    "    task_key = task_config[\"task_key\"]\n",
+    "    rolling_end_timestamp = task_config[\"dataset\"][\"kwargs\"][\"segments\"][\"test\"][1]\n",
+    "    rolling_end_datatime = rolling_end_timestamp.to_pydatetime()\n",
+    "    return task_key, rolling_end_datatime.strftime('%Y-%m-%d')\n",
+    "\n",
+    "def my_filter(task_config):\n",
+    "    # only choose the results of \"task_lgb\" and test in 2019 from all tasks\n",
+    "    task_key, rolling_end = get_task_key(task_config)\n",
+    "    if task_key==\"task_lgb\" and rolling_end.startswith('2019'):\n",
+    "        return True\n",
+    "    return False\n",
+    "\n",
+    "collector = RollingCollector(get_task_key, my_filter)\n",
+    "pred_rolling = collector(exp_name) # name tasks by \"get_task_key\" and filter tasks by \"my_filter\"\n",
+    "pred_rolling"
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%%\n"
+    }
+   }
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 2
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython2",
+   "version": "2.7.6"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 0
+}
\ No newline at end of file

From a244f87f95e70ca2f97a687be10e3e3f606517a0 Mon Sep 17 00:00:00 2001
From: lzh222333 <lzh222333@163.com>
Date: Mon, 8 Mar 2021 13:25:11 +0800
Subject: [PATCH 05/10] modified the comments

---
 qlib/workflow/task/gen.py    |  47 +++++++++------
 qlib/workflow/task/manage.py | 114 +++++++++++++++++++++++++++++------
 qlib/workflow/task/utils.py  |  47 +++++++++++++--
 3 files changed, 168 insertions(+), 40 deletions(-)

diff --git a/qlib/workflow/task/gen.py b/qlib/workflow/task/gen.py
index efbfe94a6..60fc5c221 100644
--- a/qlib/workflow/task/gen.py
+++ b/qlib/workflow/task/gen.py
@@ -17,7 +17,7 @@ def task_generator(*args, **kwargs) -> list:
 
     for example:
         There are 3 task_config(a,b,c) and 2 TaskGen(A,B). A will double the task_config and B will triple.
-        task_generator(a=a, b=b, c=c, A=A, B=B) will finally generate 18 task_config.
+        task_generator(a_key=a, b_key=b, c_key=c, A, B) will finally generate 3*2*3 = 18 task_config.
 
     Parameters
     ----------
@@ -57,27 +57,37 @@ def task_generator(*args, **kwargs) -> list:
     for gen in gen_list:
         new_task_list = []
         for task in tasks_list:
-            new_task_list.extend(gen(task))
+            new_task_list.extend(gen.generate(task))
         gen_task_list = new_task_list
     return gen_task_list
 
 
 class TaskGen(metaclass=abc.ABCMeta):
+    """
+    the base class for generate different tasks
+
+    Example 1:
+
+        input: a specific task template and rolling steps
+
+        output: rolling version of the tasks
+
+    Example 2:
+
+        input: a specific task template and losses list
+
+        output: a set of tasks with different losses
+
+    """
     @abc.abstractmethod
-    def __call__(self, *args, **kwargs) -> typing.List[dict]:
+    def generate(self, task: dict) -> typing.List[dict]:
         """
-        the base class for generate different tasks
+        generate different tasks based on a task template
 
         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
+        task: dict
+            a task template
 
         Returns
         -------
@@ -89,7 +99,7 @@ class TaskGen(metaclass=abc.ABCMeta):
 
 class RollingGen(TaskGen):
     ROLL_EX = TimeAdjuster.SHIFT_EX  # fixed start date, expanding end date
-    ROLL_SD = TimeAdjuster.SHIFT_SD  # fixed window size, slide it from start date
+    ROLL_SD = TimeAdjuster.SHIFT_SD  # fixed segments size, slide it from start date
 
     def __init__(self, step: int = 40, rtype: str = ROLL_EX):
         """
@@ -104,12 +114,13 @@ class RollingGen(TaskGen):
         """
         self.step = step
         self.rtype = rtype
-        self.ta = TimeAdjuster(future=True)  # 为了保证test最后的日期不是None, 所以这边要改一改
+        # TODO: Ask pengrong to update future date in dataset
+        self.ta = TimeAdjuster(future=True)
 
         self.test_key = "test"
         self.train_key = "train"
 
-    def __call__(self, task: dict):
+    def generate(self, task: dict):
         """
         Converting the task into a rolling task
 
@@ -153,9 +164,9 @@ class RollingGen(TaskGen):
             # calculate segments
             if prev_seg is None:
                 # First rolling
-                # 1) prepare the end porint
+                # 1) prepare the end point
                 segments = copy.deepcopy(self.ta.align_seg(t["dataset"]["kwargs"]["segments"]))
-                test_end = self.ta.max() if segments[self.test_key][1] is None else segments[self.test_key][1]
+                test_end = self.ta.last_date() if segments[self.test_key][1] is None else segments[self.test_key][1]
                 # 2) and the init test segments
                 test_start_idx = self.ta.align_idx(segments[self.test_key][0])
                 segments[self.test_key] = (self.ta.get(test_start_idx), self.ta.get(test_start_idx + self.step - 1))
@@ -164,6 +175,7 @@ class RollingGen(TaskGen):
                 try:
                     for k, seg in prev_seg.items():
                         # decide how to shift
+                        # expanding only for train data, the segments size of test data and valid data won't change
                         if k == self.train_key and self.rtype == self.ROLL_EX:
                             rtype = self.ta.SHIFT_EX
                         else:
@@ -177,6 +189,7 @@ class RollingGen(TaskGen):
                     # No more rolling
                     break
 
+            # update segments of this task
             t["dataset"]["kwargs"]["segments"] = copy.deepcopy(segments)
             prev_seg = segments
             res.append(t)
diff --git a/qlib/workflow/task/manage.py b/qlib/workflow/task/manage.py
index 1a4c341de..ae4aee147 100644
--- a/qlib/workflow/task/manage.py
+++ b/qlib/workflow/task/manage.py
@@ -1,7 +1,7 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 """
-A task consists of 2 parts
+A task 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.
@@ -26,18 +26,22 @@ from qlib import auto_init
 class TaskManager:
     """TaskManager
     here is the what will a task looks like
-    {
-        'def': pickle serialized task definition.  using pickle will make it easier
-        'filter': json-like data. This is for filtering the tasks.
-        'status': 'waiting' | 'running' | 'done'
-        'res': pickle serialized task result,
-    }
+
+    .. code-block:: python
+
+        {
+            'def': pickle serialized task definition.  using pickle will make it easier
+            'filter': json-like data. This is for filtering the tasks.
+            'status': 'waiting' | 'running' | 'done'
+            'res': pickle serialized task result,
+        }
 
     The tasks manager assume that you will only update the tasks you fetched.
     The mongo fetch one and update will make it date updating secure.
 
-    NOTE:
-    - assumption: the data in MongoDB was encoded and the data out of MongoDB was decoded
+    .. note::
+
+        assumption: the data in MongoDB was encoded and the data out of MongoDB was decoded
     """
 
     STATUS_WAITING = "waiting"
@@ -48,6 +52,14 @@ class TaskManager:
     ENCODE_FIELDS_PREFIX = ["def", "res"]
 
     def __init__(self, task_pool=None):
+        """
+        init Task Manager, remember to make the statement of MongoDB url and database name firstly.
+
+        Parameters
+        ----------
+        task_pool: str
+            the name of Collection in MongoDB
+        """
         self.mdb = get_mongodb()
         self.task_pool = task_pool
 
@@ -100,6 +112,19 @@ class TaskManager:
             task_pool.insert_one(task)
 
     def insert_task_def(self, task_def, task_pool=None):
+        """
+        insert a task to task_pool
+
+        Parameters
+        ----------
+        task_def: dict
+        task_pool: str
+            the name of Collection in MongoDB
+
+        Returns
+        -------
+
+        """
         task_pool = self._get_task_pool(task_pool)
         task = self._encode_task(
             {
@@ -111,6 +136,23 @@ class TaskManager:
         self.insert_task(task, task_pool)
 
     def create_task(self, task_def_l, task_pool=None, dry_run=False, print_nt=False):
+        """
+        if the tasks in task_def_l is new, then insert new tasks into the task_pool
+
+        Parameters
+        ----------
+        task_def_l: list
+            a list of task
+        task_pool: str
+            the name of task_pool (collection name of MongoDB)
+        dry_run: bool
+            if insert those new tasks to task pool
+        print_nt: bool
+            if print new task
+        Returns
+        -------
+
+        """
         task_pool = self._get_task_pool(task_pool)
         new_tasks = []
         for t in task_def_l:
@@ -141,7 +183,7 @@ class TaskManager:
         task = task_pool.find_one_and_update(
             query, {"$set": {"status": self.STATUS_RUNNING}}, sort=[("priority", pymongo.DESCENDING)]
         )
-        # 这里我的 priority 必须是 高数优先级更高，因为 null会被在 ASCENDING时被排在最前面
+        # null will be at the top after sorting when using ASCENDING, so the larger the number higher, the higher the priority
         if task is None:
             return None
         task["status"] = self.STATUS_RUNNING
@@ -149,6 +191,20 @@ class TaskManager:
 
     @contextmanager
     def safe_fetch_task(self, query={}, task_pool=None):
+        """
+        fetch task from task_pool using query with contextmanager
+
+        Parameters
+        ----------
+        query: dict
+            the dict of query
+        task_pool: str
+            the name of Collection in MongoDB
+
+        Returns
+        -------
+
+        """
         task = self.fetch_task(query=query, task_pool=task_pool)
         try:
             yield task
@@ -167,12 +223,20 @@ class TaskManager:
                 yield task
 
     def query(self, query={}, decode=True, task_pool=None):
-        """query
+        """
         This function may raise exception `pymongo.errors.CursorNotFound: cursor id not found` if it takes too long to iterate the generator
 
-        :param query:
-        :param decode:
-        :param task_pool:
+        Parameters
+        ----------
+        query: dict
+            the dict of query
+        decode: bool
+        task_pool: str
+            the name of Collection in MongoDB
+
+        Returns
+        -------
+
         """
         query = query.copy()
         if "_id" in query:
@@ -196,6 +260,20 @@ class TaskManager:
         task_pool.update_one({"_id": task["_id"]}, update_dict)
 
     def remove(self, query={}, task_pool=None):
+        """
+        remove the task using query
+
+        Parameters
+        ----------
+        query: dict
+            the dict of query
+        task_pool: str
+            the name of Collection in MongoDB
+
+        Returns
+        -------
+
+        """
         query = query.copy()
         task_pool = self._get_task_pool(task_pool)
         if "_id" in query:
@@ -250,15 +328,15 @@ class TaskManager:
 
 
 def run_task(task_func, task_pool, force_release=False, *args, **kwargs):
-    """run_task.
-    While task pool is not empty, use task_func to fetch and run tasks in task_pool
+    """
+    While task pool is not empty (has WAITING tasks), use task_func to fetch and run tasks in task_pool
 
     Parameters
     ----------
     task_func : def (task_def, *args, **kwargs) -> <res which will be committed>
         the function to run the task
-    task_pool :
-        The name of the task pool
+    task_pool : str
+        the name of the task pool (Collection in MongoDB)
     force_release :
         will the program force to release the resource
     args :
diff --git a/qlib/workflow/task/utils.py b/qlib/workflow/task/utils.py
index 719359d5b..5e94f55ae 100644
--- a/qlib/workflow/task/utils.py
+++ b/qlib/workflow/task/utils.py
@@ -52,11 +52,30 @@ class TimeAdjuster:
 
     def max(self):
         """
-        Return the max calendar date
+        (Deprecated)
+        Return the max calendar datetime
         """
         return max(self.cals)
 
+    def last_date(self) -> pd.Timestamp:
+        """
+        Return the last datetime in the calendar
+        """
+        return self.cals[-1]
+
     def align_idx(self, time_point, tp_type="start"):
+        """
+        align the index of time_point in the calendar
+
+        Parameters
+        ----------
+        time_point
+        tp_type : str
+
+        Returns
+        -------
+        index : int
+        """
         time_point = pd.Timestamp(time_point)
         if tp_type == "start":
             idx = bisect.bisect_left(self.cals, time_point)
@@ -68,11 +87,11 @@ class TimeAdjuster:
 
     def align_time(self, time_point, tp_type="start"):
         """
-        Align a timepoint to calendar weekdays
+        Align time_point to trade date of calendar
 
         Parameters
         ----------
-        time_point :
+        time_point
             Time point
         tp_type : str
             time point type (`"start"`, `"end"`)
@@ -80,6 +99,24 @@ class TimeAdjuster:
         return self.cals[self.align_idx(time_point, tp_type=tp_type)]
 
     def align_seg(self, segment: Union[dict, tuple]):
+        """
+        align the given date to trade date
+
+        for example:
+            input: {'train': ('2008-01-01', '2014-12-31'), 'valid': ('2015-01-01', '2016-12-31'), 'test': ('2017-01-01', '2020-08-01')}
+
+            output: {'train': (Timestamp('2008-01-02 00:00:00'), Timestamp('2014-12-31 00:00:00')),
+                    'valid': (Timestamp('2015-01-05 00:00:00'), Timestamp('2016-12-30 00:00:00')),
+                    'test': (Timestamp('2017-01-03 00:00:00'), Timestamp('2020-07-31 00:00:00'))}
+
+        Parameters
+        ----------
+        segment
+
+        Returns
+        -------
+        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()}
         elif isinstance(segment, tuple):
@@ -98,7 +135,7 @@ class TimeAdjuster:
         test_start
         days : int
             The trading days to be truncated
-            大部分情况是因为这个时间段的数据(一般是特征)会用到 `days` 天的数据
+            the data in this segment may need 'days' data
         """
         test_idx = self.align_idx(test_start)
         if isinstance(segment, tuple):
@@ -116,7 +153,7 @@ class TimeAdjuster:
 
     def shift(self, seg: tuple, step: int, rtype=SHIFT_SD):
         """
-        shift the datatiem of segment
+        shift the datatime of segment
 
         Parameters
         ----------

From def132e1407bc97585efa2d261feefd8386c34f6 Mon Sep 17 00:00:00 2001
From: lzh222333 <lzh222333@163.com>
Date: Mon, 8 Mar 2021 16:10:16 +0800
Subject: [PATCH 06/10] modified format and added TaskCollector

---
 qlib/model/trainer.py         |  6 ++--
 qlib/workflow/task/collect.py | 58 ++++++++++++++++++++++++++++++++++-
 qlib/workflow/task/gen.py     |  1 +
 qlib/workflow/task/utils.py   |  1 +
 4 files changed, 63 insertions(+), 3 deletions(-)

diff --git a/qlib/model/trainer.py b/qlib/model/trainer.py
index 71cf9061f..91061636d 100644
--- a/qlib/model/trainer.py
+++ b/qlib/model/trainer.py
@@ -6,7 +6,7 @@ from qlib.workflow import R
 from qlib.workflow.record_temp import SignalRecord
 
 
-def task_train(task_config: dict, experiment_name):
+def task_train(task_config: dict, experiment_name: str):
     """
     task based training
 
@@ -14,6 +14,8 @@ def task_train(task_config: dict, experiment_name):
     ----------
     task_config : dict
         A dict describes a task setting.
+    experiment_name: str
+        The name of experiment
     """
 
     # model initiaiton
@@ -30,7 +32,7 @@ def task_train(task_config: dict, experiment_name):
         R.save_objects(param=task_config)  # keep the original format and datatype
 
         # generate records: prediction, backtest, and analysis
-        records = task_config.get('record', [])
+        records = task_config.get("record", [])
         if isinstance(records, dict):  # prevent only one dict
             records = [records]
         for record in records:
diff --git a/qlib/workflow/task/collect.py b/qlib/workflow/task/collect.py
index 9a67d8e06..4562a1cec 100644
--- a/qlib/workflow/task/collect.py
+++ b/qlib/workflow/task/collect.py
@@ -4,6 +4,62 @@ from typing import Union
 from tqdm.auto import tqdm
 
 
+class TaskCollector:
+    """
+    Collect the record results of the finished tasks with key and filter
+    """
+
+    @staticmethod
+    def collect(
+        experiment_name: str,
+        get_key_func,
+        filter_func=None,
+    ):
+        """
+
+        Parameters
+        ----------
+        experiment_name : str
+        get_key_func : function(task: dict) -> Union[Number, str, tuple]
+            get the key of a task when collect it
+        filter_func : function(task: dict) -> bool
+            to judge a task will be collected or not
+
+        Returns
+        -------
+
+        """
+        exp = R.get_exp(experiment_name=experiment_name)
+        # filter records
+        recs = exp.list_recorders()
+
+        recs_flt = {}
+        for rid, rec in tqdm(recs.items(), desc="Loading data"):
+            params = rec.load_object("param")
+            if rec.status == rec.STATUS_FI:
+                if filter_func is None or filter_func(params):
+                    rec.params = params
+                    recs_flt[rid] = rec
+
+        # group
+        recs_group = {}
+        for _, rec in recs_flt.items():
+            params = rec.params
+            group_key = get_key_func(params)
+            recs_group.setdefault(group_key, []).append(rec)
+
+        # reduce group
+        reduce_group = {}
+        for k, rec_l in recs_group.items():
+            pred_l = []
+            for rec in rec_l:
+                pred_l.append(rec.load_object("pred.pkl").iloc[:, 0])
+            pred = pd.concat(pred_l).sort_index()
+            reduce_group[k] = pred
+
+        return reduce_group
+
+
 class RollingCollector:
     """
     Rolling Models Ensemble based on (R)ecord
@@ -13,7 +69,7 @@ class RollingCollector:
 
     # TODO: speed up this class
     def __init__(self, get_key_func, flt_func=None):
-        self.get_key_func = get_key_func  # user need to implement this method to get the key of a task based on task config
+        self.get_key_func = get_key_func  # get the key of a task based on task config
         self.flt_func = flt_func  # determine whether a task can be retained based on task config
 
     def __call__(self, exp_name) -> Union[pd.Series, dict]:
diff --git a/qlib/workflow/task/gen.py b/qlib/workflow/task/gen.py
index 60fc5c221..b1c2e0ce2 100644
--- a/qlib/workflow/task/gen.py
+++ b/qlib/workflow/task/gen.py
@@ -79,6 +79,7 @@ class TaskGen(metaclass=abc.ABCMeta):
         output: a set of tasks with different losses
 
     """
+
     @abc.abstractmethod
     def generate(self, task: dict) -> typing.List[dict]:
         """
diff --git a/qlib/workflow/task/utils.py b/qlib/workflow/task/utils.py
index 5e94f55ae..63563e2f6 100644
--- a/qlib/workflow/task/utils.py
+++ b/qlib/workflow/task/utils.py
@@ -8,6 +8,7 @@ from qlib.log import get_module_logger
 from pymongo import MongoClient
 from typing import Union
 
+
 def get_mongodb():
     """
 

From 83dbdfb45e5f429ed26186466fd8b0e56108a2ce Mon Sep 17 00:00:00 2001
From: lzh222333 <lzh222333@163.com>
Date: Tue, 9 Mar 2021 17:22:36 +0800
Subject: [PATCH 07/10] finished document and example

---
 docs/advanced/task_managment.rst              |  67 +++
 .../taskmanager/task_manager_rolling.ipynb    | 445 ++++++++++++++++++
 examples/taskmanager/task_manager_rolling.py  | 108 +++++
 examples/workflow_task_rolling.ipynb          | 177 -------
 4 files changed, 620 insertions(+), 177 deletions(-)
 create mode 100644 docs/advanced/task_managment.rst
 create mode 100644 examples/taskmanager/task_manager_rolling.ipynb
 create mode 100644 examples/taskmanager/task_manager_rolling.py
 delete mode 100644 examples/workflow_task_rolling.ipynb

diff --git a/docs/advanced/task_managment.rst b/docs/advanced/task_managment.rst
new file mode 100644
index 000000000..78ac62410
--- /dev/null
+++ b/docs/advanced/task_managment.rst
@@ -0,0 +1,67 @@
+.. _task_managment:
+
+=================================
+Task Management
+=================================
+.. currentmodule:: qlib
+
+
+Introduction
+=============
+
+The `Workflow <../component/introduction.html>`_ part introduce how to run research workflow in a loosely-coupled way. But it can only execute one ``task`` when you use ``qrun``. To automatically generate and execute different tasks, Task Management module provide a whole process including `Task Generating`_, `Task Storing`_, `Task Running`_ and `Task Collecting`_. 
+With this module, users can run their ``task`` automatically at different periods, in different losses or even by different models.
+
+An example of the entire process is shown `here <>`_.
+
+Task Generating
+===============
+A ``task`` consists of `Model`, `Dataset`, `Record` or anything added by users. 
+The specific task template can be viewed in 
+`Task Section <../component/workflow.html#task-section>`_.
+Even though the task template is fixed, Users can use ``TaskGen`` to generate different ``task`` by task template.
+
+Here is the base class of TaskGen:
+
+.. autoclass:: qlib.workflow.task.gen.TaskGen
+    :members:
+
+``Qlib`` provider a class `RollingGen<https://github.com/microsoft/qlib/tree/main/qlib/workflow/task/gen.py>`_ to generate a list of ``task`` of dataset in different date segments.
+This allows users to verify the effect of data from different periods on the model in one experiment.
+
+Task Storing
+===============
+In order to achieve higher efficiency and the possibility of cluster operation, ``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.
+
+Users need to provide the url and database of ``task`` storing like this.
+
+    .. code-block:: python
+
+        from qlib.config import C
+        C["mongo"] = {
+            "task_url" : "mongodb://localhost:27017/", # maybe you need to change it to your url
+            "task_db_name" : "rolling_db" # you can custom database name
+        }
+
+The CRUD methods of ``task`` can be found in TaskManager. More methods can be seen in the `Github<https://github.com/microsoft/qlib/tree/main/qlib/workflow/task/manage.py>`_.
+
+.. autoclass:: qlib.workflow.task.manage.TaskManager
+    :members:
+
+Task Running
+===============
+After generating and storing those ``task``, it's time to run the ``task`` in the *WAITING* status.
+``qlib`` provide a method to run those ``task`` in task pool, however users can also customize how tasks are executed.
+An easy way to get the ``task_func`` is using ``qlib.model.trainer.task_train`` directly.
+It will run the whole workflow defined by ``task``, which includes *Model*, *Dataset*, *Record*.
+
+.. autofunction:: qlib.workflow.task.manage.run_task
+
+Task Collecting
+===============
+To see the results of ``task`` after running, ``Qlib`` provide a task collector to collect the tasks by filter condition (optional).
+The collector will return a dict of filtered key (users defined by task config) and value (predict scores from ``pred.pkl``).
+
+.. autoclass:: qlib.workflow.task.collect.TaskCollector
+    :members:
\ No newline at end of file
diff --git a/examples/taskmanager/task_manager_rolling.ipynb b/examples/taskmanager/task_manager_rolling.ipynb
new file mode 100644
index 000000000..43ae5b1d1
--- /dev/null
+++ b/examples/taskmanager/task_manager_rolling.ipynb
@@ -0,0 +1,445 @@
+{
+ "cells": [
+  {
+   "cell_type": "code",
+   "execution_count": 23,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "import mlflow\n",
+    "mlflow.end_run()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 24,
+   "metadata": {
+    "collapsed": true
+   },
+   "outputs": [
+    {
+     "output_type": "stream",
+     "name": "stderr",
+     "text": [
+      "[8348:MainThread](2021-03-09 14:55:48,543) INFO - qlib.Initialization - [config.py:279] - default_conf: client.\n",
+      "[8348:MainThread](2021-03-09 14:55:50,592) WARNING - qlib.Initialization - [config.py:295] - redis connection failed(host=127.0.0.1 port=6379), cache will not be used!\n",
+      "[8348:MainThread](2021-03-09 14:55:50,597) INFO - qlib.Initialization - [__init__.py:48] - qlib successfully initialized based on client settings.\n",
+      "[8348:MainThread](2021-03-09 14:55:50,601) INFO - qlib.Initialization - [__init__.py:49] - data_path=C:\\Users\\lzh222333\\.qlib\\qlib_data\\cn_data\n"
+     ]
+    }
+   ],
+   "source": [
+    "import qlib\n",
+    "from qlib.config import REG_CN\n",
+    "from qlib.workflow.task.gen import RollingGen, task_generator\n",
+    "from qlib.workflow.task.manage import TaskManager\n",
+    "from qlib.config import C\n",
+    "\n",
+    "data_handler_template = {\n",
+    "    \"start_time\": \"2008-01-01\",\n",
+    "    \"end_time\": \"2020-08-01\",\n",
+    "    \"fit_start_time\": \"2008-01-01\",\n",
+    "    \"fit_end_time\": \"2014-12-31\",\n",
+    "    \"instruments\": 'csi100',\n",
+    "}\n",
+    "\n",
+    "dataset_template = {\n",
+    "        \"class\": \"DatasetH\",\n",
+    "        \"module_path\": \"qlib.data.dataset\",\n",
+    "        \"kwargs\": {\n",
+    "            \"handler\": {\n",
+    "                \"class\": \"Alpha158\",\n",
+    "                \"module_path\": \"qlib.contrib.data.handler\",\n",
+    "                \"kwargs\": data_handler_template,\n",
+    "            },\n",
+    "            \"segments\": {\n",
+    "                \"train\": (\"2008-01-01\", \"2014-12-31\"),\n",
+    "                \"valid\": (\"2015-01-01\", \"2016-12-31\"),\n",
+    "                \"test\": (\"2017-01-01\", \"2020-08-01\"),\n",
+    "            },\n",
+    "        },\n",
+    "    }\n",
+    "\n",
+    "record_template = [\n",
+    "    {\n",
+    "        \"class\": \"SignalRecord\",\n",
+    "        \"module_path\": \"qlib.workflow.record_temp\",\n",
+    "    },\n",
+    "    {\n",
+    "        \"class\": \"SigAnaRecord\",\n",
+    "        \"module_path\": \"qlib.workflow.record_temp\",\n",
+    "    }\n",
+    "]\n",
+    "\n",
+    "# use lgb\n",
+    "lgb_task_template = {\n",
+    "    \"model\": {\n",
+    "        \"class\": \"LGBModel\",\n",
+    "        \"module_path\": \"qlib.contrib.model.gbdt\",\n",
+    "    },\n",
+    "    \"dataset\": dataset_template,\n",
+    "    \"record\": record_template,\n",
+    "}\n",
+    "\n",
+    "# use xgboost\n",
+    "xgboost_task_template = {\n",
+    "    \"model\": {\n",
+    "        \"class\": \"XGBModel\",\n",
+    "        \"module_path\": \"qlib.contrib.model.xgboost\",\n",
+    "    },\n",
+    "    \"dataset\": dataset_template,\n",
+    "    \"record\": record_template,\n",
+    "}\n",
+    "\n",
+    "provider_uri = \"~/.qlib/qlib_data/cn_data\"  # target_dir\n",
+    "qlib.init(provider_uri=provider_uri, region=REG_CN)\n",
+    "\n",
+    "C[\"mongo\"] = {\n",
+    "    \"task_url\" : \"mongodb://localhost:27017/\", # maybe you need to change it to your url\n",
+    "    \"task_db_name\" : \"rolling_db3\"\n",
+    "}\n",
+    "\n",
+    "exp_name = 'rolling_exp3' # experiment name, will be used as the experiment in MLflow\n",
+    "task_pool = 'rolling_task3' # task pool name, will be used as the document in MongoDB"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 25,
+   "outputs": [
+    {
+     "output_type": "stream",
+     "name": "stdout",
+     "text": [
+      "[{'dataset': {'class': 'DatasetH',\n",
+      "              'kwargs': {'handler': {'class': 'Alpha158',\n",
+      "                                     'kwargs': {'end_time': '2020-08-01',\n",
+      "                                                'fit_end_time': '2014-12-31',\n",
+      "                                                'fit_start_time': '2008-01-01',\n",
+      "                                                'instruments': 'csi100',\n",
+      "                                                'start_time': '2008-01-01'},\n",
+      "                                     'module_path': 'qlib.contrib.data.handler'},\n",
+      "                         'segments': {'test': (Timestamp('2017-01-03 00:00:00'),\n",
+      "                                               Timestamp('2019-04-08 00:00:00')),\n",
+      "                                      'train': (Timestamp('2008-01-02 00:00:00'),\n",
+      "                                                Timestamp('2014-12-31 00:00:00')),\n",
+      "                                      'valid': (Timestamp('2015-01-05 00:00:00'),\n",
+      "                                                Timestamp('2016-12-30 00:00:00'))}},\n",
+      "              'module_path': 'qlib.data.dataset'},\n",
+      "  'model': {'class': 'XGBModel', 'module_path': 'qlib.contrib.model.xgboost'},\n",
+      "  'record': [{'class': 'SignalRecord',\n",
+      "              'module_path': 'qlib.workflow.record_temp'},\n",
+      "             {'class': 'SigAnaRecord',\n",
+      "              'module_path': 'qlib.workflow.record_temp'}],\n",
+      "  'task_key': 1},\n",
+      " {'dataset': {'class': 'DatasetH',\n",
+      "              'kwargs': {'handler': {'class': 'Alpha158',\n",
+      "                                     'kwargs': {'end_time': '2020-08-01',\n",
+      "                                                'fit_end_time': '2014-12-31',\n",
+      "                                                'fit_start_time': '2008-01-01',\n",
+      "                                                'instruments': 'csi100',\n",
+      "                                                'start_time': '2008-01-01'},\n",
+      "                                     'module_path': 'qlib.contrib.data.handler'},\n",
+      "                         'segments': {'test': (Timestamp('2019-04-09 00:00:00'),\n",
+      "                                               Timestamp('2021-07-12 00:00:00')),\n",
+      "                                      'train': (Timestamp('2010-04-23 00:00:00'),\n",
+      "                                                Timestamp('2017-05-24 00:00:00')),\n",
+      "                                      'valid': (Timestamp('2017-05-25 00:00:00'),\n",
+      "                                                Timestamp('2019-04-08 00:00:00'))}},\n",
+      "              'module_path': 'qlib.data.dataset'},\n",
+      "  'model': {'class': 'XGBModel', 'module_path': 'qlib.contrib.model.xgboost'},\n",
+      "  'record': [{'class': 'SignalRecord',\n",
+      "              'module_path': 'qlib.workflow.record_temp'},\n",
+      "             {'class': 'SigAnaRecord',\n",
+      "              'module_path': 'qlib.workflow.record_temp'}],\n",
+      "  'task_key': 1},\n",
+      " {'dataset': {'class': 'DatasetH',\n",
+      "              'kwargs': {'handler': {'class': 'Alpha158',\n",
+      "                                     'kwargs': {'end_time': '2020-08-01',\n",
+      "                                                'fit_end_time': '2014-12-31',\n",
+      "                                                'fit_start_time': '2008-01-01',\n",
+      "                                                'instruments': 'csi100',\n",
+      "                                                'start_time': '2008-01-01'},\n",
+      "                                     'module_path': 'qlib.contrib.data.handler'},\n",
+      "                         'segments': {'test': (Timestamp('2017-01-03 00:00:00'),\n",
+      "                                               Timestamp('2019-04-08 00:00:00')),\n",
+      "                                      'train': (Timestamp('2008-01-02 00:00:00'),\n",
+      "                                                Timestamp('2014-12-31 00:00:00')),\n",
+      "                                      'valid': (Timestamp('2015-01-05 00:00:00'),\n",
+      "                                                Timestamp('2016-12-30 00:00:00'))}},\n",
+      "              'module_path': 'qlib.data.dataset'},\n",
+      "  'model': {'class': 'LGBModel', 'module_path': 'qlib.contrib.model.gbdt'},\n",
+      "  'record': [{'class': 'SignalRecord',\n",
+      "              'module_path': 'qlib.workflow.record_temp'},\n",
+      "             {'class': 'SigAnaRecord',\n",
+      "              'module_path': 'qlib.workflow.record_temp'}],\n",
+      "  'task_key': 'task_lgb'},\n",
+      " {'dataset': {'class': 'DatasetH',\n",
+      "              'kwargs': {'handler': {'class': 'Alpha158',\n",
+      "                                     'kwargs': {'end_time': '2020-08-01',\n",
+      "                                                'fit_end_time': '2014-12-31',\n",
+      "                                                'fit_start_time': '2008-01-01',\n",
+      "                                                'instruments': 'csi100',\n",
+      "                                                'start_time': '2008-01-01'},\n",
+      "                                     'module_path': 'qlib.contrib.data.handler'},\n",
+      "                         'segments': {'test': (Timestamp('2019-04-09 00:00:00'),\n",
+      "                                               Timestamp('2021-07-12 00:00:00')),\n",
+      "                                      'train': (Timestamp('2010-04-23 00:00:00'),\n",
+      "                                                Timestamp('2017-05-24 00:00:00')),\n",
+      "                                      'valid': (Timestamp('2017-05-25 00:00:00'),\n",
+      "                                                Timestamp('2019-04-08 00:00:00'))}},\n",
+      "              'module_path': 'qlib.data.dataset'},\n",
+      "  'model': {'class': 'LGBModel', 'module_path': 'qlib.contrib.model.gbdt'},\n",
+      "  'record': [{'class': 'SignalRecord',\n",
+      "              'module_path': 'qlib.workflow.record_temp'},\n",
+      "             {'class': 'SigAnaRecord',\n",
+      "              'module_path': 'qlib.workflow.record_temp'}],\n",
+      "  'task_key': 'task_lgb'}]\n",
+      "Total Tasks, New Tasks: 4 0\n"
+     ]
+    }
+   ],
+   "source": [
+    "tasks = task_generator(\n",
+    "    xgboost_task_template, # default task name\n",
+    "    RollingGen(step=550,rtype=RollingGen.ROLL_SD), # generate different date segment\n",
+    "    task_lgb=lgb_task_template # use \"task_lgb\" as the task name\n",
+    ")\n",
+    "# Uncomment next two lines to see the generated tasks\n",
+    "from pprint import pprint\n",
+    "pprint(tasks)\n",
+    "tm = TaskManager(task_pool=task_pool)\n",
+    "tm.create_task(tasks) # all tasks will be saved to MongoDB"
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%%\n"
+    }
+   }
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 26,
+   "outputs": [
+    {
+     "output_type": "stream",
+     "name": "stderr",
+     "text": [
+      "2021-03-09 14:55:51.600 | INFO     | qlib.workflow.task.manage:run_task:355 - {'model': {'class': 'XGBModel', 'module_path': 'qlib.contrib.model.xgboost'}, 'dataset': {'class': 'DatasetH', 'module_path': 'qlib.data.dataset', 'kwargs': {'handler': {'class': 'Alpha158', 'module_path': 'qlib.contrib.data.handler', 'kwargs': {'start_time': '2008-01-01', 'end_time': '2020-08-01', 'fit_start_time': '2008-01-01', 'fit_end_time': '2014-12-31', 'instruments': 'csi100'}}, 'segments': {'train': (Timestamp('2008-01-02 00:00:00'), Timestamp('2014-12-31 00:00:00')), 'valid': (Timestamp('2015-01-05 00:00:00'), Timestamp('2016-12-30 00:00:00')), 'test': (Timestamp('2017-01-03 00:00:00'), Timestamp('2019-04-08 00:00:00'))}}}, 'record': [{'class': 'SignalRecord', 'module_path': 'qlib.workflow.record_temp'}, {'class': 'SigAnaRecord', 'module_path': 'qlib.workflow.record_temp'}], 'task_key': 1}\n",
+      "[8348:MainThread](2021-03-09 14:56:46,051) INFO - qlib.timer - [log.py:81] - Time cost: 54.448s | Loading data Done\n",
+      "[8348:MainThread](2021-03-09 14:56:46,440) INFO - qlib.timer - [log.py:81] - Time cost: 0.322s | DropnaLabel Done\n",
+      "[8348:MainThread](2021-03-09 14:56:52,461) INFO - qlib.timer - [log.py:81] - Time cost: 6.019s | CSZScoreNorm Done\n",
+      "[8348:MainThread](2021-03-09 14:56:52,464) INFO - qlib.timer - [log.py:81] - Time cost: 6.411s | fit & process data Done\n",
+      "[8348:MainThread](2021-03-09 14:56:52,468) INFO - qlib.timer - [log.py:81] - Time cost: 60.865s | Init data Done\n",
+      "[8348:MainThread](2021-03-09 14:56:52,471) INFO - qlib.workflow - [expm.py:245] - No tracking URI is provided. Use the default tracking URI.\n",
+      "[8348:MainThread](2021-03-09 14:56:52,500) INFO - qlib.workflow - [exp.py:181] - Experiment 2 starts running ...\n",
+      "[8348:MainThread](2021-03-09 14:56:52,567) INFO - qlib.workflow - [recorder.py:233] - Recorder dd6bceb6d319493686ab6565633c0b5a starts running under Experiment 2 ...\n",
+      "[0]\ttrain-rmse:1.05165\tvalid-rmse:1.05565\n",
+      "[20]\ttrain-rmse:0.97071\tvalid-rmse:1.00077\n",
+      "[40]\ttrain-rmse:0.95124\tvalid-rmse:1.00609\n",
+      "[59]\ttrain-rmse:0.93833\tvalid-rmse:1.00945\n",
+      "[8348:MainThread](2021-03-09 14:59:37,266) INFO - qlib.workflow - [record_temp.py:126] - Signal record 'pred.pkl' has been saved as the artifact of the Experiment 2\n",
+      "'The following are prediction results of the XGBModel model.'\n",
+      "                          score\n",
+      "datetime   instrument          \n",
+      "2017-01-03 SH600000   -0.103259\n",
+      "           SH600010   -0.084365\n",
+      "           SH600015   -0.107433\n",
+      "           SH600016   -0.064723\n",
+      "           SH600018   -0.038639\n",
+      "{'IC': 0.05347474869798698,\n",
+      " 'ICIR': 0.29781294430945265,\n",
+      " 'Rank IC': 0.0484064337863249,\n",
+      " 'Rank ICIR': 0.36035393716962033}\n",
+      "2021-03-09 14:59:38.633 | INFO     | qlib.workflow.task.manage:run_task:355 - {'model': {'class': 'XGBModel', 'module_path': 'qlib.contrib.model.xgboost'}, 'dataset': {'class': 'DatasetH', 'module_path': 'qlib.data.dataset', 'kwargs': {'handler': {'class': 'Alpha158', 'module_path': 'qlib.contrib.data.handler', 'kwargs': {'start_time': '2008-01-01', 'end_time': '2020-08-01', 'fit_start_time': '2008-01-01', 'fit_end_time': '2014-12-31', 'instruments': 'csi100'}}, 'segments': {'train': (Timestamp('2010-04-23 00:00:00'), Timestamp('2017-05-24 00:00:00')), 'valid': (Timestamp('2017-05-25 00:00:00'), Timestamp('2019-04-08 00:00:00')), 'test': (Timestamp('2019-04-09 00:00:00'), Timestamp('2021-07-12 00:00:00'))}}}, 'record': [{'class': 'SignalRecord', 'module_path': 'qlib.workflow.record_temp'}, {'class': 'SigAnaRecord', 'module_path': 'qlib.workflow.record_temp'}], 'task_key': 1}\n",
+      "[8348:MainThread](2021-03-09 15:00:36,591) INFO - qlib.timer - [log.py:81] - Time cost: 57.954s | Loading data Done\n",
+      "[8348:MainThread](2021-03-09 15:00:36,997) INFO - qlib.timer - [log.py:81] - Time cost: 0.338s | DropnaLabel Done\n",
+      "[8348:MainThread](2021-03-09 15:00:43,728) INFO - qlib.timer - [log.py:81] - Time cost: 6.728s | CSZScoreNorm Done\n",
+      "[8348:MainThread](2021-03-09 15:00:43,731) INFO - qlib.timer - [log.py:81] - Time cost: 7.137s | fit & process data Done\n",
+      "[8348:MainThread](2021-03-09 15:00:43,734) INFO - qlib.timer - [log.py:81] - Time cost: 65.097s | Init data Done\n",
+      "[8348:MainThread](2021-03-09 15:00:43,740) INFO - qlib.workflow - [expm.py:245] - No tracking URI is provided. Use the default tracking URI.\n",
+      "[8348:MainThread](2021-03-09 15:00:43,768) INFO - qlib.workflow - [exp.py:181] - Experiment 2 starts running ...\n",
+      "[8348:MainThread](2021-03-09 15:00:43,851) INFO - qlib.workflow - [recorder.py:233] - Recorder de2f892b569c436ba642a23e99f4f2b0 starts running under Experiment 2 ...\n",
+      "[0]\ttrain-rmse:1.05178\tvalid-rmse:1.05345\n",
+      "[20]\ttrain-rmse:0.96764\tvalid-rmse:0.99546\n",
+      "[40]\ttrain-rmse:0.94957\tvalid-rmse:0.99798\n",
+      "[57]\ttrain-rmse:0.93592\tvalid-rmse:1.00030\n",
+      "[8348:MainThread](2021-03-09 15:03:12,764) INFO - qlib.workflow - [record_temp.py:126] - Signal record 'pred.pkl' has been saved as the artifact of the Experiment 2\n",
+      "'The following are prediction results of the XGBModel model.'\n",
+      "                          score\n",
+      "datetime   instrument          \n",
+      "2019-04-09 SH600000    0.006996\n",
+      "           SH600009   -0.102482\n",
+      "           SH600010    0.016398\n",
+      "           SH600011    0.004459\n",
+      "           SH600015   -0.128315\n",
+      "{'IC': 0.013224093132176661,\n",
+      " 'ICIR': 0.08254897170570956,\n",
+      " 'Rank IC': 0.02472594591723197,\n",
+      " 'Rank ICIR': 0.16330982475433398}\n",
+      "2021-03-09 15:03:13.593 | INFO     | qlib.workflow.task.manage:run_task:355 - {'model': {'class': 'LGBModel', 'module_path': 'qlib.contrib.model.gbdt'}, 'dataset': {'class': 'DatasetH', 'module_path': 'qlib.data.dataset', 'kwargs': {'handler': {'class': 'Alpha158', 'module_path': 'qlib.contrib.data.handler', 'kwargs': {'start_time': '2008-01-01', 'end_time': '2020-08-01', 'fit_start_time': '2008-01-01', 'fit_end_time': '2014-12-31', 'instruments': 'csi100'}}, 'segments': {'train': (Timestamp('2008-01-02 00:00:00'), Timestamp('2014-12-31 00:00:00')), 'valid': (Timestamp('2015-01-05 00:00:00'), Timestamp('2016-12-30 00:00:00')), 'test': (Timestamp('2017-01-03 00:00:00'), Timestamp('2019-04-08 00:00:00'))}}}, 'record': [{'class': 'SignalRecord', 'module_path': 'qlib.workflow.record_temp'}, {'class': 'SigAnaRecord', 'module_path': 'qlib.workflow.record_temp'}], 'task_key': 'task_lgb'}\n",
+      "[8348:MainThread](2021-03-09 15:04:06,545) INFO - qlib.timer - [log.py:81] - Time cost: 52.814s | Loading data Done\n",
+      "[8348:MainThread](2021-03-09 15:04:06,919) INFO - qlib.timer - [log.py:81] - Time cost: 0.312s | DropnaLabel Done\n",
+      "[8348:MainThread](2021-03-09 15:04:12,850) INFO - qlib.timer - [log.py:81] - Time cost: 5.928s | CSZScoreNorm Done\n",
+      "[8348:MainThread](2021-03-09 15:04:12,853) INFO - qlib.timer - [log.py:81] - Time cost: 6.305s | fit & process data Done\n",
+      "[8348:MainThread](2021-03-09 15:04:12,856) INFO - qlib.timer - [log.py:81] - Time cost: 59.125s | Init data Done\n",
+      "[8348:MainThread](2021-03-09 15:04:12,859) INFO - qlib.workflow - [expm.py:245] - No tracking URI is provided. Use the default tracking URI.\n",
+      "[8348:MainThread](2021-03-09 15:04:12,888) INFO - qlib.workflow - [exp.py:181] - Experiment 2 starts running ...\n",
+      "[8348:MainThread](2021-03-09 15:04:12,958) INFO - qlib.workflow - [recorder.py:233] - Recorder 15df799127a74656829978c1b9352e60 starts running under Experiment 2 ...\n",
+      "Training until validation scores don't improve for 50 rounds\n",
+      "[20]\ttrain's l2: 0.970491\tvalid's l2: 0.987723\n",
+      "[40]\ttrain's l2: 0.957984\tvalid's l2: 0.990056\n",
+      "[60]\ttrain's l2: 0.947201\tvalid's l2: 0.991459\n",
+      "Early stopping, best iteration is:\n",
+      "[18]\ttrain's l2: 0.971834\tvalid's l2: 0.987481\n",
+      "[8348:MainThread](2021-03-09 15:04:19,847) INFO - qlib.workflow - [record_temp.py:126] - Signal record 'pred.pkl' has been saved as the artifact of the Experiment 2\n",
+      "'The following are prediction results of the LGBModel model.'\n",
+      "                          score\n",
+      "datetime   instrument          \n",
+      "2017-01-03 SH600000   -0.013089\n",
+      "           SH600010   -0.006642\n",
+      "           SH600015   -0.035137\n",
+      "           SH600016   -0.034634\n",
+      "           SH600018   -0.029493\n",
+      "{'IC': 0.05704431372255674,\n",
+      " 'ICIR': 0.28879437007622133,\n",
+      " 'Rank IC': 0.05181220321608411,\n",
+      " 'Rank ICIR': 0.3233833799543165}\n",
+      "2021-03-09 15:04:21.111 | INFO     | qlib.workflow.task.manage:run_task:355 - {'model': {'class': 'LGBModel', 'module_path': 'qlib.contrib.model.gbdt'}, 'dataset': {'class': 'DatasetH', 'module_path': 'qlib.data.dataset', 'kwargs': {'handler': {'class': 'Alpha158', 'module_path': 'qlib.contrib.data.handler', 'kwargs': {'start_time': '2008-01-01', 'end_time': '2020-08-01', 'fit_start_time': '2008-01-01', 'fit_end_time': '2014-12-31', 'instruments': 'csi100'}}, 'segments': {'train': (Timestamp('2010-04-23 00:00:00'), Timestamp('2017-05-24 00:00:00')), 'valid': (Timestamp('2017-05-25 00:00:00'), Timestamp('2019-04-08 00:00:00')), 'test': (Timestamp('2019-04-09 00:00:00'), Timestamp('2021-07-12 00:00:00'))}}}, 'record': [{'class': 'SignalRecord', 'module_path': 'qlib.workflow.record_temp'}, {'class': 'SigAnaRecord', 'module_path': 'qlib.workflow.record_temp'}], 'task_key': 'task_lgb'}\n",
+      "[8348:MainThread](2021-03-09 15:05:16,072) INFO - qlib.timer - [log.py:81] - Time cost: 54.958s | Loading data Done\n",
+      "[8348:MainThread](2021-03-09 15:05:16,466) INFO - qlib.timer - [log.py:81] - Time cost: 0.334s | DropnaLabel Done\n",
+      "[8348:MainThread](2021-03-09 15:05:22,281) INFO - qlib.timer - [log.py:81] - Time cost: 5.812s | CSZScoreNorm Done\n",
+      "[8348:MainThread](2021-03-09 15:05:22,283) INFO - qlib.timer - [log.py:81] - Time cost: 6.209s | fit & process data Done\n",
+      "[8348:MainThread](2021-03-09 15:05:22,286) INFO - qlib.timer - [log.py:81] - Time cost: 61.172s | Init data Done\n",
+      "[8348:MainThread](2021-03-09 15:05:22,291) INFO - qlib.workflow - [expm.py:245] - No tracking URI is provided. Use the default tracking URI.\n",
+      "[8348:MainThread](2021-03-09 15:05:22,317) INFO - qlib.workflow - [exp.py:181] - Experiment 2 starts running ...\n",
+      "[8348:MainThread](2021-03-09 15:05:22,386) INFO - qlib.workflow - [recorder.py:233] - Recorder 0c814539f55842b9b6310843fc5ec708 starts running under Experiment 2 ...\n",
+      "Training until validation scores don't improve for 50 rounds\n",
+      "[20]\ttrain's l2: 0.969033\tvalid's l2: 0.98571\n",
+      "[40]\ttrain's l2: 0.955399\tvalid's l2: 0.986164\n",
+      "[60]\ttrain's l2: 0.943514\tvalid's l2: 0.986301\n",
+      "Early stopping, best iteration is:\n",
+      "[26]\ttrain's l2: 0.964587\tvalid's l2: 0.985356\n",
+      "[8348:MainThread](2021-03-09 15:05:29,546) INFO - qlib.workflow - [record_temp.py:126] - Signal record 'pred.pkl' has been saved as the artifact of the Experiment 2\n",
+      "'The following are prediction results of the LGBModel model.'\n",
+      "                          score\n",
+      "datetime   instrument          \n",
+      "2019-04-09 SH600000    0.029586\n",
+      "           SH600009    0.004306\n",
+      "           SH600010   -0.004411\n",
+      "           SH600011    0.002707\n",
+      "           SH600015   -0.029124\n",
+      "{'IC': 0.020784811232504984,\n",
+      " 'ICIR': 0.11590182186569555,\n",
+      " 'Rank IC': 0.028925697036767055,\n",
+      " 'Rank ICIR': 0.16388058980901396}\n"
+     ]
+    },
+    {
+     "output_type": "execute_result",
+     "data": {
+      "text/plain": [
+       "True"
+      ]
+     },
+     "metadata": {},
+     "execution_count": 26
+    }
+   ],
+   "source": [
+    "from qlib.workflow.task.manage import run_task\n",
+    "from qlib.workflow.task.collect import TaskCollector\n",
+    "from qlib.model.trainer import task_train\n",
+    "\n",
+    "run_task(task_train, task_pool, experiment_name=exp_name) # all tasks will be trained using \"task_train\" method"
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%%\n"
+    }
+   }
+  },
+  {
+   "cell_type": "code",
+   "execution_count": 27,
+   "outputs": [
+    {
+     "output_type": "stream",
+     "name": "stderr",
+     "text": [
+      "Loading data: 100%|██████████| 4/4 [00:00<00:00, 37.38it/s]\n"
+     ]
+    },
+    {
+     "output_type": "execute_result",
+     "data": {
+      "text/plain": [
+       "{('task_lgb', '2019-04-08'): datetime    instrument\n",
+       " 2017-01-03  SH600000     -0.013089\n",
+       "             SH600010     -0.006642\n",
+       "             SH600015     -0.035137\n",
+       "             SH600016     -0.034634\n",
+       "             SH600018     -0.029493\n",
+       "                             ...   \n",
+       " 2019-04-08  SZ002415      0.049199\n",
+       "             SZ002450     -0.013450\n",
+       "             SZ002594      0.022395\n",
+       "             SZ002736      0.091433\n",
+       "             SZ300059     -0.016237\n",
+       " Name: score, Length: 55000, dtype: float64}"
+      ]
+     },
+     "metadata": {},
+     "execution_count": 27
+    }
+   ],
+   "source": [
+    "def get_task_key(task):\n",
+    "    task_key = task[\"task_key\"]\n",
+    "    rolling_end_timestamp = task[\"dataset\"][\"kwargs\"][\"segments\"][\"test\"][1]\n",
+    "    #rolling_end_datatime = rolling_end_timestamp.to_pydatetime()\n",
+    "    return task_key, rolling_end_timestamp.strftime('%Y-%m-%d')\n",
+    "\n",
+    "def my_filter(task):\n",
+    "    # only choose the results of \"task_lgb\" and test segment end in 2019 from all tasks\n",
+    "    task_key, rolling_end = get_task_key(task)\n",
+    "    if task_key==\"task_lgb\" and rolling_end.startswith('2019'):\n",
+    "        return True\n",
+    "    return False\n",
+    "\n",
+    "# name tasks by \"get_task_key\" and filter tasks by \"my_filter\"\n",
+    "pred_rolling = TaskCollector.collect(exp_name, get_task_key, my_filter) \n",
+    "pred_rolling"
+   ],
+   "metadata": {
+    "collapsed": false,
+    "pycharm": {
+     "name": "#%%\n"
+    }
+   }
+  }
+ ],
+ "metadata": {
+  "kernelspec": {
+   "display_name": "Python 3",
+   "language": "python",
+   "name": "python3"
+  },
+  "language_info": {
+   "codemirror_mode": {
+    "name": "ipython",
+    "version": 2
+   },
+   "file_extension": ".py",
+   "mimetype": "text/x-python",
+   "name": "python",
+   "nbconvert_exporter": "python",
+   "pygments_lexer": "ipython2",
+   "version": "3.6.5-final"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 0
+}
\ No newline at end of file
diff --git a/examples/taskmanager/task_manager_rolling.py b/examples/taskmanager/task_manager_rolling.py
new file mode 100644
index 000000000..7519bc4be
--- /dev/null
+++ b/examples/taskmanager/task_manager_rolling.py
@@ -0,0 +1,108 @@
+import qlib
+from qlib.config import REG_CN
+from qlib.workflow.task.gen import RollingGen, task_generator
+from qlib.workflow.task.manage import TaskManager
+from qlib.config import C
+
+data_handler_config = {
+    "start_time": "2008-01-01",
+    "end_time": "2020-08-01",
+    "fit_start_time": "2008-01-01",
+    "fit_end_time": "2014-12-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": ("2008-01-01", "2014-12-31"),
+                "valid": ("2015-01-01", "2016-12-31"),
+                "test": ("2017-01-01", "2020-08-01"),
+            },
+        },
+    }
+
+record_config = [
+    {
+        "class": "SignalRecord",
+        "module_path": "qlib.workflow.record_temp",
+    },
+    {
+        "class": "SigAnaRecord",
+        "module_path": "qlib.workflow.record_temp",
+    }
+]
+
+# use lgb
+task_lgb_config = {
+    "model": {
+        "class": "LGBModel",
+        "module_path": "qlib.contrib.model.gbdt",
+    },
+    "dataset": dataset_config,
+    "record": record_config,
+}
+
+# use xgboost
+task_xgboost_config = {
+    "model": {
+        "class": "XGBModel",
+        "module_path": "qlib.contrib.model.xgboost",
+    },
+    "dataset": dataset_config,
+    "record": record_config,
+}
+
+provider_uri = "~/.qlib/qlib_data/cn_data"  # target_dir
+qlib.init(provider_uri=provider_uri, region=REG_CN)
+
+C["mongo"] = {
+    "task_url" : "mongodb://localhost:27017/", # maybe you need to change it to your url
+    "task_db_name" : "rolling_db"
+}
+
+exp_name = 'rolling_exp' # experiment name, will be used as the experiment in MLflow
+task_pool = 'rolling_task' # task pool name, will be used as the document in MongoDB
+
+tasks = task_generator(
+    task_xgboost_config, # default task name
+    RollingGen(step=550,rtype=RollingGen.ROLL_SD), # generate different date segment
+    task_lgb=task_lgb_config # use "task_lgb" as the task name
+)
+
+# Uncomment next two lines to see the generated tasks
+# from pprint import pprint
+# pprint(tasks)
+
+tm = TaskManager(task_pool=task_pool)
+tm.create_task(tasks) # all tasks will be saved to MongoDB
+
+from qlib.workflow.task.manage import run_task
+from qlib.workflow.task.collect import RollingCollector
+from qlib.model.trainer import task_train
+
+run_task(task_train, task_pool, experiment_name=exp_name) # all tasks will be trained using "task_train" method
+
+def get_task_key(task_config):
+    task_key = task_config["task_key"]
+    rolling_end_timestamp = task_config["dataset"]["kwargs"]["segments"]["test"][1]
+    #rolling_end_datatime = rolling_end_timestamp.to_pydatetime()
+    return task_key, rolling_end_timestamp.strftime('%Y-%m-%d')
+
+def my_filter(task_config):
+    # only choose the results of "task_lgb" and test in 2019 from all tasks
+    task_key, rolling_end = get_task_key(task_config)
+    if task_key=="task_lgb" and rolling_end.startswith('2019'):
+        return True
+    return False
+
+collector = RollingCollector(get_task_key, my_filter)
+pred_rolling = collector(exp_name) # name tasks by "get_task_key" and filter tasks by "my_filter"
+print(pred_rolling)
\ No newline at end of file
diff --git a/examples/workflow_task_rolling.ipynb b/examples/workflow_task_rolling.ipynb
deleted file mode 100644
index c2d399be0..000000000
--- a/examples/workflow_task_rolling.ipynb
+++ /dev/null
@@ -1,177 +0,0 @@
-{
- "cells": [
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "metadata": {
-    "collapsed": true
-   },
-   "outputs": [],
-   "source": [
-    "import qlib\n",
-    "from qlib.config import REG_CN\n",
-    "from qlib.workflow.task.gen import RollingGen, task_generator\n",
-    "from qlib.workflow.task.manage import TaskManager\n",
-    "from qlib.config import C\n",
-    "\n",
-    "data_handler_config = {\n",
-    "    \"start_time\": \"2008-01-01\",\n",
-    "    \"end_time\": \"2020-08-01\",\n",
-    "    \"fit_start_time\": \"2008-01-01\",\n",
-    "    \"fit_end_time\": \"2014-12-31\",\n",
-    "    \"instruments\": 'csi100',\n",
-    "}\n",
-    "\n",
-    "dataset_config = {\n",
-    "        \"class\": \"DatasetH\",\n",
-    "        \"module_path\": \"qlib.data.dataset\",\n",
-    "        \"kwargs\": {\n",
-    "            \"handler\": {\n",
-    "                \"class\": \"Alpha158\",\n",
-    "                \"module_path\": \"qlib.contrib.data.handler\",\n",
-    "                \"kwargs\": data_handler_config,\n",
-    "            },\n",
-    "            \"segments\": {\n",
-    "                \"train\": (\"2008-01-01\", \"2014-12-31\"),\n",
-    "                \"valid\": (\"2015-01-01\", \"2016-12-31\"),\n",
-    "                \"test\": (\"2017-01-01\", \"2020-08-01\"),\n",
-    "            },\n",
-    "        },\n",
-    "    }\n",
-    "\n",
-    "record_config = [\n",
-    "    {\n",
-    "        \"class\": \"SignalRecord\",\n",
-    "        \"module_path\": \"qlib.workflow.record_temp\",\n",
-    "    },\n",
-    "    {\n",
-    "        \"class\": \"SigAnaRecord\",\n",
-    "        \"module_path\": \"qlib.workflow.record_temp\",\n",
-    "    }\n",
-    "]\n",
-    "\n",
-    "# use lgb\n",
-    "task_lgb_config = {\n",
-    "    \"model\": {\n",
-    "        \"class\": \"LGBModel\",\n",
-    "        \"module_path\": \"qlib.contrib.model.gbdt\",\n",
-    "    },\n",
-    "    \"dataset\": dataset_config,\n",
-    "    \"record\": record_config,\n",
-    "}\n",
-    "\n",
-    "# use xgboost\n",
-    "task_xgboost_config = {\n",
-    "    \"model\": {\n",
-    "        \"class\": \"XGBModel\",\n",
-    "        \"module_path\": \"qlib.contrib.model.xgboost\",\n",
-    "    },\n",
-    "    \"dataset\": dataset_config,\n",
-    "    \"record\": record_config,\n",
-    "}\n",
-    "provider_uri = r\"../qlib-main/qlib_data/cn_data\"\n",
-    "#provider_uri = \"~/.qlib/qlib_data/cn_data\"  # target_dir\n",
-    "qlib.init(provider_uri=provider_uri, region=REG_CN)\n",
-    "\n",
-    "C[\"mongo\"] = {\n",
-    "    \"task_url\" : \"mongodb://localhost:27017/\", # maybe you need to change it to your url\n",
-    "    \"task_db_name\" : \"rolling_db\"\n",
-    "}\n",
-    "\n",
-    "exp_name = 'rolling_exp' # experiment name, will be used as the experiment in MLflow\n",
-    "task_pool = 'rolling_task' # task pool name, will be used as the document in MongoDB"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "outputs": [],
-   "source": [
-    "tasks = task_generator(\n",
-    "    task_xgboost_config, # default task name\n",
-    "    RollingGen(step=550,rtype=RollingGen.ROLL_SD), # generate different date segment\n",
-    "    task_lgb=task_lgb_config # use \"task_lgb\" as the task name\n",
-    ")\n",
-    "# Uncomment next two lines to see the generated tasks\n",
-    "# from pprint import pprint\n",
-    "# pprint(tasks)\n",
-    "tm = TaskManager(task_pool=task_pool)\n",
-    "tm.create_task(tasks) # all tasks will be saved to MongoDB"
-   ],
-   "metadata": {
-    "collapsed": false,
-    "pycharm": {
-     "name": "#%%\n"
-    }
-   }
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "outputs": [],
-   "source": [
-    "from qlib.workflow.task.manage import run_task\n",
-    "from qlib.workflow.task.collect import RollingCollector\n",
-    "from qlib.model.trainer import task_train\n",
-    "\n",
-    "run_task(task_train, task_pool, experiment_name=exp_name) # all tasks will be trained using \"task_train\" method"
-   ],
-   "metadata": {
-    "collapsed": false,
-    "pycharm": {
-     "name": "#%%\n"
-    }
-   }
-  },
-  {
-   "cell_type": "code",
-   "execution_count": null,
-   "outputs": [],
-   "source": [
-    "def get_task_key(task_config):\n",
-    "    task_key = task_config[\"task_key\"]\n",
-    "    rolling_end_timestamp = task_config[\"dataset\"][\"kwargs\"][\"segments\"][\"test\"][1]\n",
-    "    rolling_end_datatime = rolling_end_timestamp.to_pydatetime()\n",
-    "    return task_key, rolling_end_datatime.strftime('%Y-%m-%d')\n",
-    "\n",
-    "def my_filter(task_config):\n",
-    "    # only choose the results of \"task_lgb\" and test in 2019 from all tasks\n",
-    "    task_key, rolling_end = get_task_key(task_config)\n",
-    "    if task_key==\"task_lgb\" and rolling_end.startswith('2019'):\n",
-    "        return True\n",
-    "    return False\n",
-    "\n",
-    "collector = RollingCollector(get_task_key, my_filter)\n",
-    "pred_rolling = collector(exp_name) # name tasks by \"get_task_key\" and filter tasks by \"my_filter\"\n",
-    "pred_rolling"
-   ],
-   "metadata": {
-    "collapsed": false,
-    "pycharm": {
-     "name": "#%%\n"
-    }
-   }
-  }
- ],
- "metadata": {
-  "kernelspec": {
-   "display_name": "Python 3",
-   "language": "python",
-   "name": "python3"
-  },
-  "language_info": {
-   "codemirror_mode": {
-    "name": "ipython",
-    "version": 2
-   },
-   "file_extension": ".py",
-   "mimetype": "text/x-python",
-   "name": "python",
-   "nbconvert_exporter": "python",
-   "pygments_lexer": "ipython2",
-   "version": "2.7.6"
-  }
- },
- "nbformat": 4,
- "nbformat_minor": 0
-}
\ No newline at end of file

From e2f58274ba91f1ef43e8bc87b6cc04e67416137b Mon Sep 17 00:00:00 2001
From: lzh222333 <lzh222333@163.com>
Date: Wed, 10 Mar 2021 10:58:49 +0000
Subject: [PATCH 08/10] update task manager

---
 .../taskmanager/task_manager_rolling.ipynb    | 297 +-----------------
 examples/taskmanager/task_manager_rolling.py  |   9 +-
 qlib/model/trainer.py                         |  10 +-
 qlib/workflow/task/collect.py                 |  15 +-
 qlib/workflow/task/gen.py                     |   2 +-
 5 files changed, 34 insertions(+), 299 deletions(-)

diff --git a/examples/taskmanager/task_manager_rolling.ipynb b/examples/taskmanager/task_manager_rolling.ipynb
index 43ae5b1d1..e8ec8d4a7 100644
--- a/examples/taskmanager/task_manager_rolling.ipynb
+++ b/examples/taskmanager/task_manager_rolling.ipynb
@@ -2,32 +2,11 @@
  "cells": [
   {
    "cell_type": "code",
-   "execution_count": 23,
-   "metadata": {},
-   "outputs": [],
-   "source": [
-    "import mlflow\n",
-    "mlflow.end_run()"
-   ]
-  },
-  {
-   "cell_type": "code",
-   "execution_count": 24,
+   "execution_count": null,
    "metadata": {
     "collapsed": true
    },
-   "outputs": [
-    {
-     "output_type": "stream",
-     "name": "stderr",
-     "text": [
-      "[8348:MainThread](2021-03-09 14:55:48,543) INFO - qlib.Initialization - [config.py:279] - default_conf: client.\n",
-      "[8348:MainThread](2021-03-09 14:55:50,592) WARNING - qlib.Initialization - [config.py:295] - redis connection failed(host=127.0.0.1 port=6379), cache will not be used!\n",
-      "[8348:MainThread](2021-03-09 14:55:50,597) INFO - qlib.Initialization - [__init__.py:48] - qlib successfully initialized based on client settings.\n",
-      "[8348:MainThread](2021-03-09 14:55:50,601) INFO - qlib.Initialization - [__init__.py:49] - data_path=C:\\Users\\lzh222333\\.qlib\\qlib_data\\cn_data\n"
-     ]
-    }
-   ],
+   "outputs": [],
    "source": [
     "import qlib\n",
     "from qlib.config import REG_CN\n",
@@ -96,109 +75,17 @@
     "\n",
     "C[\"mongo\"] = {\n",
     "    \"task_url\" : \"mongodb://localhost:27017/\", # maybe you need to change it to your url\n",
-    "    \"task_db_name\" : \"rolling_db3\"\n",
+    "    \"task_db_name\" : \"rolling_db\"\n",
     "}\n",
     "\n",
-    "exp_name = 'rolling_exp3' # experiment name, will be used as the experiment in MLflow\n",
-    "task_pool = 'rolling_task3' # task pool name, will be used as the document in MongoDB"
+    "exp_name = 'rolling_exp' # experiment name, will be used as the experiment in MLflow\n",
+    "task_pool = 'rolling_task' # task pool name, will be used as the document in MongoDB"
    ]
   },
   {
    "cell_type": "code",
-   "execution_count": 25,
-   "outputs": [
-    {
-     "output_type": "stream",
-     "name": "stdout",
-     "text": [
-      "[{'dataset': {'class': 'DatasetH',\n",
-      "              'kwargs': {'handler': {'class': 'Alpha158',\n",
-      "                                     'kwargs': {'end_time': '2020-08-01',\n",
-      "                                                'fit_end_time': '2014-12-31',\n",
-      "                                                'fit_start_time': '2008-01-01',\n",
-      "                                                'instruments': 'csi100',\n",
-      "                                                'start_time': '2008-01-01'},\n",
-      "                                     'module_path': 'qlib.contrib.data.handler'},\n",
-      "                         'segments': {'test': (Timestamp('2017-01-03 00:00:00'),\n",
-      "                                               Timestamp('2019-04-08 00:00:00')),\n",
-      "                                      'train': (Timestamp('2008-01-02 00:00:00'),\n",
-      "                                                Timestamp('2014-12-31 00:00:00')),\n",
-      "                                      'valid': (Timestamp('2015-01-05 00:00:00'),\n",
-      "                                                Timestamp('2016-12-30 00:00:00'))}},\n",
-      "              'module_path': 'qlib.data.dataset'},\n",
-      "  'model': {'class': 'XGBModel', 'module_path': 'qlib.contrib.model.xgboost'},\n",
-      "  'record': [{'class': 'SignalRecord',\n",
-      "              'module_path': 'qlib.workflow.record_temp'},\n",
-      "             {'class': 'SigAnaRecord',\n",
-      "              'module_path': 'qlib.workflow.record_temp'}],\n",
-      "  'task_key': 1},\n",
-      " {'dataset': {'class': 'DatasetH',\n",
-      "              'kwargs': {'handler': {'class': 'Alpha158',\n",
-      "                                     'kwargs': {'end_time': '2020-08-01',\n",
-      "                                                'fit_end_time': '2014-12-31',\n",
-      "                                                'fit_start_time': '2008-01-01',\n",
-      "                                                'instruments': 'csi100',\n",
-      "                                                'start_time': '2008-01-01'},\n",
-      "                                     'module_path': 'qlib.contrib.data.handler'},\n",
-      "                         'segments': {'test': (Timestamp('2019-04-09 00:00:00'),\n",
-      "                                               Timestamp('2021-07-12 00:00:00')),\n",
-      "                                      'train': (Timestamp('2010-04-23 00:00:00'),\n",
-      "                                                Timestamp('2017-05-24 00:00:00')),\n",
-      "                                      'valid': (Timestamp('2017-05-25 00:00:00'),\n",
-      "                                                Timestamp('2019-04-08 00:00:00'))}},\n",
-      "              'module_path': 'qlib.data.dataset'},\n",
-      "  'model': {'class': 'XGBModel', 'module_path': 'qlib.contrib.model.xgboost'},\n",
-      "  'record': [{'class': 'SignalRecord',\n",
-      "              'module_path': 'qlib.workflow.record_temp'},\n",
-      "             {'class': 'SigAnaRecord',\n",
-      "              'module_path': 'qlib.workflow.record_temp'}],\n",
-      "  'task_key': 1},\n",
-      " {'dataset': {'class': 'DatasetH',\n",
-      "              'kwargs': {'handler': {'class': 'Alpha158',\n",
-      "                                     'kwargs': {'end_time': '2020-08-01',\n",
-      "                                                'fit_end_time': '2014-12-31',\n",
-      "                                                'fit_start_time': '2008-01-01',\n",
-      "                                                'instruments': 'csi100',\n",
-      "                                                'start_time': '2008-01-01'},\n",
-      "                                     'module_path': 'qlib.contrib.data.handler'},\n",
-      "                         'segments': {'test': (Timestamp('2017-01-03 00:00:00'),\n",
-      "                                               Timestamp('2019-04-08 00:00:00')),\n",
-      "                                      'train': (Timestamp('2008-01-02 00:00:00'),\n",
-      "                                                Timestamp('2014-12-31 00:00:00')),\n",
-      "                                      'valid': (Timestamp('2015-01-05 00:00:00'),\n",
-      "                                                Timestamp('2016-12-30 00:00:00'))}},\n",
-      "              'module_path': 'qlib.data.dataset'},\n",
-      "  'model': {'class': 'LGBModel', 'module_path': 'qlib.contrib.model.gbdt'},\n",
-      "  'record': [{'class': 'SignalRecord',\n",
-      "              'module_path': 'qlib.workflow.record_temp'},\n",
-      "             {'class': 'SigAnaRecord',\n",
-      "              'module_path': 'qlib.workflow.record_temp'}],\n",
-      "  'task_key': 'task_lgb'},\n",
-      " {'dataset': {'class': 'DatasetH',\n",
-      "              'kwargs': {'handler': {'class': 'Alpha158',\n",
-      "                                     'kwargs': {'end_time': '2020-08-01',\n",
-      "                                                'fit_end_time': '2014-12-31',\n",
-      "                                                'fit_start_time': '2008-01-01',\n",
-      "                                                'instruments': 'csi100',\n",
-      "                                                'start_time': '2008-01-01'},\n",
-      "                                     'module_path': 'qlib.contrib.data.handler'},\n",
-      "                         'segments': {'test': (Timestamp('2019-04-09 00:00:00'),\n",
-      "                                               Timestamp('2021-07-12 00:00:00')),\n",
-      "                                      'train': (Timestamp('2010-04-23 00:00:00'),\n",
-      "                                                Timestamp('2017-05-24 00:00:00')),\n",
-      "                                      'valid': (Timestamp('2017-05-25 00:00:00'),\n",
-      "                                                Timestamp('2019-04-08 00:00:00'))}},\n",
-      "              'module_path': 'qlib.data.dataset'},\n",
-      "  'model': {'class': 'LGBModel', 'module_path': 'qlib.contrib.model.gbdt'},\n",
-      "  'record': [{'class': 'SignalRecord',\n",
-      "              'module_path': 'qlib.workflow.record_temp'},\n",
-      "             {'class': 'SigAnaRecord',\n",
-      "              'module_path': 'qlib.workflow.record_temp'}],\n",
-      "  'task_key': 'task_lgb'}]\n",
-      "Total Tasks, New Tasks: 4 0\n"
-     ]
-    }
-   ],
+   "execution_count": null,
+   "outputs": [],
    "source": [
     "tasks = task_generator(\n",
     "    xgboost_task_template, # default task name\n",
@@ -206,8 +93,8 @@
     "    task_lgb=lgb_task_template # use \"task_lgb\" as the task name\n",
     ")\n",
     "# Uncomment next two lines to see the generated tasks\n",
-    "from pprint import pprint\n",
-    "pprint(tasks)\n",
+    "# from pprint import pprint\n",
+    "# pprint(tasks)\n",
     "tm = TaskManager(task_pool=task_pool)\n",
     "tm.create_task(tasks) # all tasks will be saved to MongoDB"
    ],
@@ -220,133 +107,8 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 26,
-   "outputs": [
-    {
-     "output_type": "stream",
-     "name": "stderr",
-     "text": [
-      "2021-03-09 14:55:51.600 | INFO     | qlib.workflow.task.manage:run_task:355 - {'model': {'class': 'XGBModel', 'module_path': 'qlib.contrib.model.xgboost'}, 'dataset': {'class': 'DatasetH', 'module_path': 'qlib.data.dataset', 'kwargs': {'handler': {'class': 'Alpha158', 'module_path': 'qlib.contrib.data.handler', 'kwargs': {'start_time': '2008-01-01', 'end_time': '2020-08-01', 'fit_start_time': '2008-01-01', 'fit_end_time': '2014-12-31', 'instruments': 'csi100'}}, 'segments': {'train': (Timestamp('2008-01-02 00:00:00'), Timestamp('2014-12-31 00:00:00')), 'valid': (Timestamp('2015-01-05 00:00:00'), Timestamp('2016-12-30 00:00:00')), 'test': (Timestamp('2017-01-03 00:00:00'), Timestamp('2019-04-08 00:00:00'))}}}, 'record': [{'class': 'SignalRecord', 'module_path': 'qlib.workflow.record_temp'}, {'class': 'SigAnaRecord', 'module_path': 'qlib.workflow.record_temp'}], 'task_key': 1}\n",
-      "[8348:MainThread](2021-03-09 14:56:46,051) INFO - qlib.timer - [log.py:81] - Time cost: 54.448s | Loading data Done\n",
-      "[8348:MainThread](2021-03-09 14:56:46,440) INFO - qlib.timer - [log.py:81] - Time cost: 0.322s | DropnaLabel Done\n",
-      "[8348:MainThread](2021-03-09 14:56:52,461) INFO - qlib.timer - [log.py:81] - Time cost: 6.019s | CSZScoreNorm Done\n",
-      "[8348:MainThread](2021-03-09 14:56:52,464) INFO - qlib.timer - [log.py:81] - Time cost: 6.411s | fit & process data Done\n",
-      "[8348:MainThread](2021-03-09 14:56:52,468) INFO - qlib.timer - [log.py:81] - Time cost: 60.865s | Init data Done\n",
-      "[8348:MainThread](2021-03-09 14:56:52,471) INFO - qlib.workflow - [expm.py:245] - No tracking URI is provided. Use the default tracking URI.\n",
-      "[8348:MainThread](2021-03-09 14:56:52,500) INFO - qlib.workflow - [exp.py:181] - Experiment 2 starts running ...\n",
-      "[8348:MainThread](2021-03-09 14:56:52,567) INFO - qlib.workflow - [recorder.py:233] - Recorder dd6bceb6d319493686ab6565633c0b5a starts running under Experiment 2 ...\n",
-      "[0]\ttrain-rmse:1.05165\tvalid-rmse:1.05565\n",
-      "[20]\ttrain-rmse:0.97071\tvalid-rmse:1.00077\n",
-      "[40]\ttrain-rmse:0.95124\tvalid-rmse:1.00609\n",
-      "[59]\ttrain-rmse:0.93833\tvalid-rmse:1.00945\n",
-      "[8348:MainThread](2021-03-09 14:59:37,266) INFO - qlib.workflow - [record_temp.py:126] - Signal record 'pred.pkl' has been saved as the artifact of the Experiment 2\n",
-      "'The following are prediction results of the XGBModel model.'\n",
-      "                          score\n",
-      "datetime   instrument          \n",
-      "2017-01-03 SH600000   -0.103259\n",
-      "           SH600010   -0.084365\n",
-      "           SH600015   -0.107433\n",
-      "           SH600016   -0.064723\n",
-      "           SH600018   -0.038639\n",
-      "{'IC': 0.05347474869798698,\n",
-      " 'ICIR': 0.29781294430945265,\n",
-      " 'Rank IC': 0.0484064337863249,\n",
-      " 'Rank ICIR': 0.36035393716962033}\n",
-      "2021-03-09 14:59:38.633 | INFO     | qlib.workflow.task.manage:run_task:355 - {'model': {'class': 'XGBModel', 'module_path': 'qlib.contrib.model.xgboost'}, 'dataset': {'class': 'DatasetH', 'module_path': 'qlib.data.dataset', 'kwargs': {'handler': {'class': 'Alpha158', 'module_path': 'qlib.contrib.data.handler', 'kwargs': {'start_time': '2008-01-01', 'end_time': '2020-08-01', 'fit_start_time': '2008-01-01', 'fit_end_time': '2014-12-31', 'instruments': 'csi100'}}, 'segments': {'train': (Timestamp('2010-04-23 00:00:00'), Timestamp('2017-05-24 00:00:00')), 'valid': (Timestamp('2017-05-25 00:00:00'), Timestamp('2019-04-08 00:00:00')), 'test': (Timestamp('2019-04-09 00:00:00'), Timestamp('2021-07-12 00:00:00'))}}}, 'record': [{'class': 'SignalRecord', 'module_path': 'qlib.workflow.record_temp'}, {'class': 'SigAnaRecord', 'module_path': 'qlib.workflow.record_temp'}], 'task_key': 1}\n",
-      "[8348:MainThread](2021-03-09 15:00:36,591) INFO - qlib.timer - [log.py:81] - Time cost: 57.954s | Loading data Done\n",
-      "[8348:MainThread](2021-03-09 15:00:36,997) INFO - qlib.timer - [log.py:81] - Time cost: 0.338s | DropnaLabel Done\n",
-      "[8348:MainThread](2021-03-09 15:00:43,728) INFO - qlib.timer - [log.py:81] - Time cost: 6.728s | CSZScoreNorm Done\n",
-      "[8348:MainThread](2021-03-09 15:00:43,731) INFO - qlib.timer - [log.py:81] - Time cost: 7.137s | fit & process data Done\n",
-      "[8348:MainThread](2021-03-09 15:00:43,734) INFO - qlib.timer - [log.py:81] - Time cost: 65.097s | Init data Done\n",
-      "[8348:MainThread](2021-03-09 15:00:43,740) INFO - qlib.workflow - [expm.py:245] - No tracking URI is provided. Use the default tracking URI.\n",
-      "[8348:MainThread](2021-03-09 15:00:43,768) INFO - qlib.workflow - [exp.py:181] - Experiment 2 starts running ...\n",
-      "[8348:MainThread](2021-03-09 15:00:43,851) INFO - qlib.workflow - [recorder.py:233] - Recorder de2f892b569c436ba642a23e99f4f2b0 starts running under Experiment 2 ...\n",
-      "[0]\ttrain-rmse:1.05178\tvalid-rmse:1.05345\n",
-      "[20]\ttrain-rmse:0.96764\tvalid-rmse:0.99546\n",
-      "[40]\ttrain-rmse:0.94957\tvalid-rmse:0.99798\n",
-      "[57]\ttrain-rmse:0.93592\tvalid-rmse:1.00030\n",
-      "[8348:MainThread](2021-03-09 15:03:12,764) INFO - qlib.workflow - [record_temp.py:126] - Signal record 'pred.pkl' has been saved as the artifact of the Experiment 2\n",
-      "'The following are prediction results of the XGBModel model.'\n",
-      "                          score\n",
-      "datetime   instrument          \n",
-      "2019-04-09 SH600000    0.006996\n",
-      "           SH600009   -0.102482\n",
-      "           SH600010    0.016398\n",
-      "           SH600011    0.004459\n",
-      "           SH600015   -0.128315\n",
-      "{'IC': 0.013224093132176661,\n",
-      " 'ICIR': 0.08254897170570956,\n",
-      " 'Rank IC': 0.02472594591723197,\n",
-      " 'Rank ICIR': 0.16330982475433398}\n",
-      "2021-03-09 15:03:13.593 | INFO     | qlib.workflow.task.manage:run_task:355 - {'model': {'class': 'LGBModel', 'module_path': 'qlib.contrib.model.gbdt'}, 'dataset': {'class': 'DatasetH', 'module_path': 'qlib.data.dataset', 'kwargs': {'handler': {'class': 'Alpha158', 'module_path': 'qlib.contrib.data.handler', 'kwargs': {'start_time': '2008-01-01', 'end_time': '2020-08-01', 'fit_start_time': '2008-01-01', 'fit_end_time': '2014-12-31', 'instruments': 'csi100'}}, 'segments': {'train': (Timestamp('2008-01-02 00:00:00'), Timestamp('2014-12-31 00:00:00')), 'valid': (Timestamp('2015-01-05 00:00:00'), Timestamp('2016-12-30 00:00:00')), 'test': (Timestamp('2017-01-03 00:00:00'), Timestamp('2019-04-08 00:00:00'))}}}, 'record': [{'class': 'SignalRecord', 'module_path': 'qlib.workflow.record_temp'}, {'class': 'SigAnaRecord', 'module_path': 'qlib.workflow.record_temp'}], 'task_key': 'task_lgb'}\n",
-      "[8348:MainThread](2021-03-09 15:04:06,545) INFO - qlib.timer - [log.py:81] - Time cost: 52.814s | Loading data Done\n",
-      "[8348:MainThread](2021-03-09 15:04:06,919) INFO - qlib.timer - [log.py:81] - Time cost: 0.312s | DropnaLabel Done\n",
-      "[8348:MainThread](2021-03-09 15:04:12,850) INFO - qlib.timer - [log.py:81] - Time cost: 5.928s | CSZScoreNorm Done\n",
-      "[8348:MainThread](2021-03-09 15:04:12,853) INFO - qlib.timer - [log.py:81] - Time cost: 6.305s | fit & process data Done\n",
-      "[8348:MainThread](2021-03-09 15:04:12,856) INFO - qlib.timer - [log.py:81] - Time cost: 59.125s | Init data Done\n",
-      "[8348:MainThread](2021-03-09 15:04:12,859) INFO - qlib.workflow - [expm.py:245] - No tracking URI is provided. Use the default tracking URI.\n",
-      "[8348:MainThread](2021-03-09 15:04:12,888) INFO - qlib.workflow - [exp.py:181] - Experiment 2 starts running ...\n",
-      "[8348:MainThread](2021-03-09 15:04:12,958) INFO - qlib.workflow - [recorder.py:233] - Recorder 15df799127a74656829978c1b9352e60 starts running under Experiment 2 ...\n",
-      "Training until validation scores don't improve for 50 rounds\n",
-      "[20]\ttrain's l2: 0.970491\tvalid's l2: 0.987723\n",
-      "[40]\ttrain's l2: 0.957984\tvalid's l2: 0.990056\n",
-      "[60]\ttrain's l2: 0.947201\tvalid's l2: 0.991459\n",
-      "Early stopping, best iteration is:\n",
-      "[18]\ttrain's l2: 0.971834\tvalid's l2: 0.987481\n",
-      "[8348:MainThread](2021-03-09 15:04:19,847) INFO - qlib.workflow - [record_temp.py:126] - Signal record 'pred.pkl' has been saved as the artifact of the Experiment 2\n",
-      "'The following are prediction results of the LGBModel model.'\n",
-      "                          score\n",
-      "datetime   instrument          \n",
-      "2017-01-03 SH600000   -0.013089\n",
-      "           SH600010   -0.006642\n",
-      "           SH600015   -0.035137\n",
-      "           SH600016   -0.034634\n",
-      "           SH600018   -0.029493\n",
-      "{'IC': 0.05704431372255674,\n",
-      " 'ICIR': 0.28879437007622133,\n",
-      " 'Rank IC': 0.05181220321608411,\n",
-      " 'Rank ICIR': 0.3233833799543165}\n",
-      "2021-03-09 15:04:21.111 | INFO     | qlib.workflow.task.manage:run_task:355 - {'model': {'class': 'LGBModel', 'module_path': 'qlib.contrib.model.gbdt'}, 'dataset': {'class': 'DatasetH', 'module_path': 'qlib.data.dataset', 'kwargs': {'handler': {'class': 'Alpha158', 'module_path': 'qlib.contrib.data.handler', 'kwargs': {'start_time': '2008-01-01', 'end_time': '2020-08-01', 'fit_start_time': '2008-01-01', 'fit_end_time': '2014-12-31', 'instruments': 'csi100'}}, 'segments': {'train': (Timestamp('2010-04-23 00:00:00'), Timestamp('2017-05-24 00:00:00')), 'valid': (Timestamp('2017-05-25 00:00:00'), Timestamp('2019-04-08 00:00:00')), 'test': (Timestamp('2019-04-09 00:00:00'), Timestamp('2021-07-12 00:00:00'))}}}, 'record': [{'class': 'SignalRecord', 'module_path': 'qlib.workflow.record_temp'}, {'class': 'SigAnaRecord', 'module_path': 'qlib.workflow.record_temp'}], 'task_key': 'task_lgb'}\n",
-      "[8348:MainThread](2021-03-09 15:05:16,072) INFO - qlib.timer - [log.py:81] - Time cost: 54.958s | Loading data Done\n",
-      "[8348:MainThread](2021-03-09 15:05:16,466) INFO - qlib.timer - [log.py:81] - Time cost: 0.334s | DropnaLabel Done\n",
-      "[8348:MainThread](2021-03-09 15:05:22,281) INFO - qlib.timer - [log.py:81] - Time cost: 5.812s | CSZScoreNorm Done\n",
-      "[8348:MainThread](2021-03-09 15:05:22,283) INFO - qlib.timer - [log.py:81] - Time cost: 6.209s | fit & process data Done\n",
-      "[8348:MainThread](2021-03-09 15:05:22,286) INFO - qlib.timer - [log.py:81] - Time cost: 61.172s | Init data Done\n",
-      "[8348:MainThread](2021-03-09 15:05:22,291) INFO - qlib.workflow - [expm.py:245] - No tracking URI is provided. Use the default tracking URI.\n",
-      "[8348:MainThread](2021-03-09 15:05:22,317) INFO - qlib.workflow - [exp.py:181] - Experiment 2 starts running ...\n",
-      "[8348:MainThread](2021-03-09 15:05:22,386) INFO - qlib.workflow - [recorder.py:233] - Recorder 0c814539f55842b9b6310843fc5ec708 starts running under Experiment 2 ...\n",
-      "Training until validation scores don't improve for 50 rounds\n",
-      "[20]\ttrain's l2: 0.969033\tvalid's l2: 0.98571\n",
-      "[40]\ttrain's l2: 0.955399\tvalid's l2: 0.986164\n",
-      "[60]\ttrain's l2: 0.943514\tvalid's l2: 0.986301\n",
-      "Early stopping, best iteration is:\n",
-      "[26]\ttrain's l2: 0.964587\tvalid's l2: 0.985356\n",
-      "[8348:MainThread](2021-03-09 15:05:29,546) INFO - qlib.workflow - [record_temp.py:126] - Signal record 'pred.pkl' has been saved as the artifact of the Experiment 2\n",
-      "'The following are prediction results of the LGBModel model.'\n",
-      "                          score\n",
-      "datetime   instrument          \n",
-      "2019-04-09 SH600000    0.029586\n",
-      "           SH600009    0.004306\n",
-      "           SH600010   -0.004411\n",
-      "           SH600011    0.002707\n",
-      "           SH600015   -0.029124\n",
-      "{'IC': 0.020784811232504984,\n",
-      " 'ICIR': 0.11590182186569555,\n",
-      " 'Rank IC': 0.028925697036767055,\n",
-      " 'Rank ICIR': 0.16388058980901396}\n"
-     ]
-    },
-    {
-     "output_type": "execute_result",
-     "data": {
-      "text/plain": [
-       "True"
-      ]
-     },
-     "metadata": {},
-     "execution_count": 26
-    }
-   ],
+   "execution_count": null,
+   "outputs": [],
    "source": [
     "from qlib.workflow.task.manage import run_task\n",
     "from qlib.workflow.task.collect import TaskCollector\n",
@@ -363,43 +125,12 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 27,
-   "outputs": [
-    {
-     "output_type": "stream",
-     "name": "stderr",
-     "text": [
-      "Loading data: 100%|██████████| 4/4 [00:00<00:00, 37.38it/s]\n"
-     ]
-    },
-    {
-     "output_type": "execute_result",
-     "data": {
-      "text/plain": [
-       "{('task_lgb', '2019-04-08'): datetime    instrument\n",
-       " 2017-01-03  SH600000     -0.013089\n",
-       "             SH600010     -0.006642\n",
-       "             SH600015     -0.035137\n",
-       "             SH600016     -0.034634\n",
-       "             SH600018     -0.029493\n",
-       "                             ...   \n",
-       " 2019-04-08  SZ002415      0.049199\n",
-       "             SZ002450     -0.013450\n",
-       "             SZ002594      0.022395\n",
-       "             SZ002736      0.091433\n",
-       "             SZ300059     -0.016237\n",
-       " Name: score, Length: 55000, dtype: float64}"
-      ]
-     },
-     "metadata": {},
-     "execution_count": 27
-    }
-   ],
+   "execution_count": null,
+   "outputs": [],
    "source": [
     "def get_task_key(task):\n",
     "    task_key = task[\"task_key\"]\n",
     "    rolling_end_timestamp = task[\"dataset\"][\"kwargs\"][\"segments\"][\"test\"][1]\n",
-    "    #rolling_end_datatime = rolling_end_timestamp.to_pydatetime()\n",
     "    return task_key, rolling_end_timestamp.strftime('%Y-%m-%d')\n",
     "\n",
     "def my_filter(task):\n",
@@ -410,7 +141,7 @@
     "    return False\n",
     "\n",
     "# name tasks by \"get_task_key\" and filter tasks by \"my_filter\"\n",
-    "pred_rolling = TaskCollector.collect(exp_name, get_task_key, my_filter) \n",
+    "pred_rolling = TaskCollector.collect_predictions(exp_name, get_task_key, my_filter) \n",
     "pred_rolling"
    ],
    "metadata": {
diff --git a/examples/taskmanager/task_manager_rolling.py b/examples/taskmanager/task_manager_rolling.py
index 7519bc4be..db5d1817f 100644
--- a/examples/taskmanager/task_manager_rolling.py
+++ b/examples/taskmanager/task_manager_rolling.py
@@ -85,7 +85,7 @@ tm = TaskManager(task_pool=task_pool)
 tm.create_task(tasks) # all tasks will be saved to MongoDB
 
 from qlib.workflow.task.manage import run_task
-from qlib.workflow.task.collect import RollingCollector
+from qlib.workflow.task.collect import TaskCollector
 from qlib.model.trainer import task_train
 
 run_task(task_train, task_pool, experiment_name=exp_name) # all tasks will be trained using "task_train" method
@@ -93,7 +93,6 @@ run_task(task_train, task_pool, experiment_name=exp_name) # all tasks will be tr
 def get_task_key(task_config):
     task_key = task_config["task_key"]
     rolling_end_timestamp = task_config["dataset"]["kwargs"]["segments"]["test"][1]
-    #rolling_end_datatime = rolling_end_timestamp.to_pydatetime()
     return task_key, rolling_end_timestamp.strftime('%Y-%m-%d')
 
 def my_filter(task_config):
@@ -103,6 +102,6 @@ def my_filter(task_config):
         return True
     return False
 
-collector = RollingCollector(get_task_key, my_filter)
-pred_rolling = collector(exp_name) # name tasks by "get_task_key" and filter tasks by "my_filter"
-print(pred_rolling)
\ No newline at end of file
+# name tasks by "get_task_key" and filter tasks by "my_filter"
+pred_rolling = TaskCollector.collect_predictions(exp_name, get_task_key, my_filter) 
+pred_rolling
\ No newline at end of file
diff --git a/qlib/model/trainer.py b/qlib/model/trainer.py
index 91061636d..82d770b96 100644
--- a/qlib/model/trainer.py
+++ b/qlib/model/trainer.py
@@ -6,7 +6,7 @@ from qlib.workflow import R
 from qlib.workflow.record_temp import SignalRecord
 
 
-def task_train(task_config: dict, experiment_name: str):
+def task_train(task_config: dict, experiment_name: str) -> str:
     """
     task based training
 
@@ -16,6 +16,11 @@ def task_train(task_config: dict, experiment_name: str):
         A dict describes a task setting.
     experiment_name: str
         The name of experiment
+
+    Returns
+    ----------
+    rid : str
+        The id of the recorder of this task
     """
 
     # model initiaiton
@@ -29,7 +34,7 @@ def task_train(task_config: dict, experiment_name: str):
         model.fit(dataset)
         recorder = R.get_recorder()
         R.save_objects(**{"params.pkl": model})
-        R.save_objects(param=task_config)  # keep the original format and datatype
+        R.save_objects(**{"task.pkl": task_config})  # keep the original format and datatype
 
         # generate records: prediction, backtest, and analysis
         records = task_config.get("record", [])
@@ -48,3 +53,4 @@ def task_train(task_config: dict, experiment_name: str):
                 record["kwargs"].update(rconf)
                 ar = init_instance_by_config(record)
                 ar.generate()
+    return record.info["id"]
diff --git a/qlib/workflow/task/collect.py b/qlib/workflow/task/collect.py
index 4562a1cec..834189561 100644
--- a/qlib/workflow/task/collect.py
+++ b/qlib/workflow/task/collect.py
@@ -1,7 +1,7 @@
 from qlib.workflow import R
 import pandas as pd
 from typing import Union
-from tqdm.auto import tqdm
+from qlib import get_module_logger
 
 
 class TaskCollector:
@@ -10,10 +10,8 @@ class TaskCollector:
     """
 
     @staticmethod
-    def collect(
-        experiment_name: str,
-        get_key_func,
-        filter_func=None,
+    def collect_predictions(
+        experiment_name: str, get_key_func, filter_func=None,
     ):
         """
 
@@ -34,8 +32,8 @@ class TaskCollector:
         recs = exp.list_recorders()
 
         recs_flt = {}
-        for rid, rec in tqdm(recs.items(), desc="Loading data"):
-            params = rec.load_object("param")
+        for rid, rec in recs.items():
+            params = rec.load_object("task.pkl")
             if rec.status == rec.STATUS_FI:
                 if filter_func is None or filter_func(params):
                     rec.params = params
@@ -57,6 +55,7 @@ class TaskCollector:
             pred = pd.concat(pred_l).sort_index()
             reduce_group[k] = pred
 
+        get_module_logger("TaskCollector").info(f"Collect {len(reduce_group)} predictions in {experiment_name}")
         return reduce_group
 
 
@@ -82,7 +81,7 @@ class RollingCollector:
 
         recs_flt = {}
         for rid, rec in tqdm(recs.items(), desc="Loading data"):
-            params = rec.load_object("param")
+            params = rec.load_object("task.pkl")
             if rec.status == rec.STATUS_FI:
                 if self.flt_func is None or self.flt_func(params):
                     rec.params = params
diff --git a/qlib/workflow/task/gen.py b/qlib/workflow/task/gen.py
index b1c2e0ce2..19793c485 100644
--- a/qlib/workflow/task/gen.py
+++ b/qlib/workflow/task/gen.py
@@ -168,7 +168,7 @@ class RollingGen(TaskGen):
                 # 1) prepare the end point
                 segments = copy.deepcopy(self.ta.align_seg(t["dataset"]["kwargs"]["segments"]))
                 test_end = self.ta.last_date() if segments[self.test_key][1] is None else segments[self.test_key][1]
-                # 2) and the init test segments
+                # 2) and init test segments
                 test_start_idx = self.ta.align_idx(segments[self.test_key][0])
                 segments[self.test_key] = (self.ta.get(test_start_idx), self.ta.get(test_start_idx + self.step - 1))
             else:

From 2ca2071d959a007625b2879a57e84efb1507ac59 Mon Sep 17 00:00:00 2001
From: lzh222333 <lzh222333@163.com>
Date: Wed, 10 Mar 2021 17:06:08 +0000
Subject: [PATCH 09/10] format code

---
 examples/taskmanager/task_manager_rolling.py | 75 ++++++++------------
 qlib/workflow/task/collect.py                |  4 +-
 2 files changed, 34 insertions(+), 45 deletions(-)

diff --git a/examples/taskmanager/task_manager_rolling.py b/examples/taskmanager/task_manager_rolling.py
index db5d1817f..36ec81962 100644
--- a/examples/taskmanager/task_manager_rolling.py
+++ b/examples/taskmanager/task_manager_rolling.py
@@ -9,53 +9,37 @@ data_handler_config = {
     "end_time": "2020-08-01",
     "fit_start_time": "2008-01-01",
     "fit_end_time": "2014-12-31",
-    "instruments": 'csi100',
+    "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": ("2008-01-01", "2014-12-31"),
-                "valid": ("2015-01-01", "2016-12-31"),
-                "test": ("2017-01-01", "2020-08-01"),
-            },
+    "class": "DatasetH",
+    "module_path": "qlib.data.dataset",
+    "kwargs": {
+        "handler": {"class": "Alpha158", "module_path": "qlib.contrib.data.handler", "kwargs": data_handler_config,},
+        "segments": {
+            "train": ("2008-01-01", "2014-12-31"),
+            "valid": ("2015-01-01", "2016-12-31"),
+            "test": ("2017-01-01", "2020-08-01"),
         },
-    }
+    },
+}
 
 record_config = [
-    {
-        "class": "SignalRecord",
-        "module_path": "qlib.workflow.record_temp",
-    },
-    {
-        "class": "SigAnaRecord",
-        "module_path": "qlib.workflow.record_temp",
-    }
+    {"class": "SignalRecord", "module_path": "qlib.workflow.record_temp",},
+    {"class": "SigAnaRecord", "module_path": "qlib.workflow.record_temp",},
 ]
 
 # use lgb
 task_lgb_config = {
-    "model": {
-        "class": "LGBModel",
-        "module_path": "qlib.contrib.model.gbdt",
-    },
+    "model": {"class": "LGBModel", "module_path": "qlib.contrib.model.gbdt",},
     "dataset": dataset_config,
     "record": record_config,
 }
 
 # use xgboost
 task_xgboost_config = {
-    "model": {
-        "class": "XGBModel",
-        "module_path": "qlib.contrib.model.xgboost",
-    },
+    "model": {"class": "XGBModel", "module_path": "qlib.contrib.model.xgboost",},
     "dataset": dataset_config,
     "record": record_config,
 }
@@ -64,17 +48,17 @@ provider_uri = "~/.qlib/qlib_data/cn_data"  # target_dir
 qlib.init(provider_uri=provider_uri, region=REG_CN)
 
 C["mongo"] = {
-    "task_url" : "mongodb://localhost:27017/", # maybe you need to change it to your url
-    "task_db_name" : "rolling_db"
+    "task_url": "mongodb://localhost:27017/",  # maybe you need to change it to your url
+    "task_db_name": "rolling_db",
 }
 
-exp_name = 'rolling_exp' # experiment name, will be used as the experiment in MLflow
-task_pool = 'rolling_task' # task pool name, will be used as the document in MongoDB
+exp_name = "rolling_exp"  # experiment name, will be used as the experiment in MLflow
+task_pool = "rolling_task"  # task pool name, will be used as the document in MongoDB
 
 tasks = task_generator(
-    task_xgboost_config, # default task name
-    RollingGen(step=550,rtype=RollingGen.ROLL_SD), # generate different date segment
-    task_lgb=task_lgb_config # use "task_lgb" as the task name
+    task_xgboost_config,  # default task name
+    RollingGen(step=550, rtype=RollingGen.ROLL_SD),  # generate different date segment
+    task_lgb=task_lgb_config,  # use "task_lgb" as the task name
 )
 
 # Uncomment next two lines to see the generated tasks
@@ -82,26 +66,29 @@ tasks = task_generator(
 # pprint(tasks)
 
 tm = TaskManager(task_pool=task_pool)
-tm.create_task(tasks) # all tasks will be saved to MongoDB
+tm.create_task(tasks)  # all tasks will be saved to MongoDB
 
 from qlib.workflow.task.manage import run_task
 from qlib.workflow.task.collect import TaskCollector
 from qlib.model.trainer import task_train
 
-run_task(task_train, task_pool, experiment_name=exp_name) # all tasks will be trained using "task_train" method
+run_task(task_train, task_pool, experiment_name=exp_name)  # all tasks will be trained using "task_train" method
+
 
 def get_task_key(task_config):
     task_key = task_config["task_key"]
     rolling_end_timestamp = task_config["dataset"]["kwargs"]["segments"]["test"][1]
-    return task_key, rolling_end_timestamp.strftime('%Y-%m-%d')
+    return task_key, rolling_end_timestamp.strftime("%Y-%m-%d")
+
 
 def my_filter(task_config):
     # only choose the results of "task_lgb" and test in 2019 from all tasks
     task_key, rolling_end = get_task_key(task_config)
-    if task_key=="task_lgb" and rolling_end.startswith('2019'):
+    if task_key == "task_lgb" and rolling_end.startswith("2019"):
         return True
     return False
 
+
 # name tasks by "get_task_key" and filter tasks by "my_filter"
-pred_rolling = TaskCollector.collect_predictions(exp_name, get_task_key, my_filter) 
-pred_rolling
\ No newline at end of file
+pred_rolling = TaskCollector.collect_predictions(exp_name, get_task_key, my_filter)
+pred_rolling
diff --git a/qlib/workflow/task/collect.py b/qlib/workflow/task/collect.py
index 834189561..ccd6ce169 100644
--- a/qlib/workflow/task/collect.py
+++ b/qlib/workflow/task/collect.py
@@ -11,7 +11,9 @@ class TaskCollector:
 
     @staticmethod
     def collect_predictions(
-        experiment_name: str, get_key_func, filter_func=None,
+        experiment_name: str, 
+        get_key_func, 
+        filter_func=None,
     ):
         """
 

From 48f0fc147f1799b2eff4848318ebf0f222320865 Mon Sep 17 00:00:00 2001
From: lzh222333 <lzh222333@163.com>
Date: Thu, 11 Mar 2021 03:00:30 +0000
Subject: [PATCH 10/10] first version of online serving

---
 examples/taskmanager/update_online_pred.py |  77 +++++++++++
 qlib/model/trainer.py                      |   2 +-
 qlib/workflow/task/collect.py              |   4 +-
 qlib/workflow/task/update.py               | 154 +++++++++++++++++++++
 4 files changed, 234 insertions(+), 3 deletions(-)
 create mode 100644 examples/taskmanager/update_online_pred.py
 create mode 100644 qlib/workflow/task/update.py

diff --git a/examples/taskmanager/update_online_pred.py b/examples/taskmanager/update_online_pred.py
new file mode 100644
index 000000000..4dbd22b85
--- /dev/null
+++ b/examples/taskmanager/update_online_pred.py
@@ -0,0 +1,77 @@
+import qlib
+from qlib.model.trainer import task_train
+from qlib.workflow.task.update import ModelUpdater
+from qlib.config import REG_CN
+import fire
+
+data_handler_config = {
+        "start_time": "2008-01-01",
+        "end_time": "2020-08-01",
+        "fit_start_time": "2008-01-01",
+        "fit_end_time": "2014-12-31",
+        "instruments": "csi100",
+    }
+
+task = {
+    "model": {
+        "class": "LGBModel",
+        "module_path": "qlib.contrib.model.gbdt",
+        "kwargs": {
+            "loss": "mse",
+            "colsample_bytree": 0.8879,
+            "learning_rate": 0.0421,
+            "subsample": 0.8789,
+            "lambda_l1": 205.6999,
+            "lambda_l2": 580.9768,
+            "max_depth": 8,
+            "num_leaves": 210,
+            "num_threads": 20,
+        },
+    },
+    "dataset": {
+        "class": "DatasetH",
+        "module_path": "qlib.data.dataset",
+        "kwargs": {
+            "handler": {
+                "class": "Alpha158",
+                "module_path": "qlib.contrib.data.handler",
+                "kwargs": data_handler_config,
+            },
+            "segments": {
+                "train": ("2008-01-01", "2014-12-31"),
+                "valid": ("2015-01-01", "2016-12-31"),
+                "test": ("2017-01-01", "2020-08-01"),
+            },
+        },
+    },
+    "record": {"class": "SignalRecord", "module_path": "qlib.workflow.record_temp",},
+}
+
+provider_uri = "~/.qlib/qlib_data/cn_data"  # target_dir
+
+def first_train(experiment_name="online_svr"):
+
+    
+    qlib.init(provider_uri=provider_uri, region=REG_CN)
+    model_updater = ModelUpdater(experiment_name)
+
+    rid = task_train(task_config=task, experiment_name=experiment_name)
+    model_updater.reset_online_model(rid)
+
+def update_online_pred(experiment_name="online_svr"):
+    
+    qlib.init(provider_uri=provider_uri, region=REG_CN)
+    model_updater = ModelUpdater(experiment_name)
+
+    print("Here are the online models waiting for update:")
+    for rid, rec in model_updater.list_online_model().items():
+        print(rid)
+
+    model_updater.update_online_pred()
+    
+if __name__ == '__main__':
+    fire.Fire()
+    # to train a model and set it to online model, use the command below
+    # python update_online_pred.py first_train
+    # to update online predictions once a day, use the command below
+    # python update_online_pred.py update_online_pred
diff --git a/qlib/model/trainer.py b/qlib/model/trainer.py
index 82d770b96..5e62a141c 100644
--- a/qlib/model/trainer.py
+++ b/qlib/model/trainer.py
@@ -53,4 +53,4 @@ def task_train(task_config: dict, experiment_name: str) -> str:
                 record["kwargs"].update(rconf)
                 ar = init_instance_by_config(record)
                 ar.generate()
-    return record.info["id"]
+    return recorder.info["id"]
diff --git a/qlib/workflow/task/collect.py b/qlib/workflow/task/collect.py
index ccd6ce169..059871ab1 100644
--- a/qlib/workflow/task/collect.py
+++ b/qlib/workflow/task/collect.py
@@ -11,8 +11,8 @@ class TaskCollector:
 
     @staticmethod
     def collect_predictions(
-        experiment_name: str, 
-        get_key_func, 
+        experiment_name: str,
+        get_key_func,
         filter_func=None,
     ):
         """
diff --git a/qlib/workflow/task/update.py b/qlib/workflow/task/update.py
new file mode 100644
index 000000000..f9d03efbc
--- /dev/null
+++ b/qlib/workflow/task/update.py
@@ -0,0 +1,154 @@
+from typing import Union
+from qlib.workflow import R
+from tqdm.auto import tqdm
+from qlib.data import D
+import pandas as pd
+from qlib.utils import init_instance_by_config
+from qlib import get_module_logger
+from qlib.workflow import R
+
+
+class ModelUpdater:
+    """
+    The model updater to re-train model or update predictions
+    """
+
+    ONLINE_TAG = "online_model"
+    ONLINE_TAG_TRUE = "True"
+    ONLINE_TAG_FALSE = "False"
+
+    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.exp = R.get_exp(experiment_name=experiment_name)
+        self.logger = get_module_logger("ModelUpdater")
+
+    def set_online_model(self, rid: str):
+        """online model will be identified at the tags of the record
+
+        Parameters
+        ----------
+        rid : str
+            the id of a record
+        """
+        rec = self.exp.get_recorder(recorder_id=rid)
+        rec.set_tags(**{self.ONLINE_TAG: self.ONLINE_TAG_TRUE})
+
+    def cancel_online_model(self, rid: str):
+        rec = self.exp.get_recorder(recorder_id=rid)
+        rec.set_tags(**{self.ONLINE_TAG: self.ONLINE_TAG_FALSE})
+
+    def cancel_all_online_model(self):
+        recs = self.exp.list_recorders()
+        for rid, rec in recs.items():
+            self.cancel_online_model(rid)
+
+    def reset_online_model(self, rids: Union[str, list]):
+        """cancel all online model and reset the given model to online model
+
+        Parameters
+        ----------
+        rids : Union[str, list]
+            the name of a record or the list of the name of records
+        """
+        self.cancel_all_online_model()
+        if isinstance(rids, str):
+            rids = [rids]
+        for rid in rids:
+            self.set_online_model(rid)
+
+    def update_pred(self, rid: str):
+        """update predictions to the latest day in Calendar based on rid
+
+        Parameters
+        ----------
+        rid : str
+            the id of the record
+        """
+        rec = self.exp.get_recorder(recorder_id=rid)
+        old_pred = rec.load_object("pred.pkl")
+        last_end = old_pred.index.get_level_values("datetime").max()
+        task_config = rec.load_object("task.pkl")
+
+        # updated to the latest trading day
+        cal = D.calendar(start_time=last_end + pd.Timedelta(days=1), end_time=None)
+
+        if len(cal) == 0:
+            self.logger.info(f"All prediction in {rid} of {self.exp_name} are latest. No need to update.")
+            return
+
+        start_time, end_time = cal[0], cal[-1]
+        task_config["dataset"]["kwargs"]["segments"]["test"] = (start_time, end_time)
+        task_config["dataset"]["kwargs"]["handler"]["kwargs"]["end_time"] = end_time
+
+        dataset = init_instance_by_config(task_config["dataset"])
+
+        model = rec.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()
+
+        rec.save_objects(**{"pred.pkl": cb_pred})
+
+        self.logger.info(f"Finish updating new {new_pred.shape[0]} predictions in {rid} of {self.exp_name}.")
+
+    def update_all_pred(self, 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.pkl")
+                    if task_config["model"]["class"]=="LGBModel":
+                        return True
+                    return False
+
+        Parameters
+        ----------
+        filter_func : function, optional
+            the filter function to decide whether this record will be updated, by default None
+
+        Returns
+        ----------
+        cnt: int
+            the count of updated record
+
+        """
+        cnt = 0
+        recs = self.exp.list_recorders()
+        for rid, rec in recs.items():
+            if rec.status == rec.STATUS_FI:
+                if filter_func != None and filter_func(rec) == False:
+                    # records that should be filtered out
+                    continue
+                self.update_pred(rid)
+                cnt += 1
+        return cnt
+
+    def online_filter(self, record):
+        tags = record.list_tags()
+        if tags[self.ONLINE_TAG] == self.ONLINE_TAG_TRUE:
+            return True
+        return False
+
+    def update_online_pred(self):
+        """update all online model predictions to the latest day in Calendar."""
+        cnt = self.update_all_pred(self.online_filter)
+        self.logger.info(f"Finish updating {cnt} online model predictions of {self.exp_name}.")
+
+    def list_online_model(self):
+        recs = self.exp.list_recorders()
+        online_rec = {}
+        for rid, rec in recs.items():
+            if self.online_filter(rec):
+                online_rec[rid] = rec
+        return online_rec