Advanced AI Trading Guide: The Complete ML Library, Model & Framework Reference for Quants

qlib_init:
  provider_uri: "~/.qlib/qlib_data/custom_nse"   # see note below on pointing Qlib at NSE data
  region: in                                       # generic region tag; India isn't a stock Qlib region out of the box

market: &market custom_nse_500
benchmark: &benchmark NIFTY500

data_handler_config: &data_handler_config
  start_time: 2010-01-01
  end_time: 2025-12-31
  fit_start_time: 2010-01-01
  fit_end_time: 2019-12-31
  instruments: *market

task:
  model:
    class: LGBModel
    module_path: qlib.contrib.model.gbdt
    kwargs:
      loss: binary
      num_leaves: 31
      learning_rate: 0.03
      monotone_constraints: [1, 0, -1, 0]
  dataset:
    class: DatasetH
    module_path: qlib.data.dataset
    kwargs:
      handler:
        class: Alpha158
        module_path: qlib.contrib.data.handler
        kwargs: *data_handler_config
      segments:
        train: [2010-01-01, 2019-12-31]
        valid: [2020-01-01, 2021-12-31]
        test:  [2022-01-01, 2025-12-31]

# run with:  qrun this_config.yaml

pip install mlfinpy

from mlfinpy.labeling import get_events, add_vertical_barrier, get_bins
from mlfinpy.util import get_daily_vol

# causal, EWMA-based daily volatility -- uses data strictly up to t, no leakage
daily_vol = get_daily_vol(close=prices, lookback=50)

vertical_barriers = add_vertical_barrier(
    t_events=signal_dates, close=prices, num_days=10       # time-limit barrier
)

triple_barrier_events = get_events(
    close=prices,
    t_events=signal_dates,             # your primary (Pine) strategy's signal dates
    pt_sl=[2.0, 1.5],                  # profit-take / stop-loss as multiples of daily_vol
    target=daily_vol,
    min_ret=0.0005,
    num_threads=4,
    vertical_barrier_times=vertical_barriers,
)

labels = get_bins(triple_barrier_events, prices)
# labels['bin']    -> 1 = profit barrier hit first, 0 = stop/time barrier hit first
# labels['ret']    -> realized return at the barrier that was hit
# labels['t1']     -> the barrier-touch timestamp (needed for purging downstream)

pip install skfolio

from skfolio.model_selection import CombinatorialPurgedCV, cross_val_predict
from sklearn.pipeline import Pipeline
import lightgbm as lgb

meta_label_pipeline = Pipeline([
    ("model", lgb.LGBMClassifier(num_leaves=31, learning_rate=0.03,
                                  min_child_samples=50)),
])

# n_folds=10, n_test_folds=2  ->  C(10,2) = 45 combinatorial train/test paths
# purged_size / embargo_size sized to the max holding horizon (t1 - t0), NOT a flat 1%
cv = CombinatorialPurgedCV(n_folds=10, n_test_folds=2,
                            purged_size=10, embargo_size=10)

# returns an empirical DISTRIBUTION of out-of-sample paths, not one score --
# use it to compute PBO and the Deflated Sharpe Ratio, not a single train/test split
oos_paths = cross_val_predict(estimator=meta_label_pipeline, X=X, y=y,
                               cv=cv, sample_weight=w)

from hmmlearn.hmm import GaussianHMM

model = GaussianHMM(n_components=2, covariance_type="diag", n_iter=1000)
model.fit(X_full)                    # fit on the ENTIRE history -- future leaks into the fit
regimes = model.predict(X_full)      # Viterbi = globally-smoothed decode over the whole path
# regimes[t] used information from t+1 ... T. This is look-ahead bias.
# Any backtest of a regime gate built this way is invalid and will not reproduce live.

from hmmlearn.hmm import GaussianHMM
import numpy as np

regime_at_t = np.full(len(X), np.nan)
model, refit_every, min_train = None, 63, 500     # e.g. refit quarterly

for t in range(min_train, len(X)):
    if model is None or t % refit_every == 0:
        model = GaussianHMM(n_components=2, covariance_type="diag", n_iter=200)
        model.fit(X[:t])                          # ONLY data strictly before t

    # forward-algorithm FILTERED posterior using data up to and including t --
    # no smoothing, no future information, no re-fit on data that includes t's label use
    posteriors = model.predict_proba(X[:t + 1])
    regime_at_t[t] = posteriors[-1].argmax()       # today's regime, knowable today

# regime_at_t is safe to merge into the feature frame for both backtest and live use.
# Expect a MUCH smaller, laggier drawdown benefit than the full-sample-smoothed version --
# that gap IS the size of the leakage bug you just closed.

