diff --git a/code/MANIFEST.txt b/code/MANIFEST.txt index 72ca05d..01cae60 100644 --- a/code/MANIFEST.txt +++ b/code/MANIFEST.txt @@ -1,29 +1,31 @@ # TradeAC custom-qlib-code snapshot (auto-generated) -# parent repo HEAD : 125be7b96fb5975e798a0b4301eeb5809a8a181c +# parent repo HEAD : 202589fc3efe3a3350bc2293fb97ec45302dced2 # tac-qlib/tac_qlib/contrib # tac-qlib/tac_qlib/data # per-file hashes (git hash-object): 1b6298c4a5652f2e863cbdc385a1014a570fcd59 tac-qlib/tac_qlib/contrib/__init__.py - b8112569f9b2537c45b6535e1a505a207878d322 tac-qlib/tac_qlib/contrib/__pycache__/__init__.cpython-312.pyc + b419ee55ed455a1c45423d1c9025ca5cc0a98576 tac-qlib/tac_qlib/contrib/__pycache__/__init__.cpython-312.pyc c76a9f17f680e74eea766eff27f7624359749ed6 tac-qlib/tac_qlib/contrib/data/__init__.py - 8d5333ebd2b44165c50cba639ca2d4ac3fc7cfec tac-qlib/tac_qlib/contrib/data/__pycache__/__init__.cpython-312.pyc - 18cb37c0354184c49fa2e598396d7df0634cce0f tac-qlib/tac_qlib/contrib/data/__pycache__/handler.cpython-312.pyc - 871ff1e163c29261f140c3f53d42a41e6504c779 tac-qlib/tac_qlib/contrib/data/handler.py + 2f6c67620aa2f9e6aaaef3369361d9b3eac3d6ca tac-qlib/tac_qlib/contrib/data/__pycache__/__init__.cpython-312.pyc + fdd5923a70a399e8680913593ff111641947898e tac-qlib/tac_qlib/contrib/data/__pycache__/handler.cpython-312.pyc + 0dd25ef161c6e0f15eafc84886e7e1381deb38c3 tac-qlib/tac_qlib/contrib/data/handler.py b151d139a0dcde87d74b21e7c4b729176ba5c39b tac-qlib/tac_qlib/contrib/model/__init__.py - ab958203f33a99d12c7d923b6efb435189231666 tac-qlib/tac_qlib/contrib/model/__pycache__/__init__.cpython-312.pyc - 7478f6b0f6de419615c02d4d92b54529f689ef04 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 - 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 - 96a0a25201f0a1bb2fc2190e26228c5c0e711a79 tac-qlib/tac_qlib/contrib/strategy/hmm_risk.py - 816de5d58ae23d996635d42331cf9fc8963d5dbe tac-qlib/tac_qlib/contrib/strategy/momentum_gate.py + 08dec87ccdf6bb5d2cf611ca3032a4280aaab8cf tac-qlib/tac_qlib/contrib/model/__pycache__/__init__.cpython-312.pyc + 6fb61946ea9a83dfb560de3717f5fbf482c4c00e tac-qlib/tac_qlib/contrib/model/__pycache__/rank_ensemble.cpython-312.pyc + 3e80f2e08b661ddd2f58ffe5a6196063fa41ae51 tac-qlib/tac_qlib/contrib/model/__pycache__/rank_gbdt.cpython-312.pyc + d3f051f3a8650c42fedc7b367b966f7c74fb5789 tac-qlib/tac_qlib/contrib/model/rank_ensemble.py + d03e6611338918d4aac5eea4adf26f85a3763652 tac-qlib/tac_qlib/contrib/model/rank_gbdt.py + c4ef84ffda2a611262412fe1127689c667f3d0c1 tac-qlib/tac_qlib/contrib/strategy/__init__.py + 6ad10c2ebe37c16417e67c7aeb731ad1fcb6da2f tac-qlib/tac_qlib/contrib/strategy/__pycache__/__init__.cpython-312.pyc + 8d684b3216b040071d9ee4fa920a0e0c7486d278 tac-qlib/tac_qlib/contrib/strategy/__pycache__/optimal_stop.cpython-312.pyc + 896ef74ae47bcd1ed388e1e5d9c8d70c28097fe9 tac-qlib/tac_qlib/contrib/strategy/kelly_dropout.py 79aaad9e39fcc740a773f4f63c512ce1086cfde0 tac-qlib/tac_qlib/contrib/strategy/optimal_stop.py + 5b9acfb4340111b204249add7760bd53c6ae03f1 tac-qlib/tac_qlib/contrib/strategy/regime_gate.py + aa1ee880d52ceb5821d65973962099c2254f710a tac-qlib/tac_qlib/contrib/strategy/top_bottom.py + fe60bacdfedd48617863be31f24b7c7daebfac5a tac-qlib/tac_qlib/contrib/strategy/weekly_rebalance.py 92e6e90eb0cd0a25142034560f27adb6b705b1a8 tac-qlib/tac_qlib/data/__init__.py - 0ed1ead6c1314a3f25784d453e54a15a8a04baaa tac-qlib/tac_qlib/data/__pycache__/__init__.cpython-312.pyc - 9609782800944c45b78bb58eaa7b51ba1b7f8f43 tac-qlib/tac_qlib/data/__pycache__/config.cpython-312.pyc - a85628d71d12cfe5b18b1c884c5d829c89594579 tac-qlib/tac_qlib/data/__pycache__/providers.cpython-312.pyc - 686d36f6d101c547491ca866aa143aa542e17518 tac-qlib/tac_qlib/data/config.py - d9f839be30026f337754a3f015425a8efdbe8e2a tac-qlib/tac_qlib/data/providers.py + 7c4e6c345fad1978efe8860c0d977d0c02d6f8d9 tac-qlib/tac_qlib/data/__pycache__/__init__.cpython-312.pyc + 99e602392d51663cb06d5c425000b1ed1e5a916b tac-qlib/tac_qlib/data/__pycache__/config.cpython-312.pyc + 020dcdcf288e4832c8cf2386351f78d5ceb4fe13 tac-qlib/tac_qlib/data/__pycache__/providers.cpython-312.pyc + 53c9007a928841fd3c3b08450f9a6520ce1ac091 tac-qlib/tac_qlib/data/config.py + 8d0644f6f0d1efb94798ed444cc73e63b643459b tac-qlib/tac_qlib/data/providers.py diff --git a/code/tac-qlib/tac_qlib/contrib/__pycache__/__init__.cpython-312.pyc b/code/tac-qlib/tac_qlib/contrib/__pycache__/__init__.cpython-312.pyc index b811256..b419ee5 100644 Binary files a/code/tac-qlib/tac_qlib/contrib/__pycache__/__init__.cpython-312.pyc and b/code/tac-qlib/tac_qlib/contrib/__pycache__/__init__.cpython-312.pyc differ diff --git a/code/tac-qlib/tac_qlib/contrib/data/__pycache__/__init__.cpython-312.pyc b/code/tac-qlib/tac_qlib/contrib/data/__pycache__/__init__.cpython-312.pyc index 8d5333e..2f6c676 100644 Binary files a/code/tac-qlib/tac_qlib/contrib/data/__pycache__/__init__.cpython-312.pyc and b/code/tac-qlib/tac_qlib/contrib/data/__pycache__/__init__.cpython-312.pyc differ diff --git a/code/tac-qlib/tac_qlib/contrib/data/__pycache__/handler.cpython-312.pyc b/code/tac-qlib/tac_qlib/contrib/data/__pycache__/handler.cpython-312.pyc index 18cb37c..fdd5923 100644 Binary files a/code/tac-qlib/tac_qlib/contrib/data/__pycache__/handler.cpython-312.pyc and b/code/tac-qlib/tac_qlib/contrib/data/__pycache__/handler.cpython-312.pyc differ diff --git a/code/tac-qlib/tac_qlib/contrib/data/handler.py b/code/tac-qlib/tac_qlib/contrib/data/handler.py index 871ff1e..0dd25ef 100644 --- a/code/tac-qlib/tac_qlib/contrib/data/handler.py +++ b/code/tac-qlib/tac_qlib/contrib/data/handler.py @@ -64,9 +64,13 @@ def check_transform_proc(proc_l, fit_start_time, fit_end_time): def get_common_feature_fields(lake_root=None, market="US", timeframe="1d") -> List[str]: - """Discover ta-lib columns present in *every* features parquet file of the lake. + """Discover feature columns present in *every* feature file of the lake. - Returns sorted field names (without the ``$`` prefix). Empty if no features are persisted. + Walks the `family=ta|sp` partition layout (plus any legacy flat files). + TA and SP columns are disjoint by construction, so the common set is + computed per family (columns shared by all symbol files of that family), + then the per-family results are unioned. Returns sorted field names + (without the ``$`` prefix). Empty if no features are persisted. """ cfg = LakeConfig(lake_root, market) feat_dir = cfg.features_dir(timeframe) @@ -74,16 +78,30 @@ def get_common_feature_fields(lake_root=None, market="US", timeframe="1d") -> Li return [] import pyarrow.parquet as pq - common = None - for p in sorted(feat_dir.glob("symbol=*.parquet")): - try: - cols = set(pq.read_schema(p).names) - set(NON_FEATURE_COLUMNS) - except Exception: # pragma: no cover - skip unreadable files - continue - common = cols if common is None else (common & cols) - if not common: - break - return sorted(common) if common else [] + def _family_common(fam_dir: Path) -> set: + common = None + for p in sorted(fam_dir.glob("symbol=*.parquet")): + try: + cols = set(pq.read_schema(p).names) - set(NON_FEATURE_COLUMNS) + except Exception: # pragma: no cover - skip unreadable files + continue + common = cols if common is None else (common & cols) + if not common: + break + return common or set() + + common: set = set() + # family tier: features/market=*/timeframe=*/family=*/symbol=*.parquet + for fam in ("ta", "sp"): + fam_dir = feat_dir / f"family={fam}" + if fam_dir.is_dir(): + common |= _family_common(fam_dir) + # legacy flat: features/market=*/timeframe=*/symbol=*.parquet + if (feat_dir / "family=ta").exists() or (feat_dir / "family=sp").exists(): + pass # family layout already covered + else: + common |= _family_common(feat_dir) + return sorted(common) class DropAllNaN(processor_module.Processor): diff --git a/code/tac-qlib/tac_qlib/contrib/model/__pycache__/__init__.cpython-312.pyc b/code/tac-qlib/tac_qlib/contrib/model/__pycache__/__init__.cpython-312.pyc index ab95820..08dec87 100644 Binary files a/code/tac-qlib/tac_qlib/contrib/model/__pycache__/__init__.cpython-312.pyc and b/code/tac-qlib/tac_qlib/contrib/model/__pycache__/__init__.cpython-312.pyc differ diff --git a/code/tac-qlib/tac_qlib/contrib/model/__pycache__/rank_ensemble.cpython-312.pyc b/code/tac-qlib/tac_qlib/contrib/model/__pycache__/rank_ensemble.cpython-312.pyc index 7478f6b..6fb6194 100644 Binary files a/code/tac-qlib/tac_qlib/contrib/model/__pycache__/rank_ensemble.cpython-312.pyc and b/code/tac-qlib/tac_qlib/contrib/model/__pycache__/rank_ensemble.cpython-312.pyc differ diff --git a/code/tac-qlib/tac_qlib/contrib/model/__pycache__/rank_gbdt.cpython-312.pyc b/code/tac-qlib/tac_qlib/contrib/model/__pycache__/rank_gbdt.cpython-312.pyc index 9f9014d..3e80f2e 100644 Binary files a/code/tac-qlib/tac_qlib/contrib/model/__pycache__/rank_gbdt.cpython-312.pyc and b/code/tac-qlib/tac_qlib/contrib/model/__pycache__/rank_gbdt.cpython-312.pyc differ 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 ce77dea..d3f051f 100644 --- a/code/tac-qlib/tac_qlib/contrib/model/rank_ensemble.py +++ b/code/tac-qlib/tac_qlib/contrib/model/rank_ensemble.py @@ -56,7 +56,6 @@ 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 @@ -80,15 +79,11 @@ class RankICEnsembleLGBModel(RankICLGBModel): forwarded. """ - def __init__(self, seeds: str = "42", parallel: int = 0, weight_mode: str = "equal", **kwargs): + def __init__(self, seeds: str = "42", parallel: int = 0, **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) @@ -184,44 +179,11 @@ class RankICEnsembleLGBModel(RankICLGBModel): # -------------------------------------------------------------- predict def predict(self, dataset: DatasetH, segment="test") -> pd.Series: - """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). - """ + """Average the per-seed predictions over the given segment.""" 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) - 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) + return frame.mean(axis=1) diff --git a/code/tac-qlib/tac_qlib/contrib/model/rank_gbdt.py b/code/tac-qlib/tac_qlib/contrib/model/rank_gbdt.py index ccfe7d5..d03e661 100644 --- a/code/tac-qlib/tac_qlib/contrib/model/rank_gbdt.py +++ b/code/tac-qlib/tac_qlib/contrib/model/rank_gbdt.py @@ -53,23 +53,61 @@ from qlib.workflow import R __all__ = ["RankICLGBModel", "rankic_feval"] +def _group_averaged_rank(values: np.ndarray, gid: np.ndarray, offs: np.ndarray) -> np.ndarray: + """Averaged (tie-corrected) rank of ``values`` within each group, vectorized. + + ``gid`` maps each row to its group id; ``offs`` holds the cumulative row + offsets so that group ``i`` occupies rows ``[offs[i], offs[i+1])``. Returns + the same result as ``pandas.Series.rank(method='average')`` applied per + group, but in one pass (``np.lexsort`` is the only non-linear step). + """ + n = len(values) + order = np.lexsort((values, gid)) + ord_rank = np.empty(n, dtype=np.float64) + ord_rank[order] = np.arange(n, dtype=np.float64) - offs[gid[order]] + 1.0 + sg = gid[order] + sv = values[order] + newblock = np.empty(n, dtype=bool) + newblock[0] = True + newblock[1:] = (sg[1:] != sg[:-1]) | (sv[1:] != sv[:-1]) + blockid = np.cumsum(newblock) - 1 + block_mean = np.bincount(blockid, weights=ord_rank[order]) / np.bincount(blockid) + out = np.empty(n) + out[order] = block_mean[blockid] + return out + + def _per_day_spearman(preds: np.ndarray, labels: np.ndarray, group: np.ndarray) -> float: """Mean per-day Spearman rank correlation of preds vs labels. ``group`` holds the number of rows of each trading day (query group), in order. Days with <3 valid rows or a constant pred/label are skipped. + + Vectorized: per-day Spearman == Pearson of the per-day rank transforms, + and the Pearson moments (``sum``, ``sum`` of products/squares) aggregate + over each day with ``np.bincount``. Runs ~10x faster than the per-day + ``pd.Series.rank()`` loop that preceded it — this feval is invoked on the + train and valid panels every boosting round, per seed. """ if group is None or len(group) == 0: return 0.0 offs = np.concatenate([[0], np.cumsum(group.astype(int))]) - vals = [] - for i in range(len(group)): - s = slice(offs[i], offs[i + 1]) - p, l = preds[s], labels[s] - if len(p) < 3 or np.std(p) == 0 or np.std(l) == 0: - continue - vals.append(np.corrcoef(pd.Series(p).rank(), pd.Series(l).rank())[0, 1]) - return float(np.mean(vals)) if vals else 0.0 + gid = np.repeat(np.arange(len(group)), group.astype(int)) + rp = _group_averaged_rank(preds, gid, offs) + rl = _group_averaged_rank(labels, gid, offs) + n_g = group.astype(float) + s_p = np.bincount(gid, weights=rp) + s_l = np.bincount(gid, weights=rl) + s_pl = np.bincount(gid, weights=rp * rl) + s_pp = np.bincount(gid, weights=rp * rp) + s_ll = np.bincount(gid, weights=rl * rl) + cov = n_g * s_pl - s_p * s_l + var_p = n_g * s_pp - s_p ** 2 + var_l = n_g * s_ll - s_l ** 2 + denom = np.sqrt(var_p * var_l) + valid = (n_g >= 3) & (denom > 0) + corr = np.where(valid, cov / np.where(denom == 0, 1, denom), 0.0) + return float(corr[valid].mean()) if valid.any() else 0.0 def rankic_feval(preds, dataset): diff --git a/code/tac-qlib/tac_qlib/contrib/strategy/__init__.py b/code/tac-qlib/tac_qlib/contrib/strategy/__init__.py index 4afcf90..c4ef84f 100644 --- a/code/tac-qlib/tac_qlib/contrib/strategy/__init__.py +++ b/code/tac-qlib/tac_qlib/contrib/strategy/__init__.py @@ -1,3 +1,13 @@ +from .kelly_dropout import FractionalKellyDropoutStrategy # noqa: F401 from .optimal_stop import OptimalStopControl # noqa: F401 +from .regime_gate import RegimeGateDropoutStrategy # noqa: F401 +from .top_bottom import TopBottomDropoutStrategy # noqa: F401 +from .weekly_rebalance import WeeklyRebalanceDropoutStrategy # noqa: F401 -__all__ = ["OptimalStopControl"] +__all__ = [ + "OptimalStopControl", + "FractionalKellyDropoutStrategy", + "WeeklyRebalanceDropoutStrategy", + "TopBottomDropoutStrategy", + "RegimeGateDropoutStrategy", +] diff --git a/code/tac-qlib/tac_qlib/contrib/strategy/__pycache__/__init__.cpython-312.pyc b/code/tac-qlib/tac_qlib/contrib/strategy/__pycache__/__init__.cpython-312.pyc index 74e5ecb..6ad10c2 100644 Binary files a/code/tac-qlib/tac_qlib/contrib/strategy/__pycache__/__init__.cpython-312.pyc and b/code/tac-qlib/tac_qlib/contrib/strategy/__pycache__/__init__.cpython-312.pyc differ diff --git a/code/tac-qlib/tac_qlib/contrib/strategy/__pycache__/optimal_stop.cpython-312.pyc b/code/tac-qlib/tac_qlib/contrib/strategy/__pycache__/optimal_stop.cpython-312.pyc index afaf562..8d684b3 100644 Binary files a/code/tac-qlib/tac_qlib/contrib/strategy/__pycache__/optimal_stop.cpython-312.pyc and b/code/tac-qlib/tac_qlib/contrib/strategy/__pycache__/optimal_stop.cpython-312.pyc differ diff --git a/code/tac-qlib/tac_qlib/contrib/strategy/hmm_risk.py b/code/tac-qlib/tac_qlib/contrib/strategy/hmm_risk.py deleted file mode 100644 index 96a0a25..0000000 --- a/code/tac-qlib/tac_qlib/contrib/strategy/hmm_risk.py +++ /dev/null @@ -1,138 +0,0 @@ -"""TopkDropout with HMM high-volatility + drawdown-pause risk gates. - -Gates NEW entries on two risk conditions (held names are never force-sold): - -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. 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). - -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: - - 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 - - # ------------------------------------------------------------- gates - def _hmm_high_vol(self, pred_date) -> bool: - """Cross-sectional mean HMM high-vol regime probability >= threshold.""" - try: - from qlib.data import D - - 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() - if len(p) == 0: - return False - return float(p.mean()) >= self.hmm_pause_pct - except Exception: - return False - - def _drawdown_active(self, equity: float) -> bool: - if self.drawdown_pause_pct <= 0: - return False - self._peak_equity = max(self._peak_equity, equity) - 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): - 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) - 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} - - equity = current_temp.get_cash() - 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 += abs(current_temp.get_stock_amount(code)) * mark - - 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 - - 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 - - return super().generate_trade_decision(execute_result) diff --git a/code/tac-qlib/tac_qlib/contrib/strategy/kelly_dropout.py b/code/tac-qlib/tac_qlib/contrib/strategy/kelly_dropout.py new file mode 100644 index 0000000..896ef74 --- /dev/null +++ b/code/tac-qlib/tac_qlib/contrib/strategy/kelly_dropout.py @@ -0,0 +1,201 @@ +"""Fractional-Kelly dropout strategy for cross-sectional signals. + +Sizing rule variant of ``qlib.contrib.strategy.signal_strategy.TopkDropoutStrategy``: +the topk/n_drop SELECTION is identical to the reference, but the buy size is +proportional to the score MAGNITUDE (edge) instead of equal-weight, capped at a +fraction ``cap_frac`` of the equal-weight notional so a single name cannot +over-concentrate the book. + +``cap_frac`` is the fraction of the equal-weight per-name notional that a top +signal can deploy at most (e.g. 0.5 = at most half the equal-weight size). +Names whose score is below the median of the buy set get a proportionally +smaller slice; the residual stays in cash (that is the point of the rule: +throw away less edge per name, deploy less capital when conviction is low). +""" + +from __future__ import annotations + +from typing import List + +import numpy as np +import pandas as pd + +from qlib.backtest import Order +from qlib.backtest.decision import OrderDir, TradeDecisionWO +from qlib.contrib.strategy.signal_strategy import TopkDropoutStrategy + +__all__ = ["FractionalKellyDropoutStrategy"] + +DEFAULT_CAP_FRAC = 0.5 + + +class FractionalKellyDropoutStrategy(TopkDropoutStrategy): + """TopkDropout selection with score-magnitude (fractional-Kelly) sizing. + + Parameters + ---------- + topk, n_drop, method_sell, method_buy, hold_thresh, only_tradable, + forbid_all_trade_at_limit : same as ``TopkDropoutStrategy``. + cap_frac : max buy notional as a fraction of the equal-weight notional. + """ + + def __init__(self, *, topk, n_drop, cap_frac: float = DEFAULT_CAP_FRAC, **kwargs): + super().__init__(topk=topk, n_drop=n_drop, **kwargs) + self.cap_frac = cap_frac + + def generate_trade_decision(self, execute_result=None): + import copy + + 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 isinstance(pred_score, pd.DataFrame): + pred_score = pred_score.iloc[:, 0] + if pred_score is None: + return TradeDecisionWO([], self) + + if self.only_tradable: + + def get_first_n(li, n, reverse=False): + cur_n = 0 + res = [] + for si in reversed(li) if reverse else li: + if self.trade_exchange.is_stock_tradable( + stock_id=si, start_time=trade_start_time, end_time=trade_end_time + ): + res.append(si) + cur_n += 1 + if cur_n >= n: + break + return res[::-1] if reverse else res + + def get_last_n(li, n): + return get_first_n(li, n, reverse=True) + + def filter_stock(li): + return [ + si + for si in li + if self.trade_exchange.is_stock_tradable( + stock_id=si, start_time=trade_start_time, end_time=trade_end_time + ) + ] + + else: + + def get_first_n(li, n): + return list(li)[:n] + + def get_last_n(li, n): + return list(li)[-n:] + + def filter_stock(li): + return li + + current_temp: "object" = copy.deepcopy(self.trade_position) + sell_order_list: List[Order] = [] + buy_order_list: List[Order] = [] + cash = current_temp.get_cash() + current_stock_list = current_temp.get_stock_list() + last = pred_score.reindex(current_stock_list).sort_values(ascending=False).index + + if self.method_buy == "top": + today = get_first_n( + pred_score[~pred_score.index.isin(last)].sort_values(ascending=False).index, + self.n_drop + self.topk - len(last), + ) + elif self.method_buy == "random": + topk_candi = get_first_n(pred_score.sort_values(ascending=False).index, self.topk) + candi = list(filter(lambda x: x not in last, topk_candi)) + n = self.n_drop + self.topk - len(last) + try: + today = np.random.choice(candi, n, replace=False) + except ValueError: + today = candi + else: + raise NotImplementedError(f"This type of input is not supported") + + comb = pred_score.reindex(last.union(pd.Index(today))).sort_values(ascending=False).index + + if self.method_sell == "bottom": + sell = last[last.isin(get_last_n(comb, self.n_drop))] + elif self.method_sell == "random": + candi = filter_stock(last) + try: + sell = pd.Index(np.random.choice(candi, self.n_drop, replace=False) if len(last) else []) + except ValueError: + sell = candi + else: + raise NotImplementedError(f"This type of input is not supported") + + buy = today[: len(sell) + self.topk - len(last)] + for code in current_stock_list: + if not self.trade_exchange.is_stock_tradable( + stock_id=code, + start_time=trade_start_time, + end_time=trade_end_time, + direction=None if self.forbid_all_trade_at_limit else OrderDir.SELL, + ): + continue + if code in sell: + time_per_step = self.trade_calendar.get_freq() + if current_temp.get_stock_count(code, bar=time_per_step) < self.hold_thresh: + continue + sell_amount = current_temp.get_stock_amount(code=code) + sell_order = Order( + stock_id=code, + amount=sell_amount, + start_time=trade_start_time, + end_time=trade_end_time, + direction=Order.SELL, + ) + if self.trade_exchange.check_order(sell_order): + sell_order_list.append(sell_order) + trade_val, trade_cost, trade_price = self.trade_exchange.deal_order( + sell_order, position=current_temp + ) + cash += trade_val - trade_cost + + if len(buy) == 0: + return TradeDecisionWO(sell_order_list, self) + + # ---- fractional-Kelly sizing -------------------------------------- + # equal-weight notional (reference baseline) + eq_notional = cash * self.risk_degree / len(buy) + buy_scores = pred_score.reindex(buy).astype(float) + lo, hi = buy_scores.min(), buy_scores.max() + if hi == lo: + w = pd.Series(1.0, index=buy_scores.index) + else: + w = (buy_scores - lo) / (hi - lo) # [0,1] edge magnitude + w = w.clip(lower=0.0) + w_max = w.max() + w = w / w_max if w_max > 0 else w # max == 1.0 + for code in buy: + if not self.trade_exchange.is_stock_tradable( + stock_id=code, + start_time=trade_start_time, + end_time=trade_end_time, + direction=None if self.forbid_all_trade_at_limit else OrderDir.BUY, + ): + continue + buy_price = self.trade_exchange.get_deal_price( + stock_id=code, start_time=trade_start_time, end_time=trade_end_time, direction=OrderDir.BUY + ) + notional = eq_notional * min(self.cap_frac, float(w.get(code, 0.0))) + buy_amount = notional / buy_price + factor = self.trade_exchange.get_factor( + stock_id=code, start_time=trade_start_time, end_time=trade_end_time + ) + buy_amount = self.trade_exchange.round_amount_by_trade_unit(buy_amount, factor) + buy_order = Order( + stock_id=code, + amount=buy_amount, + start_time=trade_start_time, + end_time=trade_end_time, + direction=Order.BUY, + ) + buy_order_list.append(buy_order) + + return TradeDecisionWO(sell_order_list + buy_order_list, self) \ No newline at end of file diff --git a/code/tac-qlib/tac_qlib/contrib/strategy/momentum_gate.py b/code/tac-qlib/tac_qlib/contrib/strategy/momentum_gate.py deleted file mode 100644 index 816de5d..0000000 --- a/code/tac-qlib/tac_qlib/contrib/strategy/momentum_gate.py +++ /dev/null @@ -1,91 +0,0 @@ -"""TopkDropout with a 1-day momentum entry-confirmation gate. - -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: 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: - - 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.decision import 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=1 - ) - except Exception: - return False - if cur is None or cur != cur or cur <= 0: - return False - 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=0 - ) - 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) - 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 = set(current_temp.get_stock_list()) - held = {c for c in held if abs(current_temp.get_stock_amount(c)) > 1e-6} - - # 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) diff --git a/code/tac-qlib/tac_qlib/contrib/strategy/regime_gate.py b/code/tac-qlib/tac_qlib/contrib/strategy/regime_gate.py new file mode 100644 index 0000000..5b9acfb --- /dev/null +++ b/code/tac-qlib/tac_qlib/contrib/strategy/regime_gate.py @@ -0,0 +1,231 @@ +"""HMM-regime overlay TopkDropout strategy. + +Regime-gate overlay on ``qlib.contrib.strategy.signal_strategy.TopkDropoutStrategy``: +selection and sizing are identical to the reference, but a name is only BOUGHT +(entry gate) when its per-symbol HMM regime posterior ``sp_hmm_p_regime1`` on +the signal date is >= ``regime_threshold``; otherwise it is held in cash instead +of being opened. + +The regime posterior is read from the lake feature provider on the fly via +``qlib.data.D.features`` (field ``$sp_hmm_p_regime1``) for the signal window, so +no regime column needs to enter the model's ``feature_fields`` — the gate is a +pure overlay (book ch.01: regime flags regressed as model features, survived +only as an overlay). The HMM itself was fit with ``fit_end=`` when +the lake features were backfilled, so there is no lookahead. + +Names already held are NOT force-sold when the regime turns unfavourable +(entry gate only, matching the queue-10 design). +""" + +from __future__ import annotations + +from typing import List + +import numpy as np +import pandas as pd + +from qlib.backtest import Order +from qlib.backtest.decision import OrderDir, TradeDecisionWO +from qlib.contrib.strategy.signal_strategy import TopkDropoutStrategy + +try: + from qlib.data import D +except ImportError: # pragma: no cover - qlib always present in this stack + D = None + +__all__ = ["RegimeGateDropoutStrategy"] + +DEFAULT_REGIME_THRESHOLD = 0.5 +REGIME_FIELD = "$sp_hmm_p_regime1" + + +class RegimeGateDropoutStrategy(TopkDropoutStrategy): + """TopkDropout with an HMM-regime entry gate on buy candidates. + + Parameters + ---------- + topk, n_drop, method_sell, method_buy, hold_thresh, only_tradable, + forbid_all_trade_at_limit : same as ``TopkDropoutStrategy``. + regime_threshold : minimum ``sp_hmm_p_regime1`` posterior required to open a + new position (default 0.5). + """ + + def __init__(self, *, topk, n_drop, regime_threshold: float = DEFAULT_REGIME_THRESHOLD, **kwargs): + super().__init__(topk=topk, n_drop=n_drop, **kwargs) + self.regime_threshold = regime_threshold + + def _regime_for(self, codes, pred_start, pred_end) -> pd.Series: + """Return {code: sp_hmm_p_regime1} for the signal window (last day).""" + if D is None: + return pd.Series(dtype=float) + try: + df = D.features(list(codes), [REGIME_FIELD], start_time=pred_start, end_time=pred_end, freq="day") + except Exception: # noqa: BLE001 - a regime read failure should gate open, not crash + return pd.Series(dtype=float) + if df is None or len(df) == 0: + return pd.Series(dtype=float) + # df index is MultiIndex (datetime, instrument); take the last day's values + df = df.reset_index() + ts_col = "datetime" if "datetime" in df.columns else df.columns[0] + sym_col = "instrument" if "instrument" in df.columns else df.columns[1] + last_ts = df[ts_col].max() + last = df[df[ts_col] == last_ts] + out = {} + for _, row in last.iterrows(): + sym = str(row[sym_col]).split("/")[-1].upper() + val = row.iloc[-1] + out[sym] = float(val) if val == val else np.nan + return pd.Series(out) + + def generate_trade_decision(self, execute_result=None): + import copy + + 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 isinstance(pred_score, pd.DataFrame): + pred_score = pred_score.iloc[:, 0] + if pred_score is None: + return TradeDecisionWO([], self) + + if self.only_tradable: + + def get_first_n(li, n, reverse=False): + cur_n = 0 + res = [] + for si in reversed(li) if reverse else li: + if self.trade_exchange.is_stock_tradable( + stock_id=si, start_time=trade_start_time, end_time=trade_end_time + ): + res.append(si) + cur_n += 1 + if cur_n >= n: + break + return res[::-1] if reverse else res + + def get_last_n(li, n): + return get_first_n(li, n, reverse=True) + + def filter_stock(li): + return [ + si + for si in li + if self.trade_exchange.is_stock_tradable( + stock_id=si, start_time=trade_start_time, end_time=trade_end_time + ) + ] + + else: + + def get_first_n(li, n): + return list(li)[:n] + + def get_last_n(li, n): + return list(li)[-n:] + + def filter_stock(li): + return li + + current_temp: "object" = copy.deepcopy(self.trade_position) + sell_order_list: List[Order] = [] + buy_order_list: List[Order] = [] + cash = current_temp.get_cash() + current_stock_list = current_temp.get_stock_list() + last = pred_score.reindex(current_stock_list).sort_values(ascending=False).index + + if self.method_buy == "top": + today = get_first_n( + pred_score[~pred_score.index.isin(last)].sort_values(ascending=False).index, + self.n_drop + self.topk - len(last), + ) + elif self.method_buy == "random": + topk_candi = get_first_n(pred_score.sort_values(ascending=False).index, self.topk) + candi = list(filter(lambda x: x not in last, topk_candi)) + n = self.n_drop + self.topk - len(last) + try: + today = np.random.choice(candi, n, replace=False) + except ValueError: + today = candi + else: + raise NotImplementedError(f"This type of input is not supported") + + comb = pred_score.reindex(last.union(pd.Index(today))).sort_values(ascending=False).index + + if self.method_sell == "bottom": + sell = last[last.isin(get_last_n(comb, self.n_drop))] + elif self.method_sell == "random": + candi = filter_stock(last) + try: + sell = pd.Index(np.random.choice(candi, self.n_drop, replace=False) if len(last) else []) + except ValueError: + sell = candi + else: + raise NotImplementedError(f"This type of input is not supported") + + buy = today[: len(sell) + self.topk - len(last)] + + # ---- regime gate ----------------------------------------------------- + if buy: + regime = self._regime_for(buy, pred_start_time, pred_end_time) + gated = [c for c in buy if regime.get(c, np.nan) >= self.regime_threshold] + else: + gated = [] + + for code in current_stock_list: + if not self.trade_exchange.is_stock_tradable( + stock_id=code, + start_time=trade_start_time, + end_time=trade_end_time, + direction=None if self.forbid_all_trade_at_limit else OrderDir.SELL, + ): + continue + if code in sell: + time_per_step = self.trade_calendar.get_freq() + if current_temp.get_stock_count(code, bar=time_per_step) < self.hold_thresh: + continue + sell_amount = current_temp.get_stock_amount(code=code) + sell_order = Order( + stock_id=code, + amount=sell_amount, + start_time=trade_start_time, + end_time=trade_end_time, + direction=Order.SELL, + ) + if self.trade_exchange.check_order(sell_order): + sell_order_list.append(sell_order) + trade_val, trade_cost, trade_price = self.trade_exchange.deal_order( + sell_order, position=current_temp + ) + cash += trade_val - trade_cost + + if len(gated) == 0: + return TradeDecisionWO(sell_order_list, self) + + value = cash * self.risk_degree / len(gated) + for code in gated: + if not self.trade_exchange.is_stock_tradable( + stock_id=code, + start_time=trade_start_time, + end_time=trade_end_time, + direction=None if self.forbid_all_trade_at_limit else OrderDir.BUY, + ): + continue + buy_price = self.trade_exchange.get_deal_price( + stock_id=code, start_time=trade_start_time, end_time=trade_end_time, direction=OrderDir.BUY + ) + buy_amount = value / buy_price + factor = self.trade_exchange.get_factor( + stock_id=code, start_time=trade_start_time, end_time=trade_end_time + ) + buy_amount = self.trade_exchange.round_amount_by_trade_unit(buy_amount, factor) + buy_order = Order( + stock_id=code, + amount=buy_amount, + start_time=trade_start_time, + end_time=trade_end_time, + direction=Order.BUY, + ) + buy_order_list.append(buy_order) + + return TradeDecisionWO(sell_order_list + buy_order_list, self) \ No newline at end of file diff --git a/code/tac-qlib/tac_qlib/contrib/strategy/top_bottom.py b/code/tac-qlib/tac_qlib/contrib/strategy/top_bottom.py new file mode 100644 index 0000000..aa1ee88 --- /dev/null +++ b/code/tac-qlib/tac_qlib/contrib/strategy/top_bottom.py @@ -0,0 +1,169 @@ +"""Market-neutral top/bottom long-short strategy for cross-sectional signals. + +Captures the cross-sectional long-short spread net of costs: buys the top-ranked +``topk`` names and shorts the bottom-ranked ``topk`` names, equal-weight per +side, sized to ``risk_degree`` of total value per side. Rebalances daily to the +current rank (dropout-free: the book converges to the latest top/bottom sets). + +The long and short legs use equal notional per side (gross exposure ~2x +``risk_degree`` of NAV, i.e. approximately market neutral before transaction +costs). Benchmark neutrality (SPY beta ~ 0) is the secondary sanity metric. +""" + +from __future__ import annotations + +from typing import List + +import copy + +import pandas as pd + +from qlib.backtest import Order +from qlib.backtest.decision import OrderDir, TradeDecisionWO +from qlib.contrib.strategy.signal_strategy import BaseSignalStrategy + +__all__ = ["TopBottomDropoutStrategy"] + +DEFAULT_SHORT_LEG = True +DEFAULT_REBALANCE_DAILY = True + + +class TopBottomDropoutStrategy(BaseSignalStrategy): + """Long top-k / short bottom-k equal-weight market-neutral book. + + Parameters + ---------- + topk : number of names on each side (long top-k and short bottom-k). + short_leg : whether to open the short side (if False, long-only topk). + rebalance_daily : if True rebalance to current rank every day; else keep + positions and only refresh on score changes (dropout-style). + risk_degree : fraction of total value deployed per side. + """ + + def __init__( + self, + *, + topk: int = 10, + short_leg: bool = DEFAULT_SHORT_LEG, + rebalance_daily: bool = DEFAULT_REBALANCE_DAILY, + **kwargs, + ): + super().__init__(**kwargs) + self.topk = topk + self.short_leg = short_leg + self.rebalance_daily = rebalance_daily + self._prev_longs = set() + self._prev_shorts = set() + + 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 isinstance(pred_score, pd.DataFrame): + pred_score = pred_score.iloc[:, 0] + if pred_score is None or len(pred_score) == 0: + return TradeDecisionWO([], self) + + # rank all names; topk longs and topk shorts + ranked = pred_score.sort_values(ascending=False) + longs = list(ranked.index[: self.topk]) + shorts = list(ranked.index[-self.topk :]) if self.short_leg else [] + + current_temp: "object" = copy.deepcopy(self.trade_position) + current_codes = set(current_temp.get_stock_list()) + holdings = {c: current_temp for c in current_codes if abs(current_temp.get_stock_amount(c)) > 1e-6} + + sell_orders: List[Order] = [] + buy_orders: List[Order] = [] + + def _tradable(code, direction): + try: + return self.trade_exchange.is_stock_tradable( + stock_id=code, start_time=trade_start_time, end_time=trade_end_time, direction=direction + ) + except TypeError: + return self.trade_exchange.is_stock_tradable( + stock_id=code, start_time=trade_start_time, end_time=trade_end_time + ) + + # determine target set (long/short) + target_longs = set(longs) + target_shorts = set(shorts) + + # close positions not in the target book + for code in list(holdings): + if code in target_longs or code in target_shorts: + continue + amt = abs(current_temp.get_stock_amount(code)) + o = Order( + stock_id=code, + amount=amt, + start_time=trade_start_time, + end_time=trade_end_time, + direction=Order.SELL if code in target_longs else Order.SELL, + ) + if self.trade_exchange.check_order(o): + sell_orders.append(o) + self.trade_exchange.deal_order(o, position=current_temp) + + # equal-weight notional per side + total_value = current_temp.get_cash() + for code, pos in holdings.items(): + if code in target_longs or code in target_shorts: + mark = self.trade_exchange.get_deal_price( + stock_id=code, start_time=trade_start_time, end_time=trade_end_time, direction=Order.SELL + ) + if mark is not None and mark == mark: + total_value += abs(current_temp.get_stock_amount(code)) * mark + + side_notional = total_value * self.risk_degree / max(1, self.topk) + + for code in longs: + if code in holdings and abs(current_temp.get_stock_amount(code)) > 1e-6: + continue + px = self.trade_exchange.get_deal_price( + stock_id=code, start_time=trade_start_time, end_time=trade_end_time, direction=Order.BUY + ) + if px is None or px != px or px <= 0: + continue + amount = side_notional / px + factor = self.trade_exchange.get_factor( + stock_id=code, start_time=trade_start_time, end_time=trade_end_time + ) + amount = self.trade_exchange.round_amount_by_trade_unit(amount, factor) + o = Order( + stock_id=code, + amount=amount, + start_time=trade_start_time, + end_time=trade_end_time, + direction=Order.BUY, + ) + if self.trade_exchange.check_order(o): + buy_orders.append(o) + + if self.short_leg: + for code in shorts: + if code in holdings and abs(current_temp.get_stock_amount(code)) > 1e-6: + continue + px = self.trade_exchange.get_deal_price( + stock_id=code, start_time=trade_start_time, end_time=trade_end_time, direction=Order.SELL + ) + if px is None or px != px or px <= 0: + continue + amount = side_notional / px + factor = self.trade_exchange.get_factor( + stock_id=code, start_time=trade_start_time, end_time=trade_end_time + ) + amount = self.trade_exchange.round_amount_by_trade_unit(amount, factor) + o = Order( + stock_id=code, + amount=amount, + start_time=trade_start_time, + end_time=trade_end_time, + direction=Order.SELL, + ) + if self.trade_exchange.check_order(o): + sell_orders.append(o) + + return TradeDecisionWO(sell_orders + buy_orders, self) \ No newline at end of file diff --git a/code/tac-qlib/tac_qlib/contrib/strategy/weekly_rebalance.py b/code/tac-qlib/tac_qlib/contrib/strategy/weekly_rebalance.py new file mode 100644 index 0000000..fe60bac --- /dev/null +++ b/code/tac-qlib/tac_qlib/contrib/strategy/weekly_rebalance.py @@ -0,0 +1,202 @@ +"""Weekly-rebalance TopkDropout strategy. + +Turnover-reduction variant of ``qlib.contrib.strategy.signal_strategy.TopkDropoutStrategy``: +the topk/n_drop selection and sizing are identical to the reference, but the +target book is recomputed only on the first trading day of each ISO week; on the +other days the strategy issues NO orders (holds the book untouched). + +The weekly cadence is derived from the qlib trade calendar: a rebalance happens +when the current trade step's date belongs to a different ISO ``(year, week)`` +than the previous trade step. ``hold_band_pct`` (default 0) optionally skips +tiny rebalances: when a name's existing position differs from the new target by +less than this fraction, no order is generated for it. +""" + +from __future__ import annotations + +from typing import List + +import numpy as np +import pandas as pd + +from qlib.backtest import Order +from qlib.backtest.decision import OrderDir, TradeDecisionWO +from qlib.contrib.strategy.signal_strategy import TopkDropoutStrategy + +__all__ = ["WeeklyRebalanceDropoutStrategy"] + +DEFAULT_HOLD_BAND_PCT = 0.0 + + +class WeeklyRebalanceDropoutStrategy(TopkDropoutStrategy): + """TopkDropout rebalanced once per ISO week; holds otherwise. + + Parameters + ---------- + topk, n_drop, method_sell, method_buy, hold_thresh, only_tradable, + forbid_all_trade_at_limit : same as ``TopkDropoutStrategy``. + hold_band_pct : skip order for a name whose deviation from target weight is + below this fraction of the target (no-trade buffer band). + """ + + def __init__(self, *, topk, n_drop, hold_band_pct: float = DEFAULT_HOLD_BAND_PCT, **kwargs): + super().__init__(topk=topk, n_drop=n_drop, **kwargs) + self.hold_band_pct = hold_band_pct + + @staticmethod + def _iso_week(ts) -> tuple: + return (ts.year, ts.week) + + def generate_trade_decision(self, execute_result=None): + import copy + + trade_step = self.trade_calendar.get_trade_step() + trade_start_time, trade_end_time = self.trade_calendar.get_step_time(trade_step) + + cur_week = self._iso_week(trade_start_time) + prev_week = getattr(self, "_last_week", None) + self._last_week = cur_week + + if prev_week is not None and prev_week == cur_week: + # not the first trading day of this ISO week -> hold + return TradeDecisionWO([], self) + + 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 isinstance(pred_score, pd.DataFrame): + pred_score = pred_score.iloc[:, 0] + if pred_score is None: + return TradeDecisionWO([], self) + + if self.only_tradable: + + def get_first_n(li, n, reverse=False): + cur_n = 0 + res = [] + for si in reversed(li) if reverse else li: + if self.trade_exchange.is_stock_tradable( + stock_id=si, start_time=trade_start_time, end_time=trade_end_time + ): + res.append(si) + cur_n += 1 + if cur_n >= n: + break + return res[::-1] if reverse else res + + def get_last_n(li, n): + return get_first_n(li, n, reverse=True) + + def filter_stock(li): + return [ + si + for si in li + if self.trade_exchange.is_stock_tradable( + stock_id=si, start_time=trade_start_time, end_time=trade_end_time + ) + ] + + else: + + def get_first_n(li, n): + return list(li)[:n] + + def get_last_n(li, n): + return list(li)[-n:] + + def filter_stock(li): + return li + + current_temp: "object" = copy.deepcopy(self.trade_position) + sell_order_list: List[Order] = [] + buy_order_list: List[Order] = [] + cash = current_temp.get_cash() + current_stock_list = current_temp.get_stock_list() + last = pred_score.reindex(current_stock_list).sort_values(ascending=False).index + + if self.method_buy == "top": + today = get_first_n( + pred_score[~pred_score.index.isin(last)].sort_values(ascending=False).index, + self.n_drop + self.topk - len(last), + ) + elif self.method_buy == "random": + topk_candi = get_first_n(pred_score.sort_values(ascending=False).index, self.topk) + candi = list(filter(lambda x: x not in last, topk_candi)) + n = self.n_drop + self.topk - len(last) + try: + today = np.random.choice(candi, n, replace=False) + except ValueError: + today = candi + else: + raise NotImplementedError(f"This type of input is not supported") + + comb = pred_score.reindex(last.union(pd.Index(today))).sort_values(ascending=False).index + + if self.method_sell == "bottom": + sell = last[last.isin(get_last_n(comb, self.n_drop))] + elif self.method_sell == "random": + candi = filter_stock(last) + try: + sell = pd.Index(np.random.choice(candi, self.n_drop, replace=False) if len(last) else []) + except ValueError: + sell = candi + else: + raise NotImplementedError(f"This type of input is not supported") + + buy = today[: len(sell) + self.topk - len(last)] + for code in current_stock_list: + if not self.trade_exchange.is_stock_tradable( + stock_id=code, + start_time=trade_start_time, + end_time=trade_end_time, + direction=None if self.forbid_all_trade_at_limit else OrderDir.SELL, + ): + continue + if code in sell: + time_per_step = self.trade_calendar.get_freq() + if current_temp.get_stock_count(code, bar=time_per_step) < self.hold_thresh: + continue + sell_amount = current_temp.get_stock_amount(code=code) + sell_order = Order( + stock_id=code, + amount=sell_amount, + start_time=trade_start_time, + end_time=trade_end_time, + direction=Order.SELL, + ) + if self.trade_exchange.check_order(sell_order): + sell_order_list.append(sell_order) + trade_val, trade_cost, trade_price = self.trade_exchange.deal_order( + sell_order, position=current_temp + ) + cash += trade_val - trade_cost + + if len(buy) == 0: + return TradeDecisionWO(sell_order_list, self) + + value = cash * self.risk_degree / len(buy) + for code in buy: + if not self.trade_exchange.is_stock_tradable( + stock_id=code, + start_time=trade_start_time, + end_time=trade_end_time, + direction=None if self.forbid_all_trade_at_limit else OrderDir.BUY, + ): + continue + buy_price = self.trade_exchange.get_deal_price( + stock_id=code, start_time=trade_start_time, end_time=trade_end_time, direction=OrderDir.BUY + ) + buy_amount = value / buy_price + factor = self.trade_exchange.get_factor( + stock_id=code, start_time=trade_start_time, end_time=trade_end_time + ) + buy_amount = self.trade_exchange.round_amount_by_trade_unit(buy_amount, factor) + buy_order = Order( + stock_id=code, + amount=buy_amount, + start_time=trade_start_time, + end_time=trade_end_time, + direction=Order.BUY, + ) + buy_order_list.append(buy_order) + + return TradeDecisionWO(sell_order_list + buy_order_list, self) \ No newline at end of file diff --git a/code/tac-qlib/tac_qlib/data/__pycache__/__init__.cpython-312.pyc b/code/tac-qlib/tac_qlib/data/__pycache__/__init__.cpython-312.pyc index 0ed1ead..7c4e6c3 100644 Binary files a/code/tac-qlib/tac_qlib/data/__pycache__/__init__.cpython-312.pyc and b/code/tac-qlib/tac_qlib/data/__pycache__/__init__.cpython-312.pyc differ diff --git a/code/tac-qlib/tac_qlib/data/__pycache__/config.cpython-312.pyc b/code/tac-qlib/tac_qlib/data/__pycache__/config.cpython-312.pyc index 9609782..99e6023 100644 Binary files a/code/tac-qlib/tac_qlib/data/__pycache__/config.cpython-312.pyc and b/code/tac-qlib/tac_qlib/data/__pycache__/config.cpython-312.pyc differ diff --git a/code/tac-qlib/tac_qlib/data/__pycache__/providers.cpython-312.pyc b/code/tac-qlib/tac_qlib/data/__pycache__/providers.cpython-312.pyc index a85628d..020dcdc 100644 Binary files a/code/tac-qlib/tac_qlib/data/__pycache__/providers.cpython-312.pyc and b/code/tac-qlib/tac_qlib/data/__pycache__/providers.cpython-312.pyc differ diff --git a/code/tac-qlib/tac_qlib/data/config.py b/code/tac-qlib/tac_qlib/data/config.py index 686d36f..53c9007 100644 --- a/code/tac-qlib/tac_qlib/data/config.py +++ b/code/tac-qlib/tac_qlib/data/config.py @@ -6,10 +6,11 @@ The lake is a hive-partitioned parquet store (see ``tac-engine/skills/tradeac-la ├── market=US/ │ └── timeframe=1d/ │ └── symbol=AAPL.parquet # OHLCV bars: t, date, o, h, l, c, v, n, vw - ├── features/ # ta-lib indicators, wide format + ├── features/ # indicators, wide format, family tier │ └── market=US/ │ └── timeframe=1d/ - │ └── symbol=AAPL.parquet # t, sma_5, sma_20, rsi_14, ... + │ ├── family=ta/symbol=AAPL.parquet # t, sma_5, sma_20, rsi_14, ... + │ └── family=sp/symbol=AAPL.parquet # t, sp_ou_*, sp_hmm_*, ... ├── calendar.parquet # trading days per market ├── coverage.parquet # per (market,timeframe,symbol) loaded windows └── symbols.parquet # asset master @@ -107,8 +108,34 @@ class LakeConfig: return self.lake_root / "features" / f"market={self.market}" / f"timeframe={timeframe}" def features_path(self, timeframe: str, symbol: str) -> Path: + # Legacy flat path (no family tier). Prefer `load_features` which + # resolves the family=ta|sp partition layout. return self.features_dir(timeframe) / f"symbol={str(symbol).upper()}.parquet" + def load_features(self, timeframe: str, symbol: str) -> pd.DataFrame: + """All feature columns for a symbol, merging the `family=ta` and + `family=sp` partitions by timestamp. Returns an empty frame when no + feature files exist (legacy flat layout falls back transparently).""" + sym = str(symbol).upper() + frames = [] + for family in ("ta", "sp"): + p = self.features_dir(timeframe) / f"family={family}" / f"symbol={sym}.parquet" + if p.exists(): + frames.append(pd.read_parquet(p)) + if not frames: + flat = self.features_dir(timeframe) / f"symbol={sym}.parquet" + if flat.exists(): + return pd.read_parquet(flat) + return pd.DataFrame() + if len(frames) == 1: + return frames[0] + merged = frames[0] + for extra in frames[1:]: + merged = merged.merge(extra, on="t", how="outer", suffixes=("", "_dup")) + for c in [c for c in merged.columns if c.endswith("_dup")]: + merged = merged.drop(columns=c) + return merged + def calendar_path(self) -> Path: return self.lake_root / "calendar.parquet" diff --git a/code/tac-qlib/tac_qlib/data/providers.py b/code/tac-qlib/tac_qlib/data/providers.py index d9f839b..8d0644f 100644 --- a/code/tac-qlib/tac_qlib/data/providers.py +++ b/code/tac-qlib/tac_qlib/data/providers.py @@ -173,8 +173,7 @@ class LakeFeatureProvider(FeatureProvider): def _load_feature_df(self, instrument: str, timeframe: str) -> pd.DataFrame: key = (instrument, timeframe) if key not in self._feature_cache: - p = self.cfg.features_path(timeframe, instrument) - self._feature_cache[key] = pd.read_parquet(p) if p.exists() else pd.DataFrame() + self._feature_cache[key] = self.cfg.load_features(timeframe, instrument) return self._feature_cache[key] @staticmethod