ETF双因子轮动策略：动量与质量因子的Python实现与回测

你是不是经常看到券商研报里各种高大上的量化策略，感觉逻辑清晰、收益诱人，但真要自己动手实现时却无从下手？今天这篇文章，我要帮你彻底解决这个问题。

我们将完整复现一个经典的ETF双因子轮动策略，从数据获取到策略回测，再到实盘信号生成，提供全套可运行的Python源码。这个策略的核心逻辑很简单：动量（收益趋势）+ 质量（趋势稳定性/低波动）双维度筛选ETF，结合加权仓位分配，在市场不同阶段动态调整持仓。

但真正有价值的不只是代码本身，而是整个工程化实现的思路。很多人在复现研报策略时容易踩的坑：数据源不稳定、因子计算有偏差、回测逻辑不严谨、实盘对接复杂。本文将逐一拆解这些难点，让你不仅能跑通这个策略，更能掌握量化策略开发的完整方法论。

1. 这篇文章真正要解决的问题

为什么券商研报的策略看起来很美，但自己就是复现不出来？问题通常出在三个层面：

数据层面：研报往往使用清洗过的理想数据，而真实环境中你需要自己处理数据缺失、复权、停牌等问题。比如ETF的规模数据、净值数据如何稳定获取？

工程层面：策略逻辑转化为可维护的代码需要合理的架构设计。单因子计算、多因子合成、权重分配、风控规则，这些模块如何组织才能既清晰又高效？

回测层面：简单的收益率计算很容易，但真实的交易成本、滑点、调仓频率等因素会显著影响最终收益。如何构建一个可信的回测系统？

本文将基于Python生态中成熟的工具链（pandas、akshare、backtrader等），提供一个工业级的解决方案。适合有一定Python基础，想要系统学习量化策略开发的读者。

2. 基础概念与核心原理

2.1 什么是ETF双因子轮动？

双因子轮动策略的核心思想是通过两个维度的指标来评估ETF的投资价值，并定期调整持仓。

动量因子：衡量资产的价格趋势强度。通常用过去一段时间（如20日、60日）的收益率来表示。动量效应基于行为金融学中的"追涨杀跌"心理，趋势一旦形成往往会持续。

质量因子：衡量趋势的稳定性和风险调整后的收益质量。常用波动率、夏普比率、最大回撤等指标。质量因子帮助过滤掉高波动、高风险的趋势，选择更稳健的标的。

2.2 策略核心逻辑拆解

策略的完整流程包括：

1. 标的池选择：确定参与轮动的ETF范围（如宽基指数、行业ETF等）
2. 因子计算：定期计算每个ETF的动量因子和质量因子
3. 综合评分：将两个因子标准化后加权求和，得到综合得分
4. 权重分配：根据得分分配仓位，得分越高权重越大
5. 调仓执行：定期（如月度）调整持仓至新的权重分配

这种方法的优势在于：牛市时动量因子主导，抓住强势品种；震荡市时质量因子主导，控制回撤风险。

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. 常见问题与排查思路

在实现过程中可能会遇到的各种问题及解决方案：

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):