pip install arch

from arch import arch_model
import numpy as np

returns_pct = 100 * log_returns.dropna()      # arch expects % returns for numerical stability
# GJR-GARCH(1,1,1) with Student-t errors -- captures the leverage effect (vol reacts more to
# negative returns than positive ones), which plain GARCH(1,1) misses
am = arch_model(returns_pct, vol="GARCH", p=1, o=1, q=1, dist="t")
res = am.fit(disp="off", last_obs=train_end_date)   # fit only on data up to the current date

fcast = res.forecast(horizon=10, reindex=False)
vol_forecast_daily = np.sqrt(fcast.variance.values[-1]) / 100   # back to decimal daily vol

target_vol_daily = 0.10 / (252 ** 0.5)        # ~10% annualized portfolio vol target
position_scale = target_vol_daily / vol_forecast_daily[0]
position_size = base_size * min(position_scale, max_leverage_cap)   # always cap leverage

pip install tsfresh
from tsfresh import extract_relevant_features

# long-format df: columns [id, time, value]; y indexed by id (e.g. triple-barrier label)
features = extract_relevant_features(
    long_format_df, y=labels_by_id, column_id="id", column_sort="time",
    fdr_level=0.01,           # false-discovery-rate control -- do NOT skip this on financial data
)
# re-run the relevance filter INSIDE each CV fold in production, never on the full dataset

pip install statsforecast
from statsforecast import StatsForecast
from statsforecast.models import AutoARIMA, AutoETS

sf = StatsForecast(models=[AutoARIMA(), AutoETS()], freq="D", n_jobs=-1)
sf.fit(train_df)
baseline_forecast = sf.predict(h=10)
# your GBDT / foundation-model forecast must beat THIS, net of cost, or there is no edge

pip install mlflow
import mlflow

with mlflow.start_run(run_name="lgbm_meta_label_v37"):
    mlflow.log_params(params)                                   # every hyperparameter tried
    mlflow.log_param("cv_scheme", "CombinatorialPurgedCV(10,2)")
    mlflow.log_param("feature_set_hash", feature_set_hash)       # so re-used feature sets don't double count
    mlflow.log_metric("oos_sharpe_median", oos_sharpe_median)    # across the CPCV path distribution
    mlflow.log_metric("pbo", pbo_value)
    mlflow.log_metric("trial_number_in_family", trial_counter)   # THIS feeds the Deflated Sharpe N
    mlflow.lightgbm.log_model(model, "model")

# Query the full run history to compute N for the Deflated Sharpe Ratio -- include prior
# Pine-Script optimization trials too (see §14), not just the ML runs.

"""
Reference pipeline skeleton: data ingestion -> PickMyTrade execution.
Every preprocessing step is fit INSIDE the CPCV training fold only -- never on
the full dataset. Illustrative structure; wire in your own data store / config.
"""
import numpy as np
import pandas as pd
import lightgbm as lgb
import mlflow
import requests
from hmmlearn.hmm import GaussianHMM
from arch import arch_model
from mlfinpy.labeling import get_events, add_vertical_barrier, get_bins
from mlfinpy.util import get_daily_vol
from mlfinpy.features import frac_diff_ffd
from skfolio.model_selection import CombinatorialPurgedCV, cross_val_predict
from sklearn.pipeline import Pipeline
from qlib.contrib.data.handler import Alpha158
from tsfresh import extract_relevant_features
from nautilus_trader.backtest.engine import BacktestEngine
from nautilus_trader.trading.strategy import Strategy


# 1. DATA INGESTION -- point-in-time, survivorship-bias-free universe only
def load_point_in_time_universe(as_of_date, min_adv_usd=2_500_000):
    """Never filter by TODAY's liquidity -- that imports survivorship bias."""
    raw = data_store.query_asof(as_of_date=as_of_date)          # Parquet/DuckDB store
    return raw[raw["adv_20d_asof"] > min_adv_usd]


# 2. FEATURES -- Qlib Alpha158 for technicals, tsfresh + mlfinpy for the rest
def build_features(prices: pd.DataFrame) -> pd.DataFrame:
    alpha158 = Alpha158(instruments=prices["symbol"].unique().tolist(),
                         start_time=prices.index.min(), end_time=prices.index.max())
    tech_features = alpha158.fetch()

    tsfresh_features = extract_relevant_features(
        prices.reset_index(), y=None, column_id="symbol", column_sort="date", fdr_level=0.01,
    )

    d = find_min_ffd_order(prices["close"], adf_pvalue=0.05, min_corr=0.9)   # search d in [0.1, 1.0]
    frac_diff_price = frac_diff_ffd(prices["close"], d=d, threshold=1e-4)    # stationary, RETAINS memory

    return tech_features.join(tsfresh_features, how="inner").assign(frac_diff_price=frac_diff_price)


