Python量化选股策略实战：基于VeighNa框架的多因子模型构建与回测

在量化投资领域摩根大通的选股策略一直备受关注。作为全球顶级投行其量化模型往往融合了复杂的金融理论和先进的数据分析技术。本文将基于公开的量化投资原理使用Python和VeighNa框架还原一个类似摩根大通风格的选股策略实现过程并提供完整的可执行源码。无论你是量化投资新手还是有一定编程基础的开发者都能通过本文掌握量化选股的核心流程。我们将从数据获取、因子计算、策略构建到回测验证完整展示一个量化选股策略的开发全流程。1. 量化选股策略基础概念1.1 什么是量化选股量化选股是通过数学模型和计算机算法从大量股票中筛选出具有投资价值的标的。与传统的基本面分析和技术分析不同量化选股依赖于历史数据和统计规律通过系统性的方法消除人为情绪干扰。核心特点包括数据驱动基于历史行情、财务数据等量化指标系统化建立统一的评价标准和筛选流程可回溯策略表现可以通过历史数据验证自动化整个流程可以通过程序自动执行1.2 多因子模型理论基础多因子模型是量化选股的核心方法论其基本思想是股票收益可以由多个共同因子解释。常见的因子类型包括价值因子市盈率、市净率等估值指标成长因子营收增长率、利润增长率等动量因子近期价格表现质量因子盈利能力、财务稳健性技术因子波动率、换手率等市场行为指标1.3 VeighNa框架简介VeighNa是一套基于Python的开源量化交易系统开发框架为专业量化交易员提供一站式解决方案。其vnpy.alpha模块专门针对AI量化策略设计支持多因子机器学习策略开发、投研和实盘交易。主要功能模块dataset因子特征工程支持高效批量特征计算model预测模型训练集成多种机器学习算法strategy基于ML信号的量化交易策略构建lab完整的投研流程管理2. 环境准备与工具配置2.1 系统要求与Python环境推荐使用VeighNa Studio-4.4.0这是专为量化交易打造的Python发行版集成了VeighNa框架和量化管理平台。系统要求Windows 11以上 / Windows Server 2022以上 / Ubuntu 22.04 LTS以上Python 3.10以上64位推荐Python 3.13安装步骤# Windows系统 install.bat # Ubuntu系统 bash install.sh # MacOS系统 bash install_osx.sh2.2 数据源配置本文示例使用TuShare作为数据源这是一个免费的金融数据接口库。# 安装TuShare pip install tushare需要先注册TuShare账号获取tokenimport tushare as ts ts.set_token(你的token) # 在tushare.pro网站注册获取 pro ts.pro_api()2.3 项目结构规划创建项目目录结构quant_stock_selection/ ├── data/ # 数据存储 ├── factors/ # 因子计算模块 ├── models/ # 机器学习模型 ├── strategies/ # 交易策略 ├── utils/ # 工具函数 ├── config.py # 配置文件 └── main.py # 主程序3. 数据获取与预处理3.1 股票池构建首先我们需要确定选股范围这里以沪深300成分股为例# utils/data_loader.py import tushare as ts import pandas as pd import os from datetime import datetime, timedelta class DataLoader: def __init__(self, token): ts.set_token(token) self.pro ts.pro_api() def get_hs300_stocks(self): 获取沪深300成分股 df self.pro.index_weight( index_code000300.SH, start_date20230101, end_datedatetime.now().strftime(%Y%m%d) ) return df[con_code].unique().tolist() def get_stock_daily(self, ts_code, start_date, end_date): 获取股票日线数据 df self.pro.daily( ts_codets_code, start_datestart_date, end_dateend_date ) return df def get_basic_info(self, ts_code): 获取股票基本信息 df self.pro.daily_basic( ts_codets_code, fieldsts_code,trade_date,close,turnover_rate,volume_ratio,pe,pb,ps,total_mv ) return df3.2 历史数据下载批量下载股票历史数据# main.py - 数据下载部分 from utils.data_loader import DataLoader import time def download_history_data(): loader DataLoader(你的tushare_token) # 获取沪深300成分股 stock_list loader.get_hs300_stocks() print(f获取到{len(stock_list)}只股票) # 设置时间范围最近3年 end_date datetime.now().strftime(%Y%m%d) start_date (datetime.now() - timedelta(days3*365)).strftime(%Y%m%d) all_data [] for i, stock in enumerate(stock_list): try: print(f下载 {stock} 数据 ({i1}/{len(stock_list)})) data loader.get_stock_daily(stock, start_date, end_date) if not data.empty: all_data.append(data) time.sleep(0.1) # 控制请求频率 except Exception as e: print(f下载{stock}失败: {e}) # 合并所有数据 if all_data: combined_data pd.concat(all_data, ignore_indexTrue) combined_data.to_csv(data/stock_daily.csv, indexFalse) print(数据下载完成) return combined_data3.3 数据清洗与整理数据质量对量化策略至关重要需要进行严格的清洗# utils/data_processor.py import pandas as pd import numpy as np class DataProcessor: staticmethod def clean_data(df): 数据清洗 # 去除重复数据 df df.drop_duplicates() # 按日期和股票代码排序 df df.sort_values([ts_code, trade_date]) # 处理缺失值 df df.fillna(methodffill) # 前向填充 # 去除仍然有缺失值的行 df df.dropna() return df staticmethod def calculate_returns(df): 计算收益率 df df.copy() df[prev_close] df.groupby(ts_code)[close].shift(1) df[daily_return] (df[close] - df[prev_close]) / df[prev_close] # 计算5日、20日收益率 df[return_5d] df.groupby(ts_code)[close].pct_change(5) df[return_20d] df.groupby(ts_code)[close].pct_change(20) return df staticmethod def calculate_volatility(df, window20): 计算波动率 df df.copy() df[volatility_20d] df.groupby(ts_code)[daily_return].rolling(window).std().values return df4. 因子计算与特征工程4.1 价值因子计算价值因子关注股票的估值水平# factors/value_factors.py import pandas as pd import numpy as np class ValueFactors: staticmethod def calculate_pe_ratio(df): 计算市盈率因子 df[pe_ratio] df[pe] # PE倒数EP避免极端值影响 df[ep_ratio] 1 / df[pe_ratio].clip(lower0.1, upper100) return df staticmethod def calculate_pb_ratio(df): 计算市净率因子 df[pb_ratio] df[pb] # PB倒数BP df[bp_ratio] 1 / df[pb_ratio].clip(lower0.1, upper20) return df staticmethod def calculate_ps_ratio(df): 计算市销率因子 df[ps_ratio] df[ps] df[sp_ratio] 1 / df[ps_ratio].clip(lower0.1, upper50) return df staticmethod def calculate_dividend_yield(df): 计算股息率因子简化版 # 实际应用中需要获取分红数据 df[dividend_yield] df[pe_ratio].apply( lambda x: 0.03 if x 15 else 0.01 # 简化假设 ) return df4.2 成长因子计算成长因子关注公司的增长潜力# factors/growth_factors.py import pandas as pd import numpy as np class GrowthFactors: staticmethod def calculate_momentum(df, windows[5, 20, 60]): 计算动量因子 for window in windows: df[fmomentum_{window}d] df.groupby(ts_code)[close].pct_change(window) return df staticmethod def calculate_volume_trend(df): 计算成交量趋势 df[volume_ma5] df.groupby(ts_code)[vol].rolling(5).mean().values df[volume_ma20] df.groupby(ts_code)[vol].rolling(20).mean().values df[volume_trend] df[volume_ma5] / df[volume_ma20] - 1 return df staticmethod def calculate_breakout_signals(df): 计算突破信号 df[high_20d] df.groupby(ts_code)[high].rolling(20).max().values df[low_20d] df.groupby(ts_code)[low].rolling(20).min().values df[breakout_high] (df[close] df[high_20d]).astype(int) df[breakout_low] (df[close] df[low_20d]).astype(int) return df4.3 质量因子计算质量因子关注公司的财务健康状况# factors/quality_factors.py import pandas as pd import numpy as np class QualityFactors: staticmethod def calculate_profitability(df): 计算盈利质量因子 # 使用换手率作为盈利质量的代理指标简化 df[profitability] 1 / df[turnover_rate].clip(lower0.1, upper100) return df staticmethod def calculate_stability(df, window60): 计算稳定性因子 df[return_stability] 1 / df.groupby(ts_code)[daily_return].rolling(window).std().values return df staticmethod def calculate_size_factor(df): 计算规模因子 df[size_factor] np.log(df[total_mv]) # 对数市值 return df4.4 因子标准化与合成将不同量纲的因子标准化并合成综合得分# factors/factor_combiner.py import pandas as pd import numpy as np from sklearn.preprocessing import StandardScaler class FactorCombiner: def __init__(self): self.scaler StandardScaler() def normalize_factors(self, df, factor_columns): 因子标准化 for col in factor_columns: # 去除极端值 df[col] df[col].clip( lowerdf[col].quantile(0.01), upperdf[col].quantile(0.99) ) # 标准化 df[col] self.scaler.fit_transform(df[[col]]) return df def combine_factors(self, df, factor_weights): 因子合成 df[composite_score] 0 total_weight sum(factor_weights.values()) for factor, weight in factor_weights.items(): if factor in df.columns: df[composite_score] df[factor] * weight / total_weight return df def calculate_factor_ranks(self, df, date_coltrade_date): 计算因子排名 df[factor_rank] df.groupby(date_col)[composite_score].rank( ascendingFalse, methoddense ) return df5. 机器学习模型构建5.1 特征工程完整流程# models/feature_engineer.py import pandas as pd import numpy as np from factors.value_factors import ValueFactors from factors.growth_factors import GrowthFactors from factors.quality_factors import QualityFactors from factors.factor_combiner import FactorCombiner class FeatureEngineer: def __init__(self): self.value_calculator ValueFactors() self.growth_calculator GrowthFactors() self.quality_calculator QualityFactors() self.combiner FactorCombiner() def create_features(self, df): 创建完整特征集 # 价值因子 df self.value_calculator.calculate_pe_ratio(df) df self.value_calculator.calculate_pb_ratio(df) df self.value_calculator.calculate_ps_ratio(df) df self.value_calculator.calculate_dividend_yield(df) # 成长因子 df self.growth_calculator.calculate_momentum(df) df self.growth_calculator.calculate_volume_trend(df) df self.growth_calculator.calculate_breakout_signals(df) # 质量因子 df self.quality_calculator.calculate_profitability(df) df self.quality_calculator.calculate_stability(df) df self.quality_calculator.calculate_size_factor(df) # 定义因子列 factor_columns [ ep_ratio, bp_ratio, sp_ratio, dividend_yield, momentum_5d, momentum_20d, momentum_60d, volume_trend, breakout_high, breakout_low, profitability, return_stability, size_factor ] # 因子标准化和合成 df self.combiner.normalize_factors(df, factor_columns) # 设置因子权重可根据研究调整 factor_weights { ep_ratio: 0.15, bp_ratio: 0.15, sp_ratio: 0.1, momentum_20d: 0.15, volume_trend: 0.1, profitability: 0.15, return_stability: 0.1, size_factor: 0.1 } df self.combiner.combine_factors(df, factor_weights) df self.combiner.calculate_factor_ranks(df) return df5.2 目标变量定义# models/target_definition.py import pandas as pd import numpy as np class TargetDefinition: staticmethod def calculate_future_returns(df, horizon20): 计算未来收益作为目标变量 df df.copy() df[future_return] df.groupby(ts_code)[close].shift(-horizon) / df[close] - 1 return df staticmethod def create_classification_target(df, threshold0.05): 创建分类目标是否超越基准 df[target_class] (df[future_return] threshold).astype(int) return df staticmethod def prepare_training_data(df, features, target_colfuture_return): 准备训练数据 # 去除缺失值 df_clean df.dropna(subsetfeatures [target_col]) X df_clean[features] y df_clean[target_col] return X, y, df_clean5.3 机器学习模型训练# models/ml_trainer.py import pandas as pd import numpy as np from sklearn.ensemble import RandomForestRegressor from sklearn.model_selection import TimeSeriesSplit, cross_val_score from sklearn.metrics import mean_squared_error, r2_score import warnings warnings.filterwarnings(ignore) class MLTrainer: def __init__(self): self.model RandomForestRegressor( n_estimators100, max_depth10, min_samples_split20, random_state42 ) def time_series_split(self, X, y, n_splits5): 时间序列交叉验证 tscv TimeSeriesSplit(n_splitsn_splits) return tscv.split(X) def train_model(self, X_train, y_train): 训练模型 self.model.fit(X_train, y_train) return self.model def predict(self, X): 预测 return self.model.predict(X) def evaluate_model(self, X_test, y_test): 模型评估 y_pred self.predict(X_test) mse mean_squared_error(y_test, y_pred) r2 r2_score(y_test, y_pred) print(f模型评估结果:) print(fMSE: {mse:.4f}) print(fR² Score: {r2:.4f}) return mse, r2 def get_feature_importance(self, feature_names): 获取特征重要性 importance pd.DataFrame({ feature: feature_names, importance: self.model.feature_importances_ }).sort_values(importance, ascendingFalse) return importance6. 策略实现与回测6.1 基于VeighNa的策略框架# strategies/jpm_style_strategy.py from vnpy_ctastrategy import ( CtaTemplate, StopOrder, TickData, BarData, TradeData, OrderData, BarGenerator, ArrayManager, ) class JPMStockSelectionStrategy(CtaTemplate): 摩根大通风格选股策略 author Quant Developer # 策略参数 top_n 10 # 选择前N只股票 rebalance_days 20 # 调仓周期 stop_loss 0.08 # 止损比例 take_profit 0.15 # 止盈比例 # 策略变量 current_holdings {} # 当前持仓 last_rebalance_date None # 上次调仓日期 portfolio_size 0 # 组合规模 def __init__(self, cta_engine, strategy_name, vt_symbol, setting): 初始化策略 super().__init__(cta_engine, strategy_name, vt_symbol, setting) # 创建K线合成器 self.bg BarGenerator(self.on_bar) self.am ArrayManager() # 初始化模型实际使用时需要加载训练好的模型 self.model None self.feature_engineer None def on_init(self): 策略初始化 self.write_log(策略初始化) self.load_bar(30) # 加载30天历史数据 def on_start(self): 策略启动 self.write_log(策略启动) def on_stop(self): 策略停止 self.write_log(策略停止) def on_tick(self, tick: TickData): Tick数据推送 self.bg.update_tick(tick) def on_bar(self, bar: BarData): K线数据推送 self.am.update_bar(bar) if not self.am.inited: return # 检查是否需要调仓 if self.need_rebalance(bar.datetime): self.rebalance_portfolio(bar) # 执行风险管理 self.risk_management(bar) def need_rebalance(self, current_date): 判断是否需要调仓 if self.last_rebalance_date is None: return True days_passed (current_date - self.last_rebalance_date).days return days_passed self.rebalance_days def rebalance_portfolio(self, bar): 投资组合再平衡 try: # 获取当前股票池的预测得分 stock_scores self.calculate_stock_scores(bar.datetime) # 选择得分最高的top_n只股票 selected_stocks stock_scores.nlargest(self.top_n, predicted_score) # 计算调仓逻辑 self.execute_rebalance(selected_stocks, bar) self.last_rebalance_date bar.datetime self.write_log(f投资组合再平衡完成选择{len(selected_stocks)}只股票) except Exception as e: self.write_log(f再平衡失败: {e}) def calculate_stock_scores(self, current_date): 计算股票预测得分 # 这里应该调用训练好的机器学习模型 # 简化版使用复合因子得分 scores [] # 实际应用中需要从数据库或API获取数据 # 这里使用模拟数据演示 for stock in self.get_stock_universe(): score self.calculate_composite_score(stock, current_date) scores.append({stock: stock, predicted_score: score}) return pd.DataFrame(scores) def calculate_composite_score(self, stock, date): 计算复合得分简化版 # 实际应用中应该使用训练好的模型预测 # 这里使用随机数模拟 import random return random.uniform(0, 1) def execute_rebalance(self, selected_stocks, bar): 执行调仓操作 # 计算目标权重等权重 target_weight 1.0 / len(selected_stocks) # 平仓不在选择列表中的股票 for stock, position in self.current_holdings.items(): if stock not in selected_stocks[stock].values: self.sell(stock, position.volume) # 调整现有持仓到目标权重 for _, stock_info in selected_stocks.iterrows(): stock stock_info[stock] target_value self.portfolio_size * target_weight # 计算需要调整的仓位 current_value self.current_holdings.get(stock, 0) * self.get_current_price(stock) adjustment target_value - current_value if adjustment 0: self.buy(stock, adjustment) elif adjustment 0: self.sell(stock, abs(adjustment)) def risk_management(self, bar): 风险管理 for stock, position in self.current_holdings.items(): current_price self.get_current_price(stock) cost_price position.price # 计算盈亏比例 pnl_ratio (current_price - cost_price) / cost_price # 止损检查 if pnl_ratio -self.stop_loss: self.write_log(f触发止损: {stock}, 盈亏: {pnl_ratio:.2%}) self.sell(stock, position.volume) # 止盈检查 elif pnl_ratio self.take_profit: self.write_log(f触发止盈: {stock}, 盈亏: {pnl_ratio:.2%}) self.sell(stock, position.volume)6.2 回测引擎配置# backtesting/backtest_engine.py from vnpy_ctabacktester import CtaBacktesterEngine import pandas as pd from datetime import datetime class BacktestEngine: def __init__(self): self.engine CtaBacktesterEngine() def setup_backtest(self, strategy_class, setting, vt_symbols, start_date, end_date, interval1d): 设置回测参数 self.engine.set_parameters( vt_symbolsvt_symbols, intervalinterval, startstart_date, endend_date, rate0.0003, # 手续费率 slippage0.001, # 滑点 size300, # 合约乘数 pricetick0.01, # 价格精度 capital1000000, # 初始资金 ) # 添加策略 self.engine.add_strategy(strategy_class, setting) def run_backtest(self): 运行回测 self.engine.run_backtesting() results self.engine.calculate_result() statistics self.engine.calculate_statistics() return results, statistics def show_chart(self): 显示回测图表 df pd.DataFrame(self.engine.daily_results) self.engine.show_chart()6.3 策略性能分析# analysis/performance_analyzer.py import pandas as pd import numpy as np import matplotlib.pyplot as plt class PerformanceAnalyzer: def __init__(self, daily_results): self.daily_results daily_results self.df pd.DataFrame(daily_results) def calculate_metrics(self): 计算绩效指标 if self.df.empty: return {} # 计算累计收益 self.df[cumulative_return] (1 self.df[return]).cumprod() # 年化收益率 total_days len(self.df) annual_return (self.df[cumulative_return].iloc[-1] ** (252/total_days) - 1) * 100 # 年化波动率 annual_volatility self.df[return].std() * np.sqrt(252) * 100 # 夏普比率 risk_free_rate 0.03 # 无风险利率假设3% sharpe_ratio (annual_return - risk_free_rate) / annual_volatility # 最大回撤 cumulative self.df[cumulative_return] running_max cumulative.expanding().max() drawdown (cumulative - running_max) / running_max max_drawdown drawdown.min() * 100 metrics { 年化收益率: f{annual_return:.2f}%, 年化波动率: f{annual_volatility:.2f}%, 夏普比率: f{sharpe_ratio:.2f}, 最大回撤: f{max_drawdown:.2f}%, 总交易日: total_days } return metrics def plot_performance(self): 绘制绩效图表 fig, (ax1, ax2) plt.subplots(2, 1, figsize(12, 10)) # 累计收益曲线 ax1.plot(self.df[date], self.df[cumulative_return]) ax1.set_title(累计收益曲线) ax1.set_ylabel(累计收益) ax1.grid(True) # 回撤曲线 cumulative self.df[cumulative_return] running_max cumulative.expanding().max() drawdown (cumulative - running_max) / running_max ax2.fill_between(self.df[date], drawdown * 100, 0, alpha0.3) ax2.set_title(回撤曲线) ax2.set_ylabel(回撤 (%)) ax2.set_xlabel(日期) ax2.grid(True) plt.tight_layout() plt.show()7. 完整策略执行流程7.1 主程序入口# main.py import pandas as pd from datetime import datetime, timedelta from utils.data_loader import DataLoader from utils.data_processor import DataProcessor from models.feature_engineer import FeatureEngineer from models.target_definition import TargetDefinition from models.ml_trainer import MLTrainer from strategies.jpm_style_strategy import JPMStockSelectionStrategy from backtesting.backtest_engine import BacktestEngine def main(): 主执行函数 print(开始执行摩根大通风格选股策略...) # 1. 数据准备阶段 print(阶段1: 数据准备) loader DataLoader(你的tushare_token) processor DataProcessor() # 下载数据 raw_data download_history_data() cleaned_data processor.clean_data(raw_data) processed_data processor.calculate_returns(cleaned_data) # 2. 特征工程 print(阶段2: 特征工程) feature_engineer FeatureEngineer() featured_data feature_engineer.create_features(processed_data) # 3. 模型训练 print(阶段3: 模型训练) target_def TargetDefinition() data_with_target target_def.calculate_future_returns(featured_data) # 准备特征列 feature_columns [ ep_ratio, bp_ratio, sp_ratio, dividend_yield, momentum_20d, volume_trend, profitability, return_stability, size_factor ] X, y, clean_data target_def.prepare_training_data( data_with_target, feature_columns ) # 训练模型 ml_trainer MLTrainer() ml_trainer.train_model(X, y) # 显示特征重要性 importance ml_trainer.get_feature_importance(feature_columns) print(特征重要性排序:) print(importance) # 4. 策略回测 print(阶段4: 策略回测) # 设置回测参数 backtester BacktestEngine() strategy_setting { top_n: 10, rebalance_days: 20, stop_loss: 0.08, take_profit: 0.15 } # 选择回测标的前50只股票 vt_symbols clean_data[ts_code].unique()[:50] backtester.setup_backtest( strategy_classJPMStockSelectionStrategy, settingstrategy_setting, vt_symbolsvt_symbols, start_date20230101, end_date20241231, interval1d ) # 运行回测 results, statistics backtester.run_backtest() print(回测结果统计:) for key, value in statistics.items(): print(f{key}: {value}) # 5. 性能分析 print(阶段5: 性能分析) analyzer PerformanceAnalyzer(results) metrics analyzer.calculate_metrics() print(策略绩效指标:) for key, value in metrics.items(): print(f{key}: {value}) # 绘制图表 analyzer.plot_performance() print(策略执行完成!) if __name__ __main__: main()7.2 配置文件# config.py 策略配置文件 # 数据源配置 TUSHARE_TOKEN 你的tushare_token # 策略参数配置 STRATEGY_CONFIG { top_n: 10, # 选股数量 rebalance_days: 20, # 调仓周期 stop_loss: 0.08, # 止损比例 take_profit: 0.15, # 止盈比例 max_position: 0.1, # 单只股票最大仓位 } # 因子权重配置 FACTOR_WEIGHTS { value_factors: 0.4, # 价值因子权重 growth_factors: 0.3, # 成长因子权重 quality_factors: 0.3, # 质量因子权重 } # 回测配置 BACKTEST_CONFIG { start_date: 20230101, end_date: 20241231, initial_capital: 1000000, transaction_cost: 0.001, # 交易成本 }8. 常见问题与优化建议8.1 数据质量问题处理问题1数据缺失或异常# 数据质量检查函数 def validate_data_quality(df): 数据质量验证 issues [] # 检查缺失值 missing_ratio df.isnull().sum() / len(df) for col, ratio in missing_ratio.items(): if ratio 0.1: # 缺失率超过10% issues.append(f列 {col} 缺失率: {ratio:.2%}) # 检查极端值 numeric_cols df.select_dtypes(include[np.number]).columns for col in numeric_cols: q1 df[col].quantile(0.01) q99 df[col].quantile(0.99) extreme_count ((df[col] q1) | (df[col] q99)).sum() if extreme_count len(df) * 0.05: # 极端值超过5% issues.append(f列 {col} 极端值较多) return issues解决方案使用多个数据源交叉验证建立数据质量监控机制对异常数据进行平滑处理8.2 过拟合问题防范问题2模型过拟合# 过拟合检测函数 def detect_overfitting(train_score, test_score, threshold0.1): 检测过拟合 score_gap train_score - test_score if score_gap threshold: print(f警告可能过拟合训练-测试得分差距: {score_gap:.3f}) return True return False # 使用交叉验证 def robust_cross_validation(model, X, y, cv_splits5): 稳健的交叉验证 from sklearn.model_selection import cross_val_score scores cross_val_score(model, X, y, cvcv_splits, scoringr2) return scores.mean(), scores.std()防范措施使用时间序列交叉验证简化模型复杂度增加正则化参数使用早停法8.3 策略稳定性提升问题3策略表现不稳定**优化建议