diff --git a/qlib/backtest/exchange.py b/qlib/backtest/exchange.py index ea1d012eb..e73510743 100644 --- a/qlib/backtest/exchange.py +++ b/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 diff --git a/qlib/backtest/high_performance_ds.py b/qlib/backtest/high_performance_ds.py new file mode 100644 index 000000000..8a908fbf0 --- /dev/null +++ b/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 diff --git a/qlib/backtest/report.py b/qlib/backtest/report.py index 6b64bf3b1..5f8238504 100644 --- a/qlib/backtest/report.py +++ b/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 = [], [] # 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)