init version of online serving and rolling

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

qlib/__init__.py

@@ -3,19 +3,21 @@
 
 
 __version__ = "0.6.3.99"
+__version__bak = __version__  # This version is backup for QlibConfig.reset_qlib_version
 
 
 import os
 import yaml
 import logging
 import platform
 import subprocess
+from pathlib import Path
+from .log import get_module_logger
 
 
 # init qlib
 def init(default_conf="client", **kwargs):
     from .config import C
-    from .log import get_module_logger
     from .data.cache import H
 
     H.clear()
@@ -48,7 +50,6 @@ def init(default_conf="client", **kwargs):
 
 
 def _mount_nfs_uri(C):
-    from .log import get_module_logger
 
     LOG = get_module_logger("mount nfs", level=logging.INFO)
 
@@ -151,3 +152,73 @@ def init_from_yaml_conf(conf_path, **kwargs):
     config.update(kwargs)
     default_conf = config.pop("default_conf", "client")
     init(default_conf, **config)
+
+
+def get_project_path(config_name="config.yaml") -> Path:
+    """
+    If users are building a project follow the following pattern.
+    - Qlib is a sub folder in project path
+    - There is a file named `config.yaml` in qlib.
+
+    For example:
+        If your project file system stucuture follows such a pattern
+
+            <project_path>/
+              - config.yaml
+              - ...some folders...
+                - qlib/
+
+        This folder will return <project_path>
+
+        NOTE: link is not supported here.
+
+
+    This method is often used when
+    - user want to use a relative config path instead of hard-coding qlib config path in code
+
+    Raises
+    ------
+    FileNotFoundError:
+        If project path is not found
+    """
+    cur_path = Path(__file__).absolute().resolve()
+    while True:
+        if (cur_path / config_name).exists():
+            return cur_path
+        if cur_path == cur_path.parent:
+            raise FileNotFoundError("We can't find the project path")
+        cur_path = cur_path.parent
+
+
+def auto_init(**kwargs):
+    """
+    This function will init qlib automatically with following priority
+    - Find the project configuration and init qlib
+        - The parsing process will be affected by the `conf_type` of the configuration file
+    - Init qlib with default config
+    """
+
+    try:
+        pp = get_project_path()
+    except FileNotFoundError:
+        init(**kwargs)
+    else:
+
+        conf_pp = pp / "config.yaml"
+        with conf_pp.open() as f:
+            conf = yaml.safe_load(f)
+
+        conf_type = conf.get("conf_type", "origin")
+        if conf_type == "origin":
+            # The type of config is just like original qlib config
+            init_from_yaml_conf(conf_pp, **kwargs)
+        elif conf_type == "ref":
+            # This config type will be more convenient in following scenario
+            # - There is a shared configure file and you don't want to edit it inplace.
+            # - The shared configure may be updated later and you don't want to copy it.
+            # - You have some customized config.
+            qlib_conf_path = conf["qlib_cfg"]
+            qlib_conf_update = conf.get("qlib_cfg_update")
+            init_from_yaml_conf(qlib_conf_path, **qlib_conf_update, **kwargs)
+        logger = get_module_logger("Initialization")
+        logger.info(f"Auto load project config: {conf_pp}")

qlib/config.py

@@ -33,6 +33,9 @@ def __getattr__(self, attr):
 
         raise AttributeError(f"No such {attr} in self._config")
 
+    def get(self, key, default=None):
+        return self.__dict__["_config"].get(key, default)
+
     def __setitem__(self, key, value):
         self.__dict__["_config"][key] = value
 
@@ -310,8 +313,22 @@ def register(self):
         # clean up experiment when python program ends
         experiment_exit_handler()
 
+        # Supporting user reset qlib version (useful when user want to connect to qlib server with old version)
+        self.reset_qlib_version()
+
         self._registered = True
 
+    def reset_qlib_version(self):
+        import qlib
+
+        reset_version = self.get("qlib_reset_version", None)
+        if reset_version is not None:
+            qlib.__version__ = reset_version
+        else:
+            qlib.__version__ = getattr(qlib, "__version__bak")
+            # Due to a bug? that converting __version__ to _QlibConfig__version__bak
+            # Using  __version__bak instead of __version__
+
     @property
     def registered(self):
         return self._registered

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

qlib/workflow/task/__init__.py

@@ -0,0 +1,13 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+"""
+Task related workflow is implemented in this folder
+
+A typical task workflow
+
+| Step                  | Description                                    |
+|-----------------------+------------------------------------------------|
+| TaskGen               | Generating tasks.                              |
+| TaskManager(optional) | Manage generated tasks                         |
+| run task              | retrive  tasks from TaskManager and run tasks. |
+"""