RL backtest pipeline on 5-min data

qlib/contrib/ops/high_freq.py

@@ -70,7 +70,7 @@ class DayCumsum(ElemOperator):
         Otherwise, the value is zero.
     """
 
-    def __init__(self, feature, start: str = "9:30", end: str = "14:59"):
+    def __init__(self, feature, start: str = "9:30", end: str = "14:59", data_granularity: int = 1):
         self.feature = feature
         self.start = datetime.strptime(start, "%H:%M")
         self.end = datetime.strptime(end, "%H:%M")
@@ -80,15 +80,17 @@ def __init__(self, feature, start: str = "9:30", end: str = "14:59"):
         self.noon_open = datetime.strptime("13:00", "%H:%M")
         self.noon_close = datetime.strptime("15:00", "%H:%M")
 
-        self.start_id = time_to_day_index(self.start)
-        self.end_id = time_to_day_index(self.end)
+        self.data_granularity = data_granularity
+        self.start_id = time_to_day_index(self.start) // self.data_granularity
+        self.end_id = time_to_day_index(self.end) // self.data_granularity
+        assert 240 % self.data_granularity == 0
 
     def period_cusum(self, df):
         df = df.copy()
-        assert len(df) == 240
+        assert len(df) == 240 // self.data_granularity
         df.iloc[0 : self.start_id] = 0
         df = df.cumsum()
-        df.iloc[self.end_id + 1 : 240] = 0
+        df.iloc[self.end_id + 1 : 240 // self.data_granularity] = 0
         return df
 
     def _load_internal(self, instrument, start_index, end_index, freq):

qlib/rl/contrib/backtest.py

@@ -35,7 +35,7 @@ def _get_multi_level_executor_config(
         "class": "SimulatorExecutor",
         "module_path": "qlib.backtest.executor",
         "kwargs": {
-            "time_per_step": "1min",
+            "time_per_step": "5min",
             "verbose": False,
             "trade_type": SimulatorExecutor.TT_PARAL if cash_limit is not None else SimulatorExecutor.TT_SERIAL,
             "generate_report": generate_report,
@@ -187,7 +187,7 @@ def single_with_simulator(
         exchange_config.update(
             {
                 "codes": stocks,
-                "freq": "1min",
+                "freq": "5min",
             }
         )
 
@@ -286,7 +286,7 @@ def single_with_collect_data_loop(
     exchange_config.update(
         {
             "codes": stocks,
-            "freq": "1min",
+            "freq": "5min",
         }
     )
 

qlib/rl/contrib/naive_config_parser.py

@@ -98,7 +98,7 @@ def get_backtest_config_fromfile(path: str) -> dict:
         "debug_single_day": None,
         "concurrency": -1,
         "multiplier": 1.0,
-        "output_dir": "outputs/",
+        "output_dir": "outputs_backtest/",
         "generate_report": False,
     }
     backtest_config = merge_a_into_b(a=backtest_config, b=backtest_config_default)

qlib/rl/contrib/train_onpolicy.py

@@ -132,8 +132,8 @@ def _simulator_factory_simple(order: Order) -> SingleAssetOrderExecutionSimple:
         for tag in ("train", "valid", "test")
     ]
 
+    callbacks: List[Callback] = []
     if "checkpoint_path" in trainer_config:
-        callbacks: List[Callback] = []
         callbacks.append(MetricsWriter(dirpath=Path(trainer_config["checkpoint_path"])))
         callbacks.append(
             Checkpoint(

qlib/rl/data/integration.py

@@ -82,10 +82,9 @@ def _convert_to_path(path: str | Path) -> Path:
         return path if isinstance(path, Path) else Path(path)
 
     provider_uri_map = {}
-    if "provider_uri_day" in qlib_config:
-        provider_uri_map["day"] = _convert_to_path(qlib_config["provider_uri_day"]).as_posix()
-    if "provider_uri_1min" in qlib_config:
-        provider_uri_map["1min"] = _convert_to_path(qlib_config["provider_uri_1min"]).as_posix()
+    for granularity in ["1min", "5min", "day"]:
+        if f"provider_uri_{granularity}" in qlib_config:
+            provider_uri_map[f"{granularity}"] = _convert_to_path(qlib_config[f"provider_uri_{granularity}"]).as_posix()
 
     qlib.init(
         region=REG_CN,

qlib/rl/order_execution/strategy.py

@@ -89,6 +89,7 @@ def __init__(
         exchange: Exchange,
         ticks_per_step: int,
         backtest_data: IntradayBacktestData,
+        data_granularity: int = 1,
     ) -> None:
         self.position = order.amount
         self.order = order
@@ -106,11 +107,13 @@ def __init__(
 
         self.cur_time = max(backtest_data.ticks_for_order[0], order.start_time)
         self.ticks_per_step = ticks_per_step
+        self.data_granularity = data_granularity
+        assert self.ticks_per_step % self.data_granularity == 0
 
     def _next_time(self) -> pd.Timestamp:
         current_loc = self.backtest_data.ticks_index.get_loc(self.cur_time)
-        next_loc = current_loc + self.ticks_per_step
-        next_loc = next_loc - next_loc % self.ticks_per_step
+        next_loc = current_loc + (self.ticks_per_step // self.data_granularity)
+        next_loc = next_loc - next_loc % (self.ticks_per_step // self.data_granularity)
         if (
             next_loc < len(self.backtest_data.ticks_index)
             and self.backtest_data.ticks_index[next_loc] < self.order.end_time
@@ -130,7 +133,7 @@ def update(
 
         exec_vol = np.zeros(last_step_size)
         for order, _, __, ___ in execute_result:
-            idx, _ = get_day_min_idx_range(order.start_time, order.end_time, "1min", REG_CN)
+            idx, _ = get_day_min_idx_range(order.start_time, order.end_time, f"{self.data_granularity}min", REG_CN)
             exec_vol[idx - last_step_range[0]] = order.deal_amount
 
         if exec_vol.sum() > self.position and exec_vol.sum() > 0.0:
@@ -168,7 +171,9 @@ def update(
             self.history_exec,
             self._collect_multi_order_metric(
                 order=self.order,
-                datetime=_get_all_timestamps(start_time, end_time, include_end=True),
+                datetime=_get_all_timestamps(
+                    start_time, end_time, include_end=True, granularity=ONE_MIN * self.data_granularity
+                ),
                 market_vol=market_volume,
                 market_price=market_price,
                 exec_vol=exec_vol,
@@ -296,6 +301,7 @@ def __init__(
         outer_trade_decision: BaseTradeDecision = None,
         level_infra: LevelInfrastructure = None,
         common_infra: CommonInfrastructure = None,
+        data_granularity: int = 1,
         **kwargs: Any,
     ) -> None:
         super(SAOEStrategy, self).__init__(
@@ -306,6 +312,7 @@ def __init__(
             **kwargs,
         )
 
+        self._data_granularity = data_granularity
         self.adapter_dict: Dict[tuple, SAOEStateAdapter] = {}
         self._last_step_range = (0, 0)
 
@@ -324,6 +331,7 @@ def _create_qlib_backtest_adapter(
             exchange=self.trade_exchange,
             ticks_per_step=int(pd.Timedelta(self.trade_calendar.get_freq()) / ONE_MIN),
             backtest_data=backtest_data,
+            data_granularity=self._data_granularity,
         )
 
     def reset(self, outer_trade_decision: BaseTradeDecision = None, **kwargs: Any) -> None: