Resolve PR comments; remove private classes;

qlib/rl/order_execution/interpreter.py

@@ -5,7 +5,7 @@
 
 import math
 from pathlib import Path
-from typing import Any, List, cast
+from typing import Any, List, Optional, cast
 
 import numpy as np
 import pandas as pd
@@ -185,9 +185,11 @@ class CategoricalActionInterpreter(ActionInterpreter[SAOEState, int, float]):
         Then when policy givens decision $x$, $a_x$ times order amount is the output.
         It can also be an integer $n$, in which case the list of length $n+1$ is auto-generated,
         i.e., $[0, 1/n, 2/n, \\ldots, n/n]$.
+    max_step
+        Total number of steps (an upper-bound estimation). For example, 390min / 30min-per-step = 13 steps.
     """
 
-    def __init__(self, values: int | List[float], max_step: int = None) -> None:
+    def __init__(self, values: int | List[float], max_step: Optional[int] = None) -> None:
         if isinstance(values, int):
             values = [i / values for i in range(0, values + 1)]
         self.action_values = values

qlib/rl/order_execution/network.py

@@ -138,28 +138,3 @@ def forward(self, Q, K, V):
         attn_vec = torch.einsum("ijk,ikl->ijl", attn_prob, v)
 
         return attn_vec
-
-
-class DualAttentionRNN(Recurrent):
-    """
-    Dual-attention RNN leverages features from yesterday and fuses them into features today.
-    """
-
-    def _init_extra_branches(self):
-        self.attention = Attention(self.hidden_dim, self.hidden_dim)
-        self.num_sources += 1
-
-    def _source_features(self, obs: FullHistoryObs, device: torch.device) -> Tuple[List[torch.Tensor], torch.Tensor]:
-        sources, data_out = super()._source_features(obs, device)
-
-        data_prev = obs["data_processed_prev"]
-        cur_time = obs["cur_tick"].long()
-        bs_indices = torch.arange(cur_time.size(0), device=device)
-
-        data_prev_in = self.raw_fc(data_prev)
-        data_prev_out, _ = self.prev_rnn(data_prev_in)
-        att_out = self.attention(data_out, data_prev_out, data_prev_out)
-        att_out = att_out[bs_indices, cur_time]
-        sources.insert(1, att_out)
-
-        return sources, data_out

qlib/rl/order_execution/strategy.py

@@ -254,79 +254,3 @@ def _generate_trade_decision(self, execute_result: list = None) -> BaseTradeDeci
                 order_list.append(oh.create(order.stock_id, exec_vol, order.direction))
 
         return TradeDecisionWO(order_list=order_list, strategy=self)
-
-
-class MultiplexStrategyBase(BaseStrategy, metaclass=ABCMeta):
-    def __init__(
-        self,
-        strategies: List[BaseStrategy] | List[dict],
-        outer_trade_decision: BaseTradeDecision = None,
-        level_infra: LevelInfrastructure = None,
-        common_infra: CommonInfrastructure = None,
-        trade_exchange: Exchange = None,
-    ) -> None:
-        super().__init__(
-            outer_trade_decision=outer_trade_decision,
-            level_infra=level_infra,
-            common_infra=common_infra,
-            trade_exchange=trade_exchange,
-        )
-
-        self._strategies = [init_instance_by_config(strategy, accept_types=BaseStrategy) for strategy in strategies]
-
-    def set_env(self, env: EnvWrapper | CollectDataEnvWrapper) -> None:
-        for strategy in self._strategies:
-            if hasattr(strategy, "set_env"):
-                strategy.set_env(env)
-
-
-class MultiplexStrategyOnTradeStep(MultiplexStrategyBase):
-    """To use different strategy on different step of the outer calendar"""
-
-    def __init__(
-        self,
-        strategies: List[BaseStrategy] | List[dict],
-        outer_trade_decision: BaseTradeDecision = None,
-        level_infra: LevelInfrastructure = None,
-        common_infra: CommonInfrastructure = None,
-        trade_exchange: Exchange = None,
-    ) -> None:
-        super(MultiplexStrategyOnTradeStep, self).__init__(
-            strategies=strategies,
-            outer_trade_decision=outer_trade_decision,
-            level_infra=level_infra,
-            common_infra=common_infra,
-            trade_exchange=trade_exchange,
-        )
-
-    def reset_level_infra(self, level_infra: LevelInfrastructure) -> None:
-        for strategy in self._strategies:
-            strategy.reset_level_infra(level_infra)
-
-    def reset_common_infra(self, common_infra: CommonInfrastructure) -> None:
-        for strategy in self._strategies:
-            strategy.reset_common_infra(common_infra)
-
-    def reset(self, outer_trade_decision: BaseTradeDecision = None, **kwargs: Any) -> None:
-        super().reset(outer_trade_decision=outer_trade_decision, **kwargs)
-
-        if outer_trade_decision is not None:
-            strategy = self._get_current_strategy()
-            strategy.reset(outer_trade_decision=outer_trade_decision, **kwargs)
-
-    def generate_trade_decision(self, execute_result: list = None) -> BaseTradeDecision:
-        if self.outer_trade_decision is not None:
-            strategy = self._get_current_strategy()
-            return strategy.generate_trade_decision(execute_result=execute_result)
-        else:
-            return TradeDecisionWO([], self)
-
-    def post_exe_step(self, execute_result: list) -> None:
-        if self.outer_trade_decision is not None:
-            strategy = self._get_current_strategy()
-            if isinstance(strategy, RLStrategy):
-                strategy.post_exe_step(execute_result=execute_result)
-
-    def _get_current_strategy(self) -> BaseStrategy:
-        outer_calendar = self.outer_trade_decision.strategy.trade_calendar
-        return self._strategies[outer_calendar.get_trade_step()]