企业官网建设流程全解析

在量化投资领域，很多投资者都有过这样的经历：看到券商研报中精美的策略回测曲线和诱人的收益数据，却苦于没有完整的代码实现，只能停留在理论层面。本文将带你完整复现一个经典的ETF双因子轮动策略，从数据获取到回测分析，提供全套可运行的Python源码，让你真正把研报理论转化为实战工具。

本文适合有一定Python基础的量化投资爱好者，无论是学生、个人投资者还是专业开发者，都能通过本文掌握ETF轮动策略的完整实现流程。学完后你将能够独立完成数据获取、因子计算、策略构建和回测分析的全套工作。

1. ETF双因子轮动策略核心概念

1.1 什么是ETF轮动策略

ETF轮动策略是一种基于市场风格轮动的投资方法，通过在不同类型的ETF之间进行切换，捕捉不同市场环境下的投资机会。与传统买入持有策略不同，轮动策略强调动态调整，根据市场信号主动切换持仓品种。

轮动策略的核心逻辑在于：不同资产类别（如股票、债券、商品）或不同行业板块在不同市场周期中表现存在差异。通过识别当前市场的主导风格，可以将资金配置到相对强势的品种中，实现超额收益。

1.2 双因子模型理论基础

双因子轮动策略通常结合动量因子和质量因子进行综合评估：

动量因子：衡量资产价格的趋势强度，基于"强者恒强"的市场现象。常见的动量指标包括：

• 短期收益率（如20日收益率）
• 长期收益率（如120日收益率）
• 动量突破指标（如价格突破均线）

质量因子：评估资产的稳健性和风险特征，确保选择的ETF具有较好的基本面支撑。常用质量指标包括：

• 波动率（历史波动率越低，质量越高）
• 最大回撤（控制下行风险）
• 夏普比率（风险调整后收益）

双因子结合的优势在于：动量因子捕捉趋势机会，质量因子过滤虚假信号，两者结合可以提高策略的稳定性和抗风险能力。

1.3 策略适用场景与风险提示

ETF双因子轮动策略特别适合震荡市和结构性行情，在单边牛市或熊市中表现相对平缓。策略的有效性依赖于市场风格的持续性，如果市场风格快速切换，可能会产生较高的换手成本和策略滞后。

重要风险提示：本文提供的策略仅用于技术学习和研究目的，不构成任何投资建议。量化策略存在过拟合风险，历史回测不代表未来表现，实际投资需谨慎评估风险承受能力。

2. 环境准备与数据源配置

2.1 Python环境要求

本策略基于Python 3.8+环境开发，需要安装以下核心库：

# 基础数据处理库 pip install pandas>=1.3.0 pip install numpy>=1.21.0 # 金融数据获取和回测库 pip install akshare>=1.2.0 # 免费金融数据接口 pip install backtrader>=1.9.0 # 专业回测框架 # 可视化库 pip install matplotlib>=3.5.0 pip install seaborn>=0.11.0 # 日期处理 pip install datetime

建议使用Jupyter Notebook或VS Code进行代码开发和调试，便于分步验证各个环节的输出结果。

2.2 数据源选择与配置

本策略使用akshare作为主要数据源，这是一个开源的金融数据接口库，提供丰富的A股、基金、期货等数据。akshare的优势在于完全免费、数据更新及时、接口稳定。

import akshare as ak import pandas as pd import numpy as np from datetime import datetime, timedelta import warnings warnings.filterwarnings('ignore') # 验证akshare数据接口是否正常 def test_akshare_connection(): try: # 测试获取沪深300ETF数据 test_data = ak.fund_etf_hist_sina(symbol="sh510300") print("akshare数据接口连接正常") print(f"最新数据日期：{test_data['date'].iloc[-1]}") return True except Exception as e: print(f"数据接口异常：{e}") return False # 运行连接测试 test_akshare_connection()

2.3 ETF池选择标准

构建ETF轮动策略的第一步是确定备选ETF池。选择标准包括：

• 规模较大（通常>10亿元），流动性好
• 跟踪误差小，代表性强
• 覆盖不同资产类别和行业板块

# 主要ETF列表及代码映射 ETF_POOL = { '510300': '沪深300ETF', # 大盘宽基 '510500': '中证500ETF', # 中小盘 '512100': '中证1000ETF', # 小盘股 '512880': '证券ETF', # 券商板块 '515000': '科技ETF', # 科技板块 '512690': '酒ETF', # 消费板块 '518880': '黄金ETF', # 商品黄金 '511010': '国债ETF', # 债券防御 } # ETF分类字典 ETF_CATEGORY = { '权益类': ['510300', '510500', '512100'], '行业类': ['512880', '515000', '512690'], '防御类': ['518880', '511010'] }

3. 数据获取与预处理模块

3.1 历史数据批量下载

实现一个稳健的数据下载函数，处理网络异常、数据缺失等边界情况：

def fetch_etf_data(etf_code, start_date='20180101', end_date=None): """ 获取ETF历史行情数据 """ if end_date is None: end_date = datetime.now().strftime('%Y%m%d') max_retries = 3 for attempt in range(max_retries): try: # 使用akshare获取ETF历史数据 df = ak.fund_etf_hist_sina(symbol=f"sh{etf_code}") df['code'] = etf_code df['date'] = pd.to_datetime(df['date']) df = df.sort_values('date').reset_index(drop=True) # 数据质量检查 if len(df) < 100: print(f"警告：{etf_code}数据量不足") return None # 处理缺失值 df = df.fillna(method='ffill') print(f"成功获取{etf_code}数据，时间范围：{df['date'].min()} 至 {df['date'].max()}") return df except Exception as e: print(f"第{attempt+1}次尝试获取{etf_code}数据失败：{e}") if attempt == max_retries - 1: print(f"获取{etf_code}数据最终失败") return None time.sleep(2) # 等待2秒后重试 # 批量下载所有ETF数据 def fetch_all_etf_data(etf_pool, start_date='20180101'): """ 批量获取ETF池中所有品种的历史数据 """ all_data = {} for code in etf_pool.keys(): print(f"正在下载 {code} 数据...") data = fetch_etf_data(code, start_date) if data is not None: all_data[code] = data time.sleep(1) # 避免请求过于频繁 print(f"数据下载完成，成功获取{len(all_data)}个ETF数据") return all_data # 执行数据下载 etf_data_dict = fetch_all_etf_data(ETF_POOL)

3.2 数据清洗与对齐

不同ETF的交易日可能不完全一致，需要进行数据对齐处理：

def align_etf_data(etf_data_dict): """ 对齐所有ETF的交易日期，确保时间轴一致 """ # 找出所有ETF的共同交易日期 all_dates = None for code, data in etf_data_dict.items(): if all_dates is None: all_dates = set(data['date']) else: all_dates = all_dates.intersection(set(data['date'])) # 转换为排序后的日期列表 common_dates = sorted(list(all_dates)) print(f"共同交易日数量：{len(common_dates)}") # 按共同交易日对齐数据 aligned_data = {} for code, data in etf_data_dict.items(): aligned_df = data[data['date'].isin(common_dates)].sort_values('date') aligned_df = aligned_df.reset_index(drop=True) aligned_data[code] = aligned_df return aligned_data, common_dates # 执行数据对齐 aligned_etf_data, trading_dates = align_etf_data(etf_data_dict)

3.3 基础指标计算

计算每个ETF的收益率、波动率等基础指标：

def calculate_basic_indicators(aligned_etf_data): """ 计算每个ETF的基础技术指标 """ indicators_dict = {} for code, data in aligned_etf_data.items(): df = data.copy() df = df.sort_values('date').reset_index(drop=True) # 计算日收益率 df['daily_return'] = df['close'].pct_change() # 计算滚动波动率（20日） df['volatility_20d'] = df['daily_return'].rolling(20).std() # 计算不同周期的动量收益率 df['return_5d'] = df['close'].pct_change(5) # 周动量 df['return_20d'] = df['close'].pct_change(20) # 月动量 df['return_60d'] = df['close'].pct_change(60) # 季动量 # 计算均线指标 df['ma20'] = df['close'].rolling(20).mean() df['ma60'] = df['close'].rolling(60).mean() # 价格与均线关系 df['price_vs_ma20'] = df['close'] / df['ma20'] - 1 df['price_vs_ma60'] = df['close'] / df['ma60'] - 1 indicators_dict[code] = df return indicators_dict # 计算技术指标 etf_indicators = calculate_basic_indicators(aligned_etf_data)

4. 双因子计算与合成

4.1 动量因子计算

动量因子衡量价格趋势强度，采用多周期动量合成方法：

def calculate_momentum_factor(etf_indicators): """ 计算综合动量因子 """ momentum_results = {} for code, data in etf_indicators.items(): df = data.copy() # 标准化不同周期的动量指标 df['momentum_short'] = df['return_5d'].rolling(10).mean() # 短期动量 df['momentum_medium'] = df['return_20d'].rolling(5).mean() # 中期动量 df['momentum_long'] = df['return_60d'] # 长期动量 # 动量因子合成（加权平均） # 短期权重0.3，中期权重0.4，长期权重0.3 df['momentum_factor'] = (df['momentum_short'] * 0.3 + df['momentum_medium'] * 0.4 + df['momentum_long'] * 0.3) # 动量因子排名（横截面比较） momentum_results[code] = df[['date', 'momentum_factor']] return momentum_results # 计算动量因子 momentum_factors = calculate_momentum_factor(etf_indicators)

4.2 质量因子计算

质量因子关注风险调整后收益和稳定性：

def calculate_quality_factor(etf_indicators, lookback_period=60): """ 计算质量因子：基于波动率、最大回撤、夏普比率等 """ quality_results = {} for code, data in etf_indicators.items(): df = data.copy() # 计算滚动波动率（年化） df['volatility_annual'] = df['daily_return'].rolling(lookback_period).std() * np.sqrt(252) # 计算滚动夏普比率（假设无风险收益率为3%） risk_free_rate = 0.03 / 252 # 日化无风险利率 df['excess_return'] = df['daily_return'] - risk_free_rate df['sharpe_ratio'] = (df['excess_return'].rolling(lookback_period).mean() * 252 / df['volatility_annual']) # 计算最大回撤 df['rolling_max'] = df['close'].rolling(lookback_period).max() df['drawdown'] = (df['close'] - df['rolling_max']) / df['rolling_max'] df['max_drawdown'] = df['drawdown'].rolling(lookback_period).min() # 质量因子合成（波动率越低越好，夏普越高越好，回撤越小越好） # 标准化处理 df['volatility_score'] = 1 / (1 + df['volatility_annual']) # 波动率倒数 df['sharpe_score'] = df['sharpe_ratio'].clip(lower=-3, upper=3) # 截断异常值 df['drawdown_score'] = 1 / (1 - df['max_drawdown']) # 回撤保护 # 质量因子合成 df['quality_factor'] = (df['volatility_score'] * 0.4 + df['sharpe_score'] * 0.4 + df['drawdown_score'] * 0.2) quality_results[code] = df[['date', 'quality_factor', 'volatility_annual', 'sharpe_ratio', 'max_drawdown']] return quality_results # 计算质量因子 quality_factors = calculate_quality_factor(etf_indicators)

4.3 因子合成与标准化

将动量因子和质量因子合成为综合得分：

def synthesize_factors(momentum_factors, quality_factors): """ 合成双因子为综合评分 """ # 获取所有ETF代码 etf_codes = list(momentum_factors.keys()) # 创建综合数据框 all_dates = momentum_factors[etf_codes[0]]['date'] synthetic_scores = pd.DataFrame({'date': all_dates}) for code in etf_codes: # 合并动量因子和质量因子 merged_df = pd.merge(momentum_factors[code], quality_factors[code], on='date') # 横截面标准化（每日排名） def cross_section_rank(df): dates = df['date'].unique() ranked_dfs = [] for date in dates: day_data = df[df['date'] == date].copy() # 动量因子排名（越高越好） day_data['momentum_rank'] = day_data['momentum_factor'].rank(ascending=False) # 质量因子排名（越高越好） day_data['quality_rank'] = day_data['quality_factor'].rank(ascending=False) # 综合得分（动量权重0.6，质量权重0.4） day_data['composite_score'] = (day_data['momentum_rank'] * 0.6 + day_data['quality_rank'] * 0.4) ranked_dfs.append(day_data) return pd.concat(ranked_dfs).sort_values('date') # 应用横截面排名 ranked_data = cross_section_rank(merged_df) synthetic_scores[code] = ranked_data['composite_score'].values return synthetic_scores # 合成双因子 composite_scores = synthesize_factors(momentum_factors, quality_factors) print("双因子合成完成，综合评分数据形状：", composite_scores.shape)

5. 轮动策略逻辑实现

5.1 持仓选择规则

基于综合得分选择最优ETF进行轮动：

class ETFRotationStrategy: """ ETF双因子轮动策略核心类 """ def __init__(self, top_k=2, rebalance_freq=20): """ 初始化参数 top_k: 选择排名前K的ETF rebalance_freq: 调仓频率（交易日） """ self.top_k = top_k self.rebalance_freq = rebalance_freq self.current_holdings = [] self.rebalance_dates = [] def select_etfs(self, composite_scores, date): """ 根据综合得分选择ETF """ # 获取指定日期的得分数据 date_scores = composite_scores[composite_scores['date'] == date].iloc[0] # 提取ETF代码和得分 etf_scores = [] for code in ETF_POOL.keys(): if code in date_scores.index: score = date_scores[code] if not pd.isna(score): etf_scores.append((code, score)) # 按得分排序，选择前top_k个 etf_scores.sort(key=lambda x: x[1], ascending=False) selected_etfs = [etf[0] for etf in etf_scores[:self.top_k]] return selected_etfs def calculate_weights(self, selected_etfs, composite_scores, date): """ 计算持仓权重（基于得分加权） """ date_scores = composite_scores[composite_scores['date'] == date].iloc[0] # 计算选中ETF的总得分 total_score = sum(date_scores[etf] for etf in selected_etfs) # 按得分比例分配权重 weights = {} for etf in selected_etfs: weights[etf] = date_scores[etf] / total_score return weights # 初始化策略 strategy = ETFRotationStrategy(top_k=2, rebalance_freq=20)

5.2 调仓逻辑与交易规则

实现完整的调仓决策流程：

def implement_rotation_strategy(composite_scores, etf_indicators, start_date='2019-01-01', end_date=None): """ 执行轮动策略回测 """ if end_date is None: end_date = datetime.now().strftime('%Y-%m-%d') # 转换日期格式 start_date = pd.to_datetime(start_date) end_date = pd.to_datetime(end_date) # 过滤时间范围 date_range = composite_scores[(composite_scores['date'] >= start_date) & (composite_scores['date'] <= end_date)] # 初始化持仓记录 portfolio_records = [] current_holdings = {} last_rebalance_date = None strategy = ETFRotationStrategy(top_k=2, rebalance_freq=20) for current_date in date_range['date'].unique(): current_date = pd.to_datetime(current_date) # 检查是否需要调仓（首个交易日或达到调仓频率） if (last_rebalance_date is None or (current_date - last_rebalance_date).days >= strategy.rebalance_freq): # 选择ETF并计算权重 selected_etfs = strategy.select_etfs(composite_scores, current_date) weights = strategy.calculate_weights(selected_etfs, composite_scores, current_date) # 更新持仓 current_holdings = weights last_rebalance_date = current_date strategy.rebalance_dates.append(current_date) # 计算当日组合收益 daily_return = 0 for etf, weight in current_holdings.items(): # 获取该ETF当日收益率 etf_data = etf_indicators[etf] date_return = etf_data[etf_data['date'] == current_date]['daily_return'] if len(date_return) > 0: daily_return += weight * date_return.iloc[0] # 记录组合表现 portfolio_records.append({ 'date': current_date, 'daily_return': daily_return, 'holdings': current_holdings.copy() }) # 转换为DataFrame portfolio_df = pd.DataFrame(portfolio_records) # 计算累计收益 portfolio_df['cumulative_return'] = (1 + portfolio_df['daily_return']).cumprod() - 1 return portfolio_df, strategy.rebalance_dates # 执行策略回测 portfolio_results, rebalance_dates = implement_rotation_strategy( composite_scores, etf_indicators, start_date='2019-01-01' )

