Refine DDG-DA (#1472)

* Run ddg-da successfully

* Support include valid; More parameters

* Support L2 reg & visualization

* Blackformat

* Enable fill_method

* Support specify handler & optim dataset

* Fix Pylint

tests/data_mid_layer_tests/README.md

@@ -0,0 +1,5 @@
+# Introduction
+The middle layers of data, which mainly includes
+- Handler
+    - processors
+- Datasets

tests/data_mid_layer_tests/test_dataset.py

@@ -0,0 +1,107 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import unittest
+import pytest
+import sys
+from qlib.tests import TestAutoData
+from qlib.data.dataset import TSDatasetH
+import numpy as np
+import time
+from qlib.data.dataset.handler import DataHandlerLP
+
+
+class TestDataset(TestAutoData):
+    @pytest.mark.slow
+    def testTSDataset(self):
+        tsdh = TSDatasetH(
+            handler={
+                "class": "Alpha158",
+                "module_path": "qlib.contrib.data.handler",
+                "kwargs": {
+                    "start_time": "2017-01-01",
+                    "end_time": "2020-08-01",
+                    "fit_start_time": "2017-01-01",
+                    "fit_end_time": "2017-12-31",
+                    "instruments": "csi300",
+                    "infer_processors": [
+                        {"class": "FilterCol", "kwargs": {"col_list": ["RESI5", "WVMA5", "RSQR5"]}},
+                        {"class": "RobustZScoreNorm", "kwargs": {"fields_group": "feature", "clip_outlier": "true"}},
+                        {"class": "Fillna", "kwargs": {"fields_group": "feature"}},
+                    ],
+                    "learn_processors": [
+                        "DropnaLabel",
+                        {"class": "CSRankNorm", "kwargs": {"fields_group": "label"}},  # CSRankNorm
+                    ],
+                },
+            },
+            segments={
+                "train": ("2017-01-01", "2017-12-31"),
+                "valid": ("2018-01-01", "2018-12-31"),
+                "test": ("2019-01-01", "2020-08-01"),
+            },
+        )
+        tsds_train = tsdh.prepare("train", data_key=DataHandlerLP.DK_L)  # Test the correctness
+        tsds = tsdh.prepare("valid", data_key=DataHandlerLP.DK_L)
+
+        t = time.time()
+        for idx in np.random.randint(0, len(tsds_train), size=2000):
+            _ = tsds_train[idx]
+        print(f"2000 sample takes {time.time() - t}s")
+
+        t = time.time()
+        for _ in range(20):
+            data = tsds_train[np.random.randint(0, len(tsds_train), size=2000)]
+        print(data.shape)
+        print(f"2000 sample(batch index) * 20 times takes {time.time() - t}s")
+
+        # The dimension of sample is same as tabular data, but it will return timeseries data of the sample
+
+        # We have two method to get the time-series of a sample
+
+        # 1) sample by int index directly
+        tsds[len(tsds) - 1]
+
+        # 2) sample by <datetime,instrument> index
+        data_from_ds = tsds["2017-12-31", "SZ300315"]
+
+        # Check the data
+        # Get data from DataFrame Directly
+        data_from_df = (
+            tsdh.handler.fetch(data_key=DataHandlerLP.DK_L)
+            .loc(axis=0)["2017-01-01":"2017-12-31", "SZ300315"]
+            .iloc[-30:]
+            .values
+        )
+
+        equal = np.isclose(data_from_df, data_from_ds)
+        self.assertTrue(equal[~np.isnan(data_from_df)].all())
+
+        if False:
+            # 3) get both index and data
+            # NOTE: We don't want to reply on pytorch, so this test can't be included. It is just a example
+            from torch.utils.data import DataLoader
+            from qlib.model.utils import IndexSampler
+
+            i = len(tsds) - 1
+            idx = tsds.get_index()
+            tsds[i]
+            idx[i]
+
+            s_w_i = IndexSampler(tsds)
+            test_loader = DataLoader(s_w_i)
+
+            s_w_i[3]
+            for data, i in test_loader:
+                break
+            print(data.shape)
+            print(idx[i])
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=10)
+
+    # User could use following code to run test when using line_profiler
+    # td = TestDataset()
+    # td.setUpClass()
+    # td.testTSDataset()

tests/data_mid_layer_tests/test_handler.py

