From 80c7230e17b5eb9c08d1e61737e79cae074680c1 Mon Sep 17 00:00:00 2001 From: zhaoli Date: Mon, 17 Aug 2026 04:08:07 +0000 Subject: [PATCH] exp 20: sync fixed MomentumGateTopk + HmmRiskTopk + rolling-IC rank_ensemble into code snapshot --- code/MANIFEST.txt | 4 +- .../tac_qlib/contrib/strategy/hmm_risk.py | 128 +++++++----------- .../contrib/strategy/momentum_gate.py | 69 ++++------ 3 files changed, 79 insertions(+), 122 deletions(-) diff --git a/code/MANIFEST.txt b/code/MANIFEST.txt index cacee30..8850ff0 100644 --- a/code/MANIFEST.txt +++ b/code/MANIFEST.txt @@ -18,8 +18,8 @@ 4afcf9058231111c412925f4c4b84e81d656db87 tac-qlib/tac_qlib/contrib/strategy/__init__.py 74e5ecbbbb20bb71fd5cd083383de4ce88476712 tac-qlib/tac_qlib/contrib/strategy/__pycache__/__init__.cpython-312.pyc afaf562aeaa12cebc8529cd916153252e7e3c38a tac-qlib/tac_qlib/contrib/strategy/__pycache__/optimal_stop.cpython-312.pyc - f3a084be7aea509ec10381af59a0c996bf66c5b6 tac-qlib/tac_qlib/contrib/strategy/hmm_risk.py - 5e1ac61ebba6bc5bfb90c6d74ca29c28ff07aa13 tac-qlib/tac_qlib/contrib/strategy/momentum_gate.py +96a0a25201f0a1bb2fc2190e26228c5c0e711a79 tac-qlib/tac_qlib/contrib/strategy/hmm_risk.py +816de5d58ae23d996635d42331cf9fc8963d5dbe tac-qlib/tac_qlib/contrib/strategy/momentum_gate.py 79aaad9e39fcc740a773f4f63c512ce1086cfde0 tac-qlib/tac_qlib/contrib/strategy/optimal_stop.py 92e6e90eb0cd0a25142034560f27adb6b705b1a8 tac-qlib/tac_qlib/data/__init__.py 0ed1ead6c1314a3f25784d453e54a15a8a04baaa tac-qlib/tac_qlib/data/__pycache__/__init__.cpython-312.pyc diff --git a/code/tac-qlib/tac_qlib/contrib/strategy/hmm_risk.py b/code/tac-qlib/tac_qlib/contrib/strategy/hmm_risk.py index f3a084b..96a0a25 100644 --- a/code/tac-qlib/tac_qlib/contrib/strategy/hmm_risk.py +++ b/code/tac-qlib/tac_qlib/contrib/strategy/hmm_risk.py @@ -1,24 +1,22 @@ """TopkDropout with HMM high-volatility + drawdown-pause risk gates. -Adds three risk controls on top of ``TopkDropoutStrategy``: +Gates NEW entries on two risk conditions (held names are never force-sold): -1. **HMM high-vol pause**: when the current day's HMM high-volatility regime - probability (``sp_hmm_p_regime1`` feature, regime-1 = high-vol) is above - ``hmm_pause_pct``, new buys are paused (existing positions held). This - encodes the time-series study's finding that HMM high-vol probability pulses - BEFORE sharp moves (regime-change cut) — pausing new exposure at the - boundary reduces drawdown from "价格过度反应". +1. **HMM high-vol pause**: when the cross-sectional mean of ``sp_hmm_p_regime1`` + (HMM high-vol regime probability) on the signal date is >= ``hmm_pause_pct``, + new buys are paused. The time-series study showed HMM high-vol probability + pulses BEFORE sharp moves (regime-change cut) — pausing new exposure at the + boundary reduces drawdown from price over-reaction. 2. **Drawdown pause**: when the account equity drawdown from its running peak - exceeds ``drawdown_pause_pct``, new buys are paused (positions kept). This - is the ``drawdown_pause_pct`` risk-limit expressed in the backtest (the - executor-side gate is documented as not expressible in a one-shot qlib - backtest — here we implement it inside the strategy). -3. **Liquidity floor**: names whose average daily dollar volume is below - ``liquidity_floor_adv`` are dropped from the tradable set (the proven risk - mitigant from exp-18: $5M floor cut drawdown 7.9%->5.4% at higher IR). + exceeds ``drawdown_pause_pct``, new buys are paused. This is the + ``drawdown_pause_pct`` risk-limit expressed inside the backtest (the pure + executor-side gate is documented as not expressible in a one-shot backtest). +3. **Liquidity floor**: names whose 20-day average daily dollar volume is below + ``liquidity_floor_adv`` are dropped from BUY candidates (the proven mitigant + from exp-18: $5M floor cut drawdown 7.9%->5.4% at higher IR). -Held names are never force-sold by these gates; only new entries are gated. -Exits remain the pure TopkDropout rule. +Implementation: pre-filter the signal score before the base TopkDropout +decision — non-held names get score 0 when any gate fires. Wired into a workflow yaml like: @@ -67,23 +65,15 @@ class HmmRiskTopk(TopkDropoutStrategy): self.drawdown_pause_pct = float(drawdown_pause_pct) self.liquidity_floor_adv = float(liquidity_floor_adv) self._peak_equity = 0.0 - self._hmm_pause_active = False - self._drawdown_pause_active = False - # ------------------------------------------------------------- state - def _high_vol_active(self, trade_start) -> bool: - """True when HMM high-vol regime probability >= pause threshold.""" + # ------------------------------------------------------------- gates + def _hmm_high_vol(self, pred_date) -> bool: + """Cross-sectional mean HMM high-vol regime probability >= threshold.""" try: from qlib.data import D - cal = D.calendar(start_time=str((trade_start - pd.Timedelta(days=10)).date()), - end_time=str(trade_start.date())) - if len(cal) == 0: - return False - ref_date = str(pd.Timestamp(cal[-1]).date()) - feat = D.features( - D.instruments("all"), ["$sp_hmm_p_regime1"], start_time=ref_date, end_time=ref_date - ) + feat = D.features(D.instruments("all"), ["$sp_hmm_p_regime1"], + start_time=pred_date, end_time=pred_date) if feat is None or len(feat) == 0: return False p = feat["$sp_hmm_p_regime1"].dropna() @@ -93,11 +83,22 @@ class HmmRiskTopk(TopkDropoutStrategy): except Exception: return False - def _update_pause(self, equity: float, trade_start) -> None: + def _drawdown_active(self, equity: float) -> bool: + if self.drawdown_pause_pct <= 0: + return False self._peak_equity = max(self._peak_equity, equity) - dd = (self._peak_equity - equity) / self._peak_equity * 100.0 if self._peak_equity > 0 else 0.0 - self._drawdown_pause_active = self.drawdown_pause_pct > 0 and dd >= self.drawdown_pause_pct - self._hmm_pause_active = self._high_vol_active(trade_start) + if self._peak_equity <= 0: + return False + dd = (self._peak_equity - equity) / self._peak_equity * 100.0 + return dd >= self.drawdown_pause_pct + + def _illiquid(self, codes, asof) -> Dict[str, bool]: + if self.liquidity_floor_adv <= 0 or not codes: + return {} + from tac_qlib.risk_limits import dollar_adv + + adv = dollar_adv(codes, market="US", asof=asof, lookback=20) + return {c: adv.get(str(c).upper(), 0.0) < self.liquidity_floor_adv for c in codes} # ------------------------------------------------------------- decision def generate_trade_decision(self, execute_result=None): @@ -107,58 +108,31 @@ class HmmRiskTopk(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) + held = {c for c in current_temp.get_stock_list() if abs(current_temp.get_stock_amount(c)) > 1e-6} - # account equity for drawdown pause equity = current_temp.get_cash() - for code in current_temp.get_stock_list(): - amt = abs(current_temp.get_stock_amount(code)) + for code in held: mark = self.trade_exchange.get_deal_price( stock_id=code, start_time=trade_start_time, end_time=trade_end_time, direction=1 ) if mark is not None and np.isfinite(mark): - equity += amt * mark - self._update_pause(equity, trade_start_time) + equity += abs(current_temp.get_stock_amount(code)) * mark - 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 - ) + hmm_pause = self._hmm_high_vol(str(pd.Timestamp(pred_start_time).date())) + dd_pause = self._drawdown_active(equity) + buys_paused = hmm_pause or dd_pause - held = set(current_temp.get_stock_list()) - held_amt = {c: abs(current_temp.get_stock_amount(c)) for c in held} + pred_score = pred_score.copy() + if buys_paused or self.liquidity_floor_adv > 0: + new_codes = [c for c in pred_score.index if c not in held] + illiquid = self._illiquid(new_codes, str(pd.Timestamp(pred_start_time).date())) + for code in new_codes: + if buys_paused or illiquid.get(code, False): + pred_score[code] = -1e9 # cannot enter today - # liquidity floor: drop names below the ADV floor from BUY candidates - illiquid: Dict[str, bool] = {} - if self.liquidity_floor_adv > 0: - from tac_qlib.risk_limits import dollar_adv - - codes = [c for c in target_weight_position if c not in held] - if codes: - adv = dollar_adv(codes, market="US", asof=str(pd.Timestamp(trade_start_time).date()), lookback=20) - for c in codes: - illiquid[c] = adv.get(str(c).upper(), 0.0) < self.liquidity_floor_adv - - buys_paused = self._hmm_pause_active or self._drawdown_pause_active - - gated = {} - for code, w in target_weight_position.items(): - is_new = code not in held or held_amt.get(code, 0.0) <= 1e-6 - if is_new: - if buys_paused: - continue # risk gate: no new entries - if illiquid.get(code, False): - continue # liquidity floor: drop illiquid buy candidate - 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) + return super().generate_trade_decision(execute_result) diff --git a/code/tac-qlib/tac_qlib/contrib/strategy/momentum_gate.py b/code/tac-qlib/tac_qlib/contrib/strategy/momentum_gate.py index 5e1ac61..816de5d 100644 --- a/code/tac-qlib/tac_qlib/contrib/strategy/momentum_gate.py +++ b/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)