6. 回测系统与绩效评估

6.1 回测结果分析

实现专业的绩效评估指标计算：

def calculate_performance_metrics(portfolio_df, risk_free_rate=0.03): """ 计算策略绩效指标 """ returns = portfolio_df['daily_return'].dropna() cumulative_return = portfolio_df['cumulative_return'].iloc[-1] # 年化收益率 total_days = len(returns) years = total_days / 252 annual_return = (1 + cumulative_return) ** (1/years) - 1 # 年化波动率 annual_volatility = returns.std() * np.sqrt(252) # 夏普比率 excess_returns = returns - risk_free_rate/252 sharpe_ratio = excess_returns.mean() / returns.std() * np.sqrt(252) # 最大回撤 cumulative = (1 + returns).cumprod() rolling_max = cumulative.expanding().max() drawdown = (cumulative - rolling_max) / rolling_max max_drawdown = drawdown.min() # 卡玛比率（年化收益/最大回撤） calmar_ratio = annual_return / abs(max_drawdown) if max_drawdown != 0 else 0 # 胜率 win_rate = (returns > 0).mean() metrics = { '累计收益': cumulative_return, '年化收益': annual_return, '年化波动率': annual_volatility, '夏普比率': sharpe_ratio, '最大回撤': max_drawdown, '卡玛比率': calmar_ratio, '胜率': win_rate, '总交易日': total_days } return metrics, drawdown # 计算绩效指标 performance_metrics, drawdown_series = calculate_performance_metrics(portfolio_results) print("=== 策略绩效报告 ===") for metric, value in performance_metrics.items(): if isinstance(value, float): print(f"{metric}: {value:.4f}" if abs(value) < 1 else f"{metric}: {value:.2f}") else: print(f"{metric}: {value}")

6.2 可视化分析

制作专业的策略表现图表：

import matplotlib.pyplot as plt import seaborn as sns def plot_strategy_performance(portfolio_df, drawdown_series, rebalance_dates): """ 绘制策略表现图表 """ plt.style.use('seaborn-v0_8') fig, axes = plt.subplots(2, 2, figsize=(15, 10)) # 1. 累计收益曲线 axes[0, 0].plot(portfolio_df['date'], portfolio_df['cumulative_return'], label='双因子轮动策略', linewidth=2) axes[0, 0].set_title('累计收益率曲线') axes[0, 0].set_ylabel('累计收益') axes[0, 0].legend() axes[0, 0].grid(True, alpha=0.3) # 标记调仓日期 for rebalance_date in rebalance_dates: axes[0, 0].axvline(x=rebalance_date, color='red', alpha=0.3, linestyle='--') # 2. 回撤曲线 axes[0, 1].fill_between(portfolio_df['date'], drawdown_series, 0, alpha=0.3, color='red', label='回撤') axes[0, 1].plot(portfolio_df['date'], drawdown_series, color='red', linewidth=1) axes[0, 1].set_title('最大回撤曲线') axes[0, 1].set_ylabel('回撤幅度') axes[0, 1].legend() axes[0, 1].grid(True, alpha=0.3) # 3. 月度收益分布 portfolio_df['year_month'] = portfolio_df['date'].dt.to_period('M') monthly_returns = portfolio_df.groupby('year_month')['daily_return'].apply( lambda x: (1 + x).prod() - 1 ) axes[1, 0].bar(monthly_returns.index.astype(str), monthly_returns.values, alpha=0.7, color='steelblue') axes[1, 0].set_title('月度收益率分布') axes[1, 0].set_ylabel('月度收益') plt.setp(axes[1, 0].xaxis.get_majorticklabels(), rotation=45) axes[1, 0].grid(True, alpha=0.3) # 4. 持仓分布统计 all_holdings = [] for holdings in portfolio_df['holdings']: all_holdings.extend(holdings.keys()) holding_counts = pd.Series(all_holdings).value_counts() axes[1, 1].pie(holding_counts.values, labels=[ETF_POOL[code] for code in holding_counts.index], autopct='%1.1f%%', startangle=90) axes[1, 1].set_title('ETF持仓分布统计') plt.tight_layout() plt.show() # 绘制绩效图表 plot_strategy_performance(portfolio_results, drawdown_series, rebalance_dates)