@@ -0,0 +1,37 @@
+import os
+import pickle
+import shutil
+import unittest
+from qlib.tests import TestAutoData
+from qlib.data import D
+from qlib.data.dataset.handler import DataHandlerLP
+
+
+class HandlerTests(TestAutoData):
+    def to_str(self, obj):
+        return "".join(str(obj).split())
+
+    def test_handler_df(self):
+        df = D.features(["sh600519"], start_time="20190101", end_time="20190201", fields=["$close"])
+        dh = DataHandlerLP.from_df(df)
+        print(dh.fetch())
+        self.assertTrue(dh._data.equals(df))
+        self.assertTrue(dh._infer is dh._data)
+        self.assertTrue(dh._learn is dh._data)
+        self.assertTrue(dh.data_loader._data is dh._data)
+        fname = "_handler_test.pkl"
+        dh.to_pickle(fname, dump_all=True)
+
+        with open(fname, "rb") as f:
+            dh_d = pickle.load(f)
+
+        self.assertTrue(dh_d._data.equals(df))
+        self.assertTrue(dh_d._infer is dh_d._data)
+        self.assertTrue(dh_d._learn is dh_d._data)
+        # Data loader will no longer be useful
+        self.assertTrue("_data" not in dh_d.data_loader.__dict__.keys())
+        os.remove(fname)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/data_mid_layer_tests/test_handler_storage.py

@@ -0,0 +1,114 @@
+import unittest
+import time
+import numpy as np
+from qlib.data import D
+from qlib.tests import TestAutoData
+
+from qlib.data.dataset.handler import DataHandlerLP
+from qlib.contrib.data.handler import check_transform_proc
+from qlib.log import TimeInspector
+
+
+class TestHandler(DataHandlerLP):
+    def __init__(
+        self,
+        instruments="csi300",
+        start_time=None,
+        end_time=None,
+        infer_processors=[],
+        learn_processors=[],
+        fit_start_time=None,
+        fit_end_time=None,
+        drop_raw=True,
+    ):
+
+        infer_processors = check_transform_proc(infer_processors, fit_start_time, fit_end_time)
+        learn_processors = check_transform_proc(learn_processors, fit_start_time, fit_end_time)
+
+        data_loader = {
+            "class": "QlibDataLoader",
+            "kwargs": {
+                "freq": "day",
+                "config": self.get_feature_config(),
+                "swap_level": False,
+            },
+        }
+
+        super().__init__(
+            instruments=instruments,
+            start_time=start_time,
+            end_time=end_time,
+            data_loader=data_loader,
+            infer_processors=infer_processors,
+            learn_processors=learn_processors,
+            drop_raw=drop_raw,
+        )
+
+    def get_feature_config(self):
+        fields = ["Ref($open, 1)", "Ref($close, 1)", "Ref($volume, 1)", "$open", "$close", "$volume"]
+        names = ["open_0", "close_0", "volume_0", "open_1", "close_1", "volume_1"]
+        return fields, names
+
+
+class TestHandlerStorage(TestAutoData):
+
+    market = "all"
+
+    start_time = "2010-01-01"
+    end_time = "2020-12-31"
+    train_end_time = "2015-12-31"
+    test_start_time = "2016-01-01"
+
+    data_handler_kwargs = {
+        "start_time": start_time,
+        "end_time": end_time,
+        "fit_start_time": start_time,
+        "fit_end_time": train_end_time,
+        "instruments": market,
+    }
+
+    def test_handler_storage(self):
+        # init data handler
+        data_handler = TestHandler(**self.data_handler_kwargs)
+
+        # init data handler with hasing storage
+        data_handler_hs = TestHandler(**self.data_handler_kwargs, infer_processors=["HashStockFormat"])
+
+        fetch_start_time = "2019-01-01"
+        fetch_end_time = "2019-12-31"
+        instruments = D.instruments(market=self.market)
+        instruments = D.list_instruments(
+            instruments=instruments, start_time=fetch_start_time, end_time=fetch_end_time, as_list=True
+        )
+
+        with TimeInspector.logt("random fetch with DataFrame Storage"):
+
+            # single stock
+            for i in range(100):
+                random_index = np.random.randint(len(instruments), size=1)[0]
+                fetch_stock = instruments[random_index]
+                data_handler.fetch(selector=(fetch_stock, slice(fetch_start_time, fetch_end_time)), level=None)
+
+            # multi stocks
+            for i in range(100):
+                random_indexs = np.random.randint(len(instruments), size=5)
+                fetch_stocks = [instruments[_index] for _index in random_indexs]
+                data_handler.fetch(selector=(fetch_stocks, slice(fetch_start_time, fetch_end_time)), level=None)
+
+        with TimeInspector.logt("random fetch with HashingStock Storage"):
+
+            # single stock
+            for i in range(100):
+                random_index = np.random.randint(len(instruments), size=1)[0]
+                fetch_stock = instruments[random_index]
+                data_handler_hs.fetch(selector=(fetch_stock, slice(fetch_start_time, fetch_end_time)), level=None)
+
+            # multi stocks
+            for i in range(100):
+                random_indexs = np.random.randint(len(instruments), size=5)
+                fetch_stocks = [instruments[_index] for _index in random_indexs]
+                data_handler_hs.fetch(selector=(fetch_stocks, slice(fetch_start_time, fetch_end_time)), level=None)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/data_mid_layer_tests/test_processor.py

