Merge pull request #650 from microsoft/backtest_improve

Improve the backtest design and APIs

qlib/contrib/data/utils/sepdf.py

@@ -0,0 +1,183 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+import pandas as pd
+from typing import Dict, Iterable
+
+
+def align_index(df_dict, join):
+    res = {}
+    for k, df in df_dict.items():
+        if join is not None and k != join:
+            df = df.reindex(df_dict[join].index)
+        res[k] = df
+    return res
+
+
+# Mocking the pd.DataFrame class
+class SepDataFrame:
+    """
+    (Sep)erate DataFrame
+    We usually concat multiple dataframe to be processed together(Such as feature, label, weight, filter).
+    However, they are usally be used seperately at last.
+    This will result in extra cost for concating and spliting data(reshaping and copying data in the memory is very expensive)
+
+    SepDataFrame tries to act like a DataFrame whose column with multiindex
+    """
+
+    def __init__(self, df_dict: Dict[str, pd.DataFrame], join: str, skip_align=False):
+        """
+        initialize the data based on the dataframe dictionary
+
+        Parameters
+        ----------
+        df_dict : Dict[str, pd.DataFrame]
+            dataframe dictionary
+        join : str
+            how to join the data
+            It will reindex the dataframe based on the join key.
+            If join is None, the reindex step will be skipped
+
+        skip_align :
+            for some cases, we can improve performance by skipping aligning index
+        """
+        self.join = join
+
+        if skip_align:
+            self._df_dict = df_dict
+        else:
+            self._df_dict = align_index(df_dict, join)
+
+    @property
+    def loc(self):
+        return SDFLoc(self, join=self.join)
+
+    @property
+    def index(self):
+        return self._df_dict[self.join].index
+
+    def apply_each(self, method: str, skip_align=True, *args, **kwargs):
+        """
+        Assumptions:
+        - inplace methods will return None
+        """
+        inplace = False
+        df_dict = {}
+        for k, df in self._df_dict.items():
+            df_dict[k] = getattr(df, method)(*args, **kwargs)
+            if df_dict[k] is None:
+                inplace = True
+        if not inplace:
+            return SepDataFrame(df_dict=df_dict, join=self.join, skip_align=skip_align)
+
+    def sort_index(self, *args, **kwargs):
+        return self.apply_each("sort_index", True, *args, **kwargs)
+
+    def copy(self, *args, **kwargs):
+        return self.apply_each("copy", True, *args, **kwargs)
+
+    def _update_join(self):
+        if self.join not in self:
+            self.join = next(iter(self._df_dict.keys()))
+
+    def __getitem__(self, item):
+        return self._df_dict[item]
+
+    def __setitem__(self, item: str, df: pd.DataFrame):
+        # TODO: consider the join behavior
+        self._df_dict[item] = df
+
+    def __delitem__(self, item: str):
+        del self._df_dict[item]
+        self._update_join()
+
+    def __contains__(self, item):
+        return item in self._df_dict
+
+    def __len__(self):
+        return len(self._df_dict[self.join])
+
+    def droplevel(self, *args, **kwargs):
+        raise NotImplementedError(f"Please implement the `droplevel` method")
+
+    @property
+    def columns(self):
+        dfs = []
+        for k, df in self._df_dict.items():
+            df = df.head(0)
+            df.columns = pd.MultiIndex.from_product([[k], df.columns])
+            dfs.append(df)
+        return pd.concat(dfs, axis=1).columns
+
+    # Useless methods
+    @staticmethod
+    def merge(df_dict: Dict[str, pd.DataFrame], join: str):
+        all_df = df_dict[join]
+        for k, df in df_dict.items():
+            if k != join:
+                all_df = all_df.join(df)
+        return all_df
+
+
+class SDFLoc:
+    """Mock Class"""
+
+    def __init__(self, sdf: SepDataFrame, join):
+        self._sdf = sdf
+        self.axis = None
+        self.join = join
+
+    def __call__(self, axis):
+        self.axis = axis
+        return self
+
+    def __getitem__(self, args):
+        if self.axis == 1:
+            if isinstance(args, str):
+                return self._sdf[args]
+            elif isinstance(args, (tuple, list)):
+                new_df_dict = {k: self._sdf[k] for k in args}
+                return SepDataFrame(new_df_dict, join=self.join if self.join in args else args[0], skip_align=True)
+            else:
+                raise NotImplementedError(f"This type of input is not supported")
+        elif self.axis == 0:
+            return SepDataFrame(
+                {k: df.loc(axis=0)[args] for k, df in self._sdf._df_dict.items()}, join=self.join, skip_align=True
+            )
+        else:
+            df = self._sdf
+            if isinstance(args, tuple):
+                ax0, *ax1 = args
+                if len(ax1) == 0:
+                    ax1 = None
+                if ax1 is not None:
+                    df = df.loc(axis=1)[ax1]
+                if ax0 is not None:
+                    df = df.loc(axis=0)[ax0]
+                return df
+            else:
+                return df.loc(axis=0)[args]
+
+
+# Patch pandas DataFrame
+# Tricking isinstance to accept SepDataFrame as its subclass
+import builtins
+
+
+def _isinstance(instance, cls):
+    if isinstance_orig(instance, SepDataFrame):  # pylint: disable=E0602
+        if isinstance(cls, Iterable):
+            for c in cls:
+                if c is pd.DataFrame:
+                    return True
+        elif cls is pd.DataFrame:
+            return True
+    return isinstance_orig(instance, cls)  # pylint: disable=E0602
+
+
+builtins.isinstance_orig = builtins.isinstance
+builtins.isinstance = _isinstance
+
+if __name__ == "__main__":
+    sdf = SepDataFrame({}, join=None)
+    print(isinstance(sdf, (pd.DataFrame,)))
+    print(isinstance(sdf, pd.DataFrame))

