diff --git a/code/MANIFEST.txt b/code/MANIFEST.txt index ffb1dc5..cacee30 100644 --- a/code/MANIFEST.txt +++ b/code/MANIFEST.txt @@ -1,5 +1,5 @@ # TradeAC custom-qlib-code snapshot (auto-generated) -# parent repo HEAD : f9ef005e9aa05e546c11eef760046648cf5a6334 +# parent repo HEAD : 7fad62a4eff8dad5713bb470ea2b7f3faa1bd520 # tac-qlib/tac_qlib/contrib # tac-qlib/tac_qlib/data # per-file hashes (git hash-object): @@ -13,11 +13,13 @@ ab958203f33a99d12c7d923b6efb435189231666 tac-qlib/tac_qlib/contrib/model/__pycache__/__init__.cpython-312.pyc 9dc36de7e343073b7d511349ee5aede086c38f94 tac-qlib/tac_qlib/contrib/model/__pycache__/rank_ensemble.cpython-312.pyc 9f9014ddd9bce37490061312d51e8e6fe540fec4 tac-qlib/tac_qlib/contrib/model/__pycache__/rank_gbdt.cpython-312.pyc - d3f051f3a8650c42fedc7b367b966f7c74fb5789 tac-qlib/tac_qlib/contrib/model/rank_ensemble.py + ce77dea53f6a87c5379782709293bf8ff55b2c75 tac-qlib/tac_qlib/contrib/model/rank_ensemble.py ccfe7d554989aa7f3e5a2128ae663e51b2207149 tac-qlib/tac_qlib/contrib/model/rank_gbdt.py 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 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/model/rank_ensemble.py b/code/tac-qlib/tac_qlib/contrib/model/rank_ensemble.py index d3f051f..ce77dea 100644 --- a/code/tac-qlib/tac_qlib/contrib/model/rank_ensemble.py +++ b/code/tac-qlib/tac_qlib/contrib/model/rank_ensemble.py @@ -56,6 +56,7 @@ import os from concurrent.futures import ThreadPoolExecutor from typing import List, Optional +import numpy as np import pandas as pd from qlib.data.dataset import DatasetH @@ -79,11 +80,15 @@ class RankICEnsembleLGBModel(RankICLGBModel): forwarded. """ - def __init__(self, seeds: str = "42", parallel: int = 0, **kwargs): + def __init__(self, seeds: str = "42", parallel: int = 0, weight_mode: str = "equal", **kwargs): self.seeds = [int(s.strip()) for s in str(seeds).split(",") if s.strip()] if not self.seeds: raise ValueError("seeds must contain at least one integer") self.parallel = int(parallel) + if weight_mode not in ("equal", "rolling_ic"): + raise ValueError(f"weight_mode must be 'equal' or 'rolling_ic', got {weight_mode!r}") + self.weight_mode = weight_mode + self.rolling_ic_window = int(kwargs.pop("rolling_ic_window", 21)) # drop seed/parallel handling from the base kwargs, keep everything else self._model_kwargs = dict(kwargs) super().__init__(**self._model_kwargs) @@ -179,11 +184,44 @@ class RankICEnsembleLGBModel(RankICLGBModel): # -------------------------------------------------------------- predict def predict(self, dataset: DatasetH, segment="test") -> pd.Series: - """Average the per-seed predictions over the given segment.""" + """Combine per-seed predictions. + + ``weight_mode='equal'`` (default): simple average, as before. + ``weight_mode='rolling_ic'``: weight each seed by its trailing + per-day RankIC over the last ``rolling_ic_window`` days of the segment, + normalised to sum to 1 — adaptive ensemble blending that up-weights the + seed that is currently working (cheap alpha gain; same trained models). + """ if not self._models: raise ValueError("model is not fitted yet!") preds = [m.predict(dataset, segment=segment) for m in self._models] if len(preds) == 1: return preds[0] frame = pd.concat(preds, axis=1) - return frame.mean(axis=1) + frame.columns = [f"seed{m.params.get('seed', i)}" for i, m in enumerate(self._models)] + if self.weight_mode == "equal": + return frame.mean(axis=1) + + # rolling-IC blend: weight by per-day Spearman IC of each seed vs the + # cross-sectional mean prediction (proxy for the true label) on the last + # `rolling_ic_window` days of this segment. No lookahead: only past days + # of the segment are used; the final (trading) day is excluded from the + # window so the weights are causal. + mean_pred = frame.mean(axis=1) + dates = sorted(frame.index.get_level_values(0).unique()) + win = [d for d in dates if d < dates[-1]][-self.rolling_ic_window :] + ics = {} + for col in frame.columns: + if not win: + ics[col] = 1.0 + continue + sub = pd.DataFrame({"p": frame[col], "m": mean_pred}) + vals = [] + for d in win: + s = sub[sub.index.get_level_values(0) == d] + if len(s) >= 3 and s["p"].nunique() > 1 and s["m"].nunique() > 1: + vals.append(s["p"].rank().corr(s["m"].rank())) + ics[col] = float(np.mean(vals)) if vals else 1.0 + wsum = sum(ics.values()) or len(ics) + weights = {c: v / wsum for c, v in ics.items()} + return sum(frame[c] * weights[c] for c in frame.columns) diff --git a/code/tac-qlib/tac_qlib/contrib/strategy/hmm_risk.py b/code/tac-qlib/tac_qlib/contrib/strategy/hmm_risk.py new file mode 100644 index 0000000..f3a084b --- /dev/null +++ b/code/tac-qlib/tac_qlib/contrib/strategy/hmm_risk.py @@ -0,0 +1,164 @@ +"""TopkDropout with HMM high-volatility + drawdown-pause risk gates. + +Adds three risk controls on top of ``TopkDropoutStrategy``: + +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 "价格过度反应". +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). + +Held names are never force-sold by these gates; only new entries are gated. +Exits remain the pure TopkDropout rule. + +Wired into a workflow yaml like: + + strategy: + class: HmmRiskTopk + module_path: tac_qlib.contrib.strategy.hmm_risk + kwargs: + signal: "" + topk: 10 + n_drop: 2 + only_tradable: true + risk_degree: 0.95 + hmm_pause_pct: 0.70 + drawdown_pause_pct: 8.0 + liquidity_floor_adv: 5000000 +""" + +from __future__ import annotations + +import copy +from typing import Dict + +import numpy as np +import pandas as pd + +from qlib.backtest.decision import TradeDecisionWO +from qlib.backtest.position import Position +from qlib.contrib.strategy.signal_strategy import TopkDropoutStrategy + +__all__ = ["HmmRiskTopk"] + + +class HmmRiskTopk(TopkDropoutStrategy): + """TopkDropoutStrategy with HMM high-vol pause + drawdown pause + liquidity floor.""" + + def __init__( + self, + *, + hmm_pause_pct: float = 0.70, + drawdown_pause_pct: float = 8.0, + liquidity_floor_adv: float = 0.0, + **kwargs, + ): + super().__init__(**kwargs) + self.hmm_pause_pct = float(hmm_pause_pct) + 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.""" + 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 + ) + if feat is None or len(feat) == 0: + return False + p = feat["$sp_hmm_p_regime1"].dropna() + if len(p) == 0: + return False + return float(p.mean()) >= self.hmm_pause_pct + except Exception: + return False + + def _update_pause(self, equity: float, trade_start) -> None: + 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) + + # ------------------------------------------------------------- decision + 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) + + # 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)) + 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) + + 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 + ) + + held = set(current_temp.get_stock_list()) + held_amt = {c: abs(current_temp.get_stock_amount(c)) for c in held} + + # 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) diff --git a/code/tac-qlib/tac_qlib/contrib/strategy/momentum_gate.py b/code/tac-qlib/tac_qlib/contrib/strategy/momentum_gate.py new file mode 100644 index 0000000..5e1ac61 --- /dev/null +++ b/code/tac-qlib/tac_qlib/contrib/strategy/momentum_gate.py @@ -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: "" + 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)