@@ -0,0 +1,75 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import unittest
+import numpy as np
+from qlib.data import D
+from qlib.tests import TestAutoData
+from qlib.data.dataset.processor import MinMaxNorm, ZScoreNorm, CSZScoreNorm, CSZFillna
+
+
+class TestProcessor(TestAutoData):
+    TEST_INST = "SH600519"
+
+    def test_MinMaxNorm(self):
+        def normalize(df):
+            min_val = np.nanmin(df.values, axis=0)
+            max_val = np.nanmax(df.values, axis=0)
+            ignore = min_val == max_val
+            for _i, _con in enumerate(ignore):
+                if _con:
+                    max_val[_i] = 1
+                    min_val[_i] = 0
+            df.loc(axis=1)[df.columns] = (df.values - min_val) / (max_val - min_val)
+            return df
+
+        origin_df = D.features([self.TEST_INST], ["$high", "$open", "$low", "$close"]).tail(10)
+        origin_df["test"] = 0
+        df = origin_df.copy()
+        mmn = MinMaxNorm(fields_group=None, fit_start_time="2021-05-31", fit_end_time="2021-06-11")
+        mmn.fit(df)
+        mmn.__call__(df)
+        origin_df = normalize(origin_df)
+        assert (df == origin_df).all().all()
+
+    def test_ZScoreNorm(self):
+        def normalize(df):
+            mean_train = np.nanmean(df.values, axis=0)
+            std_train = np.nanstd(df.values, axis=0)
+            ignore = std_train == 0
+            for _i, _con in enumerate(ignore):
+                if _con:
+                    std_train[_i] = 1
+                    mean_train[_i] = 0
+            df.loc(axis=1)[df.columns] = (df.values - mean_train) / std_train
+            return df
+
+        origin_df = D.features([self.TEST_INST], ["$high", "$open", "$low", "$close"]).tail(10)
+        origin_df["test"] = 0
+        df = origin_df.copy()
+        zsn = ZScoreNorm(fields_group=None, fit_start_time="2021-05-31", fit_end_time="2021-06-11")
+        zsn.fit(df)
+        zsn.__call__(df)
+        origin_df = normalize(origin_df)
+        assert (df == origin_df).all().all()
+
+    def test_CSZFillna(self):
+        origin_df = D.features(D.instruments(market="csi300"), fields=["$high", "$open", "$low", "$close"])
+        origin_df = origin_df.groupby("datetime", group_keys=False).apply(lambda x: x[97:99])[228:238]
+        df = origin_df.copy()
+        CSZFillna(fields_group=None).__call__(df)
+        assert ~df[1:2].isna().all().all() and origin_df[1:2].isna().all().all()
+
+    def test_CSZScoreNorm(self):
+        origin_df = D.features(D.instruments(market="csi300"), fields=["$high", "$open", "$low", "$close"])
+        origin_df = origin_df.groupby("datetime", group_keys=False).apply(lambda x: x[10:12])[50:60]
+        df = origin_df.copy()
+        CSZScoreNorm(fields_group=None).__call__(df)
+        # If we use the formula directly on the original data, we cannot get the correct result,
+        # because the original data is processed by `groupby`, so we use the method of slicing,
+        # taking the 2nd group of data from the original data, to calculate and compare.
+        assert (df[2:4] == ((origin_df[2:4] - origin_df[2:4].mean()).div(origin_df[2:4].std()))).all().all()
+
+
+if __name__ == "__main__":
+    unittest.main()