Fix vacuous IC test assertion; surface silent deal-price fallback; warn on processor fit-range leakage

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

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):
         """

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

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), <fallback>)`` to invalidate partial windows.
+
     Parameters
     ----------
     feature : Expression

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):