Python XGBoost实战：股票走势预测的完整量化策略实现

免责声明：本文所有内容仅用于交流学习，不构成任何投资建议！投资有风险，入市需谨慎！

策略介绍

XGBoost 是一种高效、灵活的梯度提升机器学习算法，它在传统梯度提升决策树的基础上，引入了二阶导数优化、正则化项和并行计算等创新，显著提升了模型的性能和速度。该算法能自动处理缺失值，支持自定义损失函数，并提供了丰富的超参数以控制过拟合。其核心优势在于精度高、可解释性强，在各大数据科学竞赛中表现出色，广泛应用于分类、回归等任务，是许多量化研究者的重要工具。

Optuna 是一个专为机器学习设计的自动超参数优化框架。它采用基于贝叶斯优化的 TPE 算法，能智能地搜索超参数空间，相比传统的网格搜索和随机搜索，效率更高、效果更好。它允许用户通过简单的代码定义搜索空间，并支持并行计算、提前剪枝等高级功能，大幅降低了调参的时间和计算成本。其设计灵活，可与 XGBoost、神经网络等多种模型无缝集成。

基于 XGBoost 预测股票走势的完整流程通常包括以下步骤：

• 数据准备：收集历史股价、成交量、财务指标、市场情绪等多维度数据。
• 特征工程：构造技术指标（如移动平均线、RSI、MACD）、滞后特征及市场相关变量，将股票预测问题转化为监督学习问题，例如预测未来N日的涨跌方向。
• 模型训练与调优：使用 Optuna 优化 XGBoost 的超参数（如 max_depth, learning_rate），并通过交叉验证或时间序列验证来避免过拟合，常以夏普比率或信息系数作为评估指标。
• 回测与验证：在历史数据上模拟交易，评估策略的盈利能力、稳定性和风险，整个过程需严格避免未来数据泄露。

注意：股票市场受宏观政策、突发事件等复杂因素影响，仅依靠历史数据预测存在固有局限性。建议将此模型作为多因子模型的一部分，并始终结合严格的风险管理，持续迭代特征与模型。

策略实现

获取A股数据

首先，我们需要获取股票的历史行情数据。以下是一个使用 akshare 库获取A股日线数据的函数示例：

def get_stock_data_ashares(symbol='000001', start_date='20180101', end_date=None, adjust='qfq'):
    """
    获取A股历史数据
    参数：
    symbol：股票代码，如'000001'（平安银行），'600519'（贵州茅台）
    start_date：开始日期，格式'yyyyMMdd'
    end_date：结束日期，默认为今天
    adjust：复权方式，'qfq'前复权，'hfq'后复权，''不复权
    """
    if end_date is None:
        end_date = datetime.now().strftime('%Y%m%d')
    try:
        # 获取日线数据
        df = ak.stock_zh_a_hist(symbol=symbol, period='daily', start_date=start_date, end_date=end_date, adjust=adjust)

        if df.empty:
            print(f"未获取到数据：{symbol}")
            return pd.DataFrame()
        # 重命名列以保持一致
        df = df.rename(columns={
            '日期': 'date',
            '开盘': 'open',
            '最高': 'high',
            '最低': 'low',
            '收盘': 'close',
            '成交量': 'volume',
            '成交额': 'amount',
            '振幅': 'amplitude',
            '涨跌幅': 'pct_chg',
            '涨跌额': 'change',
            '换手率': 'turnover'
        })
        # 确保数据类型
        df['date'] = pd.to_datetime(df['date'])
        numeric_cols = ['open', 'high', 'low', 'close', 'volume', 'amount', 'pct_chg', 'change', 'turnover', 'amplitude']
        for col in numeric_cols:
            if col in df.columns:
                df[col] = pd.to_numeric(df[col], errors='coerce')
        # 按日期排序
        df = df.sort_values('date').reset_index(drop=True)
        print(f"股票 {symbol}：获取到 {len(df)} 条数据，从 {df['date'].min()} 到 {df['date'].max()}")
        return df
    except Exception as e:
        print(f"获取数据时出错：{e}")
        return pd.DataFrame()