6.3 基准对比分析

与常见基准策略进行对比：

def benchmark_comparison(etf_indicators, portfolio_results, benchmark_code='510300'): """ 与基准ETF进行对比分析 """ # 获取基准ETF数据 benchmark_data = etf_indicators[benchmark_code] benchmark_data = benchmark_data[benchmark_data['date'].isin(portfolio_results['date'])] # 计算基准收益 benchmark_data = benchmark_data.sort_values('date') benchmark_data['daily_return'] = benchmark_data['close'].pct_change() benchmark_data['cumulative_return'] = (1 + benchmark_data['daily_return']).cumprod() - 1 # 合并数据 comparison_df = pd.merge(portfolio_results[['date', 'cumulative_return']], benchmark_data[['date', 'cumulative_return']], on='date', suffixes=('_strategy', '_benchmark')) # 计算超额收益 comparison_df['excess_return'] = (comparison_df['cumulative_return_strategy'] - comparison_df['cumulative_return_benchmark']) # 绘制对比图 plt.figure(figsize=(12, 6)) plt.plot(comparison_df['date'], comparison_df['cumulative_return_strategy'], label='双因子轮动策略', linewidth=2) plt.plot(comparison_df['date'], comparison_df['cumulative_return_benchmark'], label=f'{ETF_POOL[benchmark_code]}基准', linewidth=2) plt.plot(comparison_df['date'], comparison_df['excess_return'], label='超额收益', linestyle='--', alpha=0.8) plt.title('策略与基准对比') plt.ylabel('累计收益') plt.legend() plt.grid(True, alpha=0.3) plt.show() # 计算超额收益指标 excess_metrics = { '累计超额收益': comparison_df['excess_return'].iloc[-1], '年化超额收益': (1 + comparison_df['excess_return'].iloc[-1]) ** (252/len(comparison_df)) - 1, '信息比率': comparison_df['excess_return'].iloc[-1] / comparison_df['excess_return'].std() } print("=== 超额收益分析 ===") for metric, value in excess_metrics.items(): print(f"{metric}: {value:.4f}") return comparison_df # 执行基准对比 benchmark_comparison_result = benchmark_comparison(etf_indicators, portfolio_results)

7. 策略优化与参数敏感性分析

7.1 关键参数优化

通过网格搜索寻找最优参数组合：

def parameter_optimization(composite_scores, etf_indicators): """ 参数优化：调仓频率和持仓数量的网格搜索 """ rebalance_freqs = [10, 20, 30, 40] # 调仓频率 top_k_options = [1, 2, 3] # 持仓数量 results = [] for freq in rebalance_freqs: for top_k in top_k_options: # 临时修改策略参数 strategy = ETFRotationStrategy(top_k=top_k, rebalance_freq=freq) # 执行回测 portfolio_df, _ = implement_rotation_strategy( composite_scores, etf_indicators, start_date='2019-01-01' ) # 计算绩效 metrics, _ = calculate_performance_metrics(portfolio_df) results.append({ 'rebalance_freq': freq, 'top_k': top_k, 'annual_return': metrics['年化收益'], 'sharpe_ratio': metrics['夏普比率'], 'max_drawdown': metrics['最大回撤'] }) # 转换为DataFrame分析 results_df = pd.DataFrame(results) # 找出夏普比率最高的参数组合 best_sharpe = results_df.loc[results_df['sharpe_ratio'].idxmax()] print("最优参数组合（夏普比率最大化）：") print(f"调仓频率：{best_sharpe['rebalance_freq']}天，持仓数量：{best_sharpe['top_k']}个") print(f"年化收益：{best_sharpe['annual_return']:.4f}，夏普比率：{best_sharpe['sharpe_ratio']:.4f}") return results_df # 执行参数优化 optimization_results = parameter_optimization(composite_scores, etf_indicators)

