update handler & fix some bugs

qlib/contrib/backtest/position.py

@@ -160,7 +160,7 @@ class Position:
     def save_position(self, path, last_trade_date):
         path = pathlib.Path(path)
         p = copy.deepcopy(self.position)
-        cash = pd.Series()
+        cash = pd.Series(dtype=np.float)
         cash["init_cash"] = self.init_cash
         cash["cash"] = p["cash"]
         cash["today_account_value"] = p["today_account_value"]

qlib/data/client.py

@@ -7,7 +7,7 @@ from __future__ import print_function
 
 import socketio
 
-from .. import __version__
+import qlib
 from ..log import get_module_logger
 import pickle
 
@@ -59,7 +59,7 @@ class Client(object):
         msg_queue: Queue
             The queue to pass the messsage after callback
         """
-        head_info = {"version": __version__}
+        head_info = {"version": qlib.__version__}
 
         def request_callback(*args):
             """callback_wrapper

qlib/data/data.py

@@ -640,7 +640,7 @@ class LocalFeatureProvider(FeatureProvider):
         uri_data = self._uri_data.format(instrument.lower(), field, freq)
         if not os.path.exists(uri_data):
             get_module_logger("data").warning("WARN: data not found for %s.%s" % (instrument, field))
-            return pd.Series()
+            return pd.Series(dtype=np.float32)
             # raise ValueError('uri_data not found: ' + uri_data)
         # load
         series = read_bin(uri_data, start_index, end_index)

qlib/data/dataset/handler.py

@@ -5,6 +5,7 @@
 import abc
 import bisect
 import logging
+import warnings
 from typing import Union, Tuple, List, Iterator, Optional
 
 import pandas as pd
@@ -61,7 +62,9 @@ class DataHandler(Serializable):
 
         # Setup data loader
         assert data_loader is not None  # to make start_time end_time could have None default value
-        self.data_loader = init_instance_by_config(data_loader, data_loader_module, accept_types=DataLoader)
+        self.data_loader = init_instance_by_config(data_loader,
+                                                   None if 'module_path' in data_loader else data_loader_module,
+                                                   accept_types=DataLoader)
 
         self.instruments = instruments
         self.start_time = start_time
@@ -224,12 +227,12 @@ class DataHandlerLP(DataHandler):
 
     # process type
     PTYPE_I = "independent"
-    # - _proc_infer_df will processed by infer_processors
+    # - self._infer will processed by infer_processors
-    # - _proc_learn_df will be processed by learn_processors
+    # - self._learn will be processed by learn_processors
     PTYPE_A = "append"
