update env & strategy, add workflow

examples/highfreq/data/README.md

@@ -0,0 +1,28 @@
+# High-Frequency Dataset
+
+This dataset is an example for RL high frequency trading.
+
+## Get High-Frequency Data
+
+Get high-frequency data by running the following command:
+```bash
+    python workflow.py get_data
+```
+
+## Dump & Reload & Reinitialize the Dataset
+
+
+The High-Frequency Dataset is implemented as `qlib.data.dataset.DatasetH` in the `workflow.py`. `DatatsetH` is the subclass of [`qlib.utils.serial.Serializable`](https://qlib.readthedocs.io/en/latest/advanced/serial.html), whose state can be dumped in or loaded from disk in `pickle` format.
+
+### About Reinitialization
+
+After reloading `Dataset` from disk, `Qlib` also support reinitializing the dataset. It means that users can reset some states of `Dataset` or `DataHandler` such as `instruments`, `start_time`, `end_time` and `segments`, etc.,  and generate new data according to the states.
+
+The example is given in `workflow.py`, users can run the code as follows.
+
+### Run the Code
+
+Run the example by running the following command:
+```bash
+    python workflow.py dump_and_load_dataset
+```

examples/highfreq/data/highfreq_handler.py

@@ -0,0 +1,174 @@
+from qlib.data.dataset.handler import DataHandler, DataHandlerLP
+from qlib.data.dataset.processor import Processor
+from qlib.utils import get_cls_kwargs
+from qlib.log import TimeInspector
+
+
+class HighFreqHandler(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,
+    ):
+        def check_transform_proc(proc_l):
+            new_l = []
+            for p in proc_l:
+                p["kwargs"].update(
+                    {
+                        "fit_start_time": fit_start_time,
+                        "fit_end_time": fit_end_time,
+                    }
+                )
+                new_l.append(p)
+            return new_l
+
+        infer_processors = check_transform_proc(infer_processors)
+        learn_processors = check_transform_proc(learn_processors)
+
+        data_loader = {
+            "class": "QlibDataLoader",
+            "kwargs": {
+                "config": self.get_feature_config(),
+                "swap_level": False,
+                "freq": "1min",
+            },
+        }
+        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 = []
+        names = []
+
+        template_if = "If(IsNull({1}), {0}, {1})"
+        template_paused = "Select(Or(IsNull($paused), Eq($paused, 0.0)), {0})"
+        template_fillnan = "BFillNan(FFillNan({0}))"
+        # Because there is no vwap field in the yahoo data, a method similar to Simpson integration is used to approximate vwap
+        simpson_vwap = "($open + 2*$high + 2*$low + $close)/6"
+
+        def get_normalized_price_feature(price_field, shift=0):
+            """Get normalized price feature ops"""
+            if shift == 0:
+                template_norm = "Cut({0}/Ref(DayLast({1}), 240), 240, None)"
+            else:
+                template_norm = "Cut(Ref({0}, " + str(shift) + ")/Ref(DayLast({1}), 240), 240, None)"
+
+            feature_ops = template_norm.format(
+                template_if.format(
+                    template_fillnan.format(template_paused.format("$close")),
+                    template_paused.format(price_field),
+                ),
+                template_fillnan.format(template_paused.format("$close")),
+            )
+            return feature_ops
+
+        fields += [get_normalized_price_feature("$open", 0)]
+        fields += [get_normalized_price_feature("$high", 0)]
+        fields += [get_normalized_price_feature("$low", 0)]
+        fields += [get_normalized_price_feature("$close", 0)]
+        fields += [get_normalized_price_feature(simpson_vwap, 0)]
+        names += ["$open", "$high", "$low", "$close", "$vwap"]
+
+        fields += [get_normalized_price_feature("$open", 240)]
+        fields += [get_normalized_price_feature("$high", 240)]
+        fields += [get_normalized_price_feature("$low", 240)]
+        fields += [get_normalized_price_feature("$close", 240)]
+        fields += [get_normalized_price_feature(simpson_vwap, 240)]
+        names += ["$open_1", "$high_1", "$low_1", "$close_1", "$vwap_1"]
+
+        fields += [
+            "Cut({0}/Ref(DayLast(Mean({0}, 7200)), 240), 240, None)".format(
+                "If(IsNull({0}), 0, If(Or(Gt({1}, Mul(1.001, {3})), Lt({1}, Mul(0.999, {2}))), 0, {0}))".format(
+                    template_paused.format("$volume"),
+                    template_paused.format(simpson_vwap),
+                    template_paused.format("$low"),
+                    template_paused.format("$high"),
+                )
+            )
+        ]
+        names += ["$volume"]
+        fields += [
+            "Cut(Ref({0}, 240)/Ref(DayLast(Mean({0}, 7200)), 240), 240, None)".format(
+                "If(IsNull({0}), 0, If(Or(Gt({1}, Mul(1.001, {3})), Lt({1}, Mul(0.999, {2}))), 0, {0}))".format(
+                    template_paused.format("$volume"),
+                    template_paused.format(simpson_vwap),
+                    template_paused.format("$low"),
+                    template_paused.format("$high"),
+                )
+            )
+        ]
+        names += ["$volume_1"]
+
+        fields += ["Cut({0}, 240, None)".format(template_paused.format("Date($close)"))]
+        names += ["date"]
+        return fields, names
+
+
+class HighFreqBacktestHandler(DataHandler):
+    def __init__(
+        self,
+        instruments="csi300",
+        start_time=None,
+        end_time=None,
+    ):
+        data_loader = {
+            "class": "QlibDataLoader",
+            "kwargs": {
+                "config": self.get_feature_config(),
+                "swap_level": False,
+                "freq": "1min",
+            },
+        }
+        super().__init__(
+            instruments=instruments,
+            start_time=start_time,
+            end_time=end_time,
+            data_loader=data_loader,
+        )
+
+    def get_feature_config(self):
+        fields = []
+        names = []
+
+        template_if = "If(IsNull({1}), {0}, {1})"
+        template_paused = "Select(Or(IsNull($paused), Eq($paused, 0.0)), {0})"
+        template_fillnan = "BFillNan(FFillNan({0}))"
+        # Because there is no vwap field in the yahoo data, a method similar to Simpson integration is used to approximate vwap
+        simpson_vwap = "($open + 2*$high + 2*$low + $close)/6"
+        fields += [
+            "Cut({0}, 240, None)".format(template_fillnan.format(template_paused.format("$close"))),
+        ]
+        names += ["$close0"]
+        fields += [
+            "Cut({0}, 240, None)".format(
+                template_if.format(
+                    template_fillnan.format(template_paused.format("$close")),
+                    template_paused.format(simpson_vwap),
+                )
+            )
+        ]
+        names += ["$vwap0"]
+        fields += [
+            "Cut(If(IsNull({0}), 0, If(Or(Gt({1}, Mul(1.001, {3})), Lt({1}, Mul(0.999, {2}))), 0, {0})), 240, None)".format(
+                template_paused.format("$volume"),
+                template_paused.format(simpson_vwap),
+                template_paused.format("$low"),
+                template_paused.format("$high"),
+            )
+        ]
+        names += ["$volume0"]
+
+        return fields, names

examples/highfreq/data/highfreq_ops.py

@@ -0,0 +1,190 @@
+import numpy as np
+import pandas as pd
+import importlib
+from qlib.data.ops import ElemOperator, PairOperator
+from qlib.config import C
+from qlib.data.cache import H
+from qlib.data.data import Cal
+
+
+def get_calendar_day(freq="day", future=False):
+    """Load High-Freq Calendar Date Using Memcache.
+
+    Parameters
+    ----------
+    freq : str
+        frequency of read calendar file.
+    future : bool
+        whether including future trading day.
+
+    Returns
+    -------
+    _calendar:
+        array of date.
+    """
+    flag = f"{freq}_future_{future}_day"
+    if flag in H["c"]:
+        _calendar = H["c"][flag]
+    else:
+        _calendar = np.array(list(map(lambda x: x.date(), Cal.load_calendar(freq, future))))
+        H["c"][flag] = _calendar
+    return _calendar
+
+
+class DayLast(ElemOperator):
+    """DayLast Operator
+
+    Parameters
+    ----------
+    feature : Expression
+        feature instance
+
+    Returns
+    ----------
+    feature:
+        a series of that each value equals the last value of its day
+    """
+
+    def _load_internal(self, instrument, start_index, end_index, freq):
+        _calendar = get_calendar_day(freq=freq)
+        series = self.feature.load(instrument, start_index, end_index, freq)
+        return series.groupby(_calendar[series.index]).transform("last")
+
+
+class FFillNan(ElemOperator):
+    """FFillNan Operator
+
+    Parameters
+    ----------
+    feature : Expression
+        feature instance
+
+    Returns
+    ----------
+    feature:
+        a forward fill nan feature
+    """
+
+    def _load_internal(self, instrument, start_index, end_index, freq):
+        series = self.feature.load(instrument, start_index, end_index, freq)
+        return series.fillna(method="ffill")
+
+
+class BFillNan(ElemOperator):
+    """BFillNan Operator
+
+    Parameters
+    ----------
+    feature : Expression
+        feature instance
+
+    Returns
+    ----------
+    feature:
+        a backfoward fill nan feature
+    """
+
+    def _load_internal(self, instrument, start_index, end_index, freq):
+        series = self.feature.load(instrument, start_index, end_index, freq)
+        return series.fillna(method="bfill")
+
+
+class Date(ElemOperator):
+    """Date Operator
+
+    Parameters
+    ----------
+    feature : Expression
+        feature instance
+
+    Returns
+    ----------
+    feature:
+        a series of that each value is the date corresponding to feature.index
+    """
+
+    def _load_internal(self, instrument, start_index, end_index, freq):
+        _calendar = get_calendar_day(freq=freq)
+        series = self.feature.load(instrument, start_index, end_index, freq)
+        return pd.Series(_calendar[series.index], index=series.index)
+
+
+class Select(PairOperator):
+    """Select Operator
+
+    Parameters
+    ----------
+    feature_left : Expression
+        feature instance, select condition
+    feature_right : Expression
+        feature instance, select value
+
+    Returns
+    ----------
+    feature:
+        value(feature_right) that meets the condition(feature_left)
+
+    """
+
+    def _load_internal(self, instrument, start_index, end_index, freq):
+        series_condition = self.feature_left.load(instrument, start_index, end_index, freq)
+        series_feature = self.feature_right.load(instrument, start_index, end_index, freq)
+        return series_feature.loc[series_condition]
+
+
+class IsNull(ElemOperator):
+    """IsNull Operator
+
+    Parameters
+    ----------
+    feature : Expression
+        feature instance
+
+    Returns
+    ----------
+    feature:
+        A series indicating whether the feature is nan
+    """
+
+    def _load_internal(self, instrument, start_index, end_index, freq):
+        series = self.feature.load(instrument, start_index, end_index, freq)
+        return series.isnull()
+
+
+class Cut(ElemOperator):
+    """Cut Operator
+
+    Parameters
+    ----------
+    feature : Expression
+        feature instance
+    l : int
+        l > 0, delete the first l elements of feature (default is None, which means 0)
+    r : int
+        r < 0, delete the last -r elements of feature (default is None, which means 0)
+    Returns
+    ----------
+    feature:
+        A series with the first l and last -r elements deleted from the feature.
+        Note: It is deleted from the raw data, not the sliced data
+    """
+
+    def __init__(self, feature, l=None, r=None):
+        self.l = l
+        self.r = r
+        if (self.l is not None and self.l <= 0) or (self.r is not None and self.r >= 0):
+            raise ValueError("Cut operator l shoud > 0 and r should < 0")
+
+        super(Cut, self).__init__(feature)
+
+    def _load_internal(self, instrument, start_index, end_index, freq):
+        series = self.feature.load(instrument, start_index, end_index, freq)
+        return series.iloc[self.l : self.r]
+
+    def get_extended_window_size(self):
+        ll = 0 if self.l is None else self.l
+        rr = 0 if self.r is None else abs(self.r)
+        lft_etd, rght_etd = self.feature.get_extended_window_size()
+        lft_etd = lft_etd + ll
+        rght_etd = rght_etd + rr
+        return lft_etd, rght_etd

examples/highfreq/data/highfreq_processor.py

@@ -0,0 +1,72 @@
+import numpy as np
+import pandas as pd
+from qlib.data.dataset.processor import Processor
+from qlib.data.dataset.utils import fetch_df_by_index
+
+
+class HighFreqNorm(Processor):
+    def __init__(self, fit_start_time, fit_end_time):
+        self.fit_start_time = fit_start_time
+        self.fit_end_time = fit_end_time
+
+    def fit(self, df_features):
+        fetch_df = fetch_df_by_index(df_features, slice(self.fit_start_time, self.fit_end_time), level="datetime")
+        del df_features
+        df_values = fetch_df.values
+        names = {
+            "price": slice(0, 10),
+            "volume": slice(10, 12),
+        }
+        self.feature_med = {}
+        self.feature_std = {}
+        self.feature_vmax = {}
+        self.feature_vmin = {}
+        for name, name_val in names.items():
+            part_values = df_values[:, name_val].astype(np.float32)
+            if name == "volume":
+                part_values = np.log1p(part_values)
+            self.feature_med[name] = np.nanmedian(part_values)
+            part_values = part_values - self.feature_med[name]
+            self.feature_std[name] = np.nanmedian(np.absolute(part_values)) * 1.4826 + 1e-12
+            part_values = part_values / self.feature_std[name]
+            self.feature_vmax[name] = np.nanmax(part_values)
+            self.feature_vmin[name] = np.nanmin(part_values)
+
+    def __call__(self, df_features):
+        df_features.set_index("date", append=True, drop=True, inplace=True)
+        df_values = df_features.values
+        names = {
+            "price": slice(0, 10),
+            "volume": slice(10, 12),
+        }
+
+        for name, name_val in names.items():
+            if name == "volume":
+                df_values[:, name_val] = np.log1p(df_values[:, name_val])
+            df_values[:, name_val] -= self.feature_med[name]
+            df_values[:, name_val] /= self.feature_std[name]
+            slice0 = df_values[:, name_val] > 3.0
+            slice1 = df_values[:, name_val] > 3.5
+            slice2 = df_values[:, name_val] < -3.0
+            slice3 = df_values[:, name_val] < -3.5
+
+            df_values[:, name_val][slice0] = (
+                3.0 + (df_values[:, name_val][slice0] - 3.0) / (self.feature_vmax[name] - 3) * 0.5
+            )
+            df_values[:, name_val][slice1] = 3.5
+            df_values[:, name_val][slice2] = (
+                -3.0 - (df_values[:, name_val][slice2] + 3.0) / (self.feature_vmin[name] + 3) * 0.5
+            )
+            df_values[:, name_val][slice3] = -3.5
+        idx = df_features.index.droplevel("datetime").drop_duplicates()
+        idx.set_names(["instrument", "datetime"], inplace=True)
+
+        # Reshape is specifically for adapting to RL high-freq executor
+        feat = df_values[:, [0, 1, 2, 3, 4, 10]].reshape(-1, 6 * 240)
+        feat_1 = df_values[:, [5, 6, 7, 8, 9, 11]].reshape(-1, 6 * 240)
+        df_new_features = pd.DataFrame(
+            data=np.concatenate((feat, feat_1), axis=1),
+            index=idx,
+            columns=["FEATURE_%d" % i for i in range(12 * 240)],
+        ).sort_index()
+        return df_new_features

examples/highfreq/data/workflow.py

@@ -0,0 +1,184 @@
+#  Copyright (c) Microsoft Corporation.
+#  Licensed under the MIT License.
+
+import sys
+import fire
+from pathlib import Path
+
+import qlib
+import pickle
+import numpy as np
+import pandas as pd
+from qlib.config import REG_CN, HIGH_FREQ_CONFIG
+from qlib.contrib.model.gbdt import LGBModel
+from qlib.contrib.data.handler import Alpha158
+from qlib.contrib.strategy.strategy import TopkDropoutStrategy
+from qlib.contrib.evaluate import (
+    backtest as normal_backtest,
+    risk_analysis,
+)
+
+from qlib.utils import init_instance_by_config, exists_qlib_data
+from qlib.data.dataset.handler import DataHandlerLP
+from qlib.data.ops import Operators
+from qlib.data.data import Cal
+from qlib.tests.data import GetData
+
+from highfreq_ops import get_calendar_day, DayLast, FFillNan, BFillNan, Date, Select, IsNull, Cut
+
+
+class HighfreqWorkflow(object):
+
+    SPEC_CONF = {"custom_ops": [DayLast, FFillNan, BFillNan, Date, Select, IsNull, Cut], "expression_cache": None}
+
+    MARKET = "all"
+    BENCHMARK = "SH000300"
+
+    start_time = "2020-09-15 00:00:00"
+    end_time = "2021-01-18 16:00:00"
+    train_end_time = "2020-11-30 16:00:00"
+    test_start_time = "2020-12-01 00:00:00"
+
+    DATA_HANDLER_CONFIG0 = {
+        "start_time": start_time,
+        "end_time": end_time,
+        "fit_start_time": start_time,
+        "fit_end_time": train_end_time,
+        "instruments": MARKET,
+        "infer_processors": [{"class": "HighFreqNorm", "module_path": "highfreq_processor", "kwargs": {}}],
+    }
+    DATA_HANDLER_CONFIG1 = {
+        "start_time": start_time,
+        "end_time": end_time,
+        "instruments": MARKET,
+    }
+
+    task = {
+        "dataset": {
+            "class": "DatasetH",
+            "module_path": "qlib.data.dataset",
+            "kwargs": {
+                "handler": {
+                    "class": "HighFreqHandler",
+                    "module_path": "highfreq_handler",
+                    "kwargs": DATA_HANDLER_CONFIG0,
+                },
+                "segments": {
+                    "train": (start_time, train_end_time),
+                    "test": (
+                        test_start_time,
+                        end_time,
+                    ),
+                },
+            },
+        },
+        "dataset_backtest": {
+            "class": "DatasetH",
+            "module_path": "qlib.data.dataset",
+            "kwargs": {
+                "handler": {
+                    "class": "HighFreqBacktestHandler",
+                    "module_path": "highfreq_handler",
+                    "kwargs": DATA_HANDLER_CONFIG1,
+                },
+                "segments": {
+                    "train": (start_time, train_end_time),
+                    "test": (
+                        test_start_time,
+                        end_time,
+                    ),
+                },
+            },
+        },
+    }
+
+    def _init_qlib(self):
+        """initialize qlib"""
+        # use yahoo_cn_1min data
+        QLIB_INIT_CONFIG = {**HIGH_FREQ_CONFIG, **self.SPEC_CONF}
+        provider_uri = QLIB_INIT_CONFIG.get("provider_uri")
+        if not exists_qlib_data(provider_uri):
+            print(f"Qlib data is not found in {provider_uri}")
+            GetData().qlib_data(target_dir=provider_uri, interval="1min", region=REG_CN)
+        qlib.init(**QLIB_INIT_CONFIG)
+
+    def _prepare_calender_cache(self):
+        """preload the calendar for cache"""
+
+        # This code used the copy-on-write feature of Linux to avoid calculating the calendar multiple times in the subprocess
+        # This code may accelerate, but may be not useful on Windows and Mac Os
+        Cal.calendar(freq="1min")
+        get_calendar_day(freq="1min")
+
+    def get_data(self):
+        """use dataset to get highreq data"""
+        self._init_qlib()
+        self._prepare_calender_cache()
+
+        dataset = init_instance_by_config(self.task["dataset"])
+        xtrain, xtest = dataset.prepare(["train", "test"])
+        print(xtrain, xtest)
+
+        dataset_backtest = init_instance_by_config(self.task["dataset_backtest"])
+        backtest_train, backtest_test = dataset_backtest.prepare(["train", "test"])
+        print(backtest_train, backtest_test)
+
+        return
+
+    def dump_and_load_dataset(self):
+        """dump and load dataset state on disk"""
+        self._init_qlib()
+        self._prepare_calender_cache()
+        dataset = init_instance_by_config(self.task["dataset"])
+        dataset_backtest = init_instance_by_config(self.task["dataset_backtest"])
+
+        ##=============dump dataset=============
+        dataset.to_pickle(path="dataset.pkl")
+        dataset_backtest.to_pickle(path="dataset_backtest.pkl")
+
+        del dataset, dataset_backtest
+        ##=============reload dataset=============
+        with open("dataset.pkl", "rb") as file_dataset:
+            dataset = pickle.load(file_dataset)
+
+        with open("dataset_backtest.pkl", "rb") as file_dataset_backtest:
+            dataset_backtest = pickle.load(file_dataset_backtest)
+
+        self._prepare_calender_cache()
+        ##=============reinit dataset=============
+        dataset.init(
+            handler_kwargs={
+                "init_type": DataHandlerLP.IT_LS,
+                "start_time": "2021-01-19 00:00:00",
+                "end_time": "2021-01-25 16:00:00",
+            },
+            segment_kwargs={
+                "test": (
+                    "2021-01-19 00:00:00",
+                    "2021-01-25 16:00:00",
+                ),
+            },
+        )
+        dataset_backtest.init(
+            handler_kwargs={
+                "start_time": "2021-01-19 00:00:00",
+                "end_time": "2021-01-25 16:00:00",
+            },
+            segment_kwargs={
+                "test": (
+                    "2021-01-19 00:00:00",
+                    "2021-01-25 16:00:00",
+                ),
+            },
+        )
+
+        ##=============get data=============
+        xtest = dataset.prepare(["test"])
+        backtest_test = dataset_backtest.prepare(["test"])
+
+        print(xtest, backtest_test)
+        return
+
+
+if __name__ == "__main__":
+    fire.Fire(HighfreqWorkflow)