start experiment 20 (exp/20-improve-the-risk-limit-reference-signal)

2026-08-17 02:55:26 +00:00
parent 7fad62a4ef
commit 337f6e17d8
4 changed files with 317 additions and 5 deletions
@@ -0,0 +1,108 @@
+"""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.
+
+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).
+
+Wired into a workflow yaml like:
+
+    strategy:
+        class: MomentumGateTopk
+        module_path: tac_qlib.contrib.strategy.momentum_gate
+        kwargs:
+            signal: "<PRED>"
+            topk: 10
+            n_drop: 2
+            only_tradable: true
+            risk_degree: 0.95
+            min_momentum: 0.0
+"""
+
+from __future__ import annotations
+
+import copy
+
+import pandas as pd
+
+from qlib.backtest import Order
+from qlib.backtest.decision import OrderDir, TradeDecisionWO
+from qlib.backtest.position import Position
+from qlib.contrib.strategy.signal_strategy import TopkDropoutStrategy
+
+__all__ = ["MomentumGateTopk"]
+
+
+class MomentumGateTopk(TopkDropoutStrategy):
+    """TopkDropoutStrategy gated on 1-day momentum for new entries."""
+
+    def __init__(self, *, min_momentum: float = 0.0, **kwargs):
+        super().__init__(**kwargs)
+        self.min_momentum = float(min_momentum)
+
+    def _momentum_ok(self, code, trade_start, trade_end) -> bool:
+        """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
+            )
+        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
+        )
+        if prev is None or prev != prev or prev <= 0:
+            return False
+        return (cur / prev - 1.0) >= self.min_momentum
+
+    def generate_trade_decision(self, execute_result=None):
+        trade_step = self.trade_calendar.get_trade_step()
+        trade_start_time, trade_end_time = self.trade_calendar.get_step_time(trade_step)
+        pred_start_time, pred_end_time = self.trade_calendar.get_step_time(trade_step, shift=1)
+        pred_score = self.signal.get_signal(start_time=pred_start_time, end_time=pred_end_time)
+        if pred_score is None:
+            return TradeDecisionWO([], self)
+        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
+
+        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)