qlib/contrib/strategy/__init__.py

@@ -2,7 +2,7 @@
 # Licensed under the MIT License.
 
 
-from .model_strategy import (
+from .signal_strategy import (
     TopkDropoutStrategy,
     WeightStrategyBase,
 )

qlib/contrib/strategy/cost_control.py

@@ -6,7 +6,7 @@ This strategy is not well maintained
 
 
 from .order_generator import OrderGenWInteract
-from .model_strategy import WeightStrategyBase
+from .signal_strategy import WeightStrategyBase
 import copy
 
 

qlib/contrib/strategy/order_generator.py

@@ -80,18 +80,22 @@ class OrderGenWInteract(OrderGenerator):
 
         :rtype: list
         """
+        if target_weight_position is None:
+            return []
+
         # calculate current_tradable_value
         current_amount_dict = current.get_stock_amount_dict()
+
         current_total_value = trade_exchange.calculate_amount_position_value(
             amount_dict=current_amount_dict,
-            trade_start_time=trade_start_time,
-            trade_end_time=trade_end_time,
+            start_time=trade_start_time,
+            end_time=trade_end_time,
             only_tradable=False,
         )
         current_tradable_value = trade_exchange.calculate_amount_position_value(
             amount_dict=current_amount_dict,
-            trade_start_time=trade_start_time,
-            trade_end_time=trade_end_time,
+            start_time=trade_start_time,
+            end_time=trade_end_time,
             only_tradable=True,
         )
         # add cash
@@ -105,9 +109,7 @@ class OrderGenWInteract(OrderGenerator):
             # value. Then just sell all the stocks
             target_amount_dict = copy.deepcopy(current_amount_dict.copy())
             for stock_id in list(target_amount_dict.keys()):
-                if trade_exchange.is_stock_tradable(
-                    stock_id, trade_start_time=trade_start_time, trade_end_time=trade_end_time
-                ):
+                if trade_exchange.is_stock_tradable(stock_id, start_time=trade_start_time, end_time=trade_end_time):
                     del target_amount_dict[stock_id]
         else:
             # consider cost rate
@@ -118,16 +120,16 @@ class OrderGenWInteract(OrderGenerator):
             target_amount_dict = trade_exchange.generate_amount_position_from_weight_position(
                 weight_position=target_weight_position,
                 cash=current_tradable_value,
-                trade_start_time=trade_start_time,
-                trade_end_time=trade_end_time,
+                start_time=trade_start_time,
+                end_time=trade_end_time,
             )
         order_list = trade_exchange.generate_order_for_target_amount_position(
             target_position=target_amount_dict,
             current_position=current_amount_dict,
-            trade_start_time=trade_start_time,
-            trade_end_time=trade_end_time,
+            start_time=trade_start_time,
+            end_time=trade_end_time,
         )
-        return TradeDecisionWO(order_list, self)
+        return order_list
 
 
 class OrderGenWOInteract(OrderGenerator):
@@ -163,8 +165,11 @@ class OrderGenWOInteract(OrderGenerator):
         :param trade_date:
         :type trade_date: pd.Timestamp
 
-        :rtype: list
+        :rtype: list of generated orders
         """
+        if target_weight_position is None:
+            return []
+
         risk_total_value = risk_degree * current.calculate_value()
 
         current_stock = current.get_stock_list()
@@ -172,13 +177,17 @@ class OrderGenWOInteract(OrderGenerator):
         for stock_id in target_weight_position:
             # Current rule will ignore the stock that not hold and cannot be traded at predict date
             if trade_exchange.is_stock_tradable(
-                stock_id=stock_id, trade_start_time=trade_start_time, trade_end_time=trade_end_time
+                stock_id=stock_id, start_time=trade_start_time, end_time=trade_end_time
+            ) and trade_exchange.is_stock_tradable(
+                stock_id=stock_id, start_time=pred_start_time, end_time=pred_end_time
             ):
                 amount_dict[stock_id] = (
                     risk_total_value
                     * target_weight_position[stock_id]
-                    / trade_exchange.get_close(stock_id, trade_start_time=pred_start_time, trade_end_time=pred_end_time)
+                    / trade_exchange.get_close(stock_id, start_time=pred_start_time, end_time=pred_end_time)
                 )
+                # TODO: Qlib use None to represent trading suspension. So last close price can't be the estimated trading price.
+                # Maybe a close price with forward fill will be a better solution.
             elif stock_id in current_stock:
                 amount_dict[stock_id] = (
                     risk_total_value * target_weight_position[stock_id] / current.get_stock_price(stock_id)
@@ -188,7 +197,7 @@ class OrderGenWOInteract(OrderGenerator):
         order_list = trade_exchange.generate_order_for_target_amount_position(
             target_position=amount_dict,
             current_position=current.get_stock_amount_dict(),
-            trade_start_time=trade_start_time,
-            trade_end_time=trade_end_time,
+            start_time=trade_start_time,
+            end_time=trade_end_time,
         )
-        return TradeDecisionWO(order_list, self)
+        return order_list

qlib/contrib/strategy/rule_strategy.py

@@ -1,3 +1,5 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
 from pathlib import Path
 import warnings
 import numpy as np

qlib/contrib/strategy/signal_strategy.py

@@ -1,27 +1,33 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
 import copy
+from qlib.backtest.signal import Signal, create_signal_from
+from typing import Dict, List, Text, Tuple, Union
+from qlib.data.dataset import Dataset
+from qlib.model.base import BaseModel
 from qlib.backtest.position import Position
 import warnings
 import numpy as np
 import pandas as pd
 
 from ...utils.resam import resam_ts_data
-from ...strategy.base import ModelStrategy
+from ...strategy.base import BaseStrategy
 from ...backtest.decision import Order, BaseTradeDecision, OrderDir, TradeDecisionWO
 
 from .order_generator import OrderGenWInteract
 
 
-class TopkDropoutStrategy(ModelStrategy):
+class TopkDropoutStrategy(BaseStrategy):
     # TODO:
     # 1. Supporting leverage the get_range_limit result from the decision
     # 2. Supporting alter_outer_trade_decision
     # 3. Supporting checking the availability of trade decision
     def __init__(
         self,
-        model,
-        dataset,
+        *,
         topk,
         n_drop,
+        signal: Union[Signal, Tuple[BaseModel, Dataset], List, Dict, Text, pd.Series, pd.DataFrame] = None,
         method_sell="bottom",
         method_buy="top",
         risk_degree=0.95,
@@ -30,6 +36,8 @@ class TopkDropoutStrategy(ModelStrategy):
         trade_exchange=None,
         level_infra=None,
         common_infra=None,
+        model=None,
+        dataset=None,
         **kwargs,
     ):
         """
@@ -39,6 +47,9 @@ class TopkDropoutStrategy(ModelStrategy):
             the number of stocks in the portfolio.
         n_drop : int
             number of stocks to be replaced in each trading date.
+        signal :
+            the information to describe a signal. Please refer to the docs of `qlib.backtest.signal.create_signal_from`
+            the decision of the strategy will base on the given signal
         method_sell : str
             dropout method_sell, random/bottom.
         method_buy : str
@@ -64,7 +75,7 @@ class TopkDropoutStrategy(ModelStrategy):
 
         """
         super(TopkDropoutStrategy, self).__init__(
-            model, dataset, level_infra=level_infra, common_infra=common_infra, trade_exchange=trade_exchange, **kwargs
+            level_infra=level_infra, common_infra=common_infra, trade_exchange=trade_exchange, **kwargs
         )
         self.topk = topk
         self.n_drop = n_drop
@@ -74,6 +85,13 @@ class TopkDropoutStrategy(ModelStrategy):
         self.hold_thresh = hold_thresh
         self.only_tradable = only_tradable
 
+        # This is trying to be compatible with previous version of qlib task config
+        if model is not None and dataset is not None:
+            warnings.warn("`model` `dataset` is deprecated; use `signal`.", DeprecationWarning)
+            signal = model, dataset
+
+        self.signal: Signal = create_signal_from(signal)
+
     def get_risk_degree(self, trade_step=None):
         """get_risk_degree
         Return the proportion of your total value you will used in investment.
@@ -87,7 +105,7 @@ class TopkDropoutStrategy(ModelStrategy):
         trade_step = self.trade_calendar.get_trade_step()
         trade_start_time, trade_end_time = self.trade_calendar.get_step_time(trade_step)
         pred_start_time, pred_end_time = self.trade_calendar.get_step_time(trade_step, shift=1)
-        pred_score = resam_ts_data(self.pred_scores, start_time=pred_start_time, end_time=pred_end_time, method="last")
+        pred_score = self.signal.get_signal(start_time=pred_start_time, end_time=pred_end_time)
         if pred_score is None:
             return TradeDecisionWO([], self)
         if self.only_tradable:
@@ -235,15 +253,15 @@ class TopkDropoutStrategy(ModelStrategy):
         return TradeDecisionWO(sell_order_list + buy_order_list, self)
 
 
-class WeightStrategyBase(ModelStrategy):
+class WeightStrategyBase(BaseStrategy):
     # TODO:
     # 1. Supporting leverage the get_range_limit result from the decision
     # 2. Supporting alter_outer_trade_decision
     # 3. Supporting checking the availability of trade decision
     def __init__(
         self,
-        model,
-        dataset,
+        *,
+        signal: Union[Signal, Tuple[BaseModel, Dataset], List, Dict, Text, pd.Series, pd.DataFrame],
         order_generator_cls_or_obj=OrderGenWInteract,
         trade_exchange=None,
         level_infra=None,
@@ -251,6 +269,9 @@ class WeightStrategyBase(ModelStrategy):
         **kwargs,
     ):
         """
+        signal :
+            the information to describe a signal. Please refer to the docs of `qlib.backtest.signal.create_signal_from`
+            the decision of the strategy will base on the given signal
         trade_exchange : Exchange
             exchange that provides market info, used to deal order and generate report
             - If `trade_exchange` is None, self.trade_exchange will be set with common_infra
@@ -260,13 +281,15 @@ class WeightStrategyBase(ModelStrategy):
                 - In minutely execution, the daily exchange is not usable, only the minutely exchange is recommended.
         """
         super(WeightStrategyBase, self).__init__(
-            model, dataset, level_infra=level_infra, common_infra=common_infra, trade_exchange=trade_exchange, **kwargs
+            level_infra=level_infra, common_infra=common_infra, trade_exchange=trade_exchange, **kwargs
         )
         if isinstance(order_generator_cls_or_obj, type):
             self.order_generator = order_generator_cls_or_obj()
         else:
             self.order_generator = order_generator_cls_or_obj
 
+        self.signal: Signal = create_signal_from(signal)
+
     def get_risk_degree(self, trade_step=None):
         """get_risk_degree
         Return the proportion of your total value you will used in investment.
@@ -298,7 +321,7 @@ class WeightStrategyBase(ModelStrategy):
         trade_step = self.trade_calendar.get_trade_step()
         trade_start_time, trade_end_time = self.trade_calendar.get_step_time(trade_step)
         pred_start_time, pred_end_time = self.trade_calendar.get_step_time(trade_step, shift=1)
-        pred_score = resam_ts_data(self.pred_scores, start_time=pred_start_time, end_time=pred_end_time, method="last")
+        pred_score = self.signal.get_signal(start_time=pred_start_time, end_time=pred_end_time)
         if pred_score is None:
             return TradeDecisionWO([], self)
         current_temp = copy.deepcopy(self.trade_position)

qlib/contrib/workflow/record_temp.py

@@ -49,7 +49,7 @@ class MultiSegRecord(RecordTemp):
 
             if save:
                 save_name = "results-{:}.pkl".format(key)
-                self.recorder.save_objects(**{save_name: results})
+                self.save(**{save_name: results})
                 logger.info(
                     "The record '{:}' has been saved as the artifact of the Experiment {:}".format(
                         save_name, self.recorder.experiment_id
@@ -79,9 +79,8 @@ class SignalMseRecord(RecordTemp):
         metrics = {"MSE": mse, "RMSE": np.sqrt(mse)}
         objects = {"mse.pkl": mse, "rmse.pkl": np.sqrt(mse)}
         self.recorder.log_metrics(**metrics)
-        self.recorder.save_objects(**objects, artifact_path=self.get_path())
+        self.save(**objects)
         logger.info("The evaluation results in SignalMseRecord is {:}".format(metrics))
 
     def list(self):
-        paths = [self.get_path("mse.pkl"), self.get_path("rmse.pkl")]
-        return paths
+        return ["mse.pkl", "rmse.pkl"]