以下是一个基于多因子量化选股策略的Python实现示例。该策略结合动量、估值、波动率等因子进行股票筛选，并包含风险管理模块。
请注意：此为教学示例，不构成投资建议，实际应用需根据市场环境调整参数。
策略框架
```python
幻方风格多因子量化策略 v1.0
包含：
1. 数据预处理模块
2. 多因子合成模块
3. 组合优化模块
4. 风险管理模块
5. 回测验证模块
… …
一、数据准备模块
```python
import pandas as pd
import numpy as np
import yfinance as yf
import talib
from scipy import stats
from pypfopt import EfficientFrontier, risk_models, expected_returns
# 获取股票池数据（示例用沪深300成分股）
def get_data(tickers, start_date, end_date):
ohlc_data = yf.download(tickers, start=start_date, end=end_date)['Adj Close']
returns = ohlc_data.pct_change().dropna()
return ohlc_data, returns
# 技术指标计算
def calculate_technical_factors(data):
factors = pd.DataFrame(index=data.index)
for ticker in data.columns:
# 动量因子 (12个月收益率)
factors[f'{ticker}_MOM'] = data[ticker].pct_change(252)
# 波动率因子 (60日波动率)
factors[f'{ticker}_VOL'] = data[ticker].pct_change().rolling(60).std()
# RSI指标
factors[f'{ticker}_RSI'] = talib.RSI(data[ticker], timeperiod=14)
# MACD差值
macd, _, _ = talib.MACD(data[ticker])
factors[f'{ticker}_MACD'] = macd
return factors.dropna()
# 示例数据获取
tickers = ['600519.SS', '000858.SZ', '601318.SS'] # 茅台、五粮液、平安
ohlc_data, returns = get_data(tickers, '2020-01-01', '2023-12-31')
tech_factors = calculate_technical_factors(ohlc_data)
```
二、因子合成模块
```python
class FactorModel:
def __init__(self, factors):
self.factors = factors
def zscore_normalization(self):
"""因子标准化处理"""
return self.factors.apply(stats.zscore)
def composite_score(self, weights):
"""
合成多因子评分
:param weights: 因子权重字典 {'MOM':0.4, 'VOL':-0.3, ...}
"""
normalized = self.zscore_normalization()
scores = pd.DataFrame(index=normalized.index)
for ticker in [col.split('_')[0] for col in normalized.columns]:
ticker_factors = normalized[[col for col in normalized.columns if ticker in col]]
score = 0
for factor, weight in weights.items():
score += ticker_factors[f'{ticker}_{factor}'] * weight
scores[ticker] = score
return scores
# 设置因子权重（需根据实际检验调整）
factor_weights = {
'MOM': 0.4, # 动量因子正向
'VOL': -0.3, # 波动率因子负向
'RSI': 0.2, # RSI正向
'MACD': 0.1 # MACD正向
}
model = FactorModel(tech_factors)
scores = model.composite_score(factor_weights)
```
三、组合优化模块
```python
def portfolio_optimization(returns, scores, top_n=2):
"""均值-方差优化组合"""
# 选择评分前top_n的股票
selected = scores.iloc[-1].nlargest(top_n).index.tolist()
# 计算预期收益和协方差矩阵
mu = expected_returns.mean_historical_return(returns[selected])
S = risk_models.sample_cov(returns[selected])
# 有效前沿优化
ef = EfficientFrontier(mu, S)
ef.add_constraint(lambda w: w >= 0.1) # 单只股票最低仓位10%
ef.add_constraint(lambda w: w <= 0.5) # 单只股票最高仓位50%
weights = ef.max_sharpe()
return ef.clean_weights()
# 运行优化
optimal_weights = portfolio_optimization(returns, scores)
print("优化后仓位分配:", optimal_weights)
```
四、风险管理模块
```python
class RiskManager:
def __init__(self, portfolio_value=1e6):
self.portfolio_value = portfolio_value
self.max_drawdown = 0.2 # 最大回撤阈值
def position_sizing(self, weights, volatility):
"""波动率调整仓位"""
scaled_weights = {}
for ticker, w in weights.items():
# 波动率倒数加权（波动率越高仓位越低）
scaled_weights[ticker] = w * (1/volatility[ticker])
total = sum(scaled_weights.values())
return {k:v/total for k,v in scaled_weights.items()}
def dynamic_stoploss(self, current_drawdown):
"""动态止损"""
if current_drawdown >= self.max_drawdown:
return "触发清仓止损"
return None
# 使用示例
volatility = tech_factors[[col for col in tech_factors.columns if 'VOL' in col]].iloc[-1]
risk_manager = RiskManager()
adjusted_weights = risk_manager.position_sizing(optimal_weights, volatility)
```
五、回测验证模块
```python
import backtrader as bt
class QuantStrategy(bt.Strategy):
params = (
('rebalance_days', 30), # 调仓周期
('top_n', 2) # 持仓数量
)
def __init__(self):
self.inds = {}
for d in self.datas:
self.inds[d] = {
'rsi': bt.indicators.RSI_Safe(d.close, period=14),
'macd': bt.indicators.MACDHisto(d.close)
}
def next(self):
# 每30天调仓
if len(self.data) % self.params.rebalance_days != 0:
return
# 计算因子评分
scores = {}
for d in self.datas:
mom = (d.close[0] - d.close[-252]) / d.close[-252]
vol = np.std([d.close[i]/d.close[i-1]-1 for i in range(-60,0)])
score = 0.4*mom - 0.3*vol + 0.2*self.inds[d]['rsi'] + 0.1*self.inds[d]['macd']
scores[d] = score
# 选择前top_n
selected = sorted(scores.items(), key=lambda x: x[1], reverse=True)[:self.params.top_n]
# 调整仓位
for d, _ in selected:
self.order_target_percent(d, target=1.0/self.params.top_n)
for d in self.datas:
if d not in [x[0] for x in selected]:
self.order_target_percent(d, target=0)
# 运行回测
cerebro = bt.Cerebro()
for ticker in tickers:
data = bt.feeds.PandasData(dataname=ohlc_data[ticker])
cerebro.adddata(data, name=ticker)
cerebro.addstrategy(QuantStrategy)
cerebro.broker.setcash(1000000)
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe')
results = cerebro.run()
print("夏普比率:", results[0].analyzers.sharpe.get_analysis())
```
六、策略优化方向
1. 因子增强：
- 加入基本面因子（PE、PB、ROE）
- 引入市场情绪因子（新闻舆情、搜索指数）
- 增加行业轮动因子
2. 风险控制升级：
```python
# 波动率预警示例
def volatility_alert(rolling_vol):
if rolling_vol > 0.4:
print("市场波动剧烈，启动防御模式")
return 0.5 # 仓位上限降为50%
return 1.0
```
3. 机器学习集成：
```python
from sklearn.ensemble import RandomForestClassifier
# 特征工程
X = tech_factors.dropna()
y = (returns.shift(-1) > 0).astype(int) # 次日涨跌标签
# 训练预测模型
model = RandomForestClassifier(n_estimators=100)
model.fit(X.iloc[:-100], y.iloc[:-100])
predictions = model.predict_proba(X.iloc[-100:])[:,1]
```
---
注意事项
1. 过拟合风险：避免在单一市场周期过度优化参数
2. 交易成本：实盘需考虑滑点和手续费（可在回测中增加2‰成本模拟）
3. 市场变化：需定期更新因子权重和股票池
4. 流动性风险：避免在小市值股票上应用高频策略
完整实盘系统还需接入以下模块：
- 实时行情接口（如通达信、TradeStation）
- 订单执行引擎（处理冰山订单、TWAP等算法）
- 异常监测系统（防止程序失控交易）
建议先用模拟盘运行3个月以上，再逐步投入实盘资金。量化交易的核心在于持续迭代——这个代码框架提供了基础结构，但需要根据实际交易数据进行持续优化。
