Fix processor bug and format

qlib/data/dataset/handler.py

@@ -16,7 +16,7 @@ from ...data import D
 from ...config import C
 from ...utils import parse_config, transform_end_date, init_instance_by_config
 from ...utils.serial import Serializable
-from .utils import get_level_index
+from .utils import get_level_index, fetch_df_by_index
 from pathlib import Path
 from .loader import DataLoader
 
@@ -99,25 +99,6 @@ class DataHandler(Serializable):
             self._data = self.data_loader.load(self.instruments, self.start_time, self.end_time)
         # TODO: cache
 
-    def _fetch_df_by_index(
-        self, df: pd.DataFrame, selector: Union[pd.Timestamp, slice, str, list], level: Union[str, int]
-    ) -> pd.DataFrame:
-        """
-        fetch data from `data` with `selector` and `level`
-
-        Parameters
-        ----------
-        selector : Union[pd.Timestamp, slice, str, list]
-            selector
-        level : Union[int, str]
-            the level to use the selector
-        """
-        # Try to get the right index
-        idx_slc = (selector, slice(None, None))
-        if get_level_index(df, level) == 1:
-            idx_slc = idx_slc[1], idx_slc[0]
-        return df.loc(axis=0)[idx_slc]
-
     CS_ALL = "__all"
 
     def _fetch_df_by_col(self, df: pd.DataFrame, col_set: str) -> pd.DataFrame:
@@ -156,7 +137,7 @@ class DataHandler(Serializable):
         -------
         pd.DataFrame:
         """
-        df = self._fetch_df_by_index(self._data, selector, level)
+        df = fetch_df_by_index(self._data, selector, level)
         df = self._fetch_df_by_col(df, col_set)
         if squeeze:
             # squeeze columns
@@ -414,7 +395,7 @@ class DataHandlerLP(DataHandler):
         pd.DataFrame:
         """
         df = self._get_df_by_key(data_key)
-        df = self._fetch_df_by_index(df, selector, level)
+        df = fetch_df_by_index(df, selector, level)
         return self._fetch_df_by_col(df, col_set)
 
     def get_cols(self, col_set=DataHandler.CS_ALL, data_key: str = DK_I) -> list:

qlib/data/dataset/processor.py

@@ -7,6 +7,7 @@ import pandas as pd
 import copy
 
 from ...log import TimeInspector
+from .utils import fetch_df_by_index
 from ...utils.serial import Serializable
 from ...utils.paral import datetime_groupby_apply
 
@@ -106,6 +107,7 @@ class ProcessInf(Processor):
 
         return replace_inf(df)
 
+
 class Fillna(Processor):
     """Process infinity  """
 
@@ -123,14 +125,15 @@ class Fillna(Processor):
 
         return fill_na(df)
 
+
 class MinMaxNorm(Processor):
     def __init__(self, fit_start_time, fit_end_time, fields_group=None):
-        # FIXME: time is not used
         self.fit_start_time = fit_start_time
         self.fit_end_time = fit_end_time
         self.fields_group = fields_group
 
     def fit(self, df):
+        df = fetch_df_by_index(df, slice(self.fit_start_time, self.fit_end_time), level="datetime")
         cols = get_group_columns(df, self.fields_group)
         self.min_val = np.nanmin(df[cols].values, axis=0)
         self.max_val = np.nanmax(df[cols].values, axis=0)
@@ -152,15 +155,15 @@ class MinMaxNorm(Processor):
 
 class ZscoreNorm(Processor):
     def __init__(self, fit_start_time, fit_end_time, fields_group=None):
-        # FIXME: time is not used
         self.fit_start_time = fit_start_time
         self.fit_end_time = fit_end_time
         self.fields_group = fields_group
 
     def fit(self, df):
+        df = fetch_df_by_index(df, slice(self.fit_start_time, self.fit_end_time), level="datetime")
         cols = get_group_columns(df, self.fields_group)
         self.mean_train = np.nanmean(df[cols].values, axis=0)
-        self.std_train = np.nanstd(df[cols].values, axis=0)
+        self.std_train = np.nanstd(_df[cols].values, axis=0)
         self.ignore = self.std_train == 0
         self.cols = cols
 

qlib/data/dataset/utils.py

@@ -29,3 +29,27 @@ def get_level_index(df: pd.DataFrame, level=Union[str, int]) -> int:
         return level
     else:
         raise NotImplementedError(f"This type of input is not supported")
+
+
+def fetch_df_by_index(
+    df: pd.DataFrame, selector: Union[pd.Timestamp, slice, str, list], level: Union[str, int]
+) -> pd.DataFrame:
+    """
+    fetch data from `data` with `selector` and `level`
+
+    Parameters
+    ----------
+    selector : Union[pd.Timestamp, slice, str, list]
+        selector
+    level : Union[int, str]
+        the level to use the selector
+
+    Returns
+    -------
+    Data of the given index.
+    """
+    # Try to get the right index
+    idx_slc = (selector, slice(None, None))
+    if get_level_index(df, level) == 1:
+        idx_slc = idx_slc[1], idx_slc[0]
+    return df.loc(axis=0)[idx_slc]