-    # - _proc_infer_df will processed by infer_processors
+    # - self._infer will processed by infer_processors
-    # - _proc_learn_df will be processed by infer_processors + learn_processors
+    # - self._learn will be processed by infer_processors + learn_processors
-    #   - (e.g. _proc_infer_df processed by learn_processors )
+    #   - (e.g. self._infer processed by learn_processors )
 
     def __init__(
         self,
@@ -265,12 +268,12 @@ class DataHandlerLP(DataHandler):
 
         process_type: str
             PTYPE_I = 'independent'
-            - _proc_infer_df will processed by infer_processors
+            - self._infer will processed by infer_processors
-            - _proc_learn_df will be processed by learn_processors
+            - self._learn will be processed by learn_processors
             PTYPE_A = 'append'
-            - _proc_infer_df will processed by infer_processors
+            - self._infer will processed by infer_processors
-            - _proc_learn_df will be processed by infer_processors + learn_processors
+            - self._learn will be processed by infer_processors + learn_processors
-              - (e.g. _proc_infer_df processed by learn_processors )
+              - (e.g. self._infer processed by learn_processors )
         """
 
         # Setup preprocessor

qlib/data/dataset/loader.py

@@ -14,7 +14,6 @@ class DataLoader(abc.ABC):
     """
     DataLoader is designed for loading raw data from original data source.
     """
-
     @abc.abstractmethod
     def load(self, instruments, start_time=None, end_time=None) -> pd.DataFrame:
         """
@@ -48,10 +47,13 @@ class DataLoader(abc.ABC):
         pass
 
 
-class QlibDataLoader(DataLoader):
+class DLWParser(DataLoader):
-    """Same as QlibDataLoader. The fields can be define by config"""
+    """
+    (D)ata(L)oader (W)ith (P)arser for features and names
 
-    def __init__(self, config: Tuple[list, tuple, dict], filter_pipe=None):
+    Extracting this class so that QlibDataLoader and other dataloaders(such as QdbDataLoader) can share the fields
+    """
+    def __init__(self, config: Tuple[list, tuple, dict]):
         """
         Parameters
         ----------
@@ -74,8 +76,6 @@ class QlibDataLoader(DataLoader):
         else:
             self.fields = self._parse_fields_info(config)
 
-        self.filter_pipe = filter_pipe
-
     def _parse_fields_info(self, fields_info: Tuple[list, tuple]) -> Tuple[list, list]:
         if isinstance(fields_info, list):
             exprs = names = fields_info
@@ -85,21 +85,62 @@ class QlibDataLoader(DataLoader):
             raise NotImplementedError(f"This type of input is not supported")
         return exprs, names
 
-    def load(self, instruments, start_time=None, end_time=None) -> pd.DataFrame:
+    @abc.abstractmethod
+    def load_group_df(self, instruments, exprs: list, names: list, start_time=None, end_time=None) -> pd.DataFrame:
+        """
+        load the dataframe for specific group
+
+        Parameters
+        ----------
+        instruments :
+            the instruments
+        exprs : list
+            The expressions to describe the content of the data
+        names : list
+            The name of the data
+
+        Returns
+        -------
+        pd.DataFrame:
+            the queried dataframe
+        """
+        pass
+
+    def load(self, instruments=None, start_time=None, end_time=None) -> pd.DataFrame:
+        if self.is_group:
+            df = pd.concat(
+                {
+                    grp: self.load_group_df(instruments, exprs, names, start_time, end_time)
+                    for grp, (exprs, names) in self.fields.items()
+                },
+                axis=1)
+        else:
+            exprs, names = self.fields
+            df = self.load_group_df(instruments, exprs, names, start_time, end_time)
+        return df
+
+
+class QlibDataLoader(DLWParser):
+    """Same as QlibDataLoader. The fields can be define by config"""
+    def __init__(self, config: Tuple[list, tuple, dict], filter_pipe=None):
+        """
+        Parameters
+        ----------
+        config : Tuple[list, tuple, dict]
+            Please refer to the doc of DLWParser
+        filter_pipe :
+            Filter pipe for the instruments
+        """
+        self.filter_pipe = filter_pipe
+        super().__init__(config)
+
+    def load_group_df(self, instruments, exprs: list, names: list, start_time=None, end_time=None) -> pd.DataFrame:
         if isinstance(instruments, str):
             instruments = D.instruments(instruments, filter_pipe=self.filter_pipe)
         elif self.filter_pipe is not None:
             warnings.warn("`filter_pipe` is not None, but it will not be used with `instruments` as list")
 
-        def _get_df(exprs, names):
+        df = D.features(instruments, exprs, start_time, end_time)
-            df = D.features(instruments, exprs, start_time, end_time)
+        df.columns = names
-            df.columns = names
-            return df
-
-        if self.is_group:
-            df = pd.concat({grp: _get_df(exprs, names) for grp, (exprs, names) in self.fields.items()}, axis=1)
-        else:
-            exprs, names = self.fields
-            df = _get_df(exprs, names)
         df = df.swaplevel().sort_index()  # NOTE: always return <datetime, instrument>
         return df

qlib/utils/__init__.py

@@ -44,7 +44,7 @@ def read_bin(file_path, start_index, end_index):
         ref_start_index = int(np.frombuffer(f.read(4), dtype="<f")[0])
         si = max(ref_start_index, start_index)
         if si > end_index:
-            return pd.Series()
+            return pd.Series(np.float32)
         # calculate offset
         f.seek(4 * (si - ref_start_index) + 4)
         # read nbytes
@@ -213,6 +213,7 @@ def init_instance_by_config(
         "ClassName":  getattr(module, config)() will be used.
     module : Python module
         Optional. It should be a python module.
+        NOTE: the "module_path" will be override by `module` arguments
 
     accept_types: Union[type, Tuple[type]]
         Optional. If the config is a instance of specific type, return the config directly.