AI 量化交易：从特征到实盘，把“预测”变成“可承受的风险”

原创

IT大佬 jzit-top

发布于 2026-10-22 15:10:48

290

AI 量化最普遍的失败不是模型不够聪明，而是回测里藏着未来信息、成本被忽略、模型一上线就衰减。专业做法是把整条链路拆成“无泄漏的特征—可验证的标签—带成本的回测—受约束的仓位”四段。本文给出可落地的实现代码与陷阱说明。

一、特征与防泄漏

特征必须严格用 t 时刻之前的信息。滚动统计用 shift(1) 隔离当日，是最低成本的防线。

import pandas as pd, numpy as np

def features(df: pd.DataFrame) -> pd.DataFrame:
    f = pd.DataFrame(index=df.index)
    ret = df["close"].pct_change()
    # 注意：全部用 shift(1)，避免用到当日收盘信息
    f["mom_5"]   = df["close"].pct_change(5).shift(1)
    f["mom_20"]  = df["close"].pct_change(20).shift(1)
    f["vol_20"]  = ret.rolling(20).std().shift(1)
    f["ma_gap"]  = (df["close"] / df["close"].rolling(20).mean() - 1).shift(1)
    f["rsi"]     = rsi(df["close"], 14).shift(1)
    f["vol_ratio"] = (df["volume"] / df["volume"].rolling(20).mean()).shift(1)
    return f.replace([np.inf, -np.inf], np.nan)

二、标签：三重障碍法

用“未来 N 日涨跌”做标签会把震荡市错标成信号。三重障碍法（López de Prado）给每个样本设止盈、止损、时间三个出口，标签由先触发者决定，更贴近真实持仓决策。

def triple_barrier(close: pd.Series, pt=0.03, sl=0.02, max_hold=10):
    labels, t_end = [], []
    for i in range(len(close)):
        entry = close.iloc[i]
        up, dn = entry * (1 + pt), entry * (1 - sl)
        label, end = 0, min(i + max_hold, len(close) - 1)  # 0=时间出场
        for j in range(i + 1, min(i + max_hold + 1, len(close))):
            p = close.iloc[j]
            if p >= up:  label, end = 1, j; break     # 止盈
            if p <= dn:  label, end = -1, end = j; break  # 止损
        labels.append(label); t_end.append(end)
    return pd.Series(labels, index=close.index), pd.Series(t_end, index=close.index)

三、交叉验证：必须 Purge

时间序列上随机 K 折会把“未来”折进“过去”。更严重的是，三重障碍的标签跨越了样本区间，训练集尾部会与验证集重叠——这就是泄漏。解法是 Purged K-Fold：在训练/验证之间挖掉一段，并剔除标签区间跨越边界的样本。

from sklearn.model_selection import BaseCrossValidator

class PurgedKFold(BaseCrossValidator):
    def __init__(self, n_splits=5, embargo=5):
        self.n_splits, self.embargo = n_splits, embargo

    def split(self, X, y=None, groups=None):
        idx = np.arange(len(X)); fold = len(idx) // self.n_splits
        for k in range(self.n_splits):
            test = idx[k*fold:(k+1)*fold]
            train = np.setdiff1d(idx, test)
            # 挖掉测试集前后 embargo 个样本
            train = train[(train < test[0] - self.embargo) |
                          (train > test[-1] + self.embargo)]
            yield train, test

    def get_n_splits(self, X=None, y=None, groups=None):
        return self.n_splits

四、训练与评估

分类器用 LightGBM 足够，关键在评估指标：不能只看准确率，要看按概率分组的收益单调性——高分组收益必须高于低分组，否则模型没有可用信息。

import lightgbm as lgb
from sklearn.metrics import roc_auc_score

def train_eval(X, y, cv):
    oof = np.zeros(len(X)); aucs = []
    for tr, te in cv.split(X):
        m = lgb.LGBMClassifier(n_estimators=300, learning_rate=0.03,
                               num_leaves=31, subsample=0.8, verbose=-1)
        m.fit(X.iloc[tr], y.iloc[tr])
        oof[te] = m.predict_proba(X.iloc[te])[:, 1]
        aucs.append(roc_auc_score((y.iloc[te] > 0).astype(int), oof[te]))

    # 分组收益单调性检验：把预测概率分 5 组，看各组平均收益
    grp = pd.qcut(oof, 5, labels=False)
    ret_by_grp = pd.Series(y.values).groupby(grp).mean()
    print("AUC:", np.mean(aucs).round(3))
    print("分组收益:\n", ret_by_grp.round(4))
    return oof

五、回测：成本与滑点不能省

向量化回测必须扣双边成本，否则高频调仓的策略收益全是假的。

def backtest(prob, ret_next, cost=0.0015, thr=0.55):
    pos = pd.Series(0, index=prob.index)
    pos[prob > thr] = 1        # 做多
    pos[prob < 1 - thr] = -1   # 做空（可选）
    pos = pos.shift(1).fillna(0)          # 信号次日生效
    turnover = pos.diff().abs().fillna(0) # 换手
    pnl = pos * ret_next - turnover * cost
    eq = (1 + pnl).cumprod()
    sharpe = pnl.mean() / pnl.std() * np.sqrt(252)
    mdd = (eq / eq.cummax() - 1).min()
    return {"sharpe": round(sharpe, 2), "mdd": round(mdd, 3),
            "turnover": round(turnover.mean(), 3), "equity": eq}

三个必查项：pos.shift(1) 保证次日成交；turnover * cost 扣换手成本；mdd 看最大回撤而非只看夏普。

六、仓位：波动率目标 + 回撤熔断

预测给方向，仓位给风险。按目标波动率反推仓位，并设回撤熔断，防止模型失效时无限亏损。

def size(pos, vol, target_vol=0.15, max_lev=1.0, dd_limit=0.15, eq=None):
    raw = pos * (target_vol / vol.clip(lower=0.05))
    lev = raw.clip(-max_lev, max_lev)
    if eq is not None:
        dd = (eq / eq.cummax() - 1).iloc[-1]
        if dd < -dd_limit:            # 回撤超限，强制降仓
            lev *= 0.3
    return lev

七、上线后的模型衰减

实盘必须监控三类指标：预测分布漂移（PSI）、滚动 IC、实际换手与滑点偏差。PSI 超 0.25 或滚动 20 日 IC 转负，就应触发降仓或重训，而不是“再等等看”。

结语

AI 量化不是“让模型猜涨跌”，而是把模型输出转化为受成本、回撤、容量约束的风险暴露。防泄漏决定回测是否可信，Purged CV 决定评估是否诚实，成本与仓位决定策略能否活到实盘。这四处做对，模型哪怕只有微弱优势也能稳定变现；做错任何一处，漂亮的回测曲线都只是统计幻觉。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

量化