# 3. LABELS -- causal EWMA vol -> ATR-scaled triple barrier (mlfinpy)
def label_events(prices: pd.Series, signal_dates: pd.DatetimeIndex) -> pd.DataFrame:
    daily_vol = get_daily_vol(close=prices, lookback=50)
    vertical_barriers = add_vertical_barrier(t_events=signal_dates, close=prices, num_days=10)
    events = get_events(close=prices, t_events=signal_dates, pt_sl=[2.0, 1.5], target=daily_vol,
                         min_ret=0.0005, num_threads=4, vertical_barrier_times=vertical_barriers)
    return get_bins(events, prices)          # -> bin (0/1), ret, t1 (needed for purging)


# 4. META-LABEL MODEL -- LightGBM inside nested Combinatorial Purged CV
def train_meta_label_model(X: pd.DataFrame, y: pd.Series, w: pd.Series, trial_counter: int):
    pipeline = Pipeline([("model", lgb.LGBMClassifier(
        num_leaves=31, learning_rate=0.03, min_child_samples=50,
        monotone_constraints=MONO_CONSTRAINTS))])

    cv = CombinatorialPurgedCV(n_folds=10, n_test_folds=2, purged_size=10, embargo_size=10)
    oos_paths = cross_val_predict(estimator=pipeline, X=X, y=y, cv=cv, sample_weight=w)

    pbo = compute_pbo(oos_paths)
    dsr = compute_deflated_sharpe(oos_paths, n_trials=get_trial_count_from_mlflow() + trial_counter)
    if pbo > 0.50 or dsr.p_value > 0.05:
        raise ValueError(f"Model rejected: PBO={pbo:.2f}, DSR p={dsr.p_value:.3f}")

    final_model = pipeline.fit(X, y, model__sample_weight=w)
    return final_model, pbo, dsr


# 5. REGIME GATE -- causal, walk-forward, FILTERED (never smoothed) HMM posterior
def causal_regime_state(vol_return_features: np.ndarray, t: int, model_cache: dict,
                         refit_every: int = 63, min_train: int = 500) -> int:
    if t < min_train:
        return 1  # default "calm" until enough history exists
    if t % refit_every == 0 or "model" not in model_cache:
        hmm = GaussianHMM(n_components=2, covariance_type="diag", n_iter=200)
        hmm.fit(vol_return_features[:t])                       # only data strictly before t
        model_cache["model"] = hmm
    posteriors = model_cache["model"].predict_proba(vol_return_features[: t + 1])
    return int(posteriors[-1].argmax())                        # filtered state at t, no look-ahead


# 6. VOL-TARGETED SIZING -- GJR-GARCH forecast -> fractional-Kelly-scaled size
def size_position(returns_pct: pd.Series, meta_label_confidence: float, regime_state: int,
                   target_vol_annual: float = 0.10, kelly_fraction: float = 0.5,
                   max_leverage: float = 1.5) -> float:
    if regime_state == 0:                                       # 0 = "stormy" -- stand aside
        return 0.0
    res = arch_model(returns_pct, vol="GARCH", p=1, o=1, q=1, dist="t").fit(disp="off")
    vol_fc = np.sqrt(res.forecast(horizon=1, reindex=False).variance.values[-1, 0]) / 100
    vol_scale = (target_vol_annual / (252 ** 0.5)) / vol_fc
    kelly_scale = kelly_fraction * max(2 * meta_label_confidence - 1, 0)   # confidence -> edge proxy
    return float(np.clip(vol_scale * kelly_scale, 0, max_leverage))


# 7. BACKTEST -- NautilusTrader (same strategy class runs backtest AND live)
class MetaLabelStrategy(Strategy):
    def __init__(self, model, hmm_cache, config):
        super().__init__(config)
        self.model, self.hmm_cache = model, hmm_cache

    def on_bar(self, bar):
        feats = self.build_live_features(bar)                   # same code path as build_features()
        confidence = self.model.predict_proba(feats)[:, 1][0]
        regime = causal_regime_state(self.vol_history, self.clock.timestamp_ns, self.hmm_cache)
        size = size_position(self.returns_pct_history, confidence, regime)
        if confidence > 0.55 and size > 0:
            self.submit_bracket_order(side="BUY", size=size, sl_atr_mult=1.5, tp_atr_mult=2.0)


def run_backtest(historical_bars, final_model, strat_config):
    engine = BacktestEngine()
    engine.add_data(historical_bars)
    engine.add_strategy(MetaLabelStrategy(model=final_model, hmm_cache={}, config=strat_config))
    return engine.run()      # the SAME strategy object later promotes to paper/live -- adapters swap only