特征工程

原始价格和成交量数据信息有限，我们需要从中构造出对预测未来走势有用的特征。这包括技术指标、统计特征和时序特征。

def calculate_technical_indicators(df):
    """计算技术指标"""
    if df.empty:
        return df
    # 复制数据，避免修改原始数据
    df = df.copy()
    # 移动平均线
    windows = [5, 10, 20, 60]
    for window in windows:
        df[f‘sma_{window}’] = df[‘close’].rolling(window=window).mean()
        df[f‘ema_{window}’] = df[‘close’].ewm(span=window, adjust=False).mean()
        # 价格与均线的关系
        df[f‘close_vs_sma_{window}’] = df[‘close’] / df[f‘sma_{window}’]
        df[f‘close_vs_ema_{window}’] = df[‘close’] / df[f‘ema_{window}’]
    # 计算收益率
    df[‘returns_1’] = df[‘close’].pct_change(1)
    df[‘returns_5’] = df[‘close’].pct_change(5)
    df[‘returns_10’] = df[‘close’].pct_change(10)
    # RSI (相对强弱指数)
    delta = df[‘close’].diff()
    gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
    loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
    rs = gain / loss
    df[‘rsi_14’] = 100 - (100 / (1 + rs))
    # MACD
    exp1 = df[‘close’].ewm(span=12, adjust=False).mean()
    exp2 = df[‘close’].ewm(span=26, adjust=False).mean()
    df[‘macd’] = exp1 - exp2
    df[‘macd_signal’] = df[‘macd’].ewm(span=9, adjust=False).mean()
    df[‘macd_hist’] = df[‘macd’] - df[‘macd_signal’]
    # 布林带
    df[‘bb_middle’] = df[‘close’].rolling(window=20).mean()
    bb_std = df[‘close’].rolling(window=20).std()
    df[‘bb_upper’] = df[‘bb_middle’] + (bb_std * 2)
    df[‘bb_lower’] = df[‘bb_middle’] - (bb_std * 2)
    df[‘bb_width’] = (df[‘bb_upper’] - df[‘bb_lower’]) / df[‘bb_middle’]
    df[‘bb_position’] = (df[‘close’] - df[‘bb_lower’]) / (df[‘bb_upper’] - df[‘bb_lower’])
    # 成交量相关
    df[‘volume_sma_20’] = df[‘volume’].rolling(window=20).mean()
    df[‘volume_ratio’] = df[‘volume’] / df[‘volume_sma_20’]
    # 价格波动特征
    df[‘high_low_ratio’] = (df[‘high’] - df[‘low’]) / df[‘close’]
    df[‘close_open_ratio’] = (df[‘close’] - df[‘open’]) / df[‘open’]
    return df.replace([np.inf, -np.inf], np.nan)

除此之外，还可以构造更丰富的特征，如高级价格位置特征、成交量加权平均价格、以及日期相关的周期性特征。

    # 收益率相关
    df[‘returns_5’] = df[‘close’].pct_change(5)
    df[‘returns_10’] = df[‘close’].pct_change(10)
    df[‘returns_20’] = df[‘close’].pct_change(20)
    # 价格位置特征
    df[‘high_low_ratio’] = (df[‘high’] - df[‘low’]) / df[‘close’]
    df[‘close_open_ratio’] = (df[‘close’] - df[‘open’]) / df[‘open’]
    # 成交量特征
    df[‘vwap’] = (df[‘amount’] / df[‘volume’]).replace([np.inf, -np.inf], np.nan)
    df[‘price_vs_vwap’] = df[‘close’] / df[‘vwap’]
    # 振幅特征
    if ‘turnover’ in df.columns:
        df[‘turnover_sma_20’] = df[‘turnover’].rolling(window=20).mean()
        df[‘turnover_ratio’] = df[‘turnover’] / df[‘turnover_sma_20’]
    # 振幅特征
    if ‘amplitude’ in df.columns:
        df[‘amplitude_sma_20’] = df[‘amplitude’].rolling(window=20).mean()
        df[‘amplitude_ratio’] = df[‘amplitude’] / df[‘amplitude_sma_20’]
    # 日历特征
    if isinstance(df.index, pd.DatetimeIndex):
        df[‘day_of_week’] = df.index.dayofweek
        df[‘month’] = df.index.month
        df[‘quarter’] = df.index.quarter
        df[‘year’] = df.index.year
        # 周期性特征
        df[‘day_sin’] = np.sin(2 * np.pi * df.index.dayofyear / 365.25)
        df[‘day_cos’] = np.cos(2 * np.pi * df.index.dayofyear / 365.25)
        df[‘month_sin’] = np.sin(2 * np.pi * df.index.month / 12)
        df[‘month_cos’] = np.cos(2 * np.pi * df.index.month / 12)
    # 清除NaN值
    df = df.replace([np.inf, -np.inf], np.nan)
return df

模型训练与优化

接下来，我们构建一个 StockPredictor 类来封装数据准备、模型训练和预测的全过程。这个类将使用 XGBoost 作为核心模型，并集成 Optuna 进行超参数优化。

class StockPredictor:
    def __init__(self, random_state=42):
        self.random_state = random_state
        self.model = None
        self.best_params = None
        self.feature_importance = None
        self.scaler = StandardScaler()

在训练模型前，超参数的选择至关重要。我们使用 Optuna 框架来自动寻找最优参数组合。

    def optimize_hyperparameters(self, X_train, y_train, X_val, y_val, n_trials=50):
        """使用Optuna优化超参数"""
        def objective(trial):
            param = {
                ‘n_estimators’: trial.suggest_int(‘n_estimators’, 100, 1000),
                ‘max_depth’: trial.suggest_int(‘max_depth’, 3, 10),
                ‘learning_rate’: trial.suggest_loguniform(‘learning_rate’, 0.01, 0.3),
                ‘subsample’: trial.suggest_uniform(‘subsample’, 0.6, 1.0),
                ‘colsample_bytree’: trial.suggest_uniform(‘colsample_bytree’, 0.6, 1.0),
                ‘gamma’: trial.suggest_loguniform(‘gamma’, 1e-8, 1.0),
                ‘reg_alpha’: trial.suggest_loguniform(‘reg_alpha’, 1e-8, 1.0),
                ‘reg_lambda’: trial.suggest_loguniform(‘reg_lambda’, 1e-8, 1.0),
                ‘min_child_weight’: trial.suggest_int(‘min_child_weight’, 1, 10),
                ‘random_state’: self.random_state
            }
            model = xgb.XGBClassifier(**param)
            model.fit(X_train, y_train, eval_set=[(X_val, y_val)], early_stopping_rounds=50, verbose=False)
            y_pred = model.predict(X_val)
            score = accuracy_score(y_val, y_pred)
            return score
        # 创建超参数研究
        study = optuna.create_study(
            direction=‘maximize’,
            sampler=optuna.samplers.TPESampler(seed=self.random_state)
        )
        study.optimize(objective, n_trials=n_trials, show_progress_bar=True)
        self.best_params = study.best_params
        print(f”最佳参数：{self.best_params}“)
        print(f”最佳得分：{study.best_value:.4f}“)
        return self.best_params
    def train(self, X_train, y_train, X_val=None, y_val=None, params=None):
        """训练模型"""
        if params is None and self.best_params is not None:
            params = self.best_params
        if params is None:
            # 默认参数
            params = {
                ’n_estimators‘: 500,
                ’max_depth‘: 6,
                ’learning_rate‘: 0.1,
                ’subsample‘: 0.8,
                ’colsample_bytree‘: 0.8,
                ’random_state‘: self.random_state,
                ’n_jobs‘: -1
            }
        # 添加回调
        callbacks = []
        if X_val is not None and y_val is not None:
            eval_set = [(X_val, y_val)]
            callbacks = [xgb.callback.EarlyStopping(rounds=50)]
        else:
            eval_set = None
        # 训练模型
        self.model = xgb.XGBClassifier(**params)
        self.model.fit(
            X_train, y_train,
            eval_set=eval_set,
            verbose=False,
            callbacks=callbacks
        )
        # 特征重要性
        self.feature_importance = pd.DataFrame({
            ‘feature’: X_train.columns,
            ‘importance’: self.model.feature_importances_
        }).sort_values(‘importance’, ascending=False)
        return self.model
    def predict(self, X, threshold=0.5):
        """预测"""
        if self.model is None:
            raise ValueError(“模型未训练”)
        y_pred_proba = self.model.predict_proba(X)[:, 1]
        y_pred = (y_pred_proba >= threshold).astype(int)
        return y_pred, y_pred_proba

对于时间序列数据，标准的随机交叉验证会导致数据泄露。我们采用前向验证来模拟真实交易场景下的滚动预测。

    def mlk_forward_validation(self, X, y, train_size=.7, step_size=0.1, optimizer=True):
        """前向验证"""
        n_samples = len(X)
        train_end = int(n_samples * train_size)
        all_predictions = []
        all_probs = []
        all_indices = []
        models_info = []
        for i in range(train_end, n_samples, step_size):
            # 训练集
            X_train = X.iloc[:i]
            y_train = y.iloc[:i]
            # 测试集
            test_start = i
            test_end = min(i + step_size, n_samples)
            X_test = X.iloc[test_start:test_end]
            y_test = y.iloc[test_start:test_end]
            if len(X_test) == 0:
                continue
            # 划分验证集
            val_size = int(0.2 * len(X_train))
            X_train_final = X_train.iloc[:-val_size]
            y_train_final = y_train.iloc[:-val_size]
            X_val = X_train.iloc[-val_size:]
            y_val = y_train.iloc[-val_size:]
            if optimizer:
                # 优化参数
                best_params = self.optimize_hyperparameters(
                    X_train_final, y_train_final, X_val, y_val, n_trials=10
                )
            else:
                best_params = None
            # 训练模型
            self.train(X_train_final, y_train_final, X_val, y_val, best_params)
            # 预测
            y_pred, y_pred_proba = self.predict(X_test)
            # 保存结果
            all_predictions.extend(y_pred)
            all_probs.extend(y_pred_proba)
            all_indices.extend(X_test.index.tolist())
            # 保存模型信息
            models_info.append({
                ‘train_end’: i,
                ‘test_start’: test_start,
                ‘test_end’: test_end,
                ‘train_samples’: len(X_train_final),
                ‘val_samples’: len(X_val),
                ‘test_samples’: len(X_test)
            })
        # 创建结果DataFrame
        results_df = pd.DataFrame({
            ‘prediction’: all_predictions,
            ‘probability’: all_probs
        }, index=all_indices)
        return results_df, models_info

完整流程整合

现在，我们将上述所有步骤整合到一个主函数中，实现端到端的策略回测流程。

