[QMT量化交易小白入门]-二十五、DeepSeek生成的年化收益率54%的动量因子评分算法（含代码解析）

本专栏主要是介绍QMT的基础用法，常见函数，写策略的方法，也会分享一些量化交易的思路，大概会写100篇左右。
QMT的相关资料较少，在使用过程中不断的摸索，遇到了一些问题，记录下来和大家一起沟通，共同进步，自己淋过雨了，希望大家都有一把伞。

文章目录

- - 相关阅读
  - 一、函数概述：优化后的ETF评分函数
  - 二、数据准备与质量检查
  - 三、数据处理：实际时间序列与线性回归分析
  - 四、年化收益率计算优化
  - 五、改进的R²计算
  - 六、动态权重调整（波动率调整）
  - 七、综合评分（加入动量因子）
  - 八、创建评分DataFrame与标准化处理
  - 九、个人观点

一、函数概述：优化后的ETF评分函数

def calculate_etf_scores(etf_pool, mkdict, lookback_window=63):"""优化后的ETF评分函数:param etf_pool: ETF列表:param mkdict: 包含各ETF历史数据的字典:param lookback_window: 观察窗口(默认63个交易日，约3个月):return: 排序后的ETF评分DataFrame"""

这个函数是整个策略的核心，它接收三个参数：etf_pool是待评估的ETF列表；mkdict是一个包含了各ETF历史数据的字典，通过它可以获取每个ETF的历史价格等信息；lookback_window是观察窗口，默认设置为63个交易日，大约相当于3个月的时间。函数的返回值是一个经过排序的ETF评分DataFrame，通过这个评分，可以确定哪些ETF更具投资价值。

二、数据准备与质量检查

score_list = []
valid_etfs = []for etf in etf_pool:try:#df = mkdict[etf].tail(63)  # 使用固定时间窗口df = mkdict[etf]# 数据质量检查if len(df) < lookback_window * 0.8:  # 允许20%数据缺失continue

初始化两个空列表score_list和valid_etfs，分别用于存储ETF的评分和有效的ETF名称。然后，通过循环遍历etf_pool中的每个ETF。在循环内部，注释掉的代码df = mkdict[etf].tail(63)原本是使用固定时间窗口来获取数据，但这里直接使用了df = mkdict[etf]获取整个历史数据，这可能是为了更全面地分析ETF的表现。接着进行数据质量检查，如果某个ETF的数据长度小于观察窗口的80%，即允许最多20%的数据缺失，那么就跳过该ETF，继续处理下一个。这一步确保了只对数据质量较好的ETF进行分析，避免了因数据不完整而导致的错误结果。

三、数据处理：实际时间序列与线性回归分析

# 使用实际时间序列（处理非交易日问题）
x = pd.to_numeric(df.index).values.reshape(-1, 1)
y = np.log(df['close'].values)# 稳健线性回归（处理异常值）
slope, intercept, r_value, _, _ = stats.linregress(x.flatten(), y)

为了处理非交易日的问题，将ETF的时间索引转换为数值型，并重塑为二维数组x，同时将ETF的收盘价取对数后赋值给y。这样处理的好处是，通过对数变换可以使数据更加稳定，减少极端值的影响。接下来，使用stats.linregress函数进行稳健线性回归分析。这个函数可以有效地处理异常值，得到回归直线的斜率slope、截距intercept以及相关系数r_value等参数。这些参数将为后续的分析和计算提供重要依据。

四、年化收益率计算优化

# 年化收益率计算优化
daily_growth = np.exp(slope) - 1
annualized_returns = (1 + daily_growth) ** 252 - 1  # 使用实际交易日数

根据线性回归得到的斜率slope，首先计算出每日增长率daily_growth，其计算公式为np.exp(slope) - 1。然后，使用实际交易日数（252天）来计算年化收益率annualized_returns，公式为(1 + daily_growth) ** 252 - 1。这种计算方法考虑了实际的交易天数，相比简单地按照日历天数计算更加准确和合理，能够更真实地反映ETF的年化收益情况。

五、改进的R²计算

# 改进的R²计算
y_pred = slope * x.flatten() + intercept
r_squared = r2_score(y, y_pred)

通过线性回归得到的斜率slope和截距intercept，可以计算出预测值y_pred。然后，使用r2_score函数计算决定系数r_squared，它衡量了模型对数据的拟合程度。与传统的R²计算方法相比，这里的改进之处在于它是在线性回归的基础上进行的，能够更准确地反映ETF价格与时间之间的线性关系，从而为评估ETF的表现提供了更可靠的依据。

六、动态权重调整（波动率调整）

# 动态权重调整（波动率调整）
volatility = np.std(np.diff(y)) * np.sqrt(252)
risk_adjusted_return = annualized_returns / (volatility + 1e-6)  # 防止除零

波动率是衡量资产风险的重要指标之一。在这里，通过计算ETF收盘价对数收益率的差分的标准差，并乘以一年的实际交易日数的平方根，得到年化波动率volatility。然后，用年化收益率annualized_returns除以波动率加上一个极小的值（1e-6，用于防止除零错误），得到风险调整后的收益率risk_adjusted_return。这种动态权重调整的方法考虑了ETF的风险因素，使得不同风险水平的ETF能够在一个相对公平的基础上进行比较。

七、综合评分（加入动量因子）

# 综合评分（加入动量因子）
momentum = df['close'].pct_change(21).iloc[-1]  # 1个月动量
score = (risk_adjusted_return * r_squared) + (0.3 * momentum)

除了考虑风险调整后的收益率和决定系数外，还加入了动量因子。动量因子反映了ETF价格变化的趋势，通常认为具有正动量的ETF在未来一段时间内表现可能会更好。这里，计算了ETF在过去1个月内的收盘价的百分比变化作为动量因子momentum。将风险调整后的收益率与决定系数的乘积加上动量因子的0.3倍作为综合评分score。这种综合评分方法综合考虑了多个因素，能够更全面地评估ETF的投资价值。

八、创建评分DataFrame与标准化处理

# 创建评分DataFrame
df_score = pd.DataFrame(index=valid_etfs, data={'score': score_list})# 分数标准化
df_score['score'] = (df_score['score'] - df_score['score'].mean()) / df_score['score'].std()
df_score = df_score.sort_values(by='score', ascending=False)