Compare commits

..
43 changed files with 184 additions and 2642 deletions
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -1,48 +0,0 @@
#!/usr/bin/env python3
"""Precompute the signal-quality gate series and save to pickle.
Usage:
python precompute_signal_quality_gate.py <pred_path> <output_path> [topk] [lookback] [threshold]
Example:
python precompute_signal_quality_gate.py \
/home/data/lake/mlruns/49/34165f27e4a34378ad54843a079a78c0/artifacts/pred.pkl \
/app/experiments/book/data/signal_quality_gate/sq_gate_5d_0.50.pkl \
10 5 0.5
"""
import sys
import pickle
from pathlib import Path
# Add tac-qlib to path
sys.path.insert(0, "/app/tac-qlib")
from tac_qlib.contrib.strategy.signal_quality_gate import compute_signal_quality_gate
if __name__ == "__main__":
if len(sys.argv) < 3:
print(__doc__)
sys.exit(1)
pred_path = sys.argv[1]
output_path = sys.argv[2]
topk = int(sys.argv[3]) if len(sys.argv) > 3 else 10
lookback = int(sys.argv[4]) if len(sys.argv) > 4 else 5
threshold = float(sys.argv[5]) if len(sys.argv) > 5 else 0.5
lake_root = "/home/data/lake"
print(f"Computing signal-quality gate: topk={topk}, lookback={lookback}, threshold={threshold}")
gate = compute_signal_quality_gate(
pred_path,
lake_root=lake_root,
topk=topk,
lookback=lookback,
threshold=threshold,
)
print(f"Gate: {gate.sum()}/{len(gate)} days open ({gate.mean():.1%})")
with open(output_path, "wb") as f:
pickle.dump(gate, f)
print(f"Saved to {output_path}")
-115
View File
@@ -1,115 +0,0 @@
# Signal-quality gate backtest for 2021
{%- set LAKE = TAC_LAKE_DIR %}
{%- set UNIVERSE = "SPY,QQQ,DIA,IWM,MDY,VTI,VOO,VEA,VWO,VT,EFA,EEM,TLT,IEF,SHY,AGG,BND,LQD,HYG,JNK,EMB,GLD,SLV,USO,UNG,DBA,DBC,XLK,XLF,XLE,XLV,XLI,XLY,XLP,XLU,XLB,XLRE,ARKK,SMH,SOXX,IBB,XBI,ITA,XAR,ICLN,TAN,FDN,IGV,ESPO,REM" %}
{%- set SP_FIELDS = "sp_ret,sp_ou_zscore,sp_ou_half_life,sp_ou_revert,sp_hmm_p_regime1,sp_hmm_state,sp_jump_ratio,sp_jump_flag,sp_jump_tail,sp_max_move,sp_rv1,sp_rv5,sp_rv22,sp_vol_ratio_5_22,sp_vol_ratio_1_22,sp_trend_slope_5,sp_trend_slope_20,sp_trend_slope_60,sp_logp,sp_hurst_exponent,sp_sig_level1_lead,sp_sig_level1_lag,sp_sig_level2_lead_lag,sp_sig_level2_lag_lead" %}
{# Gate computed on-the-fly from signal + lake close prices #}
qlib_init:
provider_uri: "{{ LAKE }}"
region: us
expression_cache: null
dataset_cache: null
calendar_provider:
class: tac_qlib.data.providers.LakeCalendarProvider
kwargs: { lake_root: "{{ LAKE }}", market: US }
instrument_provider:
class: tac_qlib.data.providers.LakeInstrumentProvider
kwargs: { lake_root: "{{ LAKE }}", market: US, markets: {} }
feature_provider:
class: tac_qlib.data.providers.LakeFeatureProvider
kwargs: { lake_root: "{{ LAKE }}", market: US }
exp_manager:
class: MLflowExpManager
module_path: qlib.workflow.expm
kwargs:
uri: "sqlite:///{{ LAKE }}/mlruns.db"
default_exp_name: "tac-rd-sq-gate-2021"
task:
model:
class: LGBModel
module_path: qlib.contrib.model.gbdt
kwargs:
loss: mse
learning_rate: 0.05
num_leaves: 15
n_estimators: 200
colsample_bytree: 0.8
subsample: 0.8
subsample_freq: 1
reg_alpha: 0.01
reg_lambda: 0.01
dataset:
class: DatasetH
module_path: qlib.data.dataset
kwargs:
handler:
class: TACHandler
module_path: tac_qlib.contrib.data.handler
kwargs:
instruments: "{{ UNIVERSE }}"
start_time: 2015-01-03
end_time: 2021-12-31
fit_start_time: 2015-01-03
fit_end_time: 2021-01-03
freq: day
lake_root: "{{ LAKE }}"
market: US
label: "Ref($close,-6)/Ref($close,-1)-1"
feature_fields: "$open,$high,$low,$close,$vwap,$volume,{{ SP_FIELDS }}"
infer_processors:
- class: DropAllNaN
kwargs: {}
- class: ProcessInf
kwargs: {}
- class: CSRankNorm
kwargs: {}
- class: ZScoreNorm
kwargs: {}
- class: Fillna
kwargs: {}
segments:
train: [2015-01-03, 2020-09-01]
valid: [2020-09-03, 2021-01-03]
test: [2021-01-04, 2021-12-31]
record:
- class: SignalRecord
module_path: qlib.workflow.record_temp
kwargs: {}
- class: SigAnaRecord
module_path: qlib.workflow.record_temp
kwargs: { ana_long_short: true, ann_scaler: 252 }
- class: PortAnaRecord
module_path: qlib.workflow.record_temp
kwargs:
config:
strategy:
class: SignalQualityGateStrategy
module_path: tac_qlib.contrib.strategy.signal_quality_gate
kwargs:
signal: "<PRED>"
lake_root: "{{ LAKE }}"
gate_topk: 10
gate_lookback: 5
gate_threshold: 0.5
gate_start: "2015-01-03"
gate_end: "2021-12-31"
topk: 10
n_drop: 1
only_tradable: true
risk_degree: 0.95
backtest:
start_time: 2021-01-04
end_time: 2021-12-31
account: 1000000
benchmark: SPY
exchange_kwargs:
codes: "{{ UNIVERSE }}"
deal_price: $close
freq: day
open_cost: 0.0005
close_cost: 0.0015
min_cost: 5.0
risk_analysis_freq: 1d
-115
View File
@@ -1,115 +0,0 @@
# Signal-quality gate backtest for 2024
{%- set LAKE = TAC_LAKE_DIR %}
{%- set UNIVERSE = "SPY,QQQ,DIA,IWM,MDY,VTI,VOO,VEA,VWO,VT,EFA,EEM,TLT,IEF,SHY,AGG,BND,LQD,HYG,JNK,EMB,GLD,SLV,USO,UNG,DBA,DBC,XLK,XLF,XLE,XLV,XLI,XLY,XLP,XLU,XLB,XLRE,ARKK,SMH,SOXX,IBB,XBI,ITA,XAR,ICLN,TAN,FDN,IGV,ESPO,REM" %}
{%- set SP_FIELDS = "sp_ret,sp_ou_zscore,sp_ou_half_life,sp_ou_revert,sp_hmm_p_regime1,sp_hmm_state,sp_jump_ratio,sp_jump_flag,sp_jump_tail,sp_max_move,sp_rv1,sp_rv5,sp_rv22,sp_vol_ratio_5_22,sp_vol_ratio_1_22,sp_trend_slope_5,sp_trend_slope_20,sp_trend_slope_60,sp_logp,sp_hurst_exponent,sp_sig_level1_lead,sp_sig_level1_lag,sp_sig_level2_lead_lag,sp_sig_level2_lag_lead" %}
{# Gate computed on-the-fly from signal + lake close prices #}
qlib_init:
provider_uri: "{{ LAKE }}"
region: us
expression_cache: null
dataset_cache: null
calendar_provider:
class: tac_qlib.data.providers.LakeCalendarProvider
kwargs: { lake_root: "{{ LAKE }}", market: US }
instrument_provider:
class: tac_qlib.data.providers.LakeInstrumentProvider
kwargs: { lake_root: "{{ LAKE }}", market: US, markets: {} }
feature_provider:
class: tac_qlib.data.providers.LakeFeatureProvider
kwargs: { lake_root: "{{ LAKE }}", market: US }
exp_manager:
class: MLflowExpManager
module_path: qlib.workflow.expm
kwargs:
uri: "sqlite:///{{ LAKE }}/mlruns.db"
default_exp_name: "tac-rd-sq-gate-2024"
task:
model:
class: LGBModel
module_path: qlib.contrib.model.gbdt
kwargs:
loss: mse
learning_rate: 0.05
num_leaves: 15
n_estimators: 200
colsample_bytree: 0.8
subsample: 0.8
subsample_freq: 1
reg_alpha: 0.01
reg_lambda: 0.01
dataset:
class: DatasetH
module_path: qlib.data.dataset
kwargs:
handler:
class: TACHandler
module_path: tac_qlib.contrib.data.handler
kwargs:
instruments: "{{ UNIVERSE }}"
start_time: 2015-01-03
end_time: 2024-12-31
fit_start_time: 2015-01-03
fit_end_time: 2024-01-03
freq: day
lake_root: "{{ LAKE }}"
market: US
label: "Ref($close,-6)/Ref($close,-1)-1"
feature_fields: "$open,$high,$low,$close,$vwap,$volume,{{ SP_FIELDS }}"
infer_processors:
- class: DropAllNaN
kwargs: {}
- class: ProcessInf
kwargs: {}
- class: CSRankNorm
kwargs: {}
- class: ZScoreNorm
kwargs: {}
- class: Fillna
kwargs: {}
segments:
train: [2015-01-03, 2023-09-01]
valid: [2023-09-03, 2024-01-03]
test: [2024-01-02, 2024-12-31]
record:
- class: SignalRecord
module_path: qlib.workflow.record_temp
kwargs: {}
- class: SigAnaRecord
module_path: qlib.workflow.record_temp
kwargs: { ana_long_short: true, ann_scaler: 252 }
- class: PortAnaRecord
module_path: qlib.workflow.record_temp
kwargs:
config:
strategy:
class: SignalQualityGateStrategy
module_path: tac_qlib.contrib.strategy.signal_quality_gate
kwargs:
signal: "<PRED>"
lake_root: "{{ LAKE }}"
gate_topk: 10
gate_lookback: 5
gate_threshold: 0.5
gate_start: "2015-01-03"
gate_end: "2024-12-31"
topk: 10
n_drop: 1
only_tradable: true
risk_degree: 0.95
backtest:
start_time: 2024-01-02
end_time: 2024-12-31
account: 1000000
benchmark: SPY
exchange_kwargs:
codes: "{{ UNIVERSE }}"
deal_price: $close
freq: day
open_cost: 0.0005
close_cost: 0.0015
min_cost: 5.0
risk_analysis_freq: 1d
-35
View File
@@ -1,35 +0,0 @@
# TradeAC custom-qlib-code snapshot (auto-generated)
# parent repo HEAD : e952feed0a66a20439f4f24ad5524233429cd0c3
# tac-qlib/tac_qlib/contrib
# tac-qlib/tac_qlib/data
# per-file hashes (git hash-object):
1b6298c4a5652f2e863cbdc385a1014a570fcd59 tac-qlib/tac_qlib/contrib/__init__.py
861592c63edd6a0853a9cb174b5970435b135fc8 tac-qlib/tac_qlib/contrib/__pycache__/__init__.cpython-312.pyc
c76a9f17f680e74eea766eff27f7624359749ed6 tac-qlib/tac_qlib/contrib/data/__init__.py
5c547a2ef92e075e550fe6d01508a2f1d3f536bc tac-qlib/tac_qlib/contrib/data/__pycache__/__init__.cpython-312.pyc
4f656130d167e79dcaaeb7783a121f0b36852374 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
b1489f2fc0dee85f0a4f90b2e6ad545ed9c8967b tac-qlib/tac_qlib/contrib/model/__pycache__/__init__.cpython-312.pyc
121ef237da1df1b8e21a561c3ad0db200b901339 tac-qlib/tac_qlib/contrib/model/__pycache__/rank_ensemble.cpython-312.pyc
74d0da348cbcc3700c96b6f4fe4391488e61efc5 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
2c2f167b693f4366a769998e3c9d4804f29e31e0 tac-qlib/tac_qlib/contrib/strategy/__init__.py
f9cd9ab729e3248542ccc490af7adc2e51c71914 tac-qlib/tac_qlib/contrib/strategy/__pycache__/__init__.cpython-312.pyc
cd3133cfbd2556b25c106e39ae97fb128df0e326 tac-qlib/tac_qlib/contrib/strategy/__pycache__/ic_gate.cpython-312.pyc
6dd1c568a2961842793674390d5abffd1a0e71b8 tac-qlib/tac_qlib/contrib/strategy/__pycache__/optimal_stop.cpython-312.pyc
03b5e4d80da00800f1b108bee0735d3d18d856d1 tac-qlib/tac_qlib/contrib/strategy/__pycache__/regime_gate.cpython-312.pyc
a6a1c21ab71b62080830df45c4784b73c1531036 tac-qlib/tac_qlib/contrib/strategy/__pycache__/signal_quality_gate.cpython-312.pyc
755e3b139496a5e22b0328db45c8199c33029fbc tac-qlib/tac_qlib/contrib/strategy/__pycache__/weekly_rebalance.cpython-312.pyc
519a1f4c05dbe0ac018ab8b779eb33d53b4dd545 tac-qlib/tac_qlib/contrib/strategy/ic_gate.py
79aaad9e39fcc740a773f4f63c512ce1086cfde0 tac-qlib/tac_qlib/contrib/strategy/optimal_stop.py
7bcee5f0b09cfa721440f1354f16f2dd9a112b12 tac-qlib/tac_qlib/contrib/strategy/regime_gate.py
16ab80b731aab6d8bb818525615d77c2d1fcb0c8 tac-qlib/tac_qlib/contrib/strategy/signal_quality_gate.py
fe60bacdfedd48617863be31f24b7c7daebfac5a tac-qlib/tac_qlib/contrib/strategy/weekly_rebalance.py
92e6e90eb0cd0a25142034560f27adb6b705b1a8 tac-qlib/tac_qlib/data/__init__.py
316bf4aa160cc8d15929ea648be03f4b4999667d tac-qlib/tac_qlib/data/__pycache__/__init__.cpython-312.pyc
554a3f29d181b64effbf49a8161b32e7f93d8d3e tac-qlib/tac_qlib/data/__pycache__/config.cpython-312.pyc
8b47f6d78ac046b6b7b2fb07bd7f3382773ffb73 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
@@ -1,11 +0,0 @@
from . import data # noqa: F401 (registers tac_qlib.contrib.data)
from . import model, strategy # noqa: F401
from .data import TACHandler # noqa: F401
from .model import RankICLGBModel # noqa: F401
from .strategy import OptimalStopControl # noqa: F401
__all__ = [
"TACHandler",
"RankICLGBModel",
"OptimalStopControl",
]
@@ -1,3 +0,0 @@
from .handler import TACHandler
__all__ = ["TACHandler"]
@@ -1,254 +0,0 @@
"""TACHandler: a qlib DataHandlerLP that builds datasets from the TradeAC lake.
This is the "custom DataHandler" entry point (Option B): the handler is referenced from the
workflow yaml's ``dataset.handler`` and reads OHLCV + pre-computed ta-lib features straight
from the lake parquet files through ``QLibDataLoader`` + the tac_qlib feature provider.
The standard qlib processor pipeline (``infer_processors`` / ``learn_processors``) still runs
on top, so existing recipes such as ``DropnaLabel``, ``CSZScoreNorm`` or ``RobustZScoreNorm``
keep working unchanged.
"""
from __future__ import annotations
import os
from inspect import getfullargspec
from typing import List, Optional, Tuple, Union
from qlib.data.dataset import processor as processor_module
from qlib.data.dataset.handler import DataHandlerLP
from qlib.utils import get_callable_kwargs
from ...data.config import (
LakeConfig,
timeframe_for_freq,
NON_FEATURE_COLUMNS,
)
DEFAULT_INFER_PROCESSORS = [
{"class": "DropAllNaN", "kwargs": {}},
{"class": "ProcessInf", "kwargs": {}},
{"class": "ZScoreNorm", "kwargs": {}},
{"class": "Fillna", "kwargs": {}},
]
DEFAULT_LEARN_PROCESSORS = [
{"class": "DropnaLabel"},
{"class": "CSZScoreNorm", "kwargs": {"fields_group": "label"}},
]
#: always include raw OHLCV; ta-lib columns are discovered from the lake and appended.
RAW_FEATURE_FIELDS = ("$open", "$high", "$low", "$close", "$vwap", "$volume")
DEFAULT_LABEL = "Ref($close,-2)/Ref($close,-1)-1"
def check_transform_proc(proc_l, fit_start_time, fit_end_time):
"""Port of ``qlib.contrib.data.handler.check_transform_proc`` (inject fit window into procs)."""
new_l = []
for p in proc_l:
if not isinstance(p, processor_module.Processor):
klass, pkwargs = get_callable_kwargs(p, processor_module)
args = getfullargspec(klass).args
if "fit_start_time" in args and "fit_end_time" in args:
assert fit_start_time is not None and fit_end_time is not None, (
"Make sure `fit_start_time` and `fit_end_time` are not None."
)
pkwargs.update({"fit_start_time": fit_start_time, "fit_end_time": fit_end_time})
proc_config = {"class": klass.__name__, "kwargs": pkwargs}
if isinstance(p, dict) and "module_path" in p:
proc_config["module_path"] = p["module_path"]
new_l.append(proc_config)
else:
new_l.append(p)
return new_l
def get_common_feature_fields(lake_root=None, market="US", timeframe="1d") -> List[str]:
"""Discover feature columns present in *every* feature file of the lake.
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)
if not feat_dir.exists():
return []
import pyarrow.parquet as pq
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):
"""Drop feature columns that are all-NaN over the fit window.
The lake can hold fully-empty indicator columns (e.g. a ta-lib output that was NaN
from the start). Such columns carry no learnable signal and make ``ZScoreNorm.fit``
warn on empty slices, so we drop them before any other processor runs. The drop set
is fixed on the fit window once (during ``fit``), then applied consistently to every
segment so train/valid/test keep identical feature columns.
"""
def __init__(self, fit_start_time=None, fit_end_time=None):
self.fit_start_time = fit_start_time
self.fit_end_time = fit_end_time
self.cols_to_drop = []
def fit(self, df=None):
if df is None or len(df) == 0:
return self
window = df
if self.fit_start_time is not None and self.fit_end_time is not None:
try:
from qlib.data.dataset.utils import fetch_df_by_index
window = fetch_df_by_index(
df, slice(self.fit_start_time, self.fit_end_time), level="datetime"
)
except Exception: # pragma: no cover - defensive
window = df
if len(window) == 0:
return self
self.cols_to_drop = [c for c in window.columns if window[c].isna().all()]
return self
def __call__(self, df):
if self.cols_to_drop:
return df.drop(columns=self.cols_to_drop, errors="ignore")
return df
class TACHandler(DataHandlerLP):
"""DataHandlerLP backed by the TradeAC parquet lake.
Parameters mirror ``Alpha158``: ``instruments``/``start_time``/``end_time``/``freq`` define
the queried window; ``feature_fields`` selects the features (default: raw OHLCV + all common
ta-lib columns found in the lake); ``label`` is a qlib expression for the target.
"""
def __init__(
self,
instruments="all",
start_time=None,
end_time=None,
freq="day",
infer_processors=DEFAULT_INFER_PROCESSORS,
learn_processors=DEFAULT_LEARN_PROCESSORS,
fit_start_time=None,
fit_end_time=None,
process_type=DataHandlerLP.PTYPE_A,
filter_pipe=None,
feature_fields=None,
label=DEFAULT_LABEL,
lake_root=None,
market="US",
**kwargs,
):
# default the processor fit window to the queried window (like Alpha158 without a split)
if fit_start_time is None:
fit_start_time = start_time
if fit_end_time is None:
fit_end_time = end_time
infer_processors = check_transform_proc(infer_processors, fit_start_time, fit_end_time)
learn_processors = check_transform_proc(learn_processors, fit_start_time, fit_end_time)
feature_fields = self._normalize_feature_fields(feature_fields, freq, lake_root, market)
if not feature_fields:
raise ValueError(
"no feature fields available for the lake; set `feature_fields` explicitly "
"(e.g. ['$close', '$rsi_14', '$sma_20'])"
)
label_expr, label_names = self._normalize_label(label)
data_loader = {
"class": "QlibDataLoader",
"kwargs": {
"config": {
"feature": (feature_fields, feature_fields),
"label": (label_expr, label_names),
},
"filter_pipe": filter_pipe,
"freq": freq,
},
}
super().__init__(
instruments=instruments,
start_time=start_time,
end_time=end_time,
data_loader=data_loader,
infer_processors=infer_processors,
learn_processors=learn_processors,
process_type=process_type,
**kwargs,
)
# ------------------------------------------------------------------ config
@staticmethod
def _normalize_feature_fields(feature_fields, freq, lake_root, market) -> List[str]:
if feature_fields is None:
common = get_common_feature_fields(lake_root, market, timeframe_for_freq(freq))
feature_fields = list(RAW_FEATURE_FIELDS) + ["$" + f for f in common if "$" + f not in RAW_FEATURE_FIELDS]
elif isinstance(feature_fields, str):
feature_fields = [f.strip() for f in feature_fields.split(",") if f.strip()]
fields = [f if f.startswith("$") else "$" + f for f in feature_fields]
# de-dup while preserving order
seen, out = set(), []
for f in fields:
if f not in seen:
seen.add(f)
out.append(f)
return out
@staticmethod
def _normalize_label(label) -> Tuple[List[str], List[str]]:
if isinstance(label, str):
return [label], ["LABEL0"]
if isinstance(label, (list, tuple)):
if len(label) == 2 and isinstance(label[0], str):
return [label[0]], list(label[1]) if isinstance(label[1], (list, tuple)) else [label[1]]
return list(label), ["LABEL%d" % i for i in range(len(label))]
raise TypeError(f"unsupported label config: {label!r}")
# ------------------------------------------------------------------ utils
def get_label_config(self):
return DEFAULT_LABEL
@staticmethod
def discover_feature_fields(lake_root=None, market="US", freq="day") -> List[str]:
return get_common_feature_fields(lake_root, market, timeframe_for_freq(freq))
__all__ = ["TACHandler", "DropAllNaN", "get_common_feature_fields"]
# Make `DropAllNaN` resolvable by bare name from processor configs (e.g. the default
# ``infer_processors`` and workflow yamls that reference it without a ``module_path``),
# mirroring how qlib registers its own processors in ``qlib.data.dataset.processor``.
processor_module.DropAllNaN = DropAllNaN
@@ -1,4 +0,0 @@
from .rank_ensemble import RankICEnsembleLGBModel # noqa: F401
from .rank_gbdt import RankICLGBModel, rankic_feval # noqa: F401
__all__ = ["RankICLGBModel", "rankic_feval", "RankICEnsembleLGBModel"]
@@ -1,189 +0,0 @@
"""Seed-ensembled LightGBM that early-stops on cross-sectional RankIC.
``RankICEnsembleLGBModel`` wraps ``RankICLGBModel`` (per-day RankIC feval +
``metric='None'`` + ``first_metric_only`` early stopping) over a seed ensemble:
one sub-model is trained per seed with identical hyper-parameters, and
predictions are averaged across seeds. This is the model class the
``tac-rd-rank-ensemble-isolated`` reference run wires into its workflow
(``module_path: tac_qlib.contrib.model.rank_ensemble``).
The ensemble inherits the RankIC early-stopping behaviour of the single-seed
model (valid RankIC drives the stopping iteration) while the seed averaging
stabilizes the prediction against any single seed's early-stopping path.
Training is parallelized: the seed sub-models train in a thread pool —
``lgb.train`` is C++ and releases the GIL, so concurrent seeds do not block on
the GIL (5 seeds ~40min/5 on this box). Measured on a 6-physical-core / 12 SMT
host: the seeds scale ~2x, not linearly — the runs are memory-bandwidth bound
and each Booster caps its threads at ``cores // workers`` so 5 concurrent
boosters don't oversubscribe; larger-core hosts scale better. The qlib data
pipeline is warmed once on the calling thread (fills the handler cache), and
each worker then prepares its **own** ``lgb.Dataset`` (independent handle, so
no concurrent ``construct()`` on a shared handle — LightGBM's ``Dataset`` is
not thread-safe to build). qlib's ``R`` recorder is also not thread-safe, so
the per-seed evaluation curves are logged on the calling thread after the pool
finishes.
Wired into a workflow yaml like:
model:
class: RankICEnsembleLGBModel
module_path: tac_qlib.contrib.model.rank_ensemble
kwargs:
loss: mse
learning_rate: 0.02
num_leaves: 31
n_estimators: 3000
num_boost_round: 3000
early_stopping_rounds: 200
min_data_in_leaf: 20
lambda_l2: 0.5
colsample_bytree: 0.8
subsample: 0.8
subsample_freq: 1
reg_alpha: 0.1
reg_lambda: 1.0
seeds: "42,7,2026,99,123"
parallel: 5
Any ``**kwargs`` other than ``seeds``/``parallel`` are forwarded unchanged to
every ``RankICLGBModel`` sub-model (same params, different ``seed``).
"""
from __future__ import annotations
import os
from concurrent.futures import ThreadPoolExecutor
from typing import List, Optional
import pandas as pd
from qlib.data.dataset import DatasetH
from qlib.data.dataset.handler import DataHandlerLP
from tac_qlib.contrib.model.rank_gbdt import RankICLGBModel
__all__ = ["RankICEnsembleLGBModel"]
class RankICEnsembleLGBModel(RankICLGBModel):
"""Seed ensemble of RankIC-early-stopping LightGBM models.
Parameters
----------
seeds : comma-separated integers, one sub-model per seed.
parallel : number of seeds to train concurrently. ``0`` (default) = auto
(all seeds, bounded by the available cores); ``1`` = sequential.
**kwargs : forwarded to every ``RankICLGBModel`` sub-model (model
hyper-parameters). ``seeds``/``parallel`` are consumed here and not
forwarded.
"""
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)
# drop seed/parallel handling from the base kwargs, keep everything else
self._model_kwargs = dict(kwargs)
super().__init__(**self._model_kwargs)
self._models: List[RankICLGBModel] = []
# --------------------------------------------------------------- helpers
@staticmethod
def _cores() -> int:
try:
return max(1, len(os.sched_getaffinity(0)))
except AttributeError:
return max(1, os.cpu_count() or 1)
def _worker_count(self) -> int:
if self.parallel > 0:
return min(len(self.seeds), self.parallel)
return min(len(self.seeds), self._cores())
# ------------------------------------------------------------------ fit
def fit(
self,
dataset: DatasetH,
num_boost_round: Optional[int] = None,
early_stopping_rounds: Optional[int] = None,
verbose_eval: int = 20,
evals_result=None,
reweighter=None,
**kwargs,
):
"""Train one RankICLGBModel per seed and keep them for prediction.
The qlib data pipeline is warmed once on this thread (handler cache),
then each seed sub-model trains in a parallel worker thread on its own
``lgb.Dataset`` (LightGBM releases the GIL in ``lgb.train``). Evals
are logged on this thread after the pool (qlib's ``R`` is not
thread-safe).
"""
n_round = num_boost_round or self.num_boost_round
n_es = early_stopping_rounds or self.early_stopping_rounds
if len(self.seeds) == 1:
m = RankICLGBModel(seed=self.seeds[0], **self._model_kwargs)
m.fit(
dataset,
num_boost_round=n_round,
early_stopping_rounds=n_es,
verbose_eval=verbose_eval,
evals_result=evals_result,
reweighter=reweighter,
**kwargs,
)
self._models = [m]
return
# Warm the qlib handler cache once on this thread so the workers'
# concurrent prepare() calls only hit cached frames (no first-write race).
proto = RankICLGBModel(seed=self.seeds[0], **self._model_kwargs)
proto._prepare_data(dataset, reweighter)
workers = self._worker_count()
# Cap per-Booster threads so concurrent seeds don't oversubscribe
# (LightGBM's num_threads=0 uses ALL cores per Booster).
per_booster = max(1, self._cores() // workers)
def fit_seed(seed):
m = RankICLGBModel(seed=seed, **self._model_kwargs)
if workers > 1 and "num_threads" not in m.params:
m.params["num_threads"] = per_booster
ds_l = m._prepare_data(dataset, reweighter)
booster, evals, names = m._train_from_datasets(
ds_l,
num_boost_round=n_round,
early_stopping_rounds=n_es,
verbose_eval=verbose_eval,
**kwargs,
)
m.model = booster
return m, evals, names
with ThreadPoolExecutor(max_workers=workers) as ex:
results = list(ex.map(fit_seed, self.seeds))
self._models = [m for m, _, _ in results]
# Merge + log evals on the main thread (qlib's R is not thread-safe).
if evals_result is not None:
for m, evals, names in results:
for k in names:
for key, val in evals.get(k, {}).items():
evals_result.setdefault(f"{k}.seed{m.params['seed']}", {})[key] = val
for m, evals, names in results:
self._log_evals(evals, names, prefix=f"seed{m.params['seed']}.")
# -------------------------------------------------------------- predict
def predict(self, dataset: DatasetH, segment="test") -> pd.Series:
"""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)
return frame.mean(axis=1)
@@ -1,238 +0,0 @@
"""LGBModel variant that early-stops on cross-sectional RankIC instead of l2.
Standard qlib ``LGBModel`` early-stops on the regression loss (mse). For
cross-sectional alpha signals the quantity we actually care about is the per-day
rank correlation (Rank IC), which mse early-stopping does not optimize for.
Experiments on the 50-ETF lake (SP-5d 55-feature panel) show that early-stopping
on a custom RankIC feval lifts RankIC 0.047 -> 0.075 vs. the mse-stopped model.
This class reuses ``LGBModel``'s data preparation but:
- tags each ``lgb.Dataset`` with per-day query ``group`` sizes so a ranking
metric can be computed per trading day;
- injects a custom ``feval`` (mean per-day Spearman of pred vs label) into
``lgb.train``; early stopping then selects the iteration that maximizes
RankIC on the valid set;
- forces ``metric='None'`` + ``first_metric_only=True`` so early-stopping
tracks RankIC only (not the regression loss).
Wired into a workflow yaml like:
model:
class: RankICLGBModel
module_path: tac_qlib.contrib.model.rank_gbdt
kwargs:
loss: mse
learning_rate: 0.03
num_leaves: 31
n_estimators: 500
...
The rank feval is used for early-stopping selection only; the objective stays
the configured loss (default mse). Set ``rank_eval=False`` to fall back to the
plain LGBModel behaviour (early-stop on the loss).
Generic: works for any cross-sectional panel whose qlib dataset index has a
``datetime`` level (each level value = one query group). The per-day groups are
derived automatically, so no universe-specific configuration is needed.
"""
from __future__ import annotations
from typing import List, Optional, Tuple
import numpy as np
import pandas as pd
import lightgbm as lgb
from qlib.data.dataset import DatasetH
from qlib.data.dataset.handler import DataHandlerLP
from qlib.contrib.model.gbdt import LGBModel
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))])
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):
"""LightGBM feval: mean RankIC (higher is better in lgb convention)."""
labels = dataset.get_label()
group = dataset.get_group()
ric = _per_day_spearman(preds, labels, group)
return "rankic", ric, True # (name, value, higher_is_better)
class RankICLGBModel(LGBModel):
"""LGBModel that early-stops on per-day RankIC via a custom feval."""
def __init__(self, rank_eval: bool = True, **kwargs):
super().__init__(**kwargs)
self.rank_eval = rank_eval
def _prepare_data(self, dataset: DatasetH, reweighter=None) -> List[Tuple[lgb.Dataset, str]]:
ds_l = []
assert "train" in dataset.segments
for key in ["train", "valid"]:
if key in dataset.segments:
df = dataset.prepare(key, col_set=["feature", "label"], data_key=DataHandlerLP.DK_L)
if df.empty:
raise ValueError("Empty data from dataset, please check your dataset config.")
x, y = df["feature"], df["label"]
if y.values.ndim == 2 and y.values.shape[1] == 1:
y = np.squeeze(y.values)
else:
raise ValueError("LightGBM doesn't support multi-label training")
if reweighter is None:
w = None
elif hasattr(reweighter, "reweight"):
w = reweighter.reweight(df)
else:
raise ValueError("Unsupported reweighter type.")
# per-day query groups: each trading day is one group
if self.rank_eval and isinstance(df.index, pd.MultiIndex) and "datetime" in df.index.names:
group = df.groupby(level="datetime").size().to_numpy(dtype=np.int32)
else:
group = None
d = lgb.Dataset(x.values, label=y, weight=w, group=group, free_raw_data=False)
ds_l.append((d, key))
return ds_l
def _train_from_datasets(
self,
ds_l: List[Tuple[lgb.Dataset, str]],
num_boost_round: Optional[int] = None,
early_stopping_rounds: Optional[int] = None,
verbose_eval: int = 20,
evals_result=None,
**kwargs,
) -> Tuple[lgb.Booster, dict, List[str]]:
"""Train a Booster from already-prepared ``lgb.Dataset`` objects.
Pure training — no ``R.log_metrics`` — so it can be called from worker
threads (qlib's ``R`` recorder is not thread-safe; the caller decides
when/where to log). Returns ``(booster, evals_result, segment_names)``.
"""
if evals_result is None:
evals_result = {}
ds, names = list(zip(*ds_l))
callbacks = [
lgb.early_stopping(
self.early_stopping_rounds if early_stopping_rounds is None else early_stopping_rounds
),
lgb.log_evaluation(period=verbose_eval),
lgb.record_evaluation(evals_result),
]
if self.rank_eval:
# early-stopping must be driven ONLY by the RankIC feval, not l2.
# metric='None' suppresses the default l2 metric; first_metric_only
# makes early_stopping track the single remaining (rankic) metric.
self.params["metric"] = "None"
self.params["first_metric_only"] = True
feval = rankic_feval
else:
self.params.pop("metric", None)
self.params.pop("first_metric_only", None)
feval = None
booster = lgb.train(
self.params,
ds[0],
num_boost_round=self.num_boost_round if num_boost_round is None else num_boost_round,
valid_sets=ds,
valid_names=names,
feval=feval,
callbacks=callbacks,
**kwargs,
)
return booster, evals_result, list(names)
def _log_evals(self, evals_result, names: List[str], prefix: str = "") -> None:
"""Log recorded evaluation curves to qlib's active recorder."""
for k in names:
for key, val in evals_result.get(k, {}).items():
name = f"{prefix}{key}.{k}"
for epoch, m in enumerate(val):
R.log_metrics(**{name.replace("@", "_"): m}, step=epoch)
def fit(
self,
dataset: DatasetH,
num_boost_round: Optional[int] = None,
early_stopping_rounds: Optional[int] = None,
verbose_eval: int = 20,
evals_result=None,
reweighter=None,
**kwargs,
):
if evals_result is None:
evals_result = {}
ds_l = self._prepare_data(dataset, reweighter)
self.model, evals_result, names = self._train_from_datasets(
ds_l,
num_boost_round=num_boost_round,
early_stopping_rounds=early_stopping_rounds,
verbose_eval=verbose_eval,
evals_result=evals_result,
**kwargs,
)
self._log_evals(evals_result, names)
@@ -1,11 +0,0 @@
from .ic_gate import ICGateTopkDropoutStrategy # noqa: F401
from .optimal_stop import OptimalStopControl # noqa: F401
from .regime_gate import RegimeGateTopkDropoutStrategy # noqa: F401
from .weekly_rebalance import WeeklyRebalanceDropoutStrategy # noqa: F401
__all__ = [
"ICGateTopkDropoutStrategy",
"OptimalStopControl",
"RegimeGateTopkDropoutStrategy",
"WeeklyRebalanceDropoutStrategy",
]
@@ -1,117 +0,0 @@
"""Realized-IC circuit breaker TopkDropout strategy.
Subclass of ``qlib.contrib.strategy.signal_strategy.TopkDropoutStrategy`` that
holds the book (issues NO orders) while the streaming realized RankIC of the
deployed signal is below threshold — i.e. the model's cross-sectional
predictions are no longer earning against realized forward returns. When the
gate is open it behaves exactly like the reference TopkDropoutStrategy.
The gate is evaluated per trade step on the trailing mean realized RankIC of
the signal over the last ``ic_window`` trading days whose label is fully
realized as of the decision date (no lookahead — a 5d fwd label ``close[t+6]/
close[t+1]-1`` is only known at ``t+6``).
Two wiring modes:
* ``ic_gate``: a precomputed ``pd.Series`` indexed by datetime of booleans
(True = gate open / trade allowed). Computed once by the caller (e.g.
``rd_backtest``) and looked up per step. Missing dates default to open.
* realized-IC self-computation: when ``ic_min_rankic`` is given but no
``ic_gate``, the strategy computes the per-date realized RankIC itself from
``self.signal`` (the pred scores) and the lake 1d bars via
``tac_qlib.risk_limits.realized_rankic_series``, then applies the same
trailing-window comparison. Works when instantiated from a workflow YAML
PortAnaRecord config (``lake_root`` / ``market`` must be provided).
"""
from __future__ import annotations
import pandas as pd
from qlib.backtest.decision import TradeDecisionWO
from qlib.contrib.strategy.signal_strategy import TopkDropoutStrategy
from tac_qlib.risk_limits import ic_circuit_breaker, realized_rankic_series
__all__ = ["ICGateTopkDropoutStrategy"]
class ICGateTopkDropoutStrategy(TopkDropoutStrategy):
"""TopkDropout with a streaming realized-IC circuit breaker.
Parameters
----------
topk, n_drop, method_sell, method_buy, hold_thresh, only_tradable,
forbid_all_trade_at_limit : same as ``TopkDropoutStrategy``.
ic_min_rankic : float — pause new trading while trailing realized RankIC is
below this threshold (0 disables the gate).
ic_window : int — trailing window for the realized RankIC mean (default 22).
ic_label_horizon : int — label horizon in trading days (default 6).
ic_min_obs : int — min realized labels before the gate arms (default 10).
ic_gate : pd.Series, optional — precomputed per-date gate (bool indexed by
datetime). When provided, it overrides self-computation.
lake_root, market : str — lake location for self-computed realized IC.
"""
def __init__(
self,
*,
topk,
n_drop,
ic_min_rankic: float = 0.0,
ic_window: int = 22,
ic_label_horizon: int = 6,
ic_min_obs: int = 10,
ic_gate=None,
lake_root: str = "",
market: str = "US",
**kwargs,
):
super().__init__(topk=topk, n_drop=n_drop, **kwargs)
self.ic_min_rankic = float(ic_min_rankic or 0.0)
self.ic_window = int(ic_window or 22)
self.ic_label_horizon = int(ic_label_horizon or 6)
self.ic_min_obs = int(ic_min_obs or 10)
self._ic_gate = ic_gate
self._realized_ic = None
self.lake_root = lake_root or ""
self.market = market or "US"
def _load_realized_ic(self):
if self._realized_ic is None:
pred_start_time, pred_end_time = self.trade_calendar.get_step_time(
self.trade_calendar.get_trade_step(), shift=-self.ic_label_horizon
)
pred = self.signal.get_signal(start_time=pred_start_time, end_time=pred_end_time)
if isinstance(pred, pd.DataFrame):
pred = pred.iloc[:, 0]
self._realized_ic = realized_rankic_series(
pred, self.lake_root, self.market, label_horizon=self.ic_label_horizon
)
return self._realized_ic
def _gate_open(self, trade_start_time) -> bool:
ts = pd.Timestamp(trade_start_time)
if self._ic_gate is not None:
# precomputed gate series: look up the latest known decision date <= ts
known = self._ic_gate[self._ic_gate.index <= ts]
if len(known):
return bool(known.iloc[-1])
return True
if self.ic_min_rankic <= 0:
return True
realized = self._load_realized_ic()
limits = {
"ic_min_rankic": self.ic_min_rankic,
"ic_window": self.ic_window,
"ic_min_obs": self.ic_min_obs,
}
tripped, _reason, _trail = ic_circuit_breaker(realized, ts, limits)
return not tripped
def generate_trade_decision(self, execute_result=None):
trade_step = self.trade_calendar.get_trade_step()
trade_start_time, _ = self.trade_calendar.get_step_time(trade_step)
if not self._gate_open(trade_start_time):
return TradeDecisionWO([], self)
return super().generate_trade_decision(execute_result)
@@ -1,217 +0,0 @@
"""Optimal-stopping / stochastic-control strategy for cross-sectional signals.
Entry is a control policy: a symbol opens a position only when its cross-sectional
signal percentile is at or above ``entry_pct`` (i.e. it is one of the top-ranked
names) and the portfolio has fewer than ``topk`` open positions.
Exit is an optimal-stopping rule: a held position is stopped (closed) when its
signal percentile falls below ``exit_pct`` (the continuation value of holding is
no longer worth the risk), OR after ``max_hold_days`` (time stop / finite
horizon), OR when the position P&L breaches ``sl`` (loss control) and the
position has been held at least ``min_hold_days``.
Sizing is fixed ``notional`` per position (equal-weight control), unlike the
TopkDropout cash-allocation heuristic.
Wired into qrun workflows like any ``BaseStrategy`` (see ``PortAnaRecord``
config). Mirrors the API usage of qlib's ``TopkDropoutStrategy``: ``Order``/
``OrderDir`` from ``qlib.backtest.decision``, ``trade_calendar`` /
``trade_exchange`` / ``trade_position`` injected by the backtest executor.
"""
from __future__ import annotations
from typing import List
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__ = ["OptimalStopControl"]
DEFAULT_NOTIONAL = 20_000.0
DEFAULT_ENTRY_PCT = 0.80
DEFAULT_EXIT_PCT = 0.50
DEFAULT_MAX_HOLD_DAYS = 10
DEFAULT_MIN_HOLD_DAYS = 2
DEFAULT_SL = -0.06
class OptimalStopControl(BaseSignalStrategy):
"""Optimal-stopping long-only strategy over a cross-sectional signal.
Parameters
----------
topk : max number of concurrent positions.
entry_pct : min cross-sectional score percentile required to OPEN (0..1).
exit_pct : held positions are stopped when score percentile < exit_pct.
max_hold_days : hard time stop (finite-horizon close).
min_hold_days : minimum holding days before stop-loss is evaluated.
notional : $ per position (equal-weight control).
sl : stop-loss threshold as fraction of entry price (<= 0), disabled if 0.
"""
def __init__(
self,
*,
signal=None,
topk: int = 10,
entry_pct: float = DEFAULT_ENTRY_PCT,
exit_pct: float = DEFAULT_EXIT_PCT,
max_hold_days: int = DEFAULT_MAX_HOLD_DAYS,
min_hold_days: int = DEFAULT_MIN_HOLD_DAYS,
notional: float = DEFAULT_NOTIONAL,
sl: float = DEFAULT_SL,
risk_degree: float = 0.95,
trade_exchange=None,
level_infra=None,
common_infra=None,
**kwargs,
):
super().__init__(
signal=signal,
trade_exchange=trade_exchange,
level_infra=level_infra,
common_infra=common_infra,
**kwargs,
)
self.topk = topk
self.entry_pct = entry_pct
self.exit_pct = exit_pct
self.max_hold_days = max_hold_days
self.min_hold_days = min_hold_days
self.notional = notional
self.sl = sl
# ------------------------------------------------------------------ utils
@staticmethod
def _pct_rank(score: pd.Series) -> pd.Series:
return score.rank(pct=True)
def _entry_price(self, pos) -> float:
# Position stores avg entry price under key "price" (see Position.position)
price = pos.position.get("price")
if price is None:
price = pos.get_stock_amount("price")
return float(price)
def _pnl_pct(self, pos, mark: float) -> float:
entry = self._entry_price(pos)
if not entry or entry != entry:
return 0.0
return mark / entry - 1.0
def _is_tradable(self, code, start, end, direction) -> bool:
try:
return self.trade_exchange.is_stock_tradable(
stock_id=code, start_time=start, end_time=end, direction=direction
)
except TypeError: # some exchanges take no direction kwarg
return self.trade_exchange.is_stock_tradable(stock_id=code, start_time=start, end_time=end)
# ------------------------------------------------------------ decision
def generate_trade_decision(self, execute_result=None):
trade_step = self.trade_calendar.get_trade_step()
trade_start, trade_end = self.trade_calendar.get_step_time(trade_step)
pred_start, pred_end = self.trade_calendar.get_step_time(trade_step, shift=1)
pred_score = self.signal.get_signal(start_time=pred_start, end_time=pred_end)
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)
pct = self._pct_rank(pred_score)
time_per_step = self.trade_calendar.get_freq()
current_temp = __import__("copy").deepcopy(self.trade_position)
holdings = {}
for code in current_temp.get_stock_list():
if abs(current_temp.get_stock_amount(code)) > 1e-6:
holdings[code] = current_temp
# ---- optimal stopping: close held positions -----------------------
sell_orders: List[Order] = []
closed_today = set()
kept = {}
for code, pos in holdings.items():
held = current_temp.get_stock_count(code, bar=time_per_step)
mark = self.trade_exchange.get_deal_price(
stock_id=code, start_time=trade_start, end_time=trade_end, direction=Order.SELL
)
if mark is None or mark != mark:
continue
rank = pct.get(code, 0.0)
stop_pnl = held >= self.min_hold_days and self.sl < 0 and self._pnl_pct(pos, mark) <= self.sl
if held >= self.max_hold_days or rank < self.exit_pct or stop_pnl:
amt = abs(current_temp.get_stock_amount(code))
o = Order(stock_id=code, amount=amt, start_time=trade_start,
end_time=trade_end, direction=Order.SELL)
if self.trade_exchange.check_order(o):
sell_orders.append(o)
self.trade_exchange.deal_order(o, position=current_temp)
closed_today.add(code)
else:
kept[code] = mark
# ---- equal-weight control: target notional per name -----------------
# candidate opens: top-ranked names whose signal pct >= entry_pct
rank_desc = pred_score.sort_values(ascending=False)
held_codes = set(kept)
opens = []
for sym in rank_desc.index:
if len(opens) >= self.topk:
break
if sym in held_codes:
continue
if pct.get(sym, 0.0) < self.entry_pct:
continue
if not self._is_tradable(sym, trade_start, trade_end, OrderDir.BUY):
continue
opens.append(sym)
targets = held_codes | set(opens)
if not targets:
return TradeDecisionWO(sell_orders, self)
# total value (cash + marked positions) -> per-target notional
total_value = current_temp.get_cash()
for code, mark in kept.items():
total_value += abs(current_temp.get_stock_amount(code)) * mark
target_notional = total_value * self.risk_degree / max(1, len(targets))
# ---- rebalance kept positions toward target weight ------------------
buy_orders: List[Order] = []
for code, mark in kept.items():
cur = abs(current_temp.get_stock_amount(code)) * mark
diff_notional = target_notional - cur
if abs(diff_notional) / target_notional < 0.02:
continue # skip tiny rebalances
amount_delta = diff_notional / mark
direction = Order.BUY if amount_delta > 0 else Order.SELL
o = Order(stock_id=code, amount=abs(amount_delta), start_time=trade_start,
end_time=trade_end, direction=direction)
if self.trade_exchange.check_order(o):
(buy_orders if direction == Order.BUY else sell_orders).append(o)
self.trade_exchange.deal_order(o, position=current_temp)
# ---- open new positions at target weight ----------------------------
for sym in opens:
px = self.trade_exchange.get_deal_price(
stock_id=sym, start_time=trade_start, end_time=trade_end, direction=OrderDir.BUY
)
if px is None or px != px or px <= 0:
continue
amount = target_notional / px
factor = self.trade_exchange.get_factor(
stock_id=sym, start_time=trade_start, end_time=trade_end
)
amount = self.trade_exchange.round_amount_by_trade_unit(amount, factor)
o = Order(stock_id=sym, amount=amount, start_time=trade_start,
end_time=trade_end, direction=Order.BUY)
if self.trade_exchange.check_order(o):
buy_orders.append(o)
return TradeDecisionWO(sell_orders + buy_orders, self)
@@ -1,215 +0,0 @@
"""Regime-gate TopkDropout strategy.
Subclass of ``qlib.contrib.strategy.signal_strategy.TopkDropoutStrategy`` that
holds the book (issues NO orders) while a regime detector says the market is in
an unfavorable state. When the gate is open it behaves exactly like the
reference TopkDropoutStrategy.
Three detector types are supported (all causal — no lookahead):
* ``dispersion``: cross-sectional standard deviation of 22-day rolling returns
across the universe. Gate closes when CS dispersion < threshold (low
dispersion means the spread between winners and losers is too narrow for
TopkDropout to exploit).
* ``vol``: cross-sectional mean of 22-day rolling realized volatility. Gate
closes when avg vol is outside a band ``[vol_low, vol_high]`` (strategy
needs moderate vol — too calm or too turbulent both hurt).
* ``hmm``: pre-computed HMM posterior for regime 1 (``sp_hmm_p_regime1``).
Gate closes when posterior < threshold (model is not confident the calm
regime is active).
The gate is provided as a precomputed ``pd.Series`` of booleans indexed by
datetime (True = trade allowed). The companion ``compute_regime_gate``
function builds this series from lake bars; call it once before backtesting
and pass the result as the ``regime_gate`` parameter.
"""
from __future__ import annotations
import pandas as pd
from qlib.backtest.decision import TradeDecisionWO
from qlib.contrib.strategy.signal_strategy import TopkDropoutStrategy
__all__ = ["RegimeGateTopkDropoutStrategy", "compute_regime_gate"]
class RegimeGateTopkDropoutStrategy(TopkDropoutStrategy):
"""TopkDropout with a regime-gate circuit breaker.
Parameters
----------
topk, n_drop, method_sell, method_buy, hold_thresh, only_tradable,
forbid_all_trade_at_limit : same as ``TopkDropoutStrategy``.
regime_gate : pd.Series — precomputed per-date gate (bool indexed by
datetime). True = trade allowed, False = no orders. Missing dates
default to open (trade allowed).
"""
def __init__(self, *, regime_gate=None, **kwargs):
super().__init__(**kwargs)
self._regime_gate = regime_gate
def _gate_open(self, trade_start_time) -> bool:
if self._regime_gate is None:
return True
ts = pd.Timestamp(trade_start_time)
known = self._regime_gate[self._regime_gate.index <= ts]
if len(known):
return bool(known.iloc[-1])
return True # default open if no history yet
def generate_trade_decision(self, execute_result=None):
trade_step = self.trade_calendar.get_trade_step()
trade_start_time, _ = self.trade_calendar.get_step_time(trade_step)
if not self._gate_open(trade_start_time):
return TradeDecisionWO([], self)
return super().generate_trade_decision(execute_result)
# ---------------------------------------------------------------------------
# Precomputation helper
# ---------------------------------------------------------------------------
def compute_regime_gate(
detector: str,
threshold: float = 0.0,
*,
lake_root: str = "",
market: str = "US",
start: str = "2015-01-03",
end: str = "2026-08-19",
vol_low: float = 0.0,
vol_high: float = 999.0,
hmm_field: str = "sp_hmm_p_regime1",
) -> pd.Series:
"""Build a per-date regime gate series from lake bars.
Parameters
----------
detector : str — ``"dispersion"``, ``"vol"``, or ``"hmm"``.
threshold : float — for ``dispersion``: min CS dispersion to allow trading.
For ``hmm``: min HMM posterior to allow trading.
Ignored for ``vol`` (uses ``vol_low``/``vol_high`` band instead).
lake_root, market : str — lake location.
start, end : str — date window.
vol_low, vol_high : float — annualized vol band for the ``vol`` detector.
hmm_field : str — HMM feature column name for the ``hmm`` detector.
Returns
-------
pd.Series — bool, indexed by datetime. True = trade allowed.
"""
from tac_qlib.data.config import LakeConfig, resolve_lake_root
cfg = LakeConfig(resolve_lake_root(lake_root or None), market)
symbols = _universe_symbols(cfg)
close_df, vol_df = _load_daily_bars(symbols, cfg, start, end)
if close_df.empty:
return pd.Series(dtype=bool)
if detector == "dispersion":
return _dispersion_gate(close_df, threshold)
elif detector == "vol":
return _vol_gate(close_df, vol_low, vol_high)
elif detector == "hmm":
return _hmm_gate(cfg, symbols, threshold, start, end, hmm_field)
else:
raise ValueError(f"Unknown detector: {detector!r}")
def _universe_symbols(cfg) -> list:
"""Read symbols from the lake symbols.parquet."""
import pathlib
sp = cfg.lake_root / "symbols.parquet"
if sp.exists():
df = pd.read_parquet(sp)
col = "symbol" if "symbol" in df.columns else df.columns[0]
return sorted(df[col].astype(str).str.upper().tolist())
return []
def _load_daily_bars(symbols, cfg, start, end):
"""Load daily close prices for all symbols into a wide DataFrame."""
closes = {}
vols = {}
for sym in symbols:
p = cfg.bar_path("1d", sym)
if not p.exists():
continue
try:
df = pd.read_parquet(p)
except Exception:
continue
if not len(df):
continue
tcol = df["t"] if "t" in df.columns else df["date"]
ts = pd.to_datetime(tcol)
df = df.assign(_t=ts).set_index("_t").sort_index()
df = df.loc[start:end]
if len(df) < 22:
continue
closes[sym] = df["c"]
if "v" in df.columns:
vols[sym] = df["v"]
close_df = pd.DataFrame(closes)
vol_df = pd.DataFrame(vols) if vols else None
return close_df, vol_df
def _dispersion_gate(close_df, threshold):
"""Cross-sectional dispersion of 22-day rolling returns."""
if close_df.empty or close_df.shape[1] < 2:
return pd.Series(dtype=bool)
ret = close_df.pct_change(22)
cs_disp = ret.std(axis=1)
gate = cs_disp >= threshold
gate.iloc[:22] = True # warmup: allow trading
return gate
def _vol_gate(close_df, vol_low, vol_high):
"""Cross-sectional mean of 22-day rolling realized vol."""
if close_df.empty or close_df.shape[1] < 2:
return pd.Series(dtype=bool)
import numpy as np
log_ret = np.log(close_df / close_df.shift(1))
rv22 = log_ret.rolling(22).std() * (252 ** 0.5)
cs_mean_vol = rv22.mean(axis=1)
gate = (cs_mean_vol >= vol_low) & (cs_mean_vol <= vol_high)
gate.iloc[:22] = True # warmup
return gate
def _hmm_gate(cfg, symbols, threshold, start, end, hmm_field):
"""HMM regime posterior gate from persisted SP features."""
feat_root = cfg.lake_root / "features"
all_posteriors = {}
for sym in symbols:
# check both ta and sp family paths
for family in ("sp", "ta"):
p = feat_root / f"market=US" / f"timeframe=1d" / f"family={family}" / f"symbol={sym}.parquet"
if not p.exists():
continue
try:
df = pd.read_parquet(p)
except Exception:
continue
if hmm_field not in df.columns:
continue
tcol = df["t"] if "t" in df.columns else df["date"]
ts = pd.to_datetime(tcol)
s = pd.Series(df[hmm_field].values, index=ts, name=sym)
s = s.loc[start:end].dropna()
if len(s) > 0:
all_posteriors[sym] = s
break
if not all_posteriors:
# no HMM features found — default open
idx = pd.date_range(start, end, freq="B")
return pd.Series(True, index=idx)
post_df = pd.DataFrame(all_posteriors)
cs_mean = post_df.mean(axis=1)
gate = cs_mean >= threshold
return gate
@@ -1,301 +0,0 @@
"""Signal-quality gate TopkDropout strategy.
Subclass of ``qlib.contrib.strategy.signal_strategy.TopkDropoutStrategy`` that
holds the book (issues NO orders) when the model's recent prediction accuracy
is below a threshold. When the gate is open it behaves exactly like the
reference TopkDropoutStrategy.
Unlike the regime gate (which asks "is the market calm?"), the signal-quality
gate asks "are my predictions accurate?" — and works across ALL years.
The gate is provided as a precomputed ``pd.Series`` of booleans indexed by
datetime (True = trade allowed). The companion ``compute_signal_quality_gate``
function builds this series from a pred.pkl and lake bars; call it once before
backtesting and pass the result as the ``signal_quality_gate`` parameter.
"""
from __future__ import annotations
import numpy as np
import pandas as pd
from qlib.backtest.decision import TradeDecisionWO
from qlib.contrib.strategy.signal_strategy import TopkDropoutStrategy
__all__ = ["SignalQualityGateStrategy", "compute_signal_quality_gate"]
class SignalQualityGateStrategy(TopkDropoutStrategy):
"""TopkDropout with signal-quality gate overlay.
When ``lake_root`` is provided the gate is computed on-the-fly from the
signal (``<PRED>``) and close prices — no precomputed gate file needed.
This ensures the gate matches the model that is actually generating the
predictions (critical when the model is retrained each year).
Parameters
----------
topk, n_drop, method_sell, method_buy, hold_thresh, only_tradable,
forbid_all_trade_at_limit : same as ``TopkDropoutStrategy``.
signal_quality_gate : pd.Series — precomputed per-date gate (bool).
signal_quality_gate_path : str — path to pickled gate Series.
lake_root : str — lake root for on-the-fly gate computation (preferred).
gate_topk, gate_lookback, gate_threshold : int/float — gate params.
gate_start, gate_end : str — date window for loading close prices.
"""
def __init__(self, *, signal_quality_gate=None, signal_quality_gate_path=None,
lake_root=None, gate_topk=10, gate_lookback=5, gate_threshold=0.5,
gate_start="2015-01-03", gate_end="2026-08-19", **kwargs):
super().__init__(**kwargs)
self._sq_gate_computed = False
if signal_quality_gate is not None:
self._sq_gate = signal_quality_gate
self._sq_gate_computed = True
elif signal_quality_gate_path is not None:
import pickle
with open(signal_quality_gate_path, "rb") as f:
self._sq_gate = pickle.load(f)
self._sq_gate_computed = True
elif lake_root is not None:
self._sq_gate = None
self._lake_root = lake_root
self._gate_topk = gate_topk
self._gate_lookback = gate_lookback
self._gate_threshold = gate_threshold
self._gate_start = gate_start
self._gate_end = gate_end
else:
self._sq_gate = None
def _compute_gate_on_fly(self):
"""Compute gate from the signal (pred.pkl) and lake close prices."""
import pickle
from pathlib import Path
signal_path = self._signal
if not Path(signal_path).exists():
return
with open(signal_path, "rb") as f:
pred = pickle.load(f)
# Handle MultiIndex DataFrame -> unstack to wide
if isinstance(pred, pd.DataFrame) and isinstance(pred.index, pd.MultiIndex):
pred = pred.iloc[:, 0]
pred.index = pd.MultiIndex.from_arrays([
pd.to_datetime(pred.index.get_level_values(0)).normalize(),
pred.index.get_level_values(1)
])
pred = pred.unstack(level=1)
elif isinstance(pred, pd.DataFrame):
pred = pred.iloc[:, 0] if pred.shape[1] >= 1 else pred.squeeze()
pred.index = pd.to_datetime(pred.index).normalize()
# Load close prices from lake
close_df = _load_close_prices(self._lake_root, "US", self._gate_start, self._gate_end)
if close_df.empty:
return
ret_df = close_df.pct_change()
ret_df.index = pd.to_datetime(ret_df.index).normalize()
pred_dates = sorted(pred.index.unique())
if len(pred_dates) < 2:
self._sq_gate = pd.Series(True, index=pd.DatetimeIndex(pred_dates))
self._sq_gate_computed = True
return
hit_rates = {}
for i in range(1, len(pred_dates)):
day = pred_dates[i]
prev_day = pred_dates[i - 1]
try:
prev_scores = pred.loc[prev_day]
except KeyError:
continue
if isinstance(prev_scores, pd.DataFrame):
prev_scores = prev_scores.iloc[:, 0]
prev_scores = prev_scores.dropna().sort_values(ascending=False)
topk_syms = list(prev_scores.index[:self._gate_topk])
if day not in ret_df.index:
continue
today_ret = ret_df.loc[day]
topk_rets = today_ret.reindex(topk_syms).dropna()
if len(topk_rets) == 0:
continue
hit_rates[day] = (topk_rets > 0).sum() / len(topk_rets)
if not hit_rates:
self._sq_gate_computed = True
return
hr_series = pd.Series(hit_rates).sort_index()
rolling_hr = hr_series.rolling(self._gate_lookback, min_periods=1).mean()
gate = rolling_hr >= self._gate_threshold
gate.iloc[:self._gate_lookback] = True
self._sq_gate = gate
self._sq_gate_computed = True
def _gate_open(self, trade_start_time) -> bool:
if self._sq_gate is None:
return True
ts = pd.Timestamp(trade_start_time)
known = self._sq_gate[self._sq_gate.index <= ts]
if len(known):
return bool(known.iloc[-1])
return True # default open if no history yet
def generate_trade_decision(self, execute_result=None):
if not self._sq_gate_computed:
self._compute_gate_on_fly()
trade_step = self.trade_calendar.get_trade_step()
trade_start_time, _ = self.trade_calendar.get_step_time(trade_step)
if not self._gate_open(trade_start_time):
return TradeDecisionWO([], self)
return super().generate_trade_decision(execute_result)
# ---------------------------------------------------------------------------
# Precomputation helper
# ---------------------------------------------------------------------------
def compute_signal_quality_gate(
pred_path: str,
*,
lake_root: str = "",
market: str = "US",
topk: int = 10,
lookback: int = 5,
threshold: float = 0.5,
start: str = "2015-01-03",
end: str = "2026-08-19",
) -> pd.Series:
"""Build a per-date signal-quality gate series from a pred.pkl and lake bars.
For each day, checks whether the model's topk picks from the previous day
had positive returns. Computes a rolling hit rate over ``lookback`` days
and opens the gate when hit rate >= ``threshold``.
Parameters
----------
pred_path : str — path to pred.pkl (from rd_train / rd_predict).
lake_root, market : str — lake location (for loading close prices).
topk : int — number of top picks to track for hit rate.
lookback : int — rolling window for hit rate computation.
threshold : float — hit rate threshold to keep trading.
start, end : str — date window for loading prices.
Returns
-------
pd.Series — bool, indexed by datetime. True = trade allowed.
"""
from pathlib import Path
import pickle
# Load pred.pkl
with open(pred_path, "rb") as f:
pred = pickle.load(f)
# Handle MultiIndex DataFrame (datetime, instrument) -> unstack to wide
if isinstance(pred, pd.DataFrame) and isinstance(pred.index, pd.MultiIndex):
pred = pred.iloc[:, 0] # take score column as Series
pred.index = pd.MultiIndex.from_arrays([
pd.to_datetime(pred.index.get_level_values(0)).normalize(),
pred.index.get_level_values(1)
])
# Unstack to wide: dates x instruments
pred = pred.unstack(level=1)
elif isinstance(pred, pd.DataFrame):
pred = pred.iloc[:, 0] if pred.shape[1] >= 1 else pred.squeeze()
pred.index = pd.to_datetime(pred.index).normalize()
# Load close prices from lake
close_df = _load_close_prices(lake_root, market, start, end)
if close_df.empty:
return pd.Series(dtype=bool)
ret_df = close_df.pct_change()
ret_df.index = pd.to_datetime(ret_df.index).normalize()
# Get sorted unique prediction dates
pred_dates = sorted(pred.index.unique())
if len(pred_dates) < 2:
return pd.Series(True, index=pd.DatetimeIndex(pred_dates))
# Compute hit rates
hit_rates = {}
for i in range(1, len(pred_dates)):
day = pred_dates[i]
# Get yesterday's topk
prev_day = pred_dates[i - 1]
try:
prev_scores = pred.loc[prev_day]
except KeyError:
continue
if isinstance(prev_scores, pd.DataFrame):
prev_scores = prev_scores.iloc[:, 0]
prev_scores = prev_scores.dropna().sort_values(ascending=False)
topk_syms = list(prev_scores.index[:topk])
# Get today's returns
if day not in ret_df.index:
continue
today_ret = ret_df.loc[day]
topk_rets = today_ret.reindex(topk_syms).dropna()
if len(topk_rets) == 0:
continue
hit_rates[day] = (topk_rets > 0).sum() / len(topk_rets)
if not hit_rates:
return pd.Series(dtype=bool)
hr_series = pd.Series(hit_rates).sort_index()
# Rolling hit rate
rolling_hr = hr_series.rolling(lookback, min_periods=1).mean()
# Gate is open when rolling hit rate >= threshold
gate = rolling_hr >= threshold
gate.iloc[:lookback] = True # warmup: allow trading
return gate
def _load_close_prices(lake_root, market, start, end):
"""Load daily close prices for all symbols into a wide DataFrame."""
from pathlib import Path
lake = Path(lake_root)
symbols_parquet = lake / "symbols.parquet"
if not symbols_parquet.exists():
return pd.DataFrame()
df = pd.read_parquet(symbols_parquet)
col = "symbol" if "symbol" in df.columns else df.columns[0]
symbols = sorted(df[col].astype(str).str.upper().tolist())
closes = {}
for sym in symbols:
p = lake / "market=US" / "timeframe=1d" / f"symbol={sym}.parquet"
if not p.exists():
continue
try:
bar = pd.read_parquet(p)
except Exception:
continue
if not len(bar):
continue
tcol = "t" if "t" in bar.columns else "date"
ts = pd.to_datetime(bar[tcol])
bar = bar.assign(_t=ts).set_index("_t").sort_index()
bar = bar.loc[start:end]
if len(bar) < 10:
continue
closes[sym] = bar["c"] if "c" in bar.columns else bar["close"]
return pd.DataFrame(closes)
@@ -1,202 +0,0 @@
"""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)
-25
View File
@@ -1,25 +0,0 @@
from .config import (
LakeConfig,
BAR_FIELD_MAP,
FREQ_TO_TIMEFRAME,
UNKNOWN_FIELD_NAMES,
timeframe_for_freq,
resolve_lake_root,
)
from .providers import (
LakeCalendarProvider,
LakeInstrumentProvider,
LakeFeatureProvider,
)
__all__ = [
"LakeConfig",
"BAR_FIELD_MAP",
"FREQ_TO_TIMEFRAME",
"UNKNOWN_FIELD_NAMES",
"timeframe_for_freq",
"resolve_lake_root",
"LakeCalendarProvider",
"LakeInstrumentProvider",
"LakeFeatureProvider",
]
-202
View File
@@ -1,202 +0,0 @@
"""TradeAC lake configuration helpers.
The lake is a hive-partitioned parquet store (see ``tac-engine/skills/tradeac-lake``):
$TAC_LAKE_DIR/
├── market=US/
│ └── timeframe=1d/
│ └── symbol=AAPL.parquet # OHLCV bars: t, date, o, h, l, c, v, n, vw
├── features/ # indicators, wide format, family tier
│ └── market=US/
│ └── timeframe=1d/
│ ├── 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
"""
from __future__ import annotations
import os
from pathlib import Path
from typing import Dict, List, Optional
import pandas as pd
#: qlib freq string (Freq.__str__) -> lake timeframe partition name
FREQ_TO_TIMEFRAME: Dict[str, str] = {
"day": "1d",
"1d": "1d",
"min": "1m",
"1min": "1m",
"5min": "5m",
"10min": "10m",
"15min": "15m",
"30min": "30m",
"hour": "1h",
"1hour": "1h",
"2hour": "2h",
"4hour": "4h",
"week": "1w",
"1week": "1w",
"month": "1M",
"1month": "1M",
}
#: bar-field map: qlib field name (without the leading ``$``) -> lake bar column
BAR_FIELD_MAP: Dict[str, str] = {
"open": "o",
"high": "h",
"low": "l",
"close": "c",
"volume": "v",
"vwap": "vw",
"avg_amount": "vw", # amount / volume
}
#: fields that qlib core/backtest queries but the lake does not store -> all-NaN
UNKNOWN_FIELD_NAMES = ("factor", "change", "trade_unit", "suspend_flag")
#: columns in the parquet files that are not features
NON_FEATURE_COLUMNS = ("t", "date", "market", "timeframe", "symbol")
def timeframe_for_freq(freq: str) -> str:
"""Map a qlib frequency (e.g. ``day``, ``1min``) to a lake timeframe (e.g. ``1d``)."""
f = str(freq).lower()
if f not in FREQ_TO_TIMEFRAME:
raise ValueError(
f"unsupported qlib freq {freq!r}; supported freqs: {sorted(set(FREQ_TO_TIMEFRAME))}"
)
return FREQ_TO_TIMEFRAME[f]
def resolve_lake_root(lake_root: Optional[str] = None) -> Path:
"""Resolve the lake root: explicit arg > ``TAC_LAKE_DIR`` (no fallback).
``TAC_LAKE_DIR`` is **mandatory** — there is deliberately no default
A missing/empty value raises so a
misconfigured environment never silently points at a wrong directory.
"""
if lake_root is None:
lake_root = os.environ.get("TAC_LAKE_DIR")
if not lake_root:
raise RuntimeError(
"TAC_LAKE_DIR is not set. Point it at the TradeAC lake root, e.g. "
"export TAC_LAKE_DIR=/home/data/lake (docker) or set an absolute "
"path in your local .env."
)
return Path(str(lake_root)).expanduser().resolve()
class LakeConfig:
"""Path helpers + cached readers for a (lake_root, market) combination."""
def __init__(self, lake_root: Optional[str] = None, market: str = "US"):
self.lake_root: Path = resolve_lake_root(lake_root)
self.market: str = (market or "US").upper()
# ---- paths --------------------------------------------------------------
def bar_dir(self, timeframe: str) -> Path:
return self.lake_root / f"market={self.market}" / f"timeframe={timeframe}"
def bar_path(self, timeframe: str, symbol: str) -> Path:
return self.bar_dir(timeframe) / f"symbol={str(symbol).upper()}.parquet"
def features_dir(self, timeframe: str) -> Path:
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"
def symbols_path(self) -> Path:
return self.lake_root / "symbols.parquet"
def coverage_path(self) -> Path:
return self.lake_root / "coverage.parquet"
# ---- metadata readers ----------------------------------------------------
def load_symbols(self) -> List[str]:
"""All symbols known to the lake (from ``symbols.parquet``)."""
p = self.symbols_path()
if not p.exists():
return []
df = pd.read_parquet(p)
if "symbol" not in df.columns:
return []
return sorted(df["symbol"].astype(str).str.upper().tolist())
def symbol_spans(self, symbol: str, timeframe: str) -> List[tuple]:
"""Listing span(s) ``[(start_iso, end_iso)]`` for a symbol from coverage.parquet."""
p = self.coverage_path()
if p.exists():
try:
df = pd.read_parquet(p)
except Exception: # pragma: no cover - defensive
df = pd.DataFrame()
if len(df):
df = df[
(df.get("market") == self.market)
& (df.get("timeframe") == timeframe)
& (df.get("symbol") == str(symbol).upper())
]
if len(df):
row = df.iloc[0]
first = pd.Timestamp(row["first_t"]).date()
last = pd.Timestamp(row["last_t"]).date()
return [(first.isoformat(), last.isoformat())]
# fallback: derive from the bar file itself
p = self.bar_path(timeframe, symbol)
if p.exists():
import pyarrow.parquet as pq
tbl = pq.read_table(p, columns=["t"])
first = pd.Timestamp(tbl.column("t")[0].as_py()).date()
last = pd.Timestamp(tbl.column("t")[-1].as_py()).date()
return [(first.isoformat(), last.isoformat())]
return [("1970-01-01", "2099-12-31")]
def load_calendar_dates(self) -> List[pd.Timestamp]:
"""Trading days (midnight timestamps) for the market, from ``calendar.parquet``."""
p = self.calendar_path()
if p.exists():
df = pd.read_parquet(p)
if "date" in df.columns:
if "market" in df.columns:
df = df[df["market"] == self.market]
dates = pd.to_datetime(df["date"]).dt.normalize().sort_values().unique()
return [pd.Timestamp(x) for x in dates]
return []
def __repr__(self) -> str: # pragma: no cover
return f"LakeConfig(lake_root={self.lake_root}, market={self.market})"
-230
View File
@@ -1,230 +0,0 @@
"""qlib data providers backed by the TradeAC parquet lake.
These providers plug into the standard qlib mechanism: ``qlib.init(calendar_provider=...,
instrument_provider=..., feature_provider=...)`` instantiates them and binds them to the
``Cal`` / ``Inst`` / ``FeatureD`` wrappers (see ``qlib.data.data.register_all_wrappers``).
The rest of qlib (``LocalDatasetProvider`` expression engine, backtest ``Exchange``) keeps
working unchanged because the interface contract is identical to the file-based providers:
- ``feature()`` returns a ``pd.Series`` indexed by the **calendar position** range
``[start_index, end_index]`` (matching ``FileFeatureStorage.__getitem__`` semantics).
- ``list_instruments()`` returns ``{symbol: [(start, end), ...]}``.
- ``load_calendar()`` returns a list of ``pd.Timestamp`` trading days.
"""
from __future__ import annotations
import bisect
from typing import Dict, List, Optional, Union
import numpy as np
import pandas as pd
from qlib.data.data import CalendarProvider, FeatureProvider, InstrumentProvider
from qlib.log import get_module_logger
from .config import (
BAR_FIELD_MAP,
LakeConfig,
UNKNOWN_FIELD_NAMES,
timeframe_for_freq,
)
logger = get_module_logger("tac_qlib.data.providers")
def _day_freq(freq: str) -> bool:
return str(freq).lower() in ("day", "1d")
def _calendar_keys(cal: List[pd.Timestamp], freq: str) -> pd.Index:
"""Convert calendar timestamps into the same key space as the lake parquet."""
if _day_freq(freq):
return pd.Index([pd.Timestamp(x).date() for x in cal])
return pd.Index([pd.Timestamp(x) for x in cal])
class LakeCalendarProvider(CalendarProvider):
"""Trading calendar read from ``<lake>/calendar.parquet`` (fallback: derived from bars)."""
def __init__(self, lake_root: Optional[str] = None, market: str = "US"):
super().__init__()
self.cfg = LakeConfig(lake_root, market)
def load_calendar(self, freq, future):
timeframe = timeframe_for_freq(freq)
if not _day_freq(freq):
raise NotImplementedError(
f"freq={freq!r} (timeframe={timeframe}) is not supported yet: the lake calendar "
f"only covers daily sessions; add a minute-level calendar to `calendar.parquet`"
)
dates = self.cfg.load_calendar_dates()
if not dates:
# Fallback: derive the trading-day set from the persisted bar files.
bar_dir = self.cfg.bar_dir(timeframe)
if bar_dir.exists():
import pyarrow.parquet as pq
cal: Dict[pd.Timestamp, None] = {}
for p in sorted(bar_dir.glob("symbol=*.parquet")):
tbl = pq.read_table(p, columns=["t"])
for v in tbl.column("t"):
cal[pd.Timestamp(v.as_py()).normalize()] = None
dates = sorted(cal.keys())
if not dates:
return []
if future:
# append the next calendar day so that "today" is a valid trade date
last = dates[-1]
dates = dates + [pd.Timestamp(last) + pd.Timedelta(days=1)]
return dates
class LakeInstrumentProvider(InstrumentProvider):
"""Instruments from ``<lake>/symbols.parquet`` with listing spans from ``coverage.parquet``."""
def __init__(
self,
lake_root: Optional[str] = None,
market: str = "US",
markets: Optional[Dict[str, list]] = None,
):
super().__init__()
self.cfg = LakeConfig(lake_root, market)
#: optional named pools, e.g. ``{"sp500": ["AAPL", "MSFT"], "etf": ["SPY"]}``.
#: ``all`` / any unregistered name resolves to every symbol in the lake.
self.markets: Dict[str, list] = markets or {}
def _resolve_symbols(self, market: Union[str, list]) -> List[str]:
if isinstance(market, (list, tuple, pd.Index, np.ndarray)):
return [str(s).upper() for s in market]
if isinstance(market, str) and "," in market:
return [s.strip().upper() for s in market.split(",") if s.strip()]
if market in self.markets:
return [str(s).upper() for s in self.markets[market]]
return self.cfg.load_symbols()
def list_instruments(self, instruments, start_time=None, end_time=None, freq="day", as_list=False):
market = instruments["market"]
timeframe = timeframe_for_freq(freq)
symbols = self._resolve_symbols(market)
if not symbols:
if as_list:
return []
return {}
# clip listing spans to the queried window (mirror of LocalInstrumentProvider)
from qlib.data.data import Cal # pylint: disable=C0415
cal = Cal.calendar(freq=freq)
start_time = pd.Timestamp(start_time or cal[0])
end_time = pd.Timestamp(end_time or cal[-1])
out: Dict[str, list] = {}
for symbol in symbols:
spans = []
for begin, end in self.cfg.symbol_spans(symbol, timeframe):
lo = max(start_time, pd.Timestamp(begin))
hi = min(end_time, pd.Timestamp(end))
if lo <= hi:
spans.append((lo, hi))
if spans:
out[symbol] = spans
filter_pipe = instruments.get("filter_pipe") or []
for filter_config in filter_pipe:
from qlib.data import filter as F # pylint: disable=C0415
filter_t = getattr(F, filter_config["filter_type"]).from_config(filter_config)
out = filter_t(out, start_time, end_time, freq)
if as_list:
return list(out)
return out
class LakeFeatureProvider(FeatureProvider):
"""Feature data from the lake parquet (OHLCV bars + pre-computed ta-lib features).
Field routing:
- ``$open/$high/$low/$close/$volume/$vwap`` -> bar parquet columns
- ``$amount`` (= v*vw), ``$avg_amount`` (= vw) -> derived from bar parquet
- ``$factor/$change/...`` -> all-NaN (not stored)
- anything else -> a ta-lib column in the features parquet
"""
def __init__(self, lake_root: Optional[str] = None, market: str = "US"):
super().__init__()
self.cfg = LakeConfig(lake_root, market)
self._bar_cache: Dict[tuple, pd.DataFrame] = {}
self._feature_cache: Dict[tuple, pd.DataFrame] = {}
# ------------------------------------------------------------------ caches
def _load_bar_df(self, instrument: str, timeframe: str) -> pd.DataFrame:
key = (instrument, timeframe)
if key not in self._bar_cache:
p = self.cfg.bar_path(timeframe, instrument)
self._bar_cache[key] = pd.read_parquet(p) if p.exists() else pd.DataFrame()
return self._bar_cache[key]
def _load_feature_df(self, instrument: str, timeframe: str) -> pd.DataFrame:
key = (instrument, timeframe)
if key not in self._feature_cache:
self._feature_cache[key] = self.cfg.load_features(timeframe, instrument)
return self._feature_cache[key]
@staticmethod
def _keys(df: pd.DataFrame, freq: str) -> pd.Index:
ts = pd.to_datetime(df["t"])
return ts.dt.date if _day_freq(freq) else ts
# ------------------------------------------------------------------ fields
def _extract(self, instrument: str, field: str, timeframe: str, freq: str) -> Optional[pd.Series]:
"""Return the field as a Series keyed by date/timestamp (None if not present in the lake)."""
bar = self._load_bar_df(instrument, timeframe)
if field in BAR_FIELD_MAP:
col = BAR_FIELD_MAP[field]
if col in bar.columns:
return bar[col].astype(float).set_axis(self._keys(bar, freq))
return None
if field == "amount":
if "v" in bar.columns and "vw" in bar.columns:
return (bar["v"] * bar["vw"]).astype(float).set_axis(self._keys(bar, freq))
return None
if field in UNKNOWN_FIELD_NAMES:
return None
feat = self._load_feature_df(instrument, timeframe)
if field in feat.columns:
return feat[field].astype(float).set_axis(self._keys(feat, freq))
return None
# ------------------------------------------------------------------ api
def _get_calendar(self, freq: str) -> List[pd.Timestamp]:
from qlib.data.data import Cal # pylint: disable=C0415
cal = Cal.calendar(freq=freq)
return list(cal)
def feature(self, instrument, field, start_index, end_index, freq):
field = str(field)[1:]
timeframe = timeframe_for_freq(freq)
cal = self._get_calendar(freq)
n = len(cal)
lo = max(0, int(start_index))
hi = min(n - 1, int(end_index))
if lo > hi:
return pd.Series(dtype=np.float32)
keys = _calendar_keys(cal[lo : hi + 1], freq)
ser = self._extract(str(instrument).upper(), field, timeframe, freq)
if ser is None:
vals = np.full(len(keys), np.nan, dtype=np.float64)
else:
vals = ser.reindex(keys).to_numpy(dtype=np.float64)
return pd.Series(vals, index=pd.RangeIndex(lo, hi + 1))
@@ -1,16 +1,16 @@
# -----------------------------------------------------------------------------
# Signal-quality gate: TopkDropout gated by rolling hit-rate of topk picks.
# ABLATION A (baseline): LightGBM with RankIC early-stopping on the 50-ETF SP-5d
# panel, using ALL 24 sp_* feature columns (ou,hmm,jump,har,trend,hurst,
# signature). Copy of the canonical workflow_lgb_sp5d_rankic.yaml with a
# distinct experiment name so the ablation runs are isolated.
#
# 1. Compute the gate: python precompute_signal_quality_gate.py <pred.pkl> <gate.pkl>
# 2. Run this workflow: rd_run_workflow config_path=<this yaml> experiment_name=<exp>
#
# The strategy loads the precomputed gate from signal_quality_gate_path.
# When hit rate >= threshold, trade; otherwise, go to cash.
# Run:
# rd_run_workflow config_path=tac-qlib/workflows/ablate_baseline_all_sp_fields.yaml \
# experiment_name=tac-rd-rank-ablate
# -----------------------------------------------------------------------------
{%- set LAKE = TAC_LAKE_DIR %}
{%- set UNIVERSE = "SPY,QQQ,DIA,IWM,MDY,VTI,VOO,VEA,VWO,VT,EFA,EEM,TLT,IEF,SHY,AGG,BND,LQD,HYG,JNK,EMB,GLD,SLV,USO,UNG,DBA,DBC,XLK,XLF,XLE,XLV,XLI,XLY,XLP,XLU,XLB,XLRE,ARKK,SMH,SOXX,IBB,XBI,ITA,XAR,ICLN,TAN,FDN,IGV,ESPO,REM" %}
{%- set SP_FIELDS = "sp_ret,sp_ou_zscore,sp_ou_half_life,sp_ou_revert,sp_hmm_p_regime1,sp_hmm_state,sp_jump_ratio,sp_jump_flag,sp_jump_tail,sp_max_move,sp_rv1,sp_rv5,sp_rv22,sp_vol_ratio_5_22,sp_vol_ratio_1_22,sp_trend_slope_5,sp_trend_slope_20,sp_trend_slope_60,sp_logp,sp_hurst_exponent,sp_sig_level1_lead,sp_sig_level1_lag,sp_sig_level2_lead_lag,sp_sig_level2_lag_lead" %}
{%- set GATE_PATH = "/app/experiments/book/data/signal_quality_gate/sq_gate_5d_0.50.pkl" %}
qlib_init:
provider_uri: "{{ LAKE }}"
@@ -40,22 +40,27 @@ qlib_init:
module_path: qlib.workflow.expm
kwargs:
uri: "sqlite:///{{ LAKE }}/mlruns.db"
default_exp_name: "tac-rd-sq-gate"
default_exp_name: "tac-rd-rank-ablate"
task:
model:
class: LGBModel
module_path: qlib.contrib.model.gbdt
class: RankICLGBModel
module_path: tac_qlib.contrib.model.rank_gbdt
kwargs:
loss: mse
learning_rate: 0.05
num_leaves: 15
n_estimators: 200
learning_rate: 0.02
num_leaves: 31
n_estimators: 3000
num_boost_round: 3000
early_stopping_rounds: 200
min_data_in_leaf: 20
lambda_l2: 0.5
colsample_bytree: 0.8
subsample: 0.8
subsample_freq: 1
reg_alpha: 0.01
reg_lambda: 0.01
reg_alpha: 0.1
reg_lambda: 1.0
seed: 42
dataset:
class: DatasetH
@@ -105,13 +110,12 @@ task:
kwargs:
config:
strategy:
class: SignalQualityGateStrategy
module_path: tac_qlib.contrib.strategy.signal_quality_gate
class: TopkDropoutStrategy
module_path: qlib.contrib.strategy
kwargs:
signal: "<PRED>"
signal_quality_gate_path: "{{ GATE_PATH }}"
topk: 10
n_drop: 1
n_drop: 2
only_tradable: true
risk_degree: 0.95
backtest:
@@ -1,44 +1,67 @@
# Signal-quality gate backtest for 2026
# -----------------------------------------------------------------------------
# ABLATION B (generic-only): same panel/model as the baseline, but feature
# fields restricted to the model-free / generic stochastic-process families
# (jump,har,trend,hurst,signature). Drops the model-specific ou (OU/AR-1
# half-life) and hmm (2-state regime) families to test whether the generic
# families alone dominate the rank dimension.
#
# Run:
# rd_run_workflow config_path=tac-qlib/workflows/ablate_generic_only_sp_fields.yaml \
# experiment_name=tac-rd-rank-ablate
# -----------------------------------------------------------------------------
{%- set LAKE = TAC_LAKE_DIR %}
{%- set UNIVERSE = "SPY,QQQ,DIA,IWM,MDY,VTI,VOO,VEA,VWO,VT,EFA,EEM,TLT,IEF,SHY,AGG,BND,LQD,HYG,JNK,EMB,GLD,SLV,USO,UNG,DBA,DBC,XLK,XLF,XLE,XLV,XLI,XLY,XLP,XLU,XLB,XLRE,ARKK,SMH,SOXX,IBB,XBI,ITA,XAR,ICLN,TAN,FDN,IGV,ESPO,REM" %}
{%- set SP_FIELDS = "sp_ret,sp_ou_zscore,sp_ou_half_life,sp_ou_revert,sp_hmm_p_regime1,sp_hmm_state,sp_jump_ratio,sp_jump_flag,sp_jump_tail,sp_max_move,sp_rv1,sp_rv5,sp_rv22,sp_vol_ratio_5_22,sp_vol_ratio_1_22,sp_trend_slope_5,sp_trend_slope_20,sp_trend_slope_60,sp_logp,sp_hurst_exponent,sp_sig_level1_lead,sp_sig_level1_lag,sp_sig_level2_lead_lag,sp_sig_level2_lag_lead" %}
{# Gate computed on-the-fly from signal + lake close prices #}
{%- set SP_FIELDS = "sp_ret,sp_jump_ratio,sp_jump_flag,sp_jump_tail,sp_max_move,sp_rv1,sp_rv5,sp_rv22,sp_vol_ratio_5_22,sp_vol_ratio_1_22,sp_trend_slope_5,sp_trend_slope_20,sp_trend_slope_60,sp_logp,sp_hurst_exponent,sp_sig_level1_lead,sp_sig_level1_lag,sp_sig_level2_lead_lag,sp_sig_level2_lag_lead" %}
qlib_init:
provider_uri: "{{ LAKE }}"
region: us
expression_cache: null
dataset_cache: null
calendar_provider:
class: tac_qlib.data.providers.LakeCalendarProvider
kwargs: { lake_root: "{{ LAKE }}", market: US }
kwargs:
lake_root: "{{ LAKE }}"
market: US
instrument_provider:
class: tac_qlib.data.providers.LakeInstrumentProvider
kwargs: { lake_root: "{{ LAKE }}", market: US, markets: {} }
kwargs:
lake_root: "{{ LAKE }}"
market: US
markets: {}
feature_provider:
class: tac_qlib.data.providers.LakeFeatureProvider
kwargs: { lake_root: "{{ LAKE }}", market: US }
kwargs:
lake_root: "{{ LAKE }}"
market: US
exp_manager:
class: MLflowExpManager
module_path: qlib.workflow.expm
kwargs:
uri: "sqlite:///{{ LAKE }}/mlruns.db"
default_exp_name: "tac-rd-sq-gate-2026"
default_exp_name: "tac-rd-rank-ablate"
task:
model:
class: LGBModel
module_path: qlib.contrib.model.gbdt
class: RankICLGBModel
module_path: tac_qlib.contrib.model.rank_gbdt
kwargs:
loss: mse
learning_rate: 0.05
num_leaves: 15
n_estimators: 200
learning_rate: 0.02
num_leaves: 31
n_estimators: 3000
num_boost_round: 3000
early_stopping_rounds: 200
min_data_in_leaf: 20
lambda_l2: 0.5
colsample_bytree: 0.8
subsample: 0.8
subsample_freq: 1
reg_alpha: 0.01
reg_lambda: 0.01
reg_alpha: 0.1
reg_lambda: 1.0
seed: 42
dataset:
class: DatasetH
@@ -52,7 +75,7 @@ task:
start_time: 2015-01-03
end_time: 2026-08-10
fit_start_time: 2015-01-03
fit_end_time: 2026-01-03
fit_end_time: 2025-09-01
freq: day
lake_root: "{{ LAKE }}"
market: US
@@ -80,24 +103,20 @@ task:
kwargs: {}
- class: SigAnaRecord
module_path: qlib.workflow.record_temp
kwargs: { ana_long_short: true, ann_scaler: 252 }
kwargs:
ana_long_short: true
ann_scaler: 252
- class: PortAnaRecord
module_path: qlib.workflow.record_temp
kwargs:
config:
strategy:
class: SignalQualityGateStrategy
module_path: tac_qlib.contrib.strategy.signal_quality_gate
class: TopkDropoutStrategy
module_path: qlib.contrib.strategy
kwargs:
signal: "<PRED>"
lake_root: "{{ LAKE }}"
gate_topk: 10
gate_lookback: 5
gate_threshold: 0.5
gate_start: "2015-01-03"
gate_end: "2026-08-19"
topk: 10
n_drop: 1
n_drop: 2
only_tradable: true
risk_degree: 0.95
backtest:
@@ -1,44 +1,74 @@
# Signal-quality gate backtest for 2023
# -----------------------------------------------------------------------------
# ISOLATION: multi-seed RankIC ensemble, ablate-B generic-only feature set.
#
# Isolates the ensemble effect on the SP-5d rank signal. Same panel, segments,
# history (full backfilled 2016+) and feature set as the exp-9 ablate-B winner
# (generic-only sp_* families: jump,har,trend,hurst,signature), but replaces the
# single RankICLGBModel with a 5-seed RankICEnsembleLGBModel (42,7,2026,99,123)
# that averages per-day predictions.
#
# Differs from exp-15 (tac-rd-rank-ensemble, mlflow exp 15) ONLY by dropping the
# TA subset (rsi_14,roc_10,macd_hist,willr_14,atr_14) and the inter-asset xr_*
# features, so any change vs exp-15 is attributable to the feature set alone,
# and any change vs exp-9 is attributable to the ensemble + full history alone.
#
# Run:
# rd_run_workflow config_path=experiments/workflows/exp12_isolation_ensemble.yaml \
# experiment_name=tac-rd-rank-ensemble-isolated
# -----------------------------------------------------------------------------
{%- set LAKE = TAC_LAKE_DIR %}
{%- set UNIVERSE = "SPY,QQQ,DIA,IWM,MDY,VTI,VOO,VEA,VWO,VT,EFA,EEM,TLT,IEF,SHY,AGG,BND,LQD,HYG,JNK,EMB,GLD,SLV,USO,UNG,DBA,DBC,XLK,XLF,XLE,XLV,XLI,XLY,XLP,XLU,XLB,XLRE,ARKK,SMH,SOXX,IBB,XBI,ITA,XAR,ICLN,TAN,FDN,IGV,ESPO,REM" %}
{%- set SP_FIELDS = "sp_ret,sp_ou_zscore,sp_ou_half_life,sp_ou_revert,sp_hmm_p_regime1,sp_hmm_state,sp_jump_ratio,sp_jump_flag,sp_jump_tail,sp_max_move,sp_rv1,sp_rv5,sp_rv22,sp_vol_ratio_5_22,sp_vol_ratio_1_22,sp_trend_slope_5,sp_trend_slope_20,sp_trend_slope_60,sp_logp,sp_hurst_exponent,sp_sig_level1_lead,sp_sig_level1_lag,sp_sig_level2_lead_lag,sp_sig_level2_lag_lead" %}
{# Gate computed on-the-fly from signal + lake close prices #}
{%- set SP_FIELDS = "sp_ret,sp_jump_ratio,sp_jump_flag,sp_jump_tail,sp_max_move,sp_rv1,sp_rv5,sp_rv22,sp_vol_ratio_5_22,sp_vol_ratio_1_22,sp_trend_slope_5,sp_trend_slope_20,sp_trend_slope_60,sp_logp,sp_hurst_exponent,sp_sig_level1_lead,sp_sig_level1_lag,sp_sig_level2_lead_lag,sp_sig_level2_lag_lead" %}
qlib_init:
provider_uri: "{{ LAKE }}"
region: us
expression_cache: null
dataset_cache: null
calendar_provider:
class: tac_qlib.data.providers.LakeCalendarProvider
kwargs: { lake_root: "{{ LAKE }}", market: US }
kwargs:
lake_root: "{{ LAKE }}"
market: US
instrument_provider:
class: tac_qlib.data.providers.LakeInstrumentProvider
kwargs: { lake_root: "{{ LAKE }}", market: US, markets: {} }
kwargs:
lake_root: "{{ LAKE }}"
market: US
markets: {}
feature_provider:
class: tac_qlib.data.providers.LakeFeatureProvider
kwargs: { lake_root: "{{ LAKE }}", market: US }
kwargs:
lake_root: "{{ LAKE }}"
market: US
exp_manager:
class: MLflowExpManager
module_path: qlib.workflow.expm
kwargs:
uri: "sqlite:///{{ LAKE }}/mlruns.db"
default_exp_name: "tac-rd-sq-gate-2023"
uri: "sqlite:///mlruns.db"
default_exp_name: "tac-rd-rank-ensemble-isolated"
task:
model:
class: LGBModel
module_path: qlib.contrib.model.gbdt
class: RankICEnsembleLGBModel
module_path: tac_qlib.contrib.model.rank_ensemble
kwargs:
loss: mse
learning_rate: 0.05
num_leaves: 15
n_estimators: 200
learning_rate: 0.02
num_leaves: 31
n_estimators: 3000
num_boost_round: 3000
early_stopping_rounds: 200
min_data_in_leaf: 20
lambda_l2: 0.5
colsample_bytree: 0.8
subsample: 0.8
subsample_freq: 1
reg_alpha: 0.01
reg_lambda: 0.01
reg_alpha: 0.1
reg_lambda: 1.0
seeds: "42,7,2026,99,123"
dataset:
class: DatasetH
@@ -50,9 +80,9 @@ task:
kwargs:
instruments: "{{ UNIVERSE }}"
start_time: 2015-01-03
end_time: 2023-12-29
fit_start_time: 2015-01-03
fit_end_time: 2023-01-03
end_time: 2026-08-14
fit_start_time: 2016-01-04
fit_end_time: 2025-09-01
freq: day
lake_root: "{{ LAKE }}"
market: US
@@ -70,9 +100,9 @@ task:
- class: Fillna
kwargs: {}
segments:
train: [2015-01-03, 2022-09-01]
valid: [2022-09-03, 2023-01-03]
test: [2023-01-03, 2023-12-29]
train: [2016-01-04, 2025-09-01]
valid: [2025-09-03, 2026-01-03]
test: [2026-01-04, 2026-08-10]
record:
- class: SignalRecord
@@ -80,29 +110,25 @@ task:
kwargs: {}
- class: SigAnaRecord
module_path: qlib.workflow.record_temp
kwargs: { ana_long_short: true, ann_scaler: 252 }
kwargs:
ana_long_short: true
ann_scaler: 252
- class: PortAnaRecord
module_path: qlib.workflow.record_temp
kwargs:
config:
strategy:
class: SignalQualityGateStrategy
module_path: tac_qlib.contrib.strategy.signal_quality_gate
class: TopkDropoutStrategy
module_path: qlib.contrib.strategy
kwargs:
signal: "<PRED>"
lake_root: "{{ LAKE }}"
gate_topk: 10
gate_lookback: 5
gate_threshold: 0.5
gate_start: "2015-01-03"
gate_end: "2023-12-29"
topk: 10
n_drop: 1
n_drop: 2
only_tradable: true
risk_degree: 0.95
backtest:
start_time: 2023-01-03
end_time: 2023-12-29
start_time: 2026-01-04
end_time: 2026-08-10
account: 1000000
benchmark: SPY
exchange_kwargs:
@@ -1,44 +1,70 @@
# Signal-quality gate backtest for 2025
# -----------------------------------------------------------------------------
# OPTSTOP VARIANT of the isolation winner (exp 12 / mlflow exp 16).
#
# Same signal pipeline as exp-12 exactly: ablate-B generic-only sp_* features
# (jump,har,trend,hurst,signature) on full backfilled 2016+ history, 5-seed
# RankICEnsembleLGBModel (42,7,2026,99,123). ONLY the portfolio construction
# changes: TopkDropout (topk=10 n_drop=2) -> OptimalStopControl stochastic
# control strategy (percentile-gated entry, optimal-stopping exits by
# percentile/time/stop-loss, equal-weight notional).
#
# Run:
# rd_run_workflow config_path=experiments/workflows/exp13_optstop_isolation.yaml \
# experiment_name=tac-rd-rank-ensemble-optstop
# -----------------------------------------------------------------------------
{%- set LAKE = TAC_LAKE_DIR %}
{%- set UNIVERSE = "SPY,QQQ,DIA,IWM,MDY,VTI,VOO,VEA,VWO,VT,EFA,EEM,TLT,IEF,SHY,AGG,BND,LQD,HYG,JNK,EMB,GLD,SLV,USO,UNG,DBA,DBC,XLK,XLF,XLE,XLV,XLI,XLY,XLP,XLU,XLB,XLRE,ARKK,SMH,SOXX,IBB,XBI,ITA,XAR,ICLN,TAN,FDN,IGV,ESPO,REM" %}
{%- set SP_FIELDS = "sp_ret,sp_ou_zscore,sp_ou_half_life,sp_ou_revert,sp_hmm_p_regime1,sp_hmm_state,sp_jump_ratio,sp_jump_flag,sp_jump_tail,sp_max_move,sp_rv1,sp_rv5,sp_rv22,sp_vol_ratio_5_22,sp_vol_ratio_1_22,sp_trend_slope_5,sp_trend_slope_20,sp_trend_slope_60,sp_logp,sp_hurst_exponent,sp_sig_level1_lead,sp_sig_level1_lag,sp_sig_level2_lead_lag,sp_sig_level2_lag_lead" %}
{# Gate computed on-the-fly from signal + lake close prices #}
{%- set SP_FIELDS = "sp_ret,sp_jump_ratio,sp_jump_flag,sp_jump_tail,sp_max_move,sp_rv1,sp_rv5,sp_rv22,sp_vol_ratio_5_22,sp_vol_ratio_1_22,sp_trend_slope_5,sp_trend_slope_20,sp_trend_slope_60,sp_logp,sp_hurst_exponent,sp_sig_level1_lead,sp_sig_level1_lag,sp_sig_level2_lead_lag,sp_sig_level2_lag_lead" %}
qlib_init:
provider_uri: "{{ LAKE }}"
region: us
expression_cache: null
dataset_cache: null
calendar_provider:
class: tac_qlib.data.providers.LakeCalendarProvider
kwargs: { lake_root: "{{ LAKE }}", market: US }
kwargs:
lake_root: "{{ LAKE }}"
market: US
instrument_provider:
class: tac_qlib.data.providers.LakeInstrumentProvider
kwargs: { lake_root: "{{ LAKE }}", market: US, markets: {} }
kwargs:
lake_root: "{{ LAKE }}"
market: US
markets: {}
feature_provider:
class: tac_qlib.data.providers.LakeFeatureProvider
kwargs: { lake_root: "{{ LAKE }}", market: US }
kwargs:
lake_root: "{{ LAKE }}"
market: US
exp_manager:
class: MLflowExpManager
module_path: qlib.workflow.expm
kwargs:
uri: "sqlite:///{{ LAKE }}/mlruns.db"
default_exp_name: "tac-rd-sq-gate-2025"
uri: "sqlite:///mlruns.db"
default_exp_name: "tac-rd-rank-ensemble-optstop"
task:
model:
class: LGBModel
module_path: qlib.contrib.model.gbdt
class: RankICEnsembleLGBModel
module_path: tac_qlib.contrib.model.rank_ensemble
kwargs:
loss: mse
learning_rate: 0.05
num_leaves: 15
n_estimators: 200
learning_rate: 0.02
num_leaves: 31
n_estimators: 3000
num_boost_round: 3000
early_stopping_rounds: 200
min_data_in_leaf: 20
lambda_l2: 0.5
colsample_bytree: 0.8
subsample: 0.8
subsample_freq: 1
reg_alpha: 0.01
reg_lambda: 0.01
reg_alpha: 0.1
reg_lambda: 1.0
seeds: "42,7,2026,99,123"
dataset:
class: DatasetH
@@ -50,9 +76,9 @@ task:
kwargs:
instruments: "{{ UNIVERSE }}"
start_time: 2015-01-03
end_time: 2025-12-31
fit_start_time: 2015-01-03
fit_end_time: 2026-01-03
end_time: 2026-08-14
fit_start_time: 2016-01-04
fit_end_time: 2025-09-01
freq: day
lake_root: "{{ LAKE }}"
market: US
@@ -70,9 +96,9 @@ task:
- class: Fillna
kwargs: {}
segments:
train: [2015-01-03, 2025-09-01]
train: [2016-01-04, 2025-09-01]
valid: [2025-09-03, 2026-01-03]
test: [2025-01-02, 2025-12-31]
test: [2026-01-04, 2026-08-10]
record:
- class: SignalRecord
@@ -80,29 +106,28 @@ task:
kwargs: {}
- class: SigAnaRecord
module_path: qlib.workflow.record_temp
kwargs: { ana_long_short: true, ann_scaler: 252 }
kwargs:
ana_long_short: true
ann_scaler: 252
- class: PortAnaRecord
module_path: qlib.workflow.record_temp
kwargs:
config:
strategy:
class: SignalQualityGateStrategy
module_path: tac_qlib.contrib.strategy.signal_quality_gate
class: OptimalStopControl
module_path: tac_qlib.contrib.strategy.optimal_stop
kwargs:
signal: "<PRED>"
lake_root: "{{ LAKE }}"
gate_topk: 10
gate_lookback: 5
gate_threshold: 0.5
gate_start: "2015-01-03"
gate_end: "2025-12-31"
topk: 10
n_drop: 1
only_tradable: true
entry_pct: 0.85
exit_pct: 0.7
max_hold_days: 10
min_hold_days: 2
sl: -0.08
risk_degree: 0.95
backtest:
start_time: 2025-01-02
end_time: 2025-12-31
start_time: 2026-01-04
end_time: 2026-08-10
account: 1000000
benchmark: SPY
exchange_kwargs: