VectorBT量化入门系列：第六章 VectorBT实战案例：机器学习预测策略-EW帮帮网

VectorBT量化入门系列：第六章 VectorBT实战案例：机器学习预测策略

本教程专为中高级开发者设计，系统讲解VectorBT技术在量化交易中的应用。通过结合Tushare数据源和TA-Lib技术指标，深度探索策略开发、回测优化与风险评估的核心方法。从数据获取到策略部署，全面提升量化交易能力，助力开发者构建高效、稳健的交易系统。
文中内容仅限技术学习与代码实践参考，市场存在不确定性，技术分析需谨慎验证，不构成任何投资建议。适合量化新手建立系统认知，为策略开发打下基础。

VectorBT

学习对象

中高级水平的开发者和数据分析师
具备 Python 编程基础和一定的数据分析能力
对量化交易和金融数据处理有一定了解
熟悉 A 股市场，了解 Tushare 数据源和 TA-Lib 技术指标

教程目标

系统学习 VectorBT 技术，掌握其在量化交易中的应用
熟练使用 Tushare 数据源获取 A 股市场数据，并使用 Parquet 文件存储
掌握基于 VectorBT 的策略开发、回测和性能评估流程
学会使用 TA-Lib 计算技术指标，并将其应用于交易策略
理解并实现多因子策略、机器学习策略等高级策略
掌握策略优化、风险管理以及策略组合的方法
能够独立构建和评估量化交易策略，并部署到生产环境

教程目录

第一章 VectorBT基础与环境搭建

1.1 VectorBT简介与应用场景
1.2 环境搭建与依赖安装
1.3 数据源与Tushare集成
1.4 数据存储与Parquet文件格式

第二章 VectorBT核心功能与数据处理

2.1 数据加载与预处理
2.2 时间序列数据处理
2.3 技术指标计算与TA-Lib集成
2.4 数据可视化与探索性分析

第三章 VectorBT策略回测基础

3.1 策略定义与实现
3.2 回测流程与关键参数
3.3 性能评估指标与解读
3.4 策略优化与参数调整

第四章高级策略开发与优化

4.1 多因子策略开发
4.2 机器学习策略集成
4.3 风险管理与交易成本模拟
4.4 策略组合与资产配置

第五章 VectorBT性能评估与分析

5.1 性能评估框架
5.2 统计指标与回测报告
5.3 敏感性分析与压力测试
5.4 策略对比与选择标准

第六章 VectorBT实战案例

6.1 策略逻辑
6.2 实现步骤
6.3 代码执行
6.4 结果分析

第六章 VectorBT实战案例：机器学习预测策略

本章将通过具体案例深入学习VectorBT的应用。我们将使用机器学习模型预测股票价格的走势。实现一个完整的股票交易策略流程，从数据预处理、特征工程、模型训练与验证、信号生成与调试到策略回测和过拟合检测。通过使用多种技术指标和LightGBM分类器，该策略能够生成有效的交易信号，并通过回测验证其性能。

LightGBM-VectorBT-6

6.1 策略逻辑

使用历史数据训练机器学习模型。
根据模型预测结果生成交易信号。

核心流程：

数据加载 -> 特征工程 -> 标签生成 -> 模型训练 -> 信号生成 -> 风险管理 -> 回测验证

数学公式：

持有期回报率

$R_{t+1→t+HOLDING\_PERIOD} = \frac{Close_{t+HOLDING\_PERIOD}}{Open_{t+1}} - 1$

示例设置：

# 全局种子设置
np.random.seed(42)
random.seed(42)

# 配置参数
HOLDING_PERIOD = 2  # 持有到 t+HOLDING_PERIOD 日收盘（跨交易日持仓）
LOOKBACK_WINDOW = 20  # 回看窗口

6.2 实现步骤

加载数据

def load_data(filepath):
    """加载并预处理数据"""
    data = pd.read_parquet(filepath)
    print(f"Data Stock: {len(data)}")
    return data

特征工程

def create_features(data):
    """为给定的数据集创建技术指标特征。
    使用closed="left"确保滚动窗口不包含当前数据
    滞后特征避免未来信息泄漏

    :param data: 包含OHLCV数据的DataFrame
    :return: 包含新增技术指标特征的DataFrame
    """
    data = data.assign(
        returns=data["close"].pct_change().shift(1),
        volatility_5=lambda x: x["returns"].rolling(5, closed="left").std().shift(1),
        volume_change=lambda x: x["vol"].pct_change(3).shift(1),
    )

    # 有效技术指标（经实证检验的因子）
    data["macd"], data["macd_signal"], _ = talib.MACD(data["close"])
    data["macd"] = data["macd"].shift(1)
    data["obv"] = talib.OBV(data["close"], data["vol"]).shift(1)

    # 布林带特征
    upper, mid, lower = talib.BBANDS(data["close"], timeperiod=20)
    data["bb_width"] = ((upper - lower) / mid).shift(1)
    data["bb_position"] = ((data["close"] - lower) / (upper - lower)).shift(1)

    # 量价背离指标
    price_high = data["high"].rolling(5).mean().shift(1)
    volume_high = data["vol"].rolling(5).mean().shift(1)
    data["pv_divergence"] = (price_high.diff() * volume_high.diff() < 0).astype(int)

    return data.dropna()  # 删除残余缺失

标签工程

def create_labels(data):
    """标签生成逻辑（严格时间对齐）：
    - 预测时间：`t`日
    - 买入时间：`t+1`开盘
    - 卖出时间：`t+HOLDING_PERIOD`收盘
    - 持仓天数：`HOLDING_PERIOD`
    - 时间轴：    t    t+1   t+HOLDING_PERIOD
                预测  买入   卖出
    - 持有期回报率：$ R_{t+1→t+HOLDING\_PERIOD} = \frac{Close_{t+HOLDING\_PERIOD}}{Open_{t+1}} - 1 $
    """
    # 计算持有期收益（严格避免未来信息）
    buy_price = data["open"].shift(-1)  # t+1开盘价
    sell_price = data["close"].shift(-HOLDING_PERIOD)  # t+HOLDING_PERIOD收盘价
    future_returns = sell_price / buy_price - 1  # 持有期回报率
    # 生成预测概率标签
    data["target"] = np.where(future_returns > 0, 1, 0)
    return data.dropna()  # 删除含有NaN值的行

特征选择

def select_features(X_train, y_train):
    """时间序列特征选择"""
    estimator = RandomForestClassifier(
        n_estimators=100,
        n_jobs=1,
        random_state=42,
        max_features="sqrt",  # 增加确定性
    )
    selector = RFE(estimator, n_features_to_select=5)
    selector.fit(X_train, y_train)
    return selector

模型训练

def train_model(X, y):
    """训练模型并进行特征选择"""

    # 时间序列交叉验证
    tscv = TimeSeriesSplit(n_splits=5)
    callbacks = [
        lgb.early_stopping(
            stopping_rounds=50, first_metric_only=True, verbose=False
        ),  # 早停回调
        lgb.log_evaluation(period=0),  # 禁用日志输出
    ]

    def objective(trial):
        params = {
            "objective": "binary",
            "metric": "auc",
            "boosting_type": "goss",
            "num_leaves": trial.suggest_int("num_leaves", 16, 64),
            "learning_rate": trial.suggest_float("lr", 0.02, 0.2, log=True),
            "max_depth": trial.suggest_int("max_depth", 3, 7),
            "min_child_samples": trial.suggest_int("min_child_samples", 20, 200),
            "reg_alpha": trial.suggest_float("reg_alpha", 1e-8, 10.0),
            "reg_lambda": trial.suggest_float("reg_lambda", 1e-8, 10.0),
            "feature_fraction": trial.suggest_float("feature_frac", 0.6, 0.95),
            "bagging_freq": trial.suggest_int("bagging_freq", 3, 7),
            "verbosity": -1,
            "force_col_wise": True,
            "random_state": 42,
            "num_threads": 1,
            "deterministic": True,
        }

        fold_scores = []  # 保存每一折的AUC
        fold_iterations = []  # 保存每个fold的最佳迭代次数

        for train_idx, valid_idx in tscv.split(X):
            # 分割数据
            X_train_fold, X_valid_fold = (
                X.iloc[train_idx],
                X.iloc[valid_idx],
            )
            y_train_fold, y_valid_fold = (
                y.iloc[train_idx],
                y.iloc[valid_idx],
            )

            # 特征选择
            selector = select_features(X_train_fold, y_train_fold)
            X_train_selected = selector.transform(X_train_fold)
            X_valid_selected = selector.transform(X_valid_fold)

            # 数据标准化
            scaler = StandardScaler()
            X_train_scaled = scaler.fit_transform(X_train_selected)
            X_valid_scaled = scaler.transform(X_valid_selected)

            # 转换为LightGBM Dataset
            train_data = lgb.Dataset(X_train_scaled, label=y_train_fold)
            valid_data = lgb.Dataset(
                X_valid_scaled, label=y_valid_fold, reference=train_data
            )

            # 训练模型（启用早停）
            model = lgb.train(
                params,
                train_data,
                valid_sets=[valid_data],
                callbacks=callbacks,
                num_boost_round=1000,
            )

            # 记录最佳迭代的AUC
            fold_scores.append(model.best_score["valid_0"]["auc"])
            # 记录最佳迭代次数
            fold_iterations.append(model.best_iteration)

        # 记录交叉验证的平均迭代次数
        trial.set_user_attr("avg_best_iteration", int(np.mean(fold_iterations)))
        # 返回平均AUC作为Optuna优化目标
        return np.mean(fold_scores)

    # 创建Optuna study并优化
    sampler = TPESampler(seed=42)  # 使用TPE采样器
    study = optuna.create_study(
        direction="maximize",
        sampler=sampler,
        pruner=optuna.pruners.MedianPruner(),  # n_warmup_steps=10
    )
    study.optimize(
        objective, n_trials=10, n_jobs=1, timeout=3600, show_progress_bar=True
    )  # 优化超参数 n_jobs=1, 禁用并行试验，保证结果可以复现

    best_auc = study.best_value  # 获取最优 AUC
    best_params = study.best_params  # 获取最优参数
    best_params.update(
        {
            "objective": "binary",
            "metric": "auc",
            "boosting_type": "goss",
            "verbosity": -1,
            "force_col_wise": True,
            "random_state": 42,
            "num_threads": 1,
            "deterministic": True,
        }
    )

    # 输出最佳参数
    print("最佳参数:", best_params)
    print("最佳AUC:", best_auc)

    # 划分训练集和测试集
    split_idx = int(len(X) * 0.8)
    X_train, X_test = X.iloc[:split_idx], X.iloc[split_idx:]
    y_train, y_test = y.iloc[:split_idx], y.iloc[split_idx:]

    best_selector = select_features(X_train, y_train)
    X_train = best_selector.transform(X_train)
    X_test = best_selector.transform(X_test)

    # 标准化：仅在训练数据上拟合，避免泄露
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)

    # 训练最终模型（可关闭早停，使用更多迭代）
    final_model = lgb.train(
        best_params,
        lgb.Dataset(X_train_scaled, label=y_train),
        num_boost_round=study.best_trial.user_attrs["avg_best_iteration"],
    )

    # 选择特征
    print(f"Selected features: {best_selector.feature_names_in_}")

    # 过拟合检查
    print("\n过拟合检测:")
    print(f"Train AUC: {roc_auc_score(y_train, final_model.predict(X_train)):.4f}")
    print(f"Test AUC: {roc_auc_score(y_test, final_model.predict(X_test)):.4f}")

    return final_model, best_selector, scaler

调试信号生成

def debug_signal_generation(data):
    """打印关键指标的统计信息，用于调试信号生成过程。

    :param data: 包含预测概率和其他指标的DataFrame
    """
    # 打印关键指标分布
    print("\n信号生成关键指标统计：")
    print(f"预测概率 > 0.5 的比例: {len(data[data['pred_proba'] > 0.5])/len(data):.2%}")
    print(
        f"成交量过滤触发比例: {len(data[data['vol'] > data['vol'].rolling(20).mean()])/len(data):.2%}"
    )
    print(
        f"处于200日均线上方比例: {len(data[data['close'] > data['close'].rolling(200).mean()])/len(data):.2%}"
    )

交易信号生成

def generate_signals(model, selector, scaler, data):
    """生成带持仓周期管理的交易信号"""
    data = data.copy()

    # 特征处理
    X = selector.transform(data[selector.feature_names_in_])
    X_scaled = scaler.transform(X)
    data["pred_proba"] = model.predict(X_scaled)

    print("\n预测概率分布:")
    print(data["pred_proba"].describe())

    # 调试信号生成
    debug_signal_generation(data)

    # 生成交易信号：简单实现示例
    data["entry_signal"] = np.where(
        data["pred_proba"] > data["pred_proba"].mean(), 1, 0
    )

    # 遍历数据，为每个买入信号设置对应的卖出信号，t+HOLDING_PERIOD收盘卖出
    data["exit_signal"] = 0
    valid_dates = data.index  # 确保索引已排序
    for entry_idx in data[data["entry_signal"] == 1].index:
        current_pos = valid_dates.get_loc(entry_idx)
        exit_pos = current_pos + HOLDING_PERIOD
        if exit_pos < len(valid_dates):
            exit_idx = valid_dates[exit_pos]
            data.loc[exit_idx, "exit_signal"] = 1

    # 信号可视化检查
    print("\n信号统计:")
    print(f"做多信号数量: {data['entry_signal'].sum()}")
    print(f"做空信号数量: {data['exit_signal'].sum()}")

    return data

风险控制

def apply_risk_management(data):
    """简单波动率仓位控制
    实现逻辑：
    1. 计算最近20个交易日的波动率中位数
    2. 当当前波动率超过中位数时，仓位降至50%
    3. 否则保持80%仓位
    """
    data = data.copy()

    # 计算波动率阈值（使用滞后数据避免未来信息）
    data["vol_threshold"] = (
        data["volatility_5"]
        .shift(1)  # 使用t-1日数据计算
        .rolling(20, min_periods=5)
        .median()  # 20日波动率中位数
    )

    # 动态仓位设置
    data["position_size"] = np.where(
        data["volatility_5"] > data["vol_threshold"],
        0.5,  # 高波动时50%仓位
        0.8,  # 正常情况80%仓位
    )

    # 强制仓位范围限制
    data["position_size"] = data["position_size"].clip(0.3, 0.8)

    return data

回测引擎

def backtest_strategy(data):
    """执行回测"""
    pf = vbt.Portfolio.from_signals(
        close=data["close"],
        open=data["open"],  # 使用开盘价作为买入价格
        entries=data["entry_signal"] == 1,
        exits=data["exit_signal"] == 1,
        freq="1D",
        init_cash=1e5,  # 初始资金 10万
        fees=0.0005,  # 佣金万五
        slippage=0.0015,  # 滑点千1.5
        size_type="percent",
        size=data["position_size"],  # 动态仓位大小
        sl_stop=0.05,  # 固定止损
    )
    return pf

结果分析

def plot_results(pf):
    """可视化策略绩效结果"""
    pf.plot(subplots=["orders", "trade_pnl", "cum_returns", "drawdowns"]).show()

主程序

if __name__ == "__main__":
    # 数据准备
    data = load_data("./data/600000.SH.parquet")
    data = create_features(data)
    data = create_labels(data)

    # 按时间分割数据集
    split_date = pd.to_datetime("2024-01-01")
    train_data = data[data.index < split_date]
    test_data = data[data.index >= split_date]

    # 检查标签分布
    print(f"Trant Data: {len(train_data)}")
    print(train_data["target"].value_counts())
    print(f"Test Data: {len(test_data)}")
    print(test_data["target"].value_counts())

    # 特征指标
    features = [
        "volatility_5",
        "volume_change",
        "macd",
        "obv",
        "bb_width",
        "bb_position",
        "pv_divergence",
    ]

    # 模型训练
    model, selector, scaler = train_model(train_data[features], train_data["target"])

    # 信号生成
    data = generate_signals(model, selector, scaler, test_data)
    data = apply_risk_management(data)

    # 回测执行
    pf = backtest_strategy(data)

    # 结果分析
    print("\n策略绩效:")
    print(pf.stats())

    # 可视化
    plot_results(pf)

6.3 代码执行

import random
import lightgbm as lgb
import numpy as np
import optuna
import pandas as pd
import talib
import vectorbt as vbt
from sklearn.ensemble import RandomForestClassifier
from sklearn.feature_selection import RFE
from sklearn.metrics import roc_auc_score
from sklearn.model_selection import TimeSeriesSplit
from sklearn.preprocessing import StandardScaler
from optuna.samplers import TPESampler

# 该策略采用机器学习驱动的量化交易框架，核心流程如下：
# 数据加载 -> 特征工程 -> 标签生成 -> 模型训练 -> 信号生成 -> 风险管理 -> 回测验证

# 全局种子设置
np.random.seed(42)
random.seed(42)

# 配置参数
HOLDING_PERIOD = 2  # 持有到 t+HOLDING_PERIOD 日收盘（跨交易日持仓）
LOOKBACK_WINDOW = 20  # 回看窗口


# ==================== 加载数据 ====================
def load_data(filepath):
    """加载并预处理数据"""
    data = pd.read_parquet(filepath)
    print(f"Data Stock: {len(data)}")
    return data


# ==================== 特征工程 ====================
def create_features(data):
    """为给定的数据集创建技术指标特征。
    使用closed="left"确保滚动窗口不包含当前数据
    滞后特征避免未来信息泄漏

    :param data: 包含OHLCV数据的DataFrame
    :return: 包含新增技术指标特征的DataFrame
    """
    data = data.assign(
        returns=data["close"].pct_change().shift(1),
        volatility_5=lambda x: x["returns"].rolling(5, closed="left").std().shift(1),
        volume_change=lambda x: x["vol"].pct_change(3).shift(1),
    )

    # 有效技术指标（经实证检验的因子）
    data["macd"], data["macd_signal"], _ = talib.MACD(data["close"])
    data["macd"] = data["macd"].shift(1)
    data["obv"] = talib.OBV(data["close"], data["vol"]).shift(1)

    # 布林带特征
    upper, mid, lower = talib.BBANDS(data["close"], timeperiod=20)
    data["bb_width"] = ((upper - lower) / mid).shift(1)
    data["bb_position"] = ((data["close"] - lower) / (upper - lower)).shift(1)

    # 量价背离指标
    price_high = data["high"].rolling(5).mean().shift(1)
    volume_high = data["vol"].rolling(5).mean().shift(1)
    data["pv_divergence"] = (price_high.diff() * volume_high.diff() < 0).astype(int)

    return data.dropna()  # 删除残余缺失


# ==================== 标签工程 ====================
def create_labels(data):
    """标签生成逻辑（严格时间对齐）：
    - 预测时间：`t`日
    - 买入时间：`t+1`开盘
    - 卖出时间：`t+HOLDING_PERIOD`收盘
    - 持仓天数：`HOLDING_PERIOD`
    - 时间轴：    t    t+1   t+HOLDING_PERIOD
                预测  买入   卖出
    - 持有期回报率：$ R_{t+1→t+HOLDING\_PERIOD} = \frac{Close_{t+HOLDING\_PERIOD}}{Open_{t+1}} - 1 $
    """
    # 计算持有期收益（严格避免未来信息）
    buy_price = data["open"].shift(-1)  # t+1开盘价
    sell_price = data["close"].shift(-HOLDING_PERIOD)  # t+HOLDING_PERIOD收盘价
    future_returns = sell_price / buy_price - 1  # 持有期回报率
    # 生成预测概率标签
    data["target"] = np.where(future_returns > 0, 1, 0)
    return data.dropna()  # 删除含有NaN值的行


# ==================== 特征选择 ====================
def select_features(X_train, y_train):
    """时间序列特征选择"""
    estimator = RandomForestClassifier(
        n_estimators=100,
        n_jobs=1,
        random_state=42,
        max_features="sqrt",  # 增加确定性
    )
    selector = RFE(estimator, n_features_to_select=5)
    selector.fit(X_train, y_train)
    return selector


# ==================== 模型训练 ====================
def train_model(X, y):
    """训练模型并进行特征选择"""

    # 时间序列交叉验证
    tscv = TimeSeriesSplit(n_splits=5)
    callbacks = [
        lgb.early_stopping(
            stopping_rounds=50, first_metric_only=True, verbose=False
        ),  # 早停回调
        lgb.log_evaluation(period=0),  # 禁用日志输出
    ]

    def objective(trial):
        params = {
            "objective": "binary",
            "metric": "auc",
            "boosting_type": "goss",
            "num_leaves": trial.suggest_int("num_leaves", 16, 64),
            "learning_rate": trial.suggest_float("lr", 0.02, 0.2, log=True),
            "max_depth": trial.suggest_int("max_depth", 3, 7),
            "min_child_samples": trial.suggest_int("min_child_samples", 20, 200),
            "reg_alpha": trial.suggest_float("reg_alpha", 1e-8, 10.0),
            "reg_lambda": trial.suggest_float("reg_lambda", 1e-8, 10.0),
            "feature_fraction": trial.suggest_float("feature_frac", 0.6, 0.95),
            "bagging_freq": trial.suggest_int("bagging_freq", 3, 7),
            "verbosity": -1,
            "force_col_wise": True,
            "random_state": 42,
            "num_threads": 1,
            "deterministic": True,
        }

        fold_scores = []  # 保存每一折的AUC
        fold_iterations = []  # 保存每个fold的最佳迭代次数

        for train_idx, valid_idx in tscv.split(X):
            # 分割数据
            X_train_fold, X_valid_fold = (
                X.iloc[train_idx],
                X.iloc[valid_idx],
            )
            y_train_fold, y_valid_fold = (
                y.iloc[train_idx],
                y.iloc[valid_idx],
            )

            # 特征选择
            selector = select_features(X_train_fold, y_train_fold)
            X_train_selected = selector.transform(X_train_fold)
            X_valid_selected = selector.transform(X_valid_fold)

            # 数据标准化
            scaler = StandardScaler()
            X_train_scaled = scaler.fit_transform(X_train_selected)
            X_valid_scaled = scaler.transform(X_valid_selected)

            # 转换为LightGBM Dataset
            train_data = lgb.Dataset(X_train_scaled, label=y_train_fold)
            valid_data = lgb.Dataset(
                X_valid_scaled, label=y_valid_fold, reference=train_data
            )

            # 训练模型（启用早停）
            model = lgb.train(
                params,
                train_data,
                valid_sets=[valid_data],
                callbacks=callbacks,
                num_boost_round=1000,
            )

            # 记录最佳迭代的AUC
            fold_scores.append(model.best_score["valid_0"]["auc"])
            # 记录最佳迭代次数
            fold_iterations.append(model.best_iteration)

        # 记录交叉验证的平均迭代次数
        trial.set_user_attr("avg_best_iteration", int(np.mean(fold_iterations)))
        # 返回平均AUC作为Optuna优化目标
        return np.mean(fold_scores)

    # 创建Optuna study并优化
    sampler = TPESampler(seed=42)  # 使用TPE采样器
    study = optuna.create_study(
        direction="maximize",
        sampler=sampler,
        pruner=optuna.pruners.MedianPruner(),  # n_warmup_steps=10
    )
    study.optimize(
        objective, n_trials=10, n_jobs=1, timeout=3600, show_progress_bar=True
    )  # 优化超参数 n_jobs=1, 禁用并行试验，保证结果可以复现

    best_auc = study.best_value  # 获取最优 AUC
    best_params = study.best_params  # 获取最优参数
    best_params.update(
        {
            "objective": "binary",
            "metric": "auc",
            "boosting_type": "goss",
            "verbosity": -1,
            "force_col_wise": True,
            "random_state": 42,
            "num_threads": 1,
            "deterministic": True,
        }
    )

    # 输出最佳参数
    print("最佳参数:", best_params)
    print("最佳AUC:", best_auc)

    # 划分训练集和测试集
    split_idx = int(len(X) * 0.8)
    X_train, X_test = X.iloc[:split_idx], X.iloc[split_idx:]
    y_train, y_test = y.iloc[:split_idx], y.iloc[split_idx:]

    best_selector = select_features(X_train, y_train)
    X_train = best_selector.transform(X_train)
    X_test = best_selector.transform(X_test)

    # 标准化：仅在训练数据上拟合，避免泄露
    scaler = StandardScaler()
    X_train_scaled = scaler.fit_transform(X_train)

    # 训练最终模型（可关闭早停，使用更多迭代）
    final_model = lgb.train(
        best_params,
        lgb.Dataset(X_train_scaled, label=y_train),
        num_boost_round=study.best_trial.user_attrs["avg_best_iteration"],
    )

    # 选择特征
    print(f"Selected features: {best_selector.feature_names_in_}")

    # 过拟合检查
    print("\n过拟合检测:")
    print(f"Train AUC: {roc_auc_score(y_train, final_model.predict(X_train)):.4f}")
    print(f"Test AUC: {roc_auc_score(y_test, final_model.predict(X_test)):.4f}")

    return final_model, best_selector, scaler


# ==================== 调试信号生成 ====================
def debug_signal_generation(data):
    """打印关键指标的统计信息，用于调试信号生成过程。

    :param data: 包含预测概率和其他指标的DataFrame
    """
    # 打印关键指标分布
    print("\n信号生成关键指标统计：")
    print(f"预测概率 > 0.5 的比例: {len(data[data['pred_proba'] > 0.5])/len(data):.2%}")
    print(
        f"成交量过滤触发比例: {len(data[data['vol'] > data['vol'].rolling(20).mean()])/len(data):.2%}"
    )
    print(
        f"处于200日均线上方比例: {len(data[data['close'] > data['close'].rolling(200).mean()])/len(data):.2%}"
    )


# ==================== 交易信号生成 ====================
def generate_signals(model, selector, scaler, data):
    """生成带持仓周期管理的交易信号"""
    data = data.copy()

    # 特征处理
    X = selector.transform(data[selector.feature_names_in_])
    X_scaled = scaler.transform(X)
    data["pred_proba"] = model.predict(X_scaled)

    print("\n预测概率分布:")
    print(data["pred_proba"].describe())

    # 调试信号生成
    debug_signal_generation(data)

    # 生成交易信号：简单实现示例
    data["entry_signal"] = np.where(
        data["pred_proba"] > data["pred_proba"].mean(), 1, 0
    )

    # 遍历数据，为每个买入信号设置对应的卖出信号，t+HOLDING_PERIOD收盘卖出
    data["exit_signal"] = 0
    valid_dates = data.index  # 确保索引已排序
    for entry_idx in data[data["entry_signal"] == 1].index:
        current_pos = valid_dates.get_loc(entry_idx)
        exit_pos = current_pos + HOLDING_PERIOD
        if exit_pos < len(valid_dates):
            exit_idx = valid_dates[exit_pos]
            data.loc[exit_idx, "exit_signal"] = 1

    # 信号可视化检查
    print("\n信号统计:")
    print(f"做多信号数量: {data['entry_signal'].sum()}")
    print(f"做空信号数量: {data['exit_signal'].sum()}")

    return data


# ==================== 风险控制 ====================
def apply_risk_management(data):
    """简单波动率仓位控制
    实现逻辑：
    1. 计算最近20个交易日的波动率中位数
    2. 当当前波动率超过中位数时，仓位降至50%
    3. 否则保持80%仓位
    """
    data = data.copy()

    # 计算波动率阈值（使用滞后数据避免未来信息）
    data["vol_threshold"] = (
        data["volatility_5"]
        .shift(1)  # 使用t-1日数据计算
        .rolling(20, min_periods=5)
        .median()  # 20日波动率中位数
    )

    # 动态仓位设置
    data["position_size"] = np.where(
        data["volatility_5"] > data["vol_threshold"],
        0.5,  # 高波动时50%仓位
        0.8,  # 正常情况80%仓位
    )

    # 强制仓位范围限制
    data["position_size"] = data["position_size"].clip(0.3, 0.8)

    return data


# ==================== 回测引擎 ====================
def backtest_strategy(data):
    """执行回测"""
    pf = vbt.Portfolio.from_signals(
        close=data["close"],
        open=data["open"],  # 使用开盘价作为买入价格
        entries=data["entry_signal"] == 1,
        exits=data["exit_signal"] == 1,
        freq="1D",
        init_cash=1e5,  # 初始资金 10万
        fees=0.0005,  # 佣金万五
        slippage=0.0015,  # 滑点千1.5
        size_type="percent",
        size=data["position_size"],  # 动态仓位大小
        sl_stop=0.05,  # 固定止损
    )
    return pf


# ==================== 结果分析 ====================
def plot_results(pf):
    """可视化策略绩效结果"""
    pf.plot(subplots=["orders", "trade_pnl", "cum_returns", "drawdowns"]).show()


# ==================== 主程序 ====================
if __name__ == "__main__":
    # 数据准备
    # 000858.SZ
    # 600000.SH
    # 600036.SH
    # 600519.SH
    data = load_data("./data/600000.SH.parquet")
    data = create_features(data)
    data = create_labels(data)

    # 按时间分割数据集
    split_date = pd.to_datetime("2024-01-01")
    train_data = data[data.index < split_date]
    test_data = data[data.index >= split_date]

    # 检查标签分布
    print(f"Trant Data: {len(train_data)}")
    print(train_data["target"].value_counts())
    print(f"Test Data: {len(test_data)}")
    print(test_data["target"].value_counts())

    features = [
        "volatility_5",
        "volume_change",
        "macd",
        "obv",
        "bb_width",
        "bb_position",
        "pv_divergence",
    ]

    # 模型训练
    model, selector, scaler = train_model(train_data[features], train_data["target"])

    # 信号生成
    data = generate_signals(model, selector, scaler, test_data)
    data = apply_risk_management(data)

    # 回测执行
    pf = backtest_strategy(data)

    # 结果分析
    print("\n策略绩效:")
    print(pf.stats())

    # 可视化
    plot_results(pf)

6.4 结果分析

Data Stock: 3596
Trant Data: 3320
target
0    1765
1    1555
Name: count, dtype: int64
Test Data: 242
target
1    129
0    113
Name: count, dtype: int64

最佳参数: {'num_leaves': 30, 'lr': 0.06695552617031866, 'max_depth': 5, 'min_child_samples': 72, 'reg_alpha': 6.118528951105265, 'reg_lambda': 1.3949386151254795, 'feature_frac': 0.7022506269873263, 'bagging_freq': 4, 'objective': 'binary', 'metric': 'auc', 'boosting_type': 'goss', 'verbosity': -1, 'force_col_wise': True, 'random_state': 42, 'num_threads': 1, 'deterministic': True}
最佳AUC: 0.5476021776135598
Selected features: ['volatility_5' 'volume_change' 'macd' 'obv' 'bb_width' 'bb_position'
 'pv_divergence']

过拟合检测:
Train AUC: 0.4930
Test AUC: 0.4848

预测概率分布:
count    242.000000
mean       0.426774
std        0.095937
min        0.211996
25%        0.356984
50%        0.419012
75%        0.491315
max        0.729235
Name: pred_proba, dtype: float64

信号生成关键指标统计：
预测概率 > 0.5 的比例: 23.14%
成交量过滤触发比例: 39.67%
处于200日均线上方比例: 17.77%

信号统计:
做多信号数量: 115
做空信号数量: 115

策略绩效:
Start                               2024-01-02 00:00:00
End                                 2024-12-31 00:00:00
Period                                242 days 00:00:00
Start Value                                    100000.0
End Value                                 115254.345896
Total Return [%]                              15.254346
Benchmark Return [%]                          55.909091
Max Gross Exposure [%]                        80.670607
Total Fees Paid                             1845.762877
Max Drawdown [%]                               5.532785
Max Drawdown Duration                  58 days 00:00:00
Total Trades                                         24
Total Closed Trades                                  24
Total Open Trades                                     0
Open Trade PnL                                      0.0
Win Rate [%]                                  54.166667
Best Trade [%]                                 6.547092
Worst Trade [%]                               -4.102633
Avg Winning Trade [%]                          3.123334
Avg Losing Trade [%]                          -1.189157
Avg Winning Trade Duration    7 days 01:50:46.153846153
Avg Losing Trade Duration     2 days 06:32:43.636363636
Profit Factor                                  2.304125
Expectancy                                   635.597746
Sharpe Ratio                                   1.954995
Calmar Ratio                                   4.315798
Omega Ratio                                    1.457148
Sortino Ratio                                  3.287022
dtype: object

总结

通过本章，你已经学习了以下内容：

数据加载与预处理
- 数据加载: 使用Parquet格式高效读取数据，支持列式存储优化。
- 特征工程: 集成TA-Lib技术指标（MACD、OBV、布林带等），并严格通过shift(1)避免未来信息泄露。例如：
```
data["macd"] = data["macd"].shift(1)  # 滞后一期
```
- 标签生成: 基于持有期收益的二分类标签，确保时间对齐：
```
buy_price = data["open"].shift(-1)  # t+1日开盘价买入
sell_price = data["close"].shift(-HOLDING_PERIOD)  # t+HOLDING_PERIOD收盘卖出
```
模型训练与优化
- 特征选择: 采用递归特征消除（RFE）筛选5个关键特征，降低过拟合风险。
- 超参数优化: 使用Optuna框架进行贝叶斯优化，结合时间序列交叉验证（TSCV）确保时序依赖性。
- 模型选择: LightGBM分类器，兼顾效率与精度，启用deterministic=True保证可复现性。
信号生成与风控
- 信号逻辑: 基于预测概率动态生成买卖信号，并严格限制持仓周期。
- 仓位管理: 波动率驱动的动态仓位调整（50%-80%），结合固定止损（5%）控制下行风险。
回测验证
- 实盘细节: 考虑交易成本（佣金0.05%+滑点0.15%），使用vectorbt进行事件驱动回测。

风险提示与免责声明
本文内容基于公开信息研究整理，不构成任何形式的投资建议。历史表现不应作为未来收益保证，市场存在不可预见的波动风险。投资者需结合自身财务状况及风险承受能力独立决策，并自行承担交易结果。作者及发布方不对任何依据本文操作导致的损失承担法律责任。市场有风险，投资须谨慎。

VectorBT量化入门系列：第六章 VectorBT实战案例：机器学习预测策略

第六章 VectorBT实战案例：机器学习预测策略

6.1 策略逻辑

6.2 实现步骤

6.3 代码执行

6.4 结果分析

总结

网站公告

今日签到

热门文章

最新发布