def run_complete_pipeline(symbol=’000001‘, start_date=’20150101‘, initial_capital=100000):
    """
    运行完整的机器学习交易策略流程
    """
    print(f”开始运行 {symbol} 的量化策略...“)
    print(”*“ * 60)
    # 1. 获取数据
    print(”步骤1: 获取股票数据...“)
    stock_data = get_stock_data_ashares(
        symbol=symbol,
        start_date=start_date,
        adjust=’qfq‘
    )
    if stock_data.empty:
        print(f”无法获取 {symbol} 的数据“)
        return
    # 2. 特征工程
    print(”步骤2: 特征工程...“)
    features_df = calculate_technical_indicators(stock_data)
    # 3. 创建标签 (例如：预测下一日是否上涨)
    print(”步骤3: 创建预测标签...“)
    features_df[‘target’] = (features_df[‘close’].shift(-1) > features_df[‘close’]).astype(int)
    features_df = features_df.dropna(subset=[‘target’])
    # 4. 准备训练数据
    print(”步骤4: 准备训练数据...“)
    predictor = StockPredictor(random_state=42)
    X = features_df.drop(columns=[‘target’, ‘date’]).select_dtypes(include=[np.number])
    y = features_df[‘target’]
    X = X.fillna(X.mean())
    # 5. 模型验证
    print(”步骤5: 运行模型验证...“)
    w_results, models_info = predictor.mlk_forward_validation(
        X, y,
        train_size=0.7,
        step_size=30,
        optimizer=True
    )
    # 6. 评估策略 (这里需要实现evaluate_strategy函数，计算收益、夏普比率等)
    print(”步骤6: 评估策略表现...“)
    # results = evaluate_strategy(features_df, w_results, initial_capital)
    # 7. 可视化
    # print(”步骤7: 可视化结果...“)
    # plot_results(results, symbol=symbol)
    print(”*“ * 60)
    print(”策略总结报告:“)
    print(”*“ * 60)
    # 打印关键指标和特征重要性
    if predictor.feature_importance is not None:
        print(f”\n最重要的5个特征:“)
        for i, row in predictor.feature_importance.head(5).iterrows():
            print(f”  {row[‘feature’]}: {row[‘importance’]:.4f}“)
    return {
        ’symbol‘: symbol,
        ’data‘: features_df,
        ’predictions‘: w_results,
        ’predictor‘: predictor,
        # ’results‘: results,
        ’feature_importance‘: predictor.feature_importance
    }
# 运行完整流程
result = run_complete_pipeline(symbol=’000001‘, start_date=’20180101‘)

多股票回测与比较

一个稳健的策略应该在多只股票上表现良好。我们可以扩展流程，方便地比较不同股票上的策略表现。

def compare_multiple_stocks(symbols, start_date=’20180101‘, initial_capital=100000):
    """比较多只股票的表现"""
    all_results = {}
    for symbol in symbols:
        print(f”处理股票 {symbol}...“)
        print(”.” * 50)
        try:
            result = run_complete_pipeline(
                symbol=symbol,
                start_date=start_date,
                initial_capital=initial_capital
            )
            if result is not None:
                # 这里假设评估函数返回了关键指标
                all_results[symbol] = {
                    ’annual_return‘: 0.0, # 替换为实际计算结果
                    ’sharpe‘: 0.0,
                    ’max_drawdown‘: 0.0,
                    ’win_rate‘: 0.0
                }
        except Exception as e:
            print(f”处理股票 {symbol} 时出错: {e}“)
            continue
    # 创建比较结果并可视化
    if all_results:
        comparison_df = pd.DataFrame(all_results).T
        comparison_df = comparison_df.sort_values(’sharpe‘, ascending=False)
        print(”\n“ + ”-“ * 60)
        print(”多只股票表现对比:“)
        print(”-“ * 60)
        print(comparison_df)
        # 可使用plotly或matplotlib进行可视化
        # fig = go.Figure(data=[...])
        # fig.show()
    return comparison_df
# 示例：比较多只股票
# symbol_list = [’000001‘, ’000002‘, ’300750‘, ’600519‘]
# comparison_result = compare_multiple_stocks(symbol_list, start_date=’20220101‘)

总结

本文详细介绍了使用 XGBoost 和 Optuna 构建股票走势预测模型的完整流程，从数据获取、特征工程、模型训练优化到策略回测。通过将金融问题转化为机器学习中的分类任务，并采用严谨的前向验证方法，我们可以在一定程度上利用历史规律。然而，必须再次强调，真实的金融市场极其复杂，任何基于历史数据的模型都有其局限性。成功的 人工智能 量化策略需要融合更多维度的数据、更精细的特征工程、严格的风险管理以及持续的迭代验证。本文代码提供了一个可供学习和扩展的框架，希望能在 云栈社区 中激发更多关于量化交易与机器学习的讨论与实践。