exp 20: sync fixed MomentumGateTopk + HmmRiskTopk + rolling-IC rank_ensemble into code snapshot

code/tac-qlib/tac_qlib/contrib/strategy/momentum_gate.py

@@ -1,21 +1,16 @@
 """TopkDropout with a 1-day momentum entry-confirmation gate.
 
-Wraps qlib's ``TopkDropoutStrategy`` and adds one rule on top of the pure
-signal ranking: a name may only be **bought** when its trailing 1-day return
-is positive (short-term momentum confirmation, Lag-1 autocorr ~ +0.45 in the
-time-series study). Held names are never force-sold by this gate — exits stay
-the TopkDropout rule (fall out of top-k / n_drop). This attacks the churn/cost
-drag: the reference TopkDropout bought and sold ~590 times in 150 days ($63.5k
-cost); momentum confirmation filters the entry side so a name that just fell
-is not immediately re-bought on rank alone.
+Gates NEW entries on short-term momentum: a name that is not currently held
+may only be bought when its trailing 1-day return is above ``min_momentum``
+(Lag-1 autocorr ~ +0.45 in the time-series study => short-term momentum
+continuation). Held names are never force-sold by this gate — exits stay the
+pure TopkDropout rule.
 
-Implementation: overrides ``generate_trade_decision`` and wraps the target
-weight dict produced by the base strategy — any BUY weight for a name whose
-1-day return <= 0 (or missing quote) is zeroed (kept at 0 weight => no entry).
-Exits (weights already held) are preserved.
-
-The 1-day return is read from the exchange's deal price over the previous
-bar (no lookahead: decision on day t uses the close of t-1).
+Implementation: override ``generate_trade_decision`` and zero out the signal
+score of any non-held name that fails the momentum check BEFORE calling the
+base TopkDropout decision, so it can never be selected as a buy candidate.
+This is a clean pre-filter: the rest of the strategy (top-k, n_drop, sizing,
+costs) is untouched.
 
 Wired into a workflow yaml like:
 
@@ -37,8 +32,7 @@ import copy
 
 import pandas as pd
 
-from qlib.backtest import Order
-from qlib.backtest.decision import OrderDir, TradeDecisionWO
+from qlib.backtest.decision import TradeDecisionWO
 from qlib.backtest.position import Position
 from qlib.contrib.strategy.signal_strategy import TopkDropoutStrategy
 
@@ -56,17 +50,16 @@ class MomentumGateTopk(TopkDropoutStrategy):
         """True when the trailing 1-day return is above the momentum floor."""
         try:
             cur = self.trade_exchange.get_deal_price(
-                stock_id=code, start_time=trade_start, end_time=trade_end, direction=OrderDir.BUY
+                stock_id=code, start_time=trade_start, end_time=trade_end, direction=1
             )
         except Exception:
             return False
         if cur is None or cur != cur or cur <= 0:
             return False
-        # previous bar: shift the window back one step
         prev_start = trade_start - pd.Timedelta(days=5)
         prev_end = trade_start - pd.Timedelta(seconds=1)
         prev = self.trade_exchange.get_deal_price(
-            stock_id=code, start_time=prev_start, end_time=prev_end, direction=OrderDir.SELL
+            stock_id=code, start_time=prev_start, end_time=prev_end, direction=0
         )
         if prev is None or prev != prev or prev <= 0:
             return False
@@ -79,30 +72,20 @@ class MomentumGateTopk(TopkDropoutStrategy):
         pred_score = self.signal.get_signal(start_time=pred_start_time, end_time=pred_end_time)
         if pred_score is None:
             return TradeDecisionWO([], self)
+        if isinstance(pred_score, pd.DataFrame):
+            pred_score = pred_score.iloc[:, 0]
+
         current_temp = copy.deepcopy(self.trade_position)
         assert isinstance(current_temp, Position)
-
-        # base topk target weights (equal-weight top-k minus n_drop)
-        target_weight_position = self.generate_target_weight_position(
-            score=pred_score, current=current_temp, trade_start_time=trade_start_time, trade_end_time=trade_end_time
-        )
-        # entry gate: zero out any NEW (not currently held) buy weight when momentum fails
         held = set(current_temp.get_stock_list())
-        gated = {}
-        for code, w in target_weight_position.items():
-            is_new = code not in held or abs(current_temp.get_stock_amount(code)) <= 1e-6
-            if is_new and not self._momentum_ok(code, trade_start_time, trade_end_time):
-                continue  # skip entry (momentum not confirmed)
-            gated[code] = w
+        held = {c for c in held if abs(current_temp.get_stock_amount(c)) > 1e-6}
 
-        order_list = self.order_generator.generate_order_list_from_target_weight_position(
-            current=current_temp,
-            trade_exchange=self.trade_exchange,
-            risk_degree=self.get_risk_degree(trade_step),
-            target_weight_position=gated,
-            pred_start_time=pred_start_time,
-            pred_end_time=pred_end_time,
-            trade_start_time=trade_start_time,
-            trade_end_time=trade_end_time,
-        )
-        return TradeDecisionWO(order_list, self)
+        # pre-filter: zero the score of non-held names that fail momentum
+        pred_score = pred_score.copy()
+        for code in pred_score.index:
+            if code in held:
+                continue  # never gate exits / re-balancing of held names
+            if not self._momentum_ok(code, trade_start_time, trade_end_time):
+                pred_score[code] = -1e9  # cannot enter today
+
+        return super().generate_trade_decision(execute_result)