qlib/qlib/backtest/report.py

# Copyright (c) Microsoft Corporation.
# Licensed under the MIT License.

from __future__ import annotations

import pathlib
from collections import OrderedDict
from typing import Any, Dict, List, Optional, Text, Tuple, Type, Union, cast

import numpy as np
import pandas as pd

import qlib.utils.index_data as idd
from qlib.backtest.decision import BaseTradeDecision, Order, OrderDir
from qlib.backtest.exchange import Exchange

from ..tests.config import CSI300_BENCH
from ..utils.resam import get_higher_eq_freq_feature, resam_ts_data
from .high_performance_ds import BaseOrderIndicator, BaseSingleMetric, NumpyOrderIndicator


class PortfolioMetrics:
    """
    Motivation:
        PortfolioMetrics is for supporting portfolio related metrics.

    Implementation:

        daily portfolio metrics of the account
        contain those followings: return, cost, turnover, account, cash, bench, value
        For each step(bar/day/minute), each column represents
        - return: the return of the portfolio generated by strategy **without transaction fee**.
        - cost: the transaction fee and slippage.
        - account: the total value of assets(cash and securities are both included) in user account based on the close price of each step.
        - cash: the amount of cash in user's account.
        - bench: the return of the benchmark
        - value: the total value of securities/stocks/instruments (cash is excluded).

        update report
    """

    def __init__(self, freq: str = "day", benchmark_config: dict = {}) -> None:
        """
        Parameters
        ----------
        freq : str
            frequency of trading bar, used for updating hold count of trading bar
        benchmark_config : dict
            config of benchmark, may including the following arguments:
            - benchmark : Union[str, list, pd.Series]
                - If `benchmark` is pd.Series, `index` is trading date; the value T is the change from T-1 to T.
                    example:
                        print(
                            D.features(D.instruments('csi500'),
                            ['$close/Ref($close, 1)-1'])['$close/Ref($close, 1)-1'].head()
                        )
                            2017-01-04    0.011693
                            2017-01-05    0.000721
                            2017-01-06   -0.004322
                            2017-01-09    0.006874
                            2017-01-10   -0.003350
                - If `benchmark` is list, will use the daily average change of the stock pool in the list as the
                    'bench'.
                - If `benchmark` is str, will use the daily change as the 'bench'.
                benchmark code, default is SH000300 CSI300
            - start_time : Union[str, pd.Timestamp], optional
                - If `benchmark` is pd.Series, it will be ignored
                - Else, it represent start time of benchmark, by default None
            - end_time : Union[str, pd.Timestamp], optional
                - If `benchmark` is pd.Series, it will be ignored
                - Else, it represent end time of benchmark, by default None

        """

        self.init_vars()
        self.init_bench(freq=freq, benchmark_config=benchmark_config)

    def init_vars(self) -> None:
        self.accounts: dict = OrderedDict()  # account position value for each trade time
        self.returns: dict = OrderedDict()  # daily return rate for each trade time
        self.total_turnovers: dict = OrderedDict()  # total turnover for each trade time
        self.turnovers: dict = OrderedDict()  # turnover for each trade time
        self.total_costs: dict = OrderedDict()  # total trade cost for each trade time
        self.costs: dict = OrderedDict()  # trade cost rate for each trade time
        self.values: dict = OrderedDict()  # value for each trade time
        self.cashes: dict = OrderedDict()
        self.benches: dict = OrderedDict()
        self.latest_pm_time: Optional[pd.TimeStamp] = None

    def init_bench(self, freq: str | None = None, benchmark_config: dict | None = None) -> None:
        if freq is not None:
            self.freq = freq
        self.benchmark_config = benchmark_config
        self.bench = self._cal_benchmark(self.benchmark_config, self.freq)

    @staticmethod
    def _cal_benchmark(benchmark_config: Optional[dict], freq: str) -> Optional[pd.Series]:
        if benchmark_config is None:
            return None
        benchmark = benchmark_config.get("benchmark", CSI300_BENCH)
        if benchmark is None:
            return None

        if isinstance(benchmark, pd.Series):
            return benchmark
        else:
            start_time = benchmark_config.get("start_time", None)
            end_time = benchmark_config.get("end_time", None)

            if freq is None:
                raise ValueError("benchmark freq can't be None!")
            _codes = benchmark if isinstance(benchmark, (list, dict)) else [benchmark]
            fields = ["$close/Ref($close,1)-1"]
            _temp_result, _ = get_higher_eq_freq_feature(_codes, fields, start_time, end_time, freq=freq)
            if len(_temp_result) == 0:
                raise ValueError(f"The benchmark {_codes} does not exist. Please provide the right benchmark")
            return _temp_result.groupby(level="datetime")[_temp_result.columns.tolist()[0]].mean().fillna(0)

    def _sample_benchmark(
        self,
        bench: pd.Series,
        trade_start_time: Union[str, pd.Timestamp],
        trade_end_time: Union[str, pd.Timestamp],
    ) -> Optional[float]:
        if self.bench is None:
            return None

        def cal_change(x):
            return (x + 1).prod()

        _ret = resam_ts_data(bench, trade_start_time, trade_end_time, method=cal_change)
        return 0.0 if _ret is None else _ret - 1

    def is_empty(self) -> bool:
        return len(self.accounts) == 0

    def get_latest_date(self) -> pd.Timestamp:
        return self.latest_pm_time

    def get_latest_account_value(self) -> float:
        return self.accounts[self.latest_pm_time]

    def get_latest_total_cost(self) -> Any:
        return self.total_costs[self.latest_pm_time]

    def get_latest_total_turnover(self) -> Any:
        return self.total_turnovers[self.latest_pm_time]

    def update_portfolio_metrics_record(
        self,
        trade_start_time: Union[str, pd.Timestamp] = None,
        trade_end_time: Union[str, pd.Timestamp] = None,
        account_value: float | None = None,
        cash: float | None = None,
        return_rate: float | None = None,
        total_turnover: float | None = None,
        turnover_rate: float | None = None,
        total_cost: float | None = None,
        cost_rate: float | None = None,
        stock_value: float | None = None,
        bench_value: float | None = None,
    ) -> None:
        # check data
        if None in [
            trade_start_time,
            account_value,
            cash,
            return_rate,
            total_turnover,
            turnover_rate,
            total_cost,
            cost_rate,
            stock_value,
        ]:
            raise ValueError(
                "None in [trade_start_time, account_value, cash, return_rate, total_turnover, turnover_rate, "
                "total_cost, cost_rate, stock_value]",
            )

        if trade_end_time is None and bench_value is None:
            raise ValueError("Both trade_end_time and bench_value is None, benchmark is not usable.")
        elif bench_value is None:
            bench_value = self._sample_benchmark(self.bench, trade_start_time, trade_end_time)

        # update pm data
        self.accounts[trade_start_time] = account_value
        self.returns[trade_start_time] = return_rate
        self.total_turnovers[trade_start_time] = total_turnover
        self.turnovers[trade_start_time] = turnover_rate
        self.total_costs[trade_start_time] = total_cost
        self.costs[trade_start_time] = cost_rate
        self.values[trade_start_time] = stock_value
        self.cashes[trade_start_time] = cash
        self.benches[trade_start_time] = bench_value
        # update pm
        self.latest_pm_time = trade_start_time
        # finish pm update in each step

    def generate_portfolio_metrics_dataframe(self) -> pd.DataFrame:
        pm = pd.DataFrame()
        pm["account"] = pd.Series(self.accounts)
        pm["return"] = pd.Series(self.returns)
        pm["total_turnover"] = pd.Series(self.total_turnovers)
        pm["turnover"] = pd.Series(self.turnovers)
        pm["total_cost"] = pd.Series(self.total_costs)
        pm["cost"] = pd.Series(self.costs)
        pm["value"] = pd.Series(self.values)
        pm["cash"] = pd.Series(self.cashes)
        pm["bench"] = pd.Series(self.benches)
        pm.index.name = "datetime"
        return pm

    def save_portfolio_metrics(self, path: str) -> None:
        r = self.generate_portfolio_metrics_dataframe()
        r.to_csv(path)

    def load_portfolio_metrics(self, path: str) -> None:
        """load pm from a file
        should have format like
        columns = ['account', 'return', 'total_turnover', 'turnover', 'cost', 'total_cost', 'value', 'cash', 'bench']
            :param
                path: str/ pathlib.Path()
        """
        with pathlib.Path(path).open("rb") as f:
            r = pd.read_csv(f, index_col=0)
        r.index = pd.DatetimeIndex(r.index)

        index = r.index
        self.init_vars()
        for trade_start_time in index:
            self.update_portfolio_metrics_record(
                trade_start_time=trade_start_time,
                account_value=r.loc[trade_start_time]["account"],
                cash=r.loc[trade_start_time]["cash"],
                return_rate=r.loc[trade_start_time]["return"],
                total_turnover=r.loc[trade_start_time]["total_turnover"],
                turnover_rate=r.loc[trade_start_time]["turnover"],
                total_cost=r.loc[trade_start_time]["total_cost"],
                cost_rate=r.loc[trade_start_time]["cost"],
                stock_value=r.loc[trade_start_time]["value"],
                bench_value=r.loc[trade_start_time]["bench"],
            )


class Indicator:
    """
    `Indicator` is implemented in a aggregate way.
    All the metrics are calculated aggregately.
    All the metrics are calculated for a separated stock and in a specific step on a specific level.

    | indicator    | desc.                                                        |
    |--------------+--------------------------------------------------------------|
    | amount       | the *target* amount given by the outer strategy              |
    | deal_amount  | the real deal amount                                         |
    | inner_amount | the total *target* amount of inner strategy                  |
    | trade_price  | the average deal price                                       |
    | trade_value  | the total trade value                                        |
    | trade_cost   | the total trade cost  (base price need drection)             |
    | trade_dir    | the trading direction                                        |
    | ffr          | full fill rate                                               |
    | pa           | price advantage                                              |
    | pos          | win rate                                                     |
    | base_price   | the price of baseline                                        |
    | base_volume  | the volume of baseline (for weighted aggregating base_price) |

    **NOTE**:
    The `base_price` and `base_volume` can't be NaN when there are not trading on that step. Otherwise
    aggregating get wrong results.

    So `base_price` will not be calculated in a aggregate way!!

    """

    def __init__(self, order_indicator_cls: Type[BaseOrderIndicator] = NumpyOrderIndicator) -> None:
        self.order_indicator_cls = order_indicator_cls

        # order indicator is metrics for a single order for a specific step
        self.order_indicator_his: dict = OrderedDict()
        self.order_indicator: BaseOrderIndicator = self.order_indicator_cls()

        # trade indicator is metrics for all orders for a specific step
        self.trade_indicator_his: dict = OrderedDict()
        self.trade_indicator: Dict[str, Optional[BaseSingleMetric]] = OrderedDict()

        self._trade_calendar = None

    # def reset(self, trade_calendar: TradeCalendarManager):
    def reset(self) -> None:
        self.order_indicator = self.order_indicator_cls()
        self.trade_indicator = OrderedDict()
        # self._trade_calendar = trade_calendar

    def record(self, trade_start_time: Union[str, pd.Timestamp]) -> None:
        self.order_indicator_his[trade_start_time] = self.get_order_indicator()
        self.trade_indicator_his[trade_start_time] = self.get_trade_indicator()

    def _update_order_trade_info(self, trade_info: List[Tuple[Order, float, float, float]]) -> None:
        amount = dict()
        deal_amount = dict()
        trade_price = dict()
        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
            deal_amount[order.stock_id] = order.deal_amount_delta
            trade_price[order.stock_id] = _trade_price
            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.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.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) -> None:
        def func(deal_amount, amount):
            # deal_amount is np.NaN or None when there is no inner decision. So full fill rate is 0.
            tmp_deal_amount = deal_amount.reindex(amount.index, 0)
            tmp_deal_amount = tmp_deal_amount.replace({np.NaN: 0})
            return tmp_deal_amount / amount

        self.order_indicator.transfer(func, "ffr")

    def update_order_indicators(self, trade_info: List[Tuple[Order, float, float, float]]) -> None:
        self._update_order_trade_info(trade_info=trade_info)
        self._update_order_fulfill_rate()

    def _agg_order_trade_info(self, inner_order_indicators: List[BaseOrderIndicator]) -> None:
        # calculate total trade amount with each inner order indicator.
        def trade_amount_func(deal_amount, trade_price):
            return deal_amount * trade_price

        for indicator in inner_order_indicators:
            indicator.transfer(trade_amount_func, "trade_price")

        # sum inner order indicators with same metric.
        all_metric = ["inner_amount", "deal_amount", "trade_price", "trade_value", "trade_cost", "trade_dir"]
        self.order_indicator_cls.sum_all_indicators(
            self.order_indicator,
            inner_order_indicators,
            all_metric,
            fill_value=0,
        )

        def func(trade_price, deal_amount):
            # trade_price is np.NaN instead of inf when deal_amount is zero.
            tmp_deal_amount = deal_amount.replace({0: np.NaN})
            return trade_price / tmp_deal_amount

        self.order_indicator.transfer(func, "trade_price")

        def func_apply(trade_dir):
            return trade_dir.apply(Order.parse_dir)

        self.order_indicator.transfer(func_apply, "trade_dir")

    def _update_trade_amount(self, outer_trade_decision: BaseTradeDecision) -> None:
        # NOTE: these indicator is designed for order execution, so the
        decision: List[Order] = cast(List[Order], outer_trade_decision.get_decision())
        if len(decision) == 0:
            self.order_indicator.assign("amount", {})
        else:
            self.order_indicator.assign("amount", {order.stock_id: order.amount_delta for order in decision})

    def _get_base_vol_pri(
        self,
        inst: str,
        trade_start_time: pd.Timestamp,
        trade_end_time: pd.Timestamp,
        direction: OrderDir,
        decision: BaseTradeDecision,
        trade_exchange: Exchange,
        pa_config: dict = {},
    ) -> Tuple[Optional[float], Optional[float]]:
        """
        Get the base volume and price information
        All the base price values are rooted from this function
        """

        agg = pa_config.get("agg", "twap").lower()
        price = pa_config.get("price", "deal_price").lower()

        if decision.trade_range is not None:
            trade_start_time, trade_end_time = decision.trade_range.clip_time_range(
                start_time=trade_start_time,
                end_time=trade_end_time,
            )

        if price == "deal_price":
            price_s = trade_exchange.get_deal_price(
                inst,
                trade_start_time,
                trade_end_time,
                direction=direction,
                method=None,
            )
        else:
            raise NotImplementedError(f"This type of input is not supported")

        # if there is no stock data during the time period
        if price_s is None:
            return None, None

        if isinstance(price_s, (int, float, np.number)):
            price_s = idd.SingleData(price_s, [trade_start_time])
        elif isinstance(price_s, idd.SingleData):
            pass
        else:
            raise NotImplementedError(f"This type of input is not supported")

        # NOTE: there are some zeros in the trading price. These cases are known meaningless
        # for aligning the previous logic, remove it.
        # remove zero and negative values.
        assert isinstance(price_s, idd.SingleData)
        price_s = price_s.loc[(price_s > 1e-08).data.astype(bool)]
        # NOTE ~(price_s < 1e-08) is different from price_s >= 1e-8
        #   ~(np.NaN < 1e-8) -> ~(False)  -> True

        assert isinstance(price_s, idd.SingleData)
        if agg == "vwap":
            volume_s = trade_exchange.get_volume(inst, trade_start_time, trade_end_time, method=None)
            if isinstance(volume_s, (int, float, np.number)):
                volume_s = idd.SingleData(volume_s, [trade_start_time])
            assert isinstance(volume_s, idd.SingleData)
            volume_s = volume_s.reindex(price_s.index)
        elif agg == "twap":
            volume_s = idd.SingleData(1, price_s.index)
        else:
            raise NotImplementedError(f"This type of input is not supported")

        assert isinstance(volume_s, idd.SingleData)
        base_volume = volume_s.sum()
        base_price = (price_s * volume_s).sum() / base_volume
        return base_price, base_volume

    def _agg_base_price(
        self,
        inner_order_indicators: List[BaseOrderIndicator],
        decision_list: List[Tuple[BaseTradeDecision, pd.Timestamp, pd.Timestamp]],
        trade_exchange: Exchange,
        pa_config: dict = {},
    ) -> None:
        """
        # NOTE:!!!!
        # Strong assumption!!!!!!
        # the correctness of the base_price relies on that the **same** exchange is used

        Parameters
        ----------
        inner_order_indicators : List[BaseOrderIndicator]
            the indicators of account of inner executor
        decision_list: List[Tuple[BaseTradeDecision, pd.Timestamp, pd.Timestamp]],
            a list of decisions according to inner_order_indicators
        trade_exchange : Exchange
            for retrieving trading price
        pa_config : dict
            For example
            {
                "agg": "twap",  # "vwap"
                "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.get_index_data("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_index_data("base_price").reindex(trade_dir.index)
                bv_s = oi.get_index_data("base_volume").reindex(trade_dir.index)

                bp_new, bv_new = {}, {}
                for pr, v, (inst, direction) in zip(bp_s.data, bv_s.data, zip(trade_dir.index, trade_dir.data)):
                    if np.isnan(pr):
                        bp_tmp, bv_tmp = self._get_base_vol_pri(
                            inst,
                            start,
                            end,
                            decision=dec,
                            direction=direction,
                            trade_exchange=trade_exchange,
                            pa_config=pa_config,
                        )
                        if (bp_tmp is not None) and (bv_tmp is not None):
                            bp_new[inst], bv_new[inst] = bp_tmp, bv_tmp
                    else:
                        bp_new[inst], bv_new[inst] = pr, v

                bp_new = idd.SingleData(bp_new)
                bv_new = idd.SingleData(bv_new)
                bp_all.append(bp_new)
                bv_all.append(bv_new)
            bp_all_multi_data = idd.concat(bp_all, axis=1)
            bv_all_multi_data = idd.concat(bv_all, axis=1)

            base_volume = bv_all_multi_data.sum(axis=1)
            self.order_indicator.assign("base_volume", base_volume.to_dict())
            self.order_indicator.assign(
                "base_price",
                ((bp_all_multi_data * bv_all_multi_data).sum(axis=1) / base_volume).to_dict(),
            )

    def _agg_order_price_advantage(self) -> None:
        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.assign("pa", {})

    def agg_order_indicators(
        self,
        inner_order_indicators: List[BaseOrderIndicator],
        decision_list: List[Tuple[BaseTradeDecision, pd.Timestamp, pd.Timestamp]],
        outer_trade_decision: BaseTradeDecision,
        trade_exchange: Exchange,
        indicator_config: dict = {},
    ) -> None:
        self._agg_order_trade_info(inner_order_indicators)
        self._update_trade_amount(outer_trade_decision)
        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)  # TODO
        self._agg_order_price_advantage()

    def _cal_trade_fulfill_rate(self, method: str = "mean") -> Optional[BaseSingleMetric]:
        if method == "mean":
            return self.order_indicator.transfer(
                lambda ffr: ffr.mean(),
            )
        elif method == "amount_weighted":
            return self.order_indicator.transfer(
                lambda ffr, deal_amount: (ffr * deal_amount.abs()).sum() / (deal_amount.abs().sum()),
            )
        elif method == "value_weighted":
            return self.order_indicator.transfer(
                lambda ffr, trade_value: (ffr * trade_value.abs()).sum() / (trade_value.abs().sum()),
            )
        else:
            raise ValueError(f"method {method} is not supported!")

    def _cal_trade_price_advantage(self, method: str = "mean") -> Optional[BaseSingleMetric]:
        if method == "mean":
            return self.order_indicator.transfer(lambda pa: pa.mean())
        elif method == "amount_weighted":
            return self.order_indicator.transfer(
                lambda pa, deal_amount: (pa * deal_amount.abs()).sum() / (deal_amount.abs().sum()),
            )
        elif method == "value_weighted":
            return self.order_indicator.transfer(
                lambda pa, trade_value: (pa * trade_value.abs()).sum() / (trade_value.abs().sum()),
            )
        else:
            raise ValueError(f"method {method} is not supported!")

    def _cal_trade_positive_rate(self) -> Optional[BaseSingleMetric]:
        def func(pa):
            return (pa > 0).sum() / pa.count()

        return self.order_indicator.transfer(func)

    def _cal_deal_amount(self) -> Optional[BaseSingleMetric]:
        def func(deal_amount):
            return deal_amount.abs().sum()

        return self.order_indicator.transfer(func)

    def _cal_trade_value(self) -> Optional[BaseSingleMetric]:
        def func(trade_value):
            return trade_value.abs().sum()

        return self.order_indicator.transfer(func)

    def _cal_trade_order_count(self) -> Optional[BaseSingleMetric]:
        def func(amount):
            return amount.count()

        return self.order_indicator.transfer(func)

    def cal_trade_indicators(
        self,
        trade_start_time: Union[str, pd.Timestamp],
        freq: str,
        indicator_config: dict = {},
    ) -> None:
        show_indicator = indicator_config.get("show_indicator", False)
        ffr_config = indicator_config.get("ffr_config", {})
        pa_config = indicator_config.get("pa_config", {})
        fulfill_rate = self._cal_trade_fulfill_rate(method=ffr_config.get("weight_method", "mean"))
        price_advantage = self._cal_trade_price_advantage(method=pa_config.get("weight_method", "mean"))
        positive_rate = self._cal_trade_positive_rate()
        deal_amount = self._cal_deal_amount()
        trade_value = self._cal_trade_value()
        order_count = self._cal_trade_order_count()
        self.trade_indicator["ffr"] = fulfill_rate
        self.trade_indicator["pa"] = price_advantage
        self.trade_indicator["pos"] = positive_rate
        self.trade_indicator["deal_amount"] = deal_amount
        self.trade_indicator["value"] = trade_value
        self.trade_indicator["count"] = order_count
        if show_indicator:
            print(
                "[Indicator({}) {}]: FFR: {}, PA: {}, POS: {}".format(
                    freq,
                    (
                        trade_start_time
                        if isinstance(trade_start_time, str)
                        else trade_start_time.strftime("%Y-%m-%d %H:%M:%S")
                    ),
                    fulfill_rate,
                    price_advantage,
                    positive_rate,
                ),
            )

    def get_order_indicator(self, raw: bool = True) -> Union[BaseOrderIndicator, Dict[Text, pd.Series]]:
        return self.order_indicator if raw else self.order_indicator.to_series()

    def get_trade_indicator(self) -> Dict[str, Optional[BaseSingleMetric]]:
        return self.trade_indicator

    def generate_trade_indicators_dataframe(self) -> pd.DataFrame:
        return pd.DataFrame.from_dict(self.trade_indicator_his, orient="index")