# 8. MLFLOW LOGGING -- every trial, forever (feeds the Deflated Sharpe trial count)
def log_run(params, pbo, dsr, backtest_results, final_model, trial_counter):
    with mlflow.start_run(run_name=f"pipeline_v{trial_counter}"):
        mlflow.log_params(params)
        mlflow.log_metric("pbo", pbo)
        mlflow.log_metric("dsr_p_value", dsr.p_value)
        mlflow.log_metric("backtest_sharpe", backtest_results.sharpe_ratio)
        mlflow.log_metric("backtest_max_dd", backtest_results.max_drawdown)
        mlflow.log_metric("trial_number_in_family", trial_counter)
        mlflow.lightgbm.log_model(final_model, "meta_label_model")


# 9. EXECUTION -- PickMyTrade webhook JSON (paper first, then tiny live)
def send_to_pickmytrade(symbol, side, qty, sl_price, tp_price, webhook_url, secret):
    payload = {
        "secret": secret, "symbol": symbol, "action": side,     # "buy" / "sell"
        "quantity": qty, "stopLoss": sl_price, "takeProfit": tp_price, "orderType": "market",
    }
    resp = requests.post(webhook_url, json=payload, timeout=5)
    resp.raise_for_status()
    return resp.json()

Review this skfolio CombinatorialPurgedCV setup [paste code]: 10 folds, 2 test folds, on triple-barrier-labeled
signals where barriers can resolve up to 10 trading days after the signal (get_events num_days=10). Check whether
purged_size and embargo_size are large enough to cover that full label-resolution window plus my slowest feature's
lookback. Flag any fold where a training sample's resolution window overlaps a test sample's, explain the leak, and
give me corrected purge/embargo values with the reasoning shown.

Here is my hmmlearn regime-detection code [paste code]. Confirm it only calls predict_proba on data up to and
including index t when producing the label used AT time t, and that it never calls hmm.decode() or reads a Viterbi
path computed over the full series. If it's using the full-sample smoothed posterior, rewrite it as a walk-forward
loop that refits every N bars on strictly past data and returns the filtered (not smoothed) posterior at each step,
matching the causal pattern, not the leaky one.

I want Microsoft Qlib's Alpha158 handler running on NSE (India) equities, not one of Qlib's built-in regions.
Here's my current workflow YAML [paste config]. Write a custom data handler/provider that points Alpha158's feature
computation at my own point-in-time OHLCV store instead of Qlib's default region data, keeping the Alpha158 feature
set intact. Explain specifically what breaks -- calendar, instrument universe, trading-day alignment -- if I just
change the region tag without supplying compatible replacements.

Here is my end-to-end pipeline: data ingestion, feature engineering, triple-barrier labeling, LightGBM meta-label
training inside CombinatorialPurgedCV, HMM regime gate, GARCH-based sizing, and NautilusTrader backtest [paste code].
Walk it stage by stage and flag every point where a transform -- scaler, feature selector, frac-diff order search, HMM
fit -- is fit on data outside the current CV training fold, or a feature could see information unavailable at signal
time. Rank findings by how much each would inflate the backtest Sharpe if left uncorrected.

Here is my TradingView Pine Script strategy [paste script]. Convert its entry logic into a Python function that
takes a pandas DataFrame of OHLCV bars and returns a boolean signal series, matching the Pine version bar-for-bar on
identical historical data. This becomes the primary-model signal feeding my meta-labeling filter, so flag anything --
security() calls with lookahead, use of the current bar's close before it closes -- that would make the two versions
diverge.

Write a Python function that takes a meta-label confidence score, ATR-based stop/target multiples, and a
position size from my vol-targeted sizing calc, and builds the JSON payload for a PickMyTrade webhook POST -- symbol,
side, quantity, stopLoss, takeProfit. Add a deterministic idempotency key (hash of symbol + signal timestamp + bar
index) as a payload field, and wrap the POST in retry-with-backoff logic that treats a timeout as "unknown, reconcile
before retrying" rather than blindly resubmitting.

{
  "idempotencyKey": "SIG-MES-2026-07-03T14:32:00Z-000481",
  "symbol": "MES",
  "side": "buy",
  "quantity": 2,
  "orderType": "market",
  "stopLoss": 5320.25,
  "takeProfit": 5364.75,
  "meta": {
    "metaLabelConfidence": 0.62,
    "atr14": 14.8,
    "slAtrMult": 1.5,
    "tpAtrMult": 2.0,
    "regimeState": 1
  }
}