diff --git a/qlib/backtest/exchange.py b/qlib/backtest/exchange.py index 69262fcbbad..d2e75570851 100644 --- a/qlib/backtest/exchange.py +++ b/qlib/backtest/exchange.py @@ -49,6 +49,7 @@ def __init__( close_cost: float = 0.0025, min_cost: float = 5.0, impact_cost: float = 0.0, + deal_price_fallback: str = "close", extra_quote: pd.DataFrame = None, quote_cls: Type[BaseQuote] = NumpyQuote, **kwargs: Any, @@ -112,6 +113,16 @@ def __init__( distinguish `not set` and `disable trade_unit` :param min_cost: min cost, default 5 :param impact_cost: market impact cost rate (a.k.a. slippage). A recommended value is 0.1. + :param deal_price_fallback: str, policy when the configured deal price (e.g. $open or $vwap) is + missing (NaN) or nearly zero (<= 1e-8) for a tradable stock. + - "close" (default): silently substitute the close price for the deal price. + This keeps backward compatibility, but **NOTE** it can be unrealistic: + e.g. an order configured to deal at open/vwap may be filled at close + price on days where open/vwap data is missing, which may inflate + backtest results if such days are systematically special + (newly listed stocks, data errors, etc.). + - "reject": treat the order as untradable (the order will not be executed), + which is more conservative and realistic when price data is missing. :param extra_quote: pandas, dataframe consists of columns: like ['$vwap', '$close', '$volume', '$factor', 'limit_sell', 'limit_buy']. The limit indicates that the etf is tradable on a specific day. @@ -163,6 +174,11 @@ def __init__( else: raise NotImplementedError(f"This type of input is not supported") + if deal_price_fallback not in ("close", "reject"): + raise ValueError(f"deal_price_fallback must be 'close' or 'reject', got {deal_price_fallback}") + self.deal_price_fallback = deal_price_fallback + self._deal_price_fallback_count = 0 + if isinstance(codes, str): codes = D.instruments(codes) self.codes = codes @@ -233,6 +249,17 @@ def get_quote_from_qlib(self) -> None: # update limit self._update_limit(self.limit_threshold) + # consistency check between suspension (NaN $close) and zero $volume + # suspended days are detected by NaN $close; if the data forward-fills prices on suspended + # days, such days will have a non-NaN close with zero volume and be wrongly treated as tradable + if "$volume" in self.quote_df.columns: + zero_vol_cnt = int(((self.quote_df["$volume"] == 0) & ~self.quote_df["$close"].isna()).sum()) + if zero_vol_cnt > 0: + self.logger.warning( + f"{zero_vol_cnt} stock-days have zero volume but non-NaN close: " + "if your data forward-fills suspended days, they will be treated as tradable." + ) + # concat extra_quote if self.extra_quote is not None: # process extra_quote @@ -423,7 +450,7 @@ def deal_order( order: Order, trade_account: Account | None = None, position: BasePosition | None = None, - dealt_order_amount: Dict[str, float] = defaultdict(float), + dealt_order_amount: Optional[Dict[str, float]] = None, ) -> Tuple[float, float, float]: """ Deal order when the actual transaction @@ -434,6 +461,8 @@ def deal_order( :param dealt_order_amount: the dealt order amount dict with the format of {stock_id: float} :return: trade_val, trade_cost, trade_price """ + if dealt_order_amount is None: + dealt_order_amount = defaultdict(float) # check order first. if not self.check_order(order): order.deal_amount = 0.0 @@ -508,9 +537,21 @@ def get_deal_price( deal_price = self.quote.get_data(stock_id, start_time, end_time, field=pstr, method=method) if method is not None and (deal_price is None or np.isnan(deal_price) or deal_price <= 1e-08): + self._deal_price_fallback_count += 1 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") - deal_price = self.get_close(stock_id, start_time, end_time, method) + if self.deal_price_fallback == "close": + self.logger.warning( + f"setting deal_price to close price " + f"(deal price fallback triggered {self._deal_price_fallback_count} times in total)" + ) + deal_price = self.get_close(stock_id, start_time, end_time, method) + else: + # "reject": the order will be treated as untradable (refer to `_calc_trade_info_by_order`) + self.logger.warning( + f"rejecting the order due to missing deal price " + f"(deal price fallback triggered {self._deal_price_fallback_count} times in total)" + ) + deal_price = np.nan return deal_price def get_factor( @@ -874,6 +915,12 @@ def _calc_trade_info_by_order( float, self.get_deal_price(order.stock_id, order.start_time, order.end_time, direction=order.direction), ) + if trade_price is None or np.isnan(trade_price): + # The deal price is unavailable (e.g. `deal_price_fallback == "reject"`), + # so the order is treated as untradable (mimicking the limitation path in `deal_order`) + order.deal_amount = 0.0 + self.logger.debug(f"Order failed due to unavailable deal price: {order}") + return np.nan, 0.0, 0.0 total_trade_val = cast(float, self.get_volume(order.stock_id, order.start_time, order.end_time)) * trade_price order.factor = self.get_factor(order.stock_id, order.start_time, order.end_time) order.deal_amount = order.amount # set to full amount and clip it step by step diff --git a/qlib/data/dataset/__init__.py b/qlib/data/dataset/__init__.py index a6cace3730f..6fc33c35459 100644 --- a/qlib/data/dataset/__init__.py +++ b/qlib/data/dataset/__init__.py @@ -120,6 +120,44 @@ def __init__( self.segments = segments.copy() self.fetch_kwargs = copy(fetch_kwargs) super().__init__(**kwargs) + self._maybe_warn_fit_leakage() + + def _maybe_warn_fit_leakage(self): + """ + Warn loudly if fit-based processors of the handler may leak non-train data. + + Processors such as `ZScoreNorm`/`MinMaxNorm`/`RobustZScoreNorm` learn statistics on + `[fit_start_time, fit_end_time]`. If `fit_end_time` reaches into a non-train segment + (e.g. valid/test), the learned statistics leak future information into training. + """ + get_all_processors = getattr(self.handler, "get_all_processors", None) + if not callable(get_all_processors): + return + logger = get_module_logger("DatasetH") + for proc in get_all_processors(): + fit_end_time = getattr(proc, "fit_end_time", None) + if fit_end_time is None: + continue + try: + fit_end_time = pd.Timestamp(fit_end_time) + except (TypeError, ValueError): + continue + for seg_name, seg in self.segments.items(): + if str(seg_name).lower() in ("train", "insample"): + continue + if not isinstance(seg, (tuple, list)) or len(seg) != 2 or seg[0] is None: + continue + try: + seg_start = pd.Timestamp(seg[0]) + except (TypeError, ValueError): + continue + if fit_end_time >= seg_start: + logger.warning( + f"POTENTIAL DATA LEAKAGE: the `fit_end_time` ({fit_end_time}) of processor " + f"{type(proc).__name__} is not earlier than the start ({seg_start}) of the " + f"non-train segment '{seg_name}'. The statistics learned by the processor " + f"may leak future information into training." + ) def config(self, handler_kwargs: dict = None, **kwargs): """ diff --git a/qlib/data/dataset/processor.py b/qlib/data/dataset/processor.py index d05dbe381c5..f123600b10e 100644 --- a/qlib/data/dataset/processor.py +++ b/qlib/data/dataset/processor.py @@ -15,6 +15,21 @@ from qlib.data import D +def check_fit_range(fit_start_time, fit_end_time): + """ + Validate the fit range of fit-based processors (e.g. `MinMaxNorm`, `ZScoreNorm`, `RobustZScoreNorm`). + + NOTE: correctly setting `fit_start_time` and `fit_end_time` is very important!!! + `fit_end_time` **must not** include any information from the test data!!! + Otherwise the learned statistics will leak future information into training. + """ + if fit_start_time is not None and fit_end_time is not None: + if pd.Timestamp(fit_start_time) > pd.Timestamp(fit_end_time): + raise ValueError( + f"fit_start_time ({fit_start_time}) must not be later than fit_end_time ({fit_end_time})" + ) + + def get_group_columns(df: pd.DataFrame, group: Union[Text, None]): """ get a group of columns from multi-index columns DataFrame @@ -112,8 +127,8 @@ def is_for_infer(self) -> bool: class DropCol(Processor): - def __init__(self, col_list=[]): - self.col_list = col_list + def __init__(self, col_list=None): + self.col_list = [] if col_list is None else col_list def __call__(self, df): if isinstance(df.columns, pd.MultiIndex): @@ -127,9 +142,9 @@ def readonly(self): class FilterCol(Processor): - def __init__(self, fields_group="feature", col_list=[]): + def __init__(self, fields_group="feature", col_list=None): self.fields_group = fields_group - self.col_list = col_list + self.col_list = [] if col_list is None else col_list def __call__(self, df): cols = get_group_columns(df, self.fields_group) @@ -197,6 +212,7 @@ class MinMaxNorm(Processor): def __init__(self, fit_start_time, fit_end_time, fields_group=None): # NOTE: correctly set the `fit_start_time` and `fit_end_time` is very important !!! # `fit_end_time` **must not** include any information from the test data!!! + check_fit_range(fit_start_time, fit_end_time) self.fit_start_time = fit_start_time self.fit_end_time = fit_end_time self.fields_group = fields_group @@ -231,6 +247,7 @@ class ZScoreNorm(Processor): def __init__(self, fit_start_time, fit_end_time, fields_group=None): # NOTE: correctly set the `fit_start_time` and `fit_end_time` is very important !!! # `fit_end_time` **must not** include any information from the test data!!! + check_fit_range(fit_start_time, fit_end_time) self.fit_start_time = fit_start_time self.fit_end_time = fit_end_time self.fields_group = fields_group @@ -273,6 +290,7 @@ class RobustZScoreNorm(Processor): def __init__(self, fit_start_time, fit_end_time, fields_group=None, clip_outlier=True): # NOTE: correctly set the `fit_start_time` and `fit_end_time` is very important !!! # `fit_end_time` **must not** include any information from the test data!!! + check_fit_range(fit_start_time, fit_end_time) self.fit_start_time = fit_start_time self.fit_end_time = fit_end_time self.fields_group = fields_group diff --git a/qlib/data/ops.py b/qlib/data/ops.py index d9a2ffbb3e3..7d41cabbbd3 100644 --- a/qlib/data/ops.py +++ b/qlib/data/ops.py @@ -716,6 +716,15 @@ class Rolling(ExpressionOps): When the window is set to 0, the behaviour of the operator should follow `expanding` Otherwise, it follows `rolling` + NOTE: the rolling window uses ``min_periods=1``, so the first ``N - 1`` values of a series + are **partial-window** results computed from fewer than ``N`` observations instead of NaN. + For example, ``Mean($close, 20)`` on a newly listed stock with only 5 days of history will + emit a "20-day" mean computed from just 1-5 observations. This is the long-standing default + behaviour and is kept for backward compatibility (downstream benchmark results depend on it). + If you want strict full-window semantics, mask the early values yourself, e.g. filter out + rows where the instrument has less than ``N`` periods of history, or use an expression such + as ``If(Ge(Count($close, N), N), Mean($close, N), )`` to invalidate partial windows. + Parameters ---------- feature : Expression diff --git a/tests/test_all_pipeline.py b/tests/test_all_pipeline.py index 7bbdaefe3c2..5f0415fb246 100644 --- a/tests/test_all_pipeline.py +++ b/tests/test_all_pipeline.py @@ -157,8 +157,8 @@ def tearDownClass(cls) -> None: @pytest.mark.slow def test_0_train(self): TestAllFlow.PRED_SCORE, ic_ric, TestAllFlow.RID, uri_path = train(self.URI_PATH) - self.assertGreaterEqual(ic_ric["ic"].all(), 0, "train failed") - self.assertGreaterEqual(ic_ric["ric"].all(), 0, "train failed") + self.assertGreater(ic_ric["ic"].mean(), 0, "train failed") + self.assertGreater(ic_ric["ric"].mean(), 0, "train failed") @pytest.mark.slow def test_1_backtest(self):