ETF双因子轮动策略：动量与质量因子的Python实现与回测
·
你是不是经常看到券商研报里各种高大上的量化策略，感觉逻辑清晰、收益诱人，但真要自己动手实现时却无从下手？今天这篇文章，我要帮你彻底解决这个问题。
我们将完整复现一个经典的ETF双因子轮动策略，从数据获取到策略回测，再到实盘信号生成，提供全套可运行的Python源码。这个策略的核心逻辑很简单： 动量（收益趋势）+ 质量（趋势稳定性/低波动） 双维度筛选ETF，结合加权仓位分配，在市场不同阶段动态调整持仓。
但真正有价值的不只是代码本身，而是整个工程化实现的思路。很多人在复现研报策略时容易踩的坑：数据源不稳定、因子计算有偏差、回测逻辑不严谨、实盘对接复杂。本文将逐一拆解这些难点，让你不仅能跑通这个策略，更能掌握量化策略开发的完整方法论。
1. 这篇文章真正要解决的问题
为什么券商研报的策略看起来很美，但自己就是复现不出来？问题通常出在三个层面：
数据层面 ：研报往往使用清洗过的理想数据，而真实环境中你需要自己处理数据缺失、复权、停牌等问题。比如ETF的规模数据、净值数据如何稳定获取？
工程层面 ：策略逻辑转化为可维护的代码需要合理的架构设计。单因子计算、多因子合成、权重分配、风控规则，这些模块如何组织才能既清晰又高效？
回测层面 ：简单的收益率计算很容易，但真实的交易成本、滑点、调仓频率等因素会显著影响最终收益。如何构建一个可信的回测系统？
本文将基于Python生态中成熟的工具链（pandas、akshare、backtrader等），提供一个工业级的解决方案。适合有一定Python基础，想要系统学习量化策略开发的读者。
2. 基础概念与核心原理
2.1 什么是ETF双因子轮动？
双因子轮动策略的核心思想是通过两个维度的指标来评估ETF的投资价值，并定期调整持仓。
动量因子 ：衡量资产的价格趋势强度。通常用过去一段时间（如20日、60日）的收益率来表示。动量效应基于行为金融学中的"追涨杀跌"心理，趋势一旦形成往往会持续。
质量因子 ：衡量趋势的稳定性和风险调整后的收益质量。常用波动率、夏普比率、最大回撤等指标。质量因子帮助过滤掉高波动、高风险的趋势，选择更稳健的标的。
2.2 策略核心逻辑拆解
策略的完整流程包括：
- 标的池选择 ：确定参与轮动的ETF范围（如宽基指数、行业ETF等）
- 因子计算 ：定期计算每个ETF的动量因子和质量因子
- 综合评分 ：将两个因子标准化后加权求和，得到综合得分
- 权重分配 ：根据得分分配仓位，得分越高权重越大
- 调仓执行 ：定期（如月度）调整持仓至新的权重分配
这种方法的优势在于：牛市时动量因子主导，抓住强势品种；震荡市时质量因子主导，控制回撤风险。
3. 环境准备与前置条件
3.1 Python环境要求
推荐使用Python 3.8+版本，主要依赖库包括：
# 创建conda环境（可选）
conda create -n quant_strategy python=3.8
conda activate quant_strategy
# 安装核心依赖
pip install pandas==1.5.3
pip install numpy==1.24.3
pip install akshare==1.10.0 # 免费金融数据接口
pip install backtrader==1.9.78.123 # 回测框架
pip install matplotlib==3.7.1 # 绘图
pip install requests==2.31.0 # 网络请求
3.2 数据源配置
我们使用akshare作为主要数据源，它提供了丰富的免费金融数据接口：
import akshare as ak
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')
# 测试数据接口是否正常
def test_data_source():
try:
# 测试获取ETF基本信息
etf_info = ak.fund_etf_spot_em()
print(f"成功获取{len(etf_info)}只ETF基本信息")
return True
except Exception as e:
print(f"数据接口测试失败: {e}")
return False
if __name__ == "__main__":
test_data_source()
4. ETF数据获取与处理模块
4.1 构建ETF标的池
首先需要确定策略操作的ETF范围。我们选择流动性好、规模较大的宽基指数ETF：
class ETFDataManager:
def __init__(self):
self.etf_pool = {
'510300': '沪深300ETF', # 沪深300
'510500': '中证500ETF', # 中证500
'159915': '创业板ETF', # 创业板
'512100': '中证1000ETF', # 中证1000
'518880': '黄金ETF', # 黄金（防御性品种）
'511010': '国债ETF', # 国债（防御性品种）
}
def get_etf_daily_data(self, etf_code, start_date, end_date):
"""
获取单只ETF的日线数据
"""
try:
# 通过akshare获取ETF日线数据
df = ak.fund_etf_hist_em(symbol=etf_code, period="每日",
start_date=start_date, end_date=end_date)
if df.empty:
return None
# 数据清洗和格式标准化
df['日期'] = pd.to_datetime(df['日期'])
df.set_index('日期', inplace=True)
df.sort_index(inplace=True)
# 重命名列以便后续处理
df.rename(columns={
'开盘': 'open',
'收盘': 'close',
'最高': 'high',
'最低': 'low',
'成交量': 'volume'
}, inplace=True)
return df[['open', 'high', 'low', 'close', 'volume']]
except Exception as e:
print(f"获取{etf_code}数据失败: {e}")
return None
def get_all_etf_data(self, start_date='20200101', end_date='20231231'):
"""
获取所有ETF标的的历史数据
"""
all_data = {}
for etf_code in self.etf_pool.keys():
print(f"正在获取{etf_code}数据...")
data = self.get_etf_daily_data(etf_code, start_date, end_date)
if data is not None and len(data) > 0:
all_data[etf_code] = data
# 添加延时避免请求过于频繁
import time
time.sleep(0.5)
print(f"成功获取{len(all_data)}只ETF数据")
return all_data
# 测试数据获取
if __name__ == "__main__":
data_manager = ETFDataManager()
etf_data = data_manager.get_all_etf_data('20230101', '20231231')
for code, data in etf_data.items():
print(f"{code}: {len(data)}个交易日数据")
4.2 数据质量检查与清洗
真实数据往往存在各种问题，需要进行严格的清洗：
def data_quality_check(etf_data):
"""
ETF数据质量检查与清洗
"""
cleaned_data = {}
for etf_code, df in etf_data.items():
# 检查数据完整性
original_length = len(df)
# 1. 去除重复日期
df = df[~df.index.duplicated(keep='first')]
# 2. 处理缺失值（前后填充）
df = df.fillna(method='ffill').fillna(method='bfill')
# 3. 检查价格合理性（去除明显异常值）
price_cols = ['open', 'high', 'low', 'close']
for col in price_cols:
# 价格应为正数且在一定范围内
df = df[df[col] > 0]
df = df[df[col] < 100] # 假设ETF价格不会超过100元
# 4. 确保时间序列连续性
all_dates = pd.date_range(start=df.index.min(), end=df.index.max(), freq='D')
df = df.reindex(all_dates)
df = df.fillna(method='ffill') # 向前填充非交易日
print(f"{etf_code}: 原始{original_length}条, 清洗后{len(df)}条")
cleaned_data[etf_code] = df
return cleaned_data
5. 双因子计算引擎
5.1 动量因子计算
动量因子我们采用经典的时间序列动量计算方法：
class FactorCalculator:
def __init__(self, lookback_periods=[20, 60, 120]):
self.lookback_periods = lookback_periods
def calculate_momentum_factor(self, price_series):
"""
计算动量因子 - 多个时间窗口的收益率
"""
momentum_factors = {}
for period in self.lookback_periods:
if len(price_series) < period:
continue
# 当前价格 / period天前价格 - 1
momentum = (price_series / price_series.shift(period) - 1)
momentum_factors[f'momentum_{period}'] = momentum
return pd.DataFrame(momentum_factors)
def calculate_quality_factor(self, price_series, window=20):
"""
计算质量因子 - 基于波动率和风险调整收益
"""
if len(price_series) < window:
return pd.DataFrame()
# 1. 波动率（年化）
returns = price_series.pct_change()
volatility = returns.rolling(window=window).std() * np.sqrt(252)
# 2. 夏普比率（简化版，假设无风险收益为0）
sharpe_ratio = returns.rolling(window=window).mean() / returns.rolling(window=window).std()
sharpe_ratio = sharpe_ratio * np.sqrt(252)
# 3. 最大回撤（滚动窗口）
rolling_max = price_series.rolling(window=window, min_periods=1).max()
drawdown = (price_series - rolling_max) / rolling_max
quality_factors = {
'volatility': volatility,
'sharpe_ratio': sharpe_ratio,
'max_drawdown': drawdown
}
return pd.DataFrame(quality_factors)
def calculate_comprehensive_score(self, etf_data):
"""
计算综合得分：动量因子 + 质量因子
"""
all_scores = {}
for etf_code, df in etf_data.items():
price_series = df['close']
# 计算动量因子
momentum_df = self.calculate_momentum_factor(price_series)
# 计算质量因子
quality_df = self.calculate_quality_factor(price_series)
if momentum_df.empty or quality_df.empty:
continue
# 因子标准化（Z-score）
factors_df = pd.concat([momentum_df, quality_df], axis=1)
factors_normalized = (factors_df - factors_df.mean()) / factors_df.std()
# 因子方向调整（我们希望动量越大越好，波动率越小越好）
factors_normalized['volatility'] = -factors_normalized['volatility'] # 波动率取负
factors_normalized['max_drawdown'] = -factors_normalized['max_drawdown'] # 回撤取负
# 综合得分（等权重）
factors_normalized['composite_score'] = factors_normalized.mean(axis=1)
all_scores[etf_code] = factors_normalized['composite_score']
return pd.DataFrame(all_scores)
# 测试因子计算
if __name__ == "__main__":
# 生成示例数据测试
dates = pd.date_range('2023-01-01', '2023-12-31', freq='D')
sample_data = pd.DataFrame({
'close': np.cumsum(np.random.randn(len(dates)) * 0.01) + 100
}, index=dates)
calculator = FactorCalculator()
momentum_factors = calculator.calculate_momentum_factor(sample_data['close'])
quality_factors = calculator.calculate_quality_factor(sample_data['close'])
print("动量因子示例:")
print(momentum_factors.tail())
print("\n质量因子示例:")
print(quality_factors.tail())
5.2 因子合成与权重分配
多个因子需要合理合成才能产生更好的效果：
class WeightAllocator:
def __init__(self, top_n=3, min_weight=0.1):
self.top_n = top_n # 选择前N只ETF
self.min_weight = min_weight # 最小权重
def calculate_weights(self, scores_df):
"""
基于综合得分计算权重分配
"""
weights_df = pd.DataFrame(index=scores_df.index, columns=scores_df.columns)
for date in scores_df.index:
daily_scores = scores_df.loc[date].dropna()
if len(daily_scores) == 0:
continue
# 选择得分最高的top_n只ETF
top_etfs = daily_scores.nlargest(self.top_n)
if len(top_etfs) == 0:
continue
# 基于得分计算权重（得分越高权重越大）
total_score = top_etfs.sum()
weights = top_etfs / total_score
# 确保每只ETF至少有最小权重
if len(weights) > 0:
min_adj_weights = weights.clip(lower=self.min_weight)
# 重新归一化
adj_weights = min_adj_weights / min_adj_weights.sum()
for etf, weight in adj_weights.items():
weights_df.loc[date, etf] = weight
return weights_df.fillna(0)
def generate_rebalance_dates(self, start_date, end_date, freq='M'):
"""
生成调仓日期（月末调仓）
"""
dates = pd.date_range(start=start_date, end=end_date, freq=freq)
# 调整为月末
rebalance_dates = [date.replace(day=1) + pd.offsets.MonthEnd(0) for date in dates]
return rebalance_dates
# 权重分配测试
def test_weight_allocation():
# 模拟得分数据
dates = pd.date_range('2023-01-01', '2023-03-01', freq='D')
etf_codes = ['510300', '510500', '159915', '512100']
# 生成随机得分
np.random.seed(42)
scores_data = np.random.randn(len(dates), len(etf_codes)) * 0.1 + 0.5
scores_df = pd.DataFrame(scores_data, index=dates, columns=etf_codes)
allocator = WeightAllocator(top_n=2, min_weight=0.2)
weights = allocator.calculate_weights(scores_df)
print("权重分配示例:")
print(weights.tail(10))
if __name__ == "__main__":
test_weight_allocation()
6. 回测系统实现
6.1 基于Backtrader的回测引擎
使用成熟的backtrader框架构建回测系统：
import backtrader as bt
import backtrader.analyzers as btanalyzers
class DualFactorRotationStrategy(bt.Strategy):
params = (
('rebalance_freq', 21), # 调仓频率（交易日）
('lookback_period', 60), # 回看周期
('top_n', 3), # 持仓数量
)
def __init__(self):
self.etf_data = {} # 存储各ETF数据
self.rebalance_day = 0
self.weights_df = None # 权重数据
# 记录交易日志
self.trade_log = []
def next(self):
# 调仓逻辑
if len(self.data) % self.params.rebalance_freq == 0:
self.rebalance_portfolio()
def rebalance_portfolio(self):
"""执行调仓操作"""
current_date = self.data.datetime.date(0)
if self.weights_df is None or current_date not in self.weights_df.index:
return
# 获取当前日期权重
daily_weights = self.weights_df.loc[current_date]
# 清空现有持仓
for data in self.datas:
self.close(data)
# 按新权重建仓
portfolio_value = self.broker.getvalue()
for etf_code, weight in daily_weights.items():
if weight > 0.01: # 权重大于1%才交易
data = self.get_etf_data(etf_code)
if data is not None:
target_value = portfolio_value * weight
price = data.close[0]
size = int(target_value / price)
if size > 0:
self.buy(data=data, size=size)
# 记录交易
self.trade_log.append({
'date': current_date,
'etf': etf_code,
'action': 'BUY',
'size': size,
'price': price,
'weight': weight
})
def get_etf_data(self, etf_code):
"""根据代码获取对应的data对象"""
for data in self.datas:
if data._name == etf_code:
return data
return None
def run_backtest(etf_data, weights_df, initial_cash=1000000):
"""
运行回测
"""
cerebro = bt.Cerebro()
cerebro.broker.setcash(initial_cash)
# 添加交易数据
for etf_code, data in etf_data.items():
# 将DataFrame转换为backtrader数据格式
data_df = data.reset_index()
data_df['datetime'] = pd.to_datetime(data_df['日期'])
data_df.set_index('datetime', inplace=True)
data_feed = bt.feeds.PandasData(
dataname=data_df,
datetime=None, # 已经设置为index
open='open',
high='high',
low='low',
close='close',
volume='volume'
)
cerebro.adddata(data_feed, name=etf_code)
# 添加策略
cerebro.addstrategy(DualFactorRotationStrategy, weights_df=weights_df)
# 添加分析器
cerebro.addanalyzer(btanalyzers.SharpeRatio, _name='sharpe')
cerebro.addanalyzer(btanalyzers.DrawDown, _name='drawdown')
cerebro.addanalyzer(btanalyzers.Returns, _name='returns')
cerebro.addanalyzer(btanalyzers.TradeAnalyzer, _name='trades')
# 运行回测
print('初始资金: %.2f' % cerebro.broker.getvalue())
results = cerebro.run()
print('最终资金: %.2f' % cerebro.broker.getvalue())
# 返回回测结果
strat = results[0]
return {
'sharpe': strat.analyzers.sharpe.get_analysis(),
'drawdown': strat.analyzers.drawdown.get_analysis(),
'returns': strat.analyzers.returns.get_analysis(),
'trades': strat.analyzers.trades.get_analysis(),
'trade_log': strat.trade_log
}
# 回测结果分析
def analyze_backtest_results(results):
"""分析回测结果"""
sharpe = results['sharpe']
drawdown = results['drawdown']
returns = results['returns']
trades = results['trades']
print("\n=== 回测结果分析 ===")
print(f"年化夏普比率: {sharpe.get('sharperatio', 0):.3f}")
print(f"最大回撤: {drawdown.get('max', {}).get('drawdown', 0):.2f}%")
print(f"年化收益率: {returns.get('rnorm100', 0):.2f}%")
print(f"总交易次数: {trades.get('total', {}).get('total', 0)}")
# 绘制交易日志
trade_df = pd.DataFrame(results['trade_log'])
if not trade_df.empty:
print(f"\n最近5次交易:")
print(trade_df.tail())
6.2 完整的策略回测流程
将各个模块组合成完整的回测流水线：
def complete_strategy_backtest(start_date='20200101', end_date='20231231'):
"""
完整的策略回测流程
"""
print("=== ETF双因子轮动策略回测开始 ===")
# 1. 数据获取
print("步骤1: 获取ETF数据...")
data_manager = ETFDataManager()
raw_etf_data = data_manager.get_all_etf_data(start_date, end_date)
# 2. 数据清洗
print("步骤2: 数据质量检查与清洗...")
cleaned_etf_data = data_quality_check(raw_etf_data)
# 3. 因子计算
print("步骤3: 计算双因子...")
factor_calculator = FactorCalculator()
scores_df = factor_calculator.calculate_comprehensive_score(cleaned_etf_data)
# 4. 权重分配
print("步骤4: 计算权重分配...")
weight_allocator = WeightAllocator(top_n=3, min_weight=0.1)
weights_df = weight_allocator.calculate_weights(scores_df)
# 5. 回测执行
print("步骤5: 执行回测...")
results = run_backtest(cleaned_etf_data, weights_df)
# 6. 结果分析
print("步骤6: 分析回测结果...")
analyze_backtest_results(results)
return {
'etf_data': cleaned_etf_data,
'scores_df': scores_df,
'weights_df': weights_df,
'backtest_results': results
}
if __name__ == "__main__":
# 运行完整回测（使用较短时间范围测试）
results = complete_strategy_backtest('20230101', '20231231')
7. 实盘信号生成与监控
7.1 生成交易信号
回测验证有效后，可以生成实盘交易信号：
class TradingSignalGenerator:
def __init__(self, model_path=None):
self.factor_calculator = FactorCalculator()
self.weight_allocator = WeightAllocator()
def generate_signals(self, etf_data):
"""
生成最新的交易信号
"""
# 计算最新因子得分
scores_df = self.factor_calculator.calculate_comprehensive_score(etf_data)
if scores_df.empty:
return None
# 获取最新得分
latest_scores = scores_df.iloc[-1].dropna()
# 计算权重
latest_weights = self.weight_allocator.calculate_weights(
pd.DataFrame([latest_scores])
).iloc[0]
# 生成交易信号
signals = {}
for etf_code, weight in latest_weights.items():
if weight > 0.01: # 权重超过1%生成信号
signals[etf_code] = {
'action': 'BUY' if weight > 0.05 else 'HOLD',
'target_weight': round(weight, 4),
'score': round(latest_scores.get(etf_code, 0), 4)
}
return signals
def generate_signal_report(self, signals, etf_data):
"""
生成信号报告
"""
if not signals:
return "暂无有效交易信号"
report = ["=== ETF双因子轮动策略交易信号 ==="]
report.append(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
report.append("\n推荐持仓:")
for etf_code, signal in signals.items():
if signal['action'] == 'BUY':
latest_price = etf_data[etf_code]['close'].iloc[-1]
report.append(
f"- {etf_code}: {signal['action']} | "
f"目标权重: {signal['target_weight']*100}% | "
f"综合得分: {signal['score']} | "
f"最新价格: {latest_price:.3f}"
)
return "\n".join(report)
# 实时信号生成示例
def demo_realtime_signals():
"""演示实时信号生成"""
data_manager = ETFDataManager()
# 获取最近一年的数据（模拟实时环境）
end_date = datetime.now().strftime('%Y%m%d')
start_date = (datetime.now() - timedelta(days=365)).strftime('%Y%m%d')
etf_data = data_manager.get_all_etf_data(start_date, end_date)
cleaned_data = data_quality_check(etf_data)
signal_generator = TradingSignalGenerator()
signals = signal_generator.generate_signals(cleaned_data)
if signals:
report = signal_generator.generate_signal_report(signals, cleaned_data)
print(report)
else:
print("无法生成有效交易信号")
if __name__ == "__main__":
demo_realtime_signals()
7.2 策略监控与调优
实盘运行需要持续的监控和优化：
class StrategyMonitor:
def __init__(self):
self.performance_history = []
def track_daily_performance(self, date, portfolio_value, signals):
"""每日业绩跟踪"""
self.performance_history.append({
'date': date,
'portfolio_value': portfolio_value,
'signals': signals.copy() if signals else {}
})
def analyze_strategy_stability(self, window=30):
"""分析策略稳定性"""
if len(self.performance_history) < window:
return "数据不足，需要更多历史记录"
recent_perf = self.performance_history[-window:]
values = [item['portfolio_value'] for item in recent_perf]
returns = np.diff(values) / values[:-1]
volatility = np.std(returns) * np.sqrt(252)
sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252) if np.std(returns) > 0 else 0
return {
'近期波动率': f"{volatility:.3f}",
'近期夏普比率': f"{sharpe:.3f}",
'收益稳定性': "良好" if volatility < 0.2 else "需关注"
}
def generate_monitoring_report(self):
"""生成监控报告"""
if not self.performance_history:
return "暂无监控数据"
latest = self.performance_history[-1]
stability = self.analyze_strategy_stability()
report = [
"=== 策略监控报告 ===",
f"报告时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
f"最新净值: {latest['portfolio_value']:,.2f}",
f"信号数量: {len(latest.get('signals', {}))}",
"\n策略稳定性分析:"
]
for key, value in stability.items():
report.append(f"- {key}: {value}")
return "\n".join(report)
8. 常见问题与排查思路
在实现过程中可能会遇到的各种问题及解决方案：
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 数据获取失败 | 网络问题/接口变更 | 检查akshare版本和网络连接 | 更新akshare，添加重试机制 |
| 因子计算异常 | 数据缺失或格式错误 | 检查数据质量和长度 | 增加数据验证步骤，处理边界情况 |
| 回测结果异常 | 前视偏差或逻辑错误 | 检查时间戳对齐情况 | 确保使用历史数据进行计算 |
| 权重分配不合理 | 因子极端值影响 | 检查因子分布情况 | 添加因子缩尾处理 |
| 交易频率过高 | 调仓周期设置过短 | 分析交易成本影响 | 调整调仓频率，添加过滤条件 |
8.1 典型错误示例与修正
错误示例1：前视偏差
# 错误写法：使用未来数据
def wrong_factor_calculation(df):
# 使用当天的收盘价计算动量（错误）
df['momentum'] = df['close'] / df['close'].shift(20) - 1
return df
# 正确写法：使用历史数据
def correct_factor_calculation(df):
# 使用前一天的收盘价计算动量（正确）
df['momentum'] = df['close'].shift(1) / df['close'].shift(21) - 1
return df
错误示例2：数据未对齐
# 错误写法：直接合并不同时间序列
def wrong_data_merge(df1, df2):
merged = pd.merge(df1, df2, left_index=True, right_index=True) # 可能丢失数据
return merged
# 正确写法：先确保时间索引对齐
def correct_data_merge(df1, df2):
# 使用相同的时间索引
common_index = df1.index.intersection(df2.index)
merged = pd.merge(df1.loc[common_index], df2.loc[common_index],
left_index=True, right_index=True)
return merged
9. 最佳实践与工程建议
9.1 代码组织规范
良好的项目结构是策略可持续迭代的基础：
etf_dual_factor/
├── data/ # 数据模块
│ ├── manager.py # 数据管理
│ └── validator.py # 数据验证
├── factors/ # 因子模块
│ ├── calculator.py # 因子计算
│ └── composite.py # 因子合成
├── portfolio/ # 组合模块
│ ├── allocator.py # 权重分配
│ └── rebalance.py # 调仓逻辑
├── backtest/ # 回测模块
│ ├── engine.py # 回测引擎
│ └── analyzer.py # 结果分析
├── live/ # 实盘模块
│ ├── signals.py # 信号生成
│ └── monitor.py # 策略监控
└── config/ # 配置模块
├── settings.py # 参数配置
└── constants.py # 常量定义
9.2 参数优化与验证
策略参数需要系统化的优化流程：
class ParameterOptimizer:
def __init__(self, etf_data):
self.etf_data = etf_data
def grid_search_parameters(self, param_grid):
"""
网格搜索最优参数
"""
best_params = None
best_sharpe = -float('inf')
results = []
# 参数组合遍历
for lookback in param_grid['lookback_periods']:
for top_n in param_grid['top_n_values']:
for min_weight in param_grid['min_weights']:
# 使用当前参数运行回测
try:
factor_calc = FactorCalculator(lookback_periods=[lookback])
scores_df = factor_calc.calculate_comprehensive_score(self.etf_data)
weight_alloc = WeightAllocator(top_n=top_n, min_weight=min_weight)
weights_df = weight_alloc.calculate_weights(scores_df)
backtest_results = run_backtest(self.etf_data, weights_df)
sharpe = backtest_results['sharpe'].get('sharperatio', -999)
results.append({
'params': {'lookback': lookback, 'top_n': top_n, 'min_weight': min_weight},
'sharpe': sharpe
})
if sharpe > best_sharpe:
best_sharpe = sharpe
best_params = {'lookback': lookback, 'top_n': top_n, 'min_weight': min_weight}
except Exception as e:
print(f"参数组合失败: {e}")
continue
return best_params, best_sharpe, results
# 参数优化示例
def demo_parameter_optimization():
"""演示参数优化过程"""
data_manager = ETFDataManager()
etf_data = data_manager.get_all_etf_data('20220101', '20231231')
cleaned_data = data_quality_check(etf_data)
optimizer = ParameterOptimizer(cleaned_data)
param_grid = {
'lookback_periods': [20, 60, 120],
'top_n_values': [2, 3, 4],
'min_weights': [0.1, 0.15, 0.2]
}
best_params, best_sharpe, all_results = optimizer.grid_search_parameters(param_grid)
print(f"最优参数: {best_params}")
print(f"最优夏普比率: {best_sharpe:.3f}")
9.3 风险控制与资金管理
实盘交易必须重视风险控制：
class RiskManager:
def __init__(self, max_position_size=0.3, max_drawdown=0.15):
self.max_position_size = max_position_size # 单标的最大仓位
self.max_drawdown = max_drawdown # 最大回撤阈值
def validate_position_weights(self, weights):
"""验证仓位权重合理性"""
# 检查单标的风险
for etf, weight in weights.items():
if weight > self.max_position_size:
print(f"警告: {etf}权重{weight:.1%}超过限制{self.max_position_size:.1%}")
return False
# 检查集中度风险
top_weights = sorted(weights.values(), reverse=True)[:2]
if len(top_weights) == 2 and sum(top_weights) > 0.6:
print(f"警告: 前两大权重合计{sum(top_weights):.1%}，集中度偏高")
return False
return True
def check_drawdown_breach(self, current_drawdown):
"""检查回撤是否超限"""
if current_drawdown > self.max_drawdown:
print(f"紧急: 当前回撤{current_drawdown:.1%}超过阈值{self.max_drawdown:.1%}")
return True
return False
def apply_emergency_measures(self, current_drawdown):
"""执行紧急风控措施"""
if self.check_drawdown_breach(current_drawdown):
码道开发者社区，聚焦华为云码道 CodeArts 代码智能体，沉淀 Agent、Skill、鸿蒙开发实战内容，供开发者查阅资料、交流技术、分享工程实践
13
0- 0
已为社区贡献86条内容
所有评论(0)