7.2 因子权重敏感性测试

测试不同因子权重对策略表现的影响：

def factor_weight_sensitivity(composite_scores, etf_indicators): """ 分析动量因子和质量因子权重的敏感性 """ momentum_weights = [0.3, 0.4, 0.5, 0.6, 0.7] # 动量因子权重 quality_weights = [0.7, 0.6, 0.5, 0.4, 0.3] # 质量因子权重 sensitivity_results = [] for mom_w, qual_w in zip(momentum_weights, quality_weights): # 重新合成因子（修改权重） def resynthesize_factors(momentum_factors, quality_factors, mom_weight, qual_weight): etf_codes = list(momentum_factors.keys()) all_dates = momentum_factors[etf_codes[0]]['date'] new_scores = pd.DataFrame({'date': all_dates}) for code in etf_codes: merged_df = pd.merge(momentum_factors[code], quality_factors[code], on='date') def new_cross_section_rank(df): dates = df['date'].unique() ranked_dfs = [] for date in dates: day_data = df[df['date'] == date].copy() day_data['momentum_rank'] = day_data['momentum_factor'].rank(ascending=False) day_data['quality_rank'] = day_data['quality_factor'].rank(ascending=False) # 使用新的权重合成 day_data['composite_score'] = (day_data['momentum_rank'] * mom_weight + day_data['quality_rank'] * qual_weight) ranked_dfs.append(day_data) return pd.concat(ranked_dfs).sort_values('date') ranked_data = new_cross_section_rank(merged_df) new_scores[code] = ranked_data['composite_score'].values return new_scores # 重新计算综合得分 new_composite_scores = resynthesize_factors(momentum_factors, quality_factors, mom_w, qual_w) # 执行回测 portfolio_df, _ = implement_rotation_strategy( new_composite_scores, etf_indicators, start_date='2019-01-01' ) # 计算绩效 metrics, _ = calculate_performance_metrics(portfolio_df) sensitivity_results.append({ 'momentum_weight': mom_w, 'quality_weight': qual_w, 'annual_return': metrics['年化收益'], 'sharpe_ratio': metrics['夏普比率'] }) sensitivity_df = pd.DataFrame(sensitivity_results) # 可视化敏感性分析 plt.figure(figsize=(10, 6)) plt.plot(sensitivity_df['momentum_weight'], sensitivity_df['sharpe_ratio'], marker='o', linewidth=2) plt.xlabel('动量因子权重') plt.ylabel('夏普比率') plt.title('因子权重敏感性分析') plt.grid(True, alpha=0.3) plt.show() return sensitivity_df # 执行敏感性分析 sensitivity_results = factor_weight_sensitivity(composite_scores, etf_indicators)

8. 实盘注意事项与风险控制

8.1 实盘交易考虑因素

将回测策略转化为实盘交易时需要关注的关键点：

class LiveTradingConsiderations: """ 实盘交易注意事项 """ def __init__(self): self.transaction_cost = 0.001 # 假设交易成本0.1% self.slippage = 0.002 # 滑点成本0.2% def calculate_realistic_returns(self, portfolio_df, rebalance_dates): """ 考虑交易成本和滑点后的真实收益 """ realistic_returns = portfolio_df.copy() total_transaction_cost = 0 for i, rebalance_date in enumerate(rebalance_dates): # 每次调仓产生交易成本 if i > 0: