Merge pull request #520 from wangwenxi-handsome/nested_decision_exe

abstract Quote class from Exchange

qlib/backtest/exchange.py

@@ -5,7 +5,7 @@
 from qlib.backtest.position import Position
 import random
 import logging
-from typing import List, Tuple, Union
+from typing import List, Tuple, Union, Callable, Iterable
 
 import numpy as np
 import pandas as pd
@@ -16,6 +16,7 @@ from ..config import C, REG_CN
 from ..utils.resam import resam_ts_data, ts_data_last
 from ..log import get_module_logger
 from .order import Order, OrderDir, OrderHelper
+from .high_performance_ds import PandasQuote
 
 
 class Exchange:
@@ -33,6 +34,7 @@ class Exchange:
         close_cost=0.0025,
         min_cost=5,
         extra_quote=None,
+        quote_cls=PandasQuote,
         **kwargs,
     ):
         """__init__
@@ -103,10 +105,11 @@ class Exchange:
 
         # TODO: the quote, trade_dates, codes are not necessray.
         # It is just for performance consideration.
+        self.limit_type = self._get_limit_type(limit_threshold)
         if limit_threshold is None:
             if C.region == REG_CN:
                 self.logger.warning(f"limit_threshold not set. The stocks hit the limit may be bought/sold")
-        elif self._get_limit_type(limit_threshold) == self.LT_FLT and abs(limit_threshold) > 0.1:
+        elif self.limit_type == self.LT_FLT and abs(limit_threshold) > 0.1:
             if C.region == REG_CN:
                 self.logger.warning(f"limit_threshold may not be set to a reasonable value")
 
@@ -128,10 +131,9 @@ class Exchange:
         # $change is for calculating the limit of the stock
 
         necessary_fields = {self.buy_price, self.sell_price, "$close", "$change", "$factor", "$volume"}
-        if self._get_limit_type(limit_threshold) == self.LT_TP_EXP:
+        if self.limit_type == self.LT_TP_EXP:
             for exp in limit_threshold:
                 necessary_fields.add(exp)
-        subscribe_fields = list(necessary_fields | set(subscribe_fields))
         all_fields = list(necessary_fields | set(subscribe_fields))
 
         self.all_fields = all_fields
@@ -141,39 +143,43 @@ class Exchange:
         self.limit_threshold: Union[Tuple[str, str], float, None] = limit_threshold
         self.volume_threshold = volume_threshold
         self.extra_quote = extra_quote
-        self.set_quote(codes, start_time, end_time)
+        self.get_quote_from_qlib()
 
-    def set_quote(self, codes, start_time, end_time):
-        if len(codes) == 0:
-            codes = D.instruments()
+        # init quote by quote_df
+        self.quote_cls = quote_cls
+        self.quote = self.quote_cls(self.quote_df)
 
-        self.quote = D.features(codes, self.all_fields, start_time, end_time, freq=self.freq, disk_cache=True).dropna(
-            subset=["$close"]
-        )
-        self.quote.columns = self.all_fields
+    def get_quote_from_qlib(self):
+        # get stock data from qlib
+        if len(self.codes) == 0:
+            self.codes = D.instruments()
+        self.quote_df = D.features(
+            self.codes, self.all_fields, self.start_time, self.end_time, freq=self.freq, disk_cache=True
+        ).dropna(subset=["$close"])
+        self.quote_df.columns = self.all_fields
 
+        # check buy_price data and sell_price data
         for attr in "buy_price", "sell_price":
             pstr = getattr(self, attr)  # price string
-            if self.quote[pstr].isna().any():
+            if self.quote_df[pstr].isna().any():
                 self.logger.warning("{} field data contains nan.".format(pstr))
 
-        if self.quote["$factor"].isna().any():
+        # update trade_w_adj_price
+        if self.quote_df["$factor"].isna().any():
             # The 'factor.day.bin' file not exists, and `factor` field contains `nan`
             # Use adjusted price
             self.trade_w_adj_price = True
             self.logger.warning("factor.day.bin file not exists or factor contains `nan`. Order using adjusted_price.")
             if self.trade_unit is not None:
                 self.logger.warning(f"trade unit {self.trade_unit} is not supported in adjusted_price mode.")
-
         else:
             # The `factor.day.bin` file exists and all data `close` and `factor` are not `nan`
             # Use normal price
             self.trade_w_adj_price = False
-
         # update limit
-        self._update_limit()
+        self._update_limit(self.limit_threshold)
 
-        quote_df = self.quote
+        # concat extra_quote
         if self.extra_quote is not None:
             # process extra_quote
             if "$close" not in self.extra_quote:
@@ -192,21 +198,15 @@ class Exchange:
             if "limit_buy" not in self.extra_quote.columns:
                 self.extra_quote["limit_buy"] = False
                 self.logger.warning("No limit_buy set for extra_quote. All stock will be able to be bought.")
-
-            assert set(self.extra_quote.columns) == set(quote_df.columns) - {"$change"}
-            quote_df = pd.concat([quote_df, self.extra_quote], sort=False, axis=0)
-
-        quote_dict = {}
-        for stock_id, stock_val in quote_df.groupby(level="instrument"):
-            quote_dict[stock_id] = stock_val.droplevel(level="instrument")
-
-        self.quote = quote_dict
+            assert set(self.extra_quote.columns) == set(self.quote_df.columns) - {"$change"}
+            self.quote_df = pd.concat([self.quote_df, extra_quote], sort=False, axis=0)
 
     LT_TP_EXP = "(exp)"  # Tuple[str, str]
     LT_FLT = "float"  # float
     LT_NONE = "none"  # none
 
     def _get_limit_type(self, limit_threshold):
+        """get limit type"""
         if isinstance(limit_threshold, Tuple):
             return self.LT_TP_EXP
         elif isinstance(limit_threshold, float):
@@ -216,19 +216,19 @@ class Exchange:
         else:
             raise NotImplementedError(f"This type of `limit_threshold` is not supported")
 
-    def _update_limit(self):
+    def _update_limit(self, limit_threshold):
         # check limit_threshold
-        lt_type = self._get_limit_type(self.limit_threshold)
-        if lt_type == self.LT_NONE:
-            self.quote["limit_buy"] = False
-            self.quote["limit_sell"] = False
-        elif lt_type == self.LT_TP_EXP:
+        limit_type = self._get_limit_type(limit_threshold)
+        if limit_type == self.LT_NONE:
+            self.quote_df["limit_buy"] = False
+            self.quote_df["limit_sell"] = False
+        elif limit_type == self.LT_TP_EXP:
             # set limit
-            self.quote["limit_buy"] = self.quote[self.limit_threshold[0]]
-            self.quote["limit_sell"] = self.quote[self.limit_threshold[1]]
-        elif lt_type == self.LT_FLT:
-            self.quote["limit_buy"] = self.quote["$change"].ge(self.limit_threshold)
-            self.quote["limit_sell"] = self.quote["$change"].le(-self.limit_threshold)  # pylint: disable=E1130
+            self.quote_df["limit_buy"] = self.quote_df[limit_threshold[0]]
+            self.quote_df["limit_sell"] = self.quote_df[limit_threshold[1]]
+        elif limit_type == self.LT_FLT:
+            self.quote_df["limit_buy"] = self.quote_df["$change"].ge(limit_threshold)
+            self.quote_df["limit_sell"] = self.quote_df["$change"].le(-limit_threshold)  # pylint: disable=E1130
 
     def check_stock_limit(self, stock_id, start_time, end_time, direction=None):
         """
@@ -242,20 +242,20 @@ class Exchange:
 
         """
         if direction is None:
-            buy_limit = resam_ts_data(self.quote[stock_id]["limit_buy"], start_time, end_time, method="all")
-            sell_limit = resam_ts_data(self.quote[stock_id]["limit_sell"], start_time, end_time, method="all")
+            buy_limit = self.quote.get_data(stock_id, start_time, end_time, fields="limit_buy", method="all")
+            sell_limit = self.quote.get_data(stock_id, start_time, end_time, fields="limit_sell", method="all")
             return buy_limit or sell_limit
         elif direction == Order.BUY:
-            return resam_ts_data(self.quote[stock_id]["limit_buy"], start_time, end_time, method="all")
+            return self.quote.get_data(stock_id, start_time, end_time, fields="limit_buy", method="all")
         elif direction == Order.SELL:
-            return resam_ts_data(self.quote[stock_id]["limit_sell"], start_time, end_time, method="all")
+            return self.quote.get_data(stock_id, start_time, end_time, fields="limit_sell", method="all")
         else:
             raise ValueError(f"direction {direction} is not supported!")
 
     def check_stock_suspended(self, stock_id, start_time, end_time):
         # is suspended
-        if stock_id in self.quote:
-            return resam_ts_data(self.quote[stock_id], start_time, end_time, method=None) is None
+        if stock_id in self.quote.get_all_stock():
+            return self.quote.get_data(stock_id, start_time, end_time) is None
         else:
             return True
 
@@ -316,13 +316,13 @@ class Exchange:
         return trade_val, trade_cost, trade_price
 
     def get_quote_info(self, stock_id, start_time, end_time, method=ts_data_last):
-        return resam_ts_data(self.quote[stock_id], start_time, end_time, method=method)
+        return self.quote.get_data(stock_id, start_time, end_time, method=method)
 
     def get_close(self, stock_id, start_time, end_time, method=ts_data_last):
-        return resam_ts_data(self.quote[stock_id]["$close"], start_time, end_time, method=method)
+        return self.quote.get_data(stock_id, start_time, end_time, fields="$close", method=method)
 
     def get_volume(self, stock_id, start_time, end_time, method="sum"):
-        return resam_ts_data(self.quote[stock_id]["$volume"], start_time, end_time, method=method)
+        return self.quote.get_data(stock_id, start_time, end_time, fields="$volume", method=method)
 
     def get_deal_price(self, stock_id, start_time, end_time, direction: OrderDir, method=ts_data_last):
         if direction == OrderDir.SELL:
@@ -331,7 +331,7 @@ class Exchange:
             pstr = self.buy_price
         else:
             raise NotImplementedError(f"This type of input is not supported")
-        deal_price = resam_ts_data(self.quote[stock_id][pstr], start_time, end_time, method=method)
+        deal_price = self.quote.get_data(stock_id, start_time, end_time, fields=pstr, method=method)
         if method is not None and (np.isclose(deal_price, 0.0) or np.isnan(deal_price)):
             self.logger.warning(f"(stock_id:{stock_id}, trade_time:{(start_time, end_time)}, {pstr}): {deal_price}!!!")
             self.logger.warning(f"setting deal_price to close price")
@@ -347,9 +347,9 @@ class Exchange:
             `float`: return factor if the factor exists
         """
         assert (start_time is not None and end_time is not None, "the time range must be given")
-        if stock_id not in self.quote:
+        if stock_id not in self.quote.get_all_stock():
             return None
-        return resam_ts_data(self.quote[stock_id]["$factor"], start_time, end_time, method=ts_data_last)
+        return self.quote.get_data(stock_id, start_time, end_time, fields="$factor", method=ts_data_last)
 
     def generate_amount_position_from_weight_position(
         self, weight_position, cash, start_time, end_time, direction=OrderDir.BUY

qlib/backtest/high_performance_ds.py

@@ -0,0 +1,419 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+
+import logging
+from typing import List, Tuple, Union, Callable, Iterable, Dict
+from collections import OrderedDict
+
+import inspect
+import pandas as pd
+
+from ..utils.resam import resam_ts_data
+from ..log import get_module_logger
+
+
+class BaseQuote:
+    def __init__(self, quote_df: pd.DataFrame):
+        self.logger = get_module_logger("online operator", level=logging.INFO)
+
+    def get_all_stock(self) -> Iterable:
+        """return all stock codes
+
+        Return
+        ------
+        Iterable
+            all stock codes
+        """
+
+        raise NotImplementedError(f"Please implement the `get_all_stock` method")
+
+    def get_data(
+        self,
+        stock_id: Union[str, list],
+        start_time: Union[pd.Timestamp, str],
+        end_time: Union[pd.Timestamp, str],
+        fields: Union[str, list] = None,
+        method: Union[str, Callable] = None,
+    ) -> Union[None, float, pd.Series, pd.DataFrame]:
+        """get the specific fields of stock data during start time and end_time,
+           and apply method to the data.
+
+           Example:
+            .. code-block::
+                                        $close      $volume
+                instrument  datetime
+                SH600000    2010-01-04  86.778313   16162960.0
+                            2010-01-05  87.433578   28117442.0
+                            2010-01-06  85.713585   23632884.0
+                            2010-01-07  83.788803   20813402.0
+                            2010-01-08  84.730675   16044853.0
+
+                SH600655    2010-01-04  2699.567383  158193.328125
+                            2010-01-08  2612.359619   77501.406250
+                            2010-01-11  2712.982422  160852.390625
+                            2010-01-12  2788.688232  164587.937500
+                            2010-01-13  2790.604004  145460.453125
+
+                print(get_data(stock_id=["SH600000", "SH600655"], start_time="2010-01-04", end_time="2010-01-05", fields=["$close", "$volume"], method="last"))
+
+                            $close      $volume
+                instrument
+                SH600000    87.433578 28117442.0
+                SH600655    2699.567383  158193.328125
+
+                print(get_data(stock_id="SH600000", start_time="2010-01-04", end_time="2010-01-05", fields=["$close", "$volume"], method="last"))
+
+                $close 87.433578
+                $volume 28117442.0
+
+                print(get_data(stock_id="SH600000", start_time="2010-01-04", end_time="2010-01-05", fields="$close", method="last"))
+
+                87.433578
+
+        Parameters
+        ----------
+        stock_id: Union[str, list]
+        start_time : Union[pd.Timestamp, str]
+            closed start time for backtest
+        end_time : Union[pd.Timestamp, str]
+            closed end time for backtest
+        fields : Union[str, List]
+            the columns of data to fetch
+        method : Union[str, Callable]
+            the method apply to data.
+            e.g ["None", "last", "all", "sum", "mean", "any", qlib/utils/resam.py/ts_data_last]
+
+        Return
+        ----------
+        Union[None, float, pd.Series, pd.DataFrame]
+            The resampled DataFrame/Series/value, return None when the resampled data is empty.
+        """
+
+        raise NotImplementedError(f"Please implement the `get_data` method")
+
+
+class PandasQuote(BaseQuote):
+    def __init__(self, quote_df: pd.DataFrame):
+        super().__init__(quote_df=quote_df)
+        quote_dict = {}
+        for stock_id, stock_val in quote_df.groupby(level="instrument"):
+            quote_dict[stock_id] = stock_val.droplevel(level="instrument")
+        self.data = quote_dict
+
+    def get_all_stock(self):
+        return self.data.keys()
+
+    def get_data(self, stock_id, start_time, end_time, fields=None, method=None):
+        if fields is None:
+            return resam_ts_data(self.data[stock_id], start_time, end_time, method=method)
+        elif isinstance(fields, (str, list)):
+            return resam_ts_data(self.data[stock_id][fields], start_time, end_time, method=method)
+        else:
+            raise ValueError(f"fields must be None, str or list")
+
+
+class BaseSingleMetric:
+    """
+    The data structure of the single metric.
+    The following methods are used for computing metrics in one indicator.
+    """
+
+    def __init__(self, metric: Union[dict, pd.Series]):
+        raise NotImplementedError(f"Please implement the `__init__` method")
+
+    def __add__(self, other: Union["BaseSingleMetric", int, float]) -> "BaseSingleMetric":
+        raise NotImplementedError(f"Please implement the `__add__` method")
+
+    def __radd__(self, other: Union["BaseSingleMetric", int, float]) -> "BaseSingleMetric":
+        return self + other
+
+    def __sub__(self, other: Union["BaseSingleMetric", int, float]) -> "BaseSingleMetric":
+        raise NotImplementedError(f"Please implement the `__sub__` method")
+
+    def __rsub__(self, other: Union["BaseSingleMetric", int, float]) -> "BaseSingleMetric":
+        raise NotImplementedError(f"Please implement the `__rsub__` method")
+
+    def __mul__(self, other: Union["BaseSingleMetric", int, float]) -> "BaseSingleMetric":
+        raise NotImplementedError(f"Please implement the `__mul__` method")
+
+    def __truediv__(self, other: Union["BaseSingleMetric", int, float]) -> "BaseSingleMetric":
+        raise NotImplementedError(f"Please implement the `__truediv__` method")
+
+    def __eq__(self, other: Union["BaseSingleMetric", int, float]) -> "BaseSingleMetric":
+        raise NotImplementedError(f"Please implement the `__eq__` method")
+
+    def __gt__(self, other: Union["BaseSingleMetric", int, float]) -> "BaseSingleMetric":
+        raise NotImplementedError(f"Please implement the `__gt__` method")
+
+    def __lt__(self, other: Union["BaseSingleMetric", int, float]) -> "BaseSingleMetric":
+        raise NotImplementedError(f"Please implement the `__lt__` method")
+
+    def __len__(self) -> int:
+        raise NotImplementedError(f"Please implement the `__len__` method")
+
+    def sum(self) -> float:
+        raise NotImplementedError(f"Please implement the `sum` method")
+
+    def mean(self) -> float:
+        raise NotImplementedError(f"Please implement the `mean` method")
+
+    def count(self) -> int:
+        """Return the count of the single metric, NaN is not included. 
+        """
+
+        raise NotImplementedError(f"Please implement the `count` method")
+
+    def abs(self) -> "BaseSingleMetric":
+        raise NotImplementedError(f"Please implement the `abs` method")
+
+    def astype(self, type: type) -> "BaseSingleMetric":
+        raise NotImplementedError(f"Please implement the `astype` method")
+
+    @property
+    def empty(self) -> bool:
+        """If metric is empyt, return True."""
+        raise NotImplementedError(f"Please implement the `empty` method")
+
+    def add(self, other: "BaseSingleMetric", fill_value: float = None) -> "BaseSingleMetric":
+        """Replace np.NaN with fill_value in two metrics and add them."""
+        raise NotImplementedError(f"Please implement the `add` method")
+
+    def map(self, map_dict: dict) -> "BaseSingleMetric":
+        """Replace the value of metric according to map_dict."""
+        raise NotImplementedError(f"Please implement the `map` method")
+
+
+class BaseOrderIndicator:
+    """
+    The data structure of order indicator.
+    !!!NOTE: There are two ways to organize the data structure. Please choose a better way.
+        1. One way is using BaseSingleMetric to represent each metric. For example, the data
+        structure of PandasOrderIndicator is Dict[str, PandasSingleMetric]. It uses
+        PandasSingleMetric based on pd.Series to represent each metric.
+        2. The another way doesn't use BaseSingleMetric to represent each metric. The data
+        structure of PandasOrderIndicator is a whole matrix. It means you are not neccesary
+        to inherit the BaseSingleMetric.
+    """
+
+    def assign(self, col: str, metric: Union[dict, pd.Series]):
+        """assign one metric.
+
+        Parameters
+        ----------
+        col : str
+            the metric name of one metric.
+        metric : Union[dict, pd.Series]
+            the metric data.
+        """
+
+        pass
+
+    def transfer(self, func: Callable, new_col: str = None) -> Union[None, BaseSingleMetric]:
+        """compute new metric with existing metrics.
+
+        Parameters
+        ----------
+        func : Callable
+            the func of computing new metric.
+            the kwargs of func will be replaced with metric data by name in this function.
+            e.g.
+                def func(pa):
+                    return (pa > 0).astype(int).sum() / pa.count()
+        new_col : str, optional
+            New metric will be assigned in the data if new_col is not None, by default None.
+
+        Return
+        ----------
+        BaseSingleMetric
+            new metric.
+        """
+
+        pass
+
+    def get_metric_series(self, metric: str) -> pd.Series:
+        """return the single metric with pd.Series format.
+
+        Parameters
+        ----------
+        metric : str
+            the metric name.
+
+        Return
+        ----------
+        pd.Series
+            the single metric.
+            If there is no metric name in the data, return pd.Series().
+        """
+
+        pass
+
+    @staticmethod
+    def sum_all_indicators(
+        indicators: list, metrics: Union[str, List[str]], fill_value: float = None
+    ) -> Dict[str, BaseSingleMetric]:
+        """sum indicators with the same metrics.
+
+        Parameters
+        ----------
+        indicators : List[BaseOrderIndicator]
+            the list of all inner indicators.
+        metrics : Union[str, List[str]]
+            all metrics needs ot be sumed.
+        fill_value : float, optional
+            fill np.NaN with value. By default None.
+
+        Return
+        ----------
+        Dict[str: PandasSingleMetric]
+            a dict of metric name and data.
+        """
+
+        pass
+
+
+class PandasSingleMetric:
+    """Each SingleMetric is based on pd.Series."""
+
+    def __init__(self, metric: Union[dict, pd.Series]):
+        if isinstance(metric, dict):
+            self.metric = pd.Series(metric)
+        elif isinstance(metric, pd.Series):
+            self.metric = metric
+        else:
+            raise ValueError(f"metric must be dict or pd.Series")
+
+    def __add__(self, other):
+        if isinstance(other, (int, float)):
+            return PandasSingleMetric(self.metric + other)
+        elif isinstance(other, PandasSingleMetric):
+            return PandasSingleMetric(self.metric + other.metric)
+        else:
+            return NotImplemented
+
+    def __sub__(self, other):
+        if isinstance(other, (int, float)):
+            return PandasSingleMetric(self.metric - other)
+        elif isinstance(other, PandasSingleMetric):
+            return PandasSingleMetric(self.metric - other.metric)
+        else:
+            return NotImplemented
+
+    def __rsub__(self, other):
+        if isinstance(other, (int, float)):
+            return PandasSingleMetric(other - self.metric)
+        elif isinstance(other, PandasSingleMetric):
+            return PandasSingleMetric(other.metric - self.metric)
+        else:
+            return NotImplemented
+
+    def __mul__(self, other):
+        if isinstance(other, (int, float)):
+            return PandasSingleMetric(self.metric * other)
+        elif isinstance(other, PandasSingleMetric):
+            return PandasSingleMetric(self.metric * other.metric)
+        else:
+            return NotImplemented
+
+    def __truediv__(self, other):
+        if isinstance(other, (int, float)):
+            return PandasSingleMetric(self.metric / other)
+        elif isinstance(other, PandasSingleMetric):
+            return PandasSingleMetric(self.metric / other.metric)
+        else:
+            return NotImplemented
+
+    def __eq__(self, other):
+        if isinstance(other, (int, float)):
+            return PandasSingleMetric(self.metric == other)
+        elif isinstance(other, PandasSingleMetric):
+            return PandasSingleMetric(self.metric == other.metric)
+        else:
+            return NotImplemented
+
+    def __gt__(self, other):
+        if isinstance(other, (int, float)):
+            return PandasSingleMetric(self.metric < other)
+        elif isinstance(other, PandasSingleMetric):
+            return PandasSingleMetric(self.metric < other.metric)
+        else:
+            return NotImplemented
+
+    def __lt__(self, other):
+        if isinstance(other, (int, float)):
+            return PandasSingleMetric(self.metric > other)
+        elif isinstance(other, PandasSingleMetric):
+            return PandasSingleMetric(self.metric > other.metric)
+        else:
+            return NotImplemented
+
+    def __len__(self):
+        return len(self.metric)
+
+    def sum(self):
+        return self.metric.sum()
+
+    def mean(self):
+        return self.metric.mean()
+
+    def count(self):
+        return self.metric.count()
+
+    def abs(self):
+        return PandasSingleMetric(self.metric.abs())
+
+    def astype(self, type):
+        return PandasSingleMetric(self.metric.astype(type))
+
+    @property
+    def empty(self):
+        return self.metric.empty
+
+    def add(self, other, fill_value=None):
+        return PandasSingleMetric(self.metric.add(other.metric, fill_value=fill_value))
+
+    def map(self, map_dict: dict):
+        return PandasSingleMetric(self.metric.apply(map_dict))
+
+
+class PandasOrderIndicator(BaseOrderIndicator):
+    """
+    The data structure is OrderedDict(str: PandasSingleMetric).
+    Each PandasSingleMetric based on pd.Series is one metric.
+    Str is the name of metric.
+    """
+
+    def __init__(self):
+        self.data: Dict[str, PandasSingleMetric] = OrderedDict()
+
+    def assign(self, col: str, metric: Union[dict, pd.Series]):
+        self.data[col] = PandasSingleMetric(metric)
+
+    def transfer(self, func: Callable, new_col: str = None) -> Union[None, PandasSingleMetric]:
+        func_sig = inspect.signature(func).parameters.keys()
+        func_kwargs = {sig: self.data[sig] for sig in func_sig}
+        tmp_metric = func(**func_kwargs)
+        if new_col is not None:
+            self.data[new_col] = tmp_metric
+        else:
+            return tmp_metric
+
+    def get_metric_series(self, metric: str) -> Union[pd.Series]:
+        if metric in self.data:
+            return self.data[metric].metric
+        else:
+            return pd.Series()
+
+    @staticmethod
+    def sum_all_indicators(
+        indicators: list, metrics: Union[str, List[str]], fill_value=None
+    ) -> Dict[str, PandasSingleMetric]:
+        metric_dict = {}
+        if isinstance(metrics, str):
+            metrics = [metrics]
+        for metric in metrics:
+            tmp_metric = PandasSingleMetric({})
+            for indicator in indicators:
+                tmp_metric = tmp_metric.add(indicator.data[metric], fill_value)
+            metric_dict[metric] = tmp_metric.metric
+        return metric_dict

qlib/backtest/report.py

@@ -5,8 +5,7 @@
 from collections import OrderedDict
 from logging import warning
 import pathlib
-from typing import Dict, List, Tuple
-import warnings
+from typing import Dict, List, Tuple, Union, Callable
 
 import numpy as np
 import pandas as pd
@@ -17,6 +16,7 @@ from qlib.backtest.exchange import Exchange
 from qlib.backtest.order import BaseTradeDecision, Order, OrderDir
 from qlib.backtest.utils import TradeCalendarManager
 
+from .high_performance_ds import PandasOrderIndicator
 from ..data import D
 from ..tests.config import CSI300_BENCH
 from ..utils.resam import get_higher_eq_freq_feature, resam_ts_data
@@ -62,6 +62,7 @@ class Report:
                 - Else, it represent end time of benchmark, by default None
 
         """
+
         self.init_vars()
         self.init_bench(freq=freq, benchmark_config=benchmark_config)
 
@@ -252,10 +253,12 @@ class Indicator:
 
     """
 
-    def __init__(self):
+    def __init__(self, order_indicator_cls=PandasOrderIndicator):
+        self.order_indicator_cls = order_indicator_cls
+
         # order indicator is metrics for a single order for a specific step
         self.order_indicator_his = OrderedDict()
-        self.order_indicator: Dict[str, pd.Series] = OrderedDict()
+        self.order_indicator = self.order_indicator_cls()
 
         # trade indicator is metrics for all orders for a specific step
         self.trade_indicator_his = OrderedDict()
@@ -265,12 +268,12 @@ class Indicator:
 
     # def reset(self, trade_calendar: TradeCalendarManager):
     def reset(self):
-        self.order_indicator = OrderedDict()
+        self.order_indicator = self.order_indicator_cls()
         self.trade_indicator = OrderedDict()
         # self._trade_calendar = trade_calendar
 
     def record(self, trade_start_time):
-        self.order_indicator_his[trade_start_time] = self.order_indicator
+        self.order_indicator_his[trade_start_time] = self.order_indicator.data
         self.trade_indicator_his[trade_start_time] = self.trade_indicator
 
     def _update_order_trade_info(self, trade_info: list):
@@ -280,6 +283,7 @@ class Indicator:
         trade_value = dict()
         trade_cost = dict()
         trade_dir = dict()
+        pa = dict()
 
         for order, _trade_val, _trade_cost, _trade_price in trade_info:
             amount[order.stock_id] = order.amount_delta
@@ -288,66 +292,52 @@ class Indicator:
             trade_value[order.stock_id] = _trade_val * order.sign
             trade_cost[order.stock_id] = _trade_cost
             trade_dir[order.stock_id] = order.direction
+            # The PA in the innermost layer is meanless
+            pa[order.stock_id] = 0
 
-        self.order_indicator["amount"] = self.order_indicator["inner_amount"] = pd.Series(amount)
-        self.order_indicator["deal_amount"] = pd.Series(deal_amount)
+        self.order_indicator.assign("amount", amount)
+        self.order_indicator.assign("inner_amount", amount)
+        self.order_indicator.assign("deal_amount", deal_amount)
         # NOTE: trade_price and baseline price will be same on the lowest-level
-        self.order_indicator["trade_price"] = pd.Series(trade_price)
-        self.order_indicator["trade_value"] = pd.Series(trade_value)
-        self.order_indicator["trade_cost"] = pd.Series(trade_cost)
-        self.order_indicator["trade_dir"] = pd.Series(trade_dir)
+        self.order_indicator.assign("trade_price", trade_price)
+        self.order_indicator.assign("trade_value", trade_value)
+        self.order_indicator.assign("trade_cost", trade_cost)
+        self.order_indicator.assign("trade_dir", trade_dir)
+        self.order_indicator.assign("pa", pa)
 
     def _update_order_fulfill_rate(self):
-        self.order_indicator["ffr"] = self.order_indicator["deal_amount"] / self.order_indicator["amount"]
+        def func(deal_amount, amount):
+            return deal_amount / amount
 
-    def _update_order_price_advantage(self):
-        # NOTE:
-        # trade_price and baseline price will be same on the lowest-level
-        # So Pa should be 0 or do nothing
-        self.order_indicator["pa"] = 0
+        self.order_indicator.transfer(func, "ffr")
 
     def update_order_indicators(self, trade_info: list):
         self._update_order_trade_info(trade_info=trade_info)
         self._update_order_fulfill_rate()
-        self._update_order_price_advantage()
 
     def _agg_order_trade_info(self, inner_order_indicators: List[Dict[str, pd.Series]]):
-        inner_amount = pd.Series()
-        deal_amount = pd.Series()
-        trade_price = pd.Series()
-        trade_value = pd.Series()
-        trade_cost = pd.Series()
-        trade_dir = pd.Series()
-        for _order_indicator in inner_order_indicators:
-            inner_amount = inner_amount.add(_order_indicator["inner_amount"], fill_value=0)
-            deal_amount = deal_amount.add(_order_indicator["deal_amount"], fill_value=0)
-            trade_price = trade_price.add(
-                _order_indicator["trade_price"] * _order_indicator["deal_amount"], fill_value=0
-            )
-            trade_value = trade_value.add(_order_indicator["trade_value"], fill_value=0)
-            trade_cost = trade_cost.add(_order_indicator["trade_cost"], fill_value=0)
-            trade_dir = trade_dir.add(_order_indicator["trade_dir"], fill_value=0)
+        all_metric = ["inner_amount", "deal_amount", "trade_price", "trade_value", "trade_cost", "trade_dir"]
+        metric_dict = self.order_indicator_cls.sum_all_indicators(inner_order_indicators, all_metric, fill_value=0)
+        for metric in metric_dict:
+            self.order_indicator.assign(metric, metric_dict[metric])
 
-        trade_dir = trade_dir.apply(Order.parse_dir)
+        def func(trade_price, deal_amount):
+            return trade_price / deal_amount
 
-        self.order_indicator["inner_amount"] = inner_amount
-        self.order_indicator["deal_amount"] = deal_amount
-        trade_price /= self.order_indicator["deal_amount"]
-        self.order_indicator["trade_price"] = trade_price
-        self.order_indicator["trade_value"] = trade_value
-        self.order_indicator["trade_cost"] = trade_cost
-        self.order_indicator["trade_dir"] = trade_dir
+        self.order_indicator.transfer(func, "trade_price")
+
+        def func_apply(trade_dir):
+            return trade_dir.map(Order.parse_dir)
+
+        self.order_indicator.transfer(func_apply, "trade_dir")
 
     def _update_trade_amount(self, outer_trade_decision: BaseTradeDecision):
         # NOTE: these indicator is designed for order execution, so the
         decision: List[Order] = outer_trade_decision.get_decision()
-        if decision is None:
-            self.order_indicator["amount"] = pd.Series()
+        if len(decision) == 0:
+            self.order_indicator.assign("amount", {})
         else:
-            self.order_indicator["amount"] = pd.Series({order.stock_id: order.amount_delta for order in decision})
-
-    def _agg_order_fulfill_rate(self):
-        self.order_indicator["ffr"] = self.order_indicator["deal_amount"] / self.order_indicator["amount"]
+            self.order_indicator.assign("amount", {order.stock_id: order.amount_delta for order in decision})
 
     def _get_base_vol_pri(
         self,
@@ -428,17 +418,16 @@ class Indicator:
                 "price": "$close",  # TODO: this is not supported now!!!!!
                                     # default to use deal price of the exchange
             }
-
         """
 
         # TODO: I think there are potentials to be optimized
-        trade_dir = self.order_indicator["trade_dir"]
+        trade_dir = self.order_indicator.get_metric_series("trade_dir")
         if len(trade_dir) > 0:
             bp_all, bv_all = [], []
             # <step, inst, (base_volume | base_price)>
             for oi, (dec, start, end) in zip(inner_order_indicators, decision_list):
-                bp_s = oi.get("base_price", pd.Series()).reindex(trade_dir.index)
-                bv_s = oi.get("base_volume", pd.Series()).reindex(trade_dir.index)
+                bp_s = oi.get_metric_series("base_price").reindex(trade_dir.index)
+                bv_s = oi.get_metric_series("base_volume").reindex(trade_dir.index)
                 bp_new, bv_new = {}, {}
                 for pr, v, (inst, direction) in zip(bp_s.values, bv_s.values, trade_dir.items()):
                     if np.isnan(pr):
@@ -462,17 +451,24 @@ class Indicator:
             bp_all = pd.concat(bp_all, axis=1)
             bv_all = pd.concat(bv_all, axis=1)
 
-            self.order_indicator["base_volume"] = bv_all.sum(axis=1)
-            self.order_indicator["base_price"] = (bp_all * bv_all).sum(axis=1) / self.order_indicator["base_volume"]
+            base_volume = bv_all.sum(axis=1)
+            self.order_indicator.assign("base_volume", base_volume)
+            self.order_indicator.assign("base_price", (bp_all * bv_all).sum(axis=1) / base_volume)
 
     def _agg_order_price_advantage(self):
-        if not self.order_indicator["trade_price"].empty:
-            sign = 1 - self.order_indicator["trade_dir"] * 2
-            self.order_indicator["pa"] = sign * (
-                self.order_indicator["trade_price"] / self.order_indicator["base_price"] - 1
-            )
+        def if_empty_func(trade_price):
+            return trade_price.empty
+
+        if_empty = self.order_indicator.transfer(if_empty_func)
+        if not if_empty:
+
+            def func(trade_dir, trade_price, base_price):
+                sign = 1 - trade_dir * 2
+                return sign * (trade_price / base_price - 1)
+
+            self.order_indicator.transfer(func, "pa")
         else:
-            self.order_indicator["pa"] = pd.Series()
+            self.order_indicator.assign("pa", {})
 
     def agg_order_indicators(
         self,
@@ -484,55 +480,74 @@ class Indicator:
     ):
         self._agg_order_trade_info(inner_order_indicators)
         self._update_trade_amount(outer_trade_decision)
-        self._agg_order_fulfill_rate()
+        self._update_order_fulfill_rate()
         pa_config = indicator_config.get("pa_config", {})
-        self._agg_base_price(inner_order_indicators, decision_list, trade_exchange, pa_config=pa_config)
+        self._agg_base_price(inner_order_indicators, decision_list, trade_exchange, pa_config=pa_config)  # TODO
         self._agg_order_price_advantage()
 
     def _cal_trade_fulfill_rate(self, method="mean"):
         if method == "mean":
-            return self.order_indicator["ffr"].mean()
+
+            def func(ffr):
+                return ffr.mean()
+
         elif method == "amount_weighted":
-            weights = self.order_indicator["deal_amount"].abs()
-            return (self.order_indicator["ffr"] * weights).sum() / weights.sum()
+
+            def func(ffr, deal_amount):
+                return (ffr * deal_amount.abs()).sum() / (deal_amount.abs().sum())
+
         elif method == "value_weighted":
-            weights = self.order_indicator["trade_value"].abs()
-            return (self.order_indicator["ffr"] * weights).sum() / weights.sum()
+
+            def func(ffr, trade_value):
+                return (ffr * trade_value.abs()).sum() / (trade_value.abs().sum())
+
         else:
             raise ValueError(f"method {method} is not supported!")
+        return self.order_indicator.transfer(func)
 
     def _cal_trade_price_advantage(self, method="mean"):
-        pa_order = self.order_indicator["pa"]
-        if isinstance(pa_order, (int, float)):
-            # pa from atomic executor
-            return pa_order
-
         if method == "mean":
-            return pa_order.mean()
+
+            def func(pa):
+                return pa.mean()
+
         elif method == "amount_weighted":
-            weights = self.order_indicator["deal_amount"].abs()
-            return (pa_order * weights).sum() / weights.sum()
+
+            def func(pa, deal_amount):
+                return (pa * deal_amount.abs()).sum() / (deal_amount.abs().sum())
+
         elif method == "value_weighted":
-            weights = self.order_indicator["trade_value"].abs()
-            return (pa_order * weights).sum() / weights.sum()
+
+            def func(pa, trade_value):
+                return (pa * trade_value.abs()).sum() / (trade_value.abs().sum())
+
         else:
             raise ValueError(f"method {method} is not supported!")
+        return self.order_indicator.transfer(func)
 
     def _cal_trade_positive_rate(self):
-        pa_order = self.order_indicator["pa"]
-        if isinstance(pa_order, (int, float)):
-            # pa from atomic executor
-            return pa_order
-        return (pa_order > 0).astype(int).sum() / pa_order.count()
+        def func(pa):
+            return (pa > 0).astype(int).sum() / pa.count()
+
+        return self.order_indicator.transfer(func)
 
     def _cal_deal_amount(self):
-        return self.order_indicator["deal_amount"].abs().sum()
+        def func(deal_amount):
+            return deal_amount.abs().sum()
+
+        return self.order_indicator.transfer(func)
 
     def _cal_trade_value(self):
-        return self.order_indicator["trade_value"].abs().sum()
+        def func(trade_value):
+            return trade_value.abs().sum()
+
+        return self.order_indicator.transfer(func)
 
     def _cal_trade_order_count(self):
-        return self.order_indicator["amount"].count()
+        def func(amount):
+            return amount.count()
+
+        return self.order_indicator.transfer(func)
 
     def cal_trade_indicators(self, trade_start_time, freq, indicator_config={}):
         show_indicator = indicator_config.get("show_indicator", False)