当前位置：首页 > news >正文

量化-因子处理

news 2025/6/23 12:34:00

超额收益：alpha和beta的含义

在量化交易中，阿尔法（Alpha）和贝塔（Beta）是衡量投资组合表现与风险的关键指标，源自资本资产定价模型（CAPM），具体定义如下：

阿尔法（Alpha）：衡量投资组合相对市场基准的超额收益，反映基金经理或策略在扣除市场整体影响后的真实盈利能力。公式为 Alpha = R_p - (R_f + \beta (R_m - R_f)) （R_p 为投资组合收益，R_f 为无风险利率，R_m 为市场基准收益
）。正Alpha说明策略跑赢市场，负Alpha则表现不佳，常用于评估选股能力与策略有效性，追求正Alpha是量化交易获取超额收益的重要目标。
贝塔（Beta）：衡量投资组合或资产相对市场的系统性风险与波动相关性，反映策略对大盘变化的敏感性，是CAPM核心概念。若Beta为1，波动与市场同步；大于1，波动比市场大、风险更高；小于1，波动更小、相对稳定
。可用于调整组合风险暴露，辅助投资者根据市场环境（如牛市选高Beta、熊市选低Beta）优化配置、管理风险。

简单来说，Alpha代表“超额收益能力”，Beta代表“市场风险暴露程度”，量化交易常通过二者结合，平衡收益与风险，追求风险调整后最优回报。
在这里插入图片描述

入门策略编写

这里我们以米筐为例
在这里插入图片描述
策略的编写至少需要具备三个函数
初始代码如下：

# 可以自己import我们平台支持的第三方python模块，比如pandas、numpy等。# 在这个方法中编写任何的初始化逻辑。context对象将会在你的算法策略的任何方法之间做传递。
def init(context):# 在context中保存全局变量context.s1 = "000001.XSHE"# 实时打印日志logger.info("RunInfo: {}".format(context.run_info))# before_trading此函数会在每天策略交易开始前被调用，当天只会被调用一次
def before_trading(context):pass# 你选择的证券的数据更新将会触发此段逻辑，例如日或分钟历史数据切片或者是实时数据切片更新
def handle_bar(context, bar_dict):# 开始编写你的主要的算法逻辑# bar_dict[order_book_id] 可以拿到某个证券的bar信息# context.portfolio 可以拿到现在的投资组合信息# 使用order_shares(id_or_ins, amount)方法进行落单# TODO: 开始编写你的算法吧！order_shares(context.s1, 1000)# after_trading函数会在每天交易结束后被调用，当天只会被调用一次
def after_trading(context):pass

一个策略当中至少要具备这四个函数，对于不需要使用的函数，我们也完全可以采用pass来表示在此处不做处理

定时器的功能与作用

在 RiceQuant 量化平台里， scheduler 定时器是用于管理策略中函数执行时机的工具，下面结合 scheduler.run_daily 详细解释并举例：

scheduler.run_daily(function) 主要功能是让指定函数每天在交易时段按规则执行一次，且需在策略初始化函数（ init ）里调用。它能帮我们在量化策略中，定时完成如每日开盘前检查资金、收盘后计算收益等周期性任务。要注意，它执行在对应时间点 handle_bar （处理 K 线数据的核心函数）之前，若定时函数运行太久，中间的 handle_bar 事件会被跳过，所以定时函数逻辑要尽量简洁高效。

使用示例（基于 RiceQuant 平台 Python 策略框架）

下面以“每日收盘后记录账户剩余现金”为例，展示完整策略结构：

# 导入 RiceQuant 平台所需模块
from rqalpha.api import *# 定义定时执行的函数，必须包含 context（策略上下文，存账户、持仓等信息 ）和 bar_dict（K 线数据字典 ）参数
def log_cash(context, bar_dict):# 从 context 中获取账户剩余现金，并用日志输出logger.info("Remaining cash: %r" % context.portfolio.cash)# 策略初始化函数，策略启动时执行一次，用于设置初始配置、定时器等
def init(context):# 调用 scheduler.run_daily，设置每天运行 log_cash 函数scheduler.run_daily(log_cash)# （可选）K 线数据处理函数，按频率（如分钟、日线）触发，这里简单示例可不写复杂逻辑
def handle_bar(context, bar_dict):pass# 以下是策略入口配置（RiceQuant 平台运行策略时需的设置 ）
config = {"base": {"start_date": "2023-01-01",  # 策略回测开始日期"end_date": "2023-12-31",    # 策略回测结束日期"frequency": "1d",           # 策略运行频率，"1d" 表示日线"accounts": {"stock": 100000          # 初始股票账户资金}},"extra": {"log_level": "info"          # 日志级别，info 能输出关键信息}
}if __name__ == "__main__":from rqalpha import run_funcrun_func(init=init, handle_bar=handle_bar, config=config)

代码流程说明：

先导入平台模块，定义 log_cash 函数，它借助 context 获取账户现金并打印。
init 函数里通过 scheduler.run_daily(log_cash) ，安排 log_cash 每天执行。
handle_bar 这里虽简单示例没复杂逻辑，但实际策略中可处理实时行情、交易信号等。
最后通过 config 配置策略运行的基础参数（时间、资金、频率等），用 run_func 启动策略，运行后每天就会执行 log_cash 记录现金情况。

除 run_daily ，RiceQuant 还有 scheduler.run_weekly （每周执行）、scheduler.run_monthly （每月执行）等，用法类似，按需替换就能实现不同周期的定时任务，这些函数定义在init函数当中，可以理解为每多少天执行一次某函数。

因子策略

选择股票的时候有一定标准，而选择的标准就叫做因子
因子策略三步走：去极值，标准化，中性化

极值处理方法

极值是指数据集中明显偏离大多数数据点的观测值，可能由数据录入错误、测量误差、自然极端现象或真实极端情况（如收入分布中的高收入群体）引起，而极值处理方法是数据预处理中的重要环节，用于识别和处理数据中的极端值（异常值），避免其对数据分析、建模或机器学习任务产生负面影响。

分位数去极值

col = 'AAPL.0'def filter_extreme_percent(series, min=0.25, max=0.75):series = series.sort_values()q = series.quantile([min, max])return np.clip(series, q.iloc[0], q.iloc[1])percentile = filter_extreme_percent(data[col])data[col].plot()
percentile.plot()

这段代码实现了一个基于分位数的去极值函数。它的核心逻辑是：先将输入的Series数据按升序排列，然后根据设定的上下分位阈值（默认25%和75%分位数）计算分位数值，最后使用numpy的clip函数将超出此分位区间的值替换为区间边界值。这种方法相比传统的3σ法则更稳健，不易受极端异常值影响，适合处理分布不规则的数据。例如，当输入数据包含离群值时，函数会将小于下分位数的值调整为下分位数值，大于上分位数的值调整为上分位数值，从而有效抑制极端值对整体统计特性的影响。使用时只需传入数据列及可选的分位阈值参数，即可返回经过去极值处理后的序列。
在这里插入图片描述

Mad法去极值

import numpy as np
import pandas as pddef filter_extreme_mad(series, n):
#n一般是取1.4826median = series.quantile(0.5)mad = ((series - median).abs().quantile(0.5))max_range = median + n * madmin_range = median - n * madreturn np.clip(series, min_range, max_range)mad = filter_extreme_mad(data[col], 1.4826)mad.plot()
data[col].plot()

MAD（绝对值差中位数法）是一种常用于异常值检测与数据修正的方法，其核心逻辑是通过“中位数+绝对偏差”的组合来识别并处理极端值，使数据更稳定、分析结果更可靠。以数据集(1, 1, 2, 2, 4, 6, 9)为例，计算步骤如下：首先将数据排序后找到中位数，该例中排序为1, 1, 2, 2, 4, 6, 9，中位数为2；接着计算每个数据点与中位数的绝对偏差，得到(1, 1, 0, 0, 2, 4, 7)；再将绝对偏差排序后取中位数，即0, 0, 1, 1, 2, 4, 7的中位数为1，此即为MAD值。处理极端值时，通常以中位数±n×MAD为区间（n一般取1.4826，使MAD与正态分布下的标准差建立对应关系），超出区间的值会被修正为边界值。该例中，上限为2+1.4826×1≈3.4826，下限为2-1.4826×1≈0.5174，原始数据中的9因大于上限被修正为3.4826，其余数据保持不变。MAD的优势在于抗极端值干扰——用中位数替代均值，对异常值不敏感，且无需假设数据服从正态分布，计算逻辑直观，适用于金融风控、传感器数据清洗等场景中快速定位异常点，通过划定“合理区间”将极端值“拉回”，从而净化数据、提升分析稳定性。
在这里插入图片描述

3sigma方法

import numpy as npdef filter_extreme_3sigma(series, n=3):mean = series.mean()std = series.std()max_range = mean + n * stdmin_range = mean - n * stdreturn np.clip(series, min_range, max_range)sigma = filter_extreme_3sigma(data[col], 1)sigma.plot()
data[col].plot()

3σ方法是基于正态分布特性的异常值处理手段，其核心逻辑是通过计算数据的均值（mean）和标准差（std）来划定合理范围，将超出“均值±n倍标准差”的数据视为异常值并截断至边界。以代码为例，函数先求出序列的均值与标准差，再根据参数n（默认3，对应覆盖约99.7%的正态分布数据）确定上下限，最后用np.clip将数据限制在区间内。例如当n=1时，范围缩小到均值±1倍标准差，仅保留约68%的数据，筛选更为严格。该方法简单直观且有统计学理论支撑，但依赖数据服从正态分布的假设，若数据存在偏态或厚尾，均值和标准差易受极端值干扰，导致阈值不准确。在这里插入图片描述

标准化

在这里插入图片描述

def standardize(series):mean = series.mean()std = series.std()return (series - mean) / std

标准化公式：
在这里插入图片描述
公式各部分说明

x i：原始数据中的第 i个观测值。
μ：原始数据的均值
σ：原始数据的标准差

在这里插入图片描述

z i：标准化后的值，也称 Z-score，表示 x i距离均值的标准差倍数

中性化

中性化的本质就是实现提纯：通过剔除数据中无关的系统性偏差或干扰因素，使目标变量仅保留所需的核心信息，如同从混合物中分离出纯净物质。在金融领域的因子中性化中，若某股票因子同时受行业属性和市值规模影响，中性化操作会通过回归分析等手段，将行业效应、市值效应等 “杂质” 从因子中剥离 —— 好比用过滤器滤除溶液中的杂质，最终得到的残差部分仅反映因子本身的超额收益特征。这种处理本质上是一种 “提纯”：去除数据中与研究目标无关的系统性影响，让变量更纯粹地体现目标信号（如阿尔法因子的真实有效性），避免无关因素对分析结果的干扰，从而提升数据的准确性和研究结论的可靠性。从更广泛的场景看，无论是经济学中的变量去趋势化，还是机器学习中剔除特征间的共线性影响，中性化都是通过系统性 “杂质” 的剥离，实现数据从 “混合态” 到 “纯净态” 的转化，这正是 “提纯” 本质的具体体现。

import pandas as pd
import numpy as np
from sklearn.linear_model import LinearRegression
import statsmodels.api as sm# 假设我们有一个DataFrame，包含股票的市盈率（PB）和市值（Market Cap）
data = {'Stock': ['A', 'B', 'C', 'D', 'E'],'PB': [1.2, 0.8, 1.5, 0.9, 1.1],'Market_Cap': [1000, 1500, 800, 1200, 1100]
}df = pd.DataFrame(data)# 为线性回归添加常数项
X = sm.add_constant(df['Market_Cap'])
Y = df['PB']# 使用statsmodels进行线性回归
model = sm.OLS(Y, X).fit()# 打印回归结果
print(model.summary())# 预测市盈率
df['Predicted_PB'] = model.predict(X)# 市盈率中性化处理
df['Neutralized_PB'] = df['PB'] - df['Predicted_PB']# 输出结果
print(df[['Stock', 'PB', 'Market_Cap', 'Predicted_PB', 'Neutralized_PB']])

这段代码实现了市值中性化处理，其核心逻辑是通过线性回归剔除市盈率（PB）中与市值相关的系统性因素。具体而言，代码将PB作为因变量、市值作为自变量构建回归模型，得到的预测值代表PB中能被市值解释的部分，而残差（原始PB减去预测值）则是剔除市值影响后的纯净PB，即中性化后的PB值。这种处理使中性化后的PB与市值的相关性显著降低，更适合用于选股或因子分析。除了线性回归外，常见的中性化方法还包括：行业中性化（按行业分组计算Z-score）、分组排序法（按市值分组后在组内排序或标准化）、正交化（用多重线性回归同时剔除市值、行业等多个因素）、分位数回归（在不同分位数上回归，对异常值更稳健）以及非线性中性化（用多项式回归或树模型处理非线性关系）。选择中性化方法时，需根据数据特性和业务场景灵活决策，例如线性关系明显时优先用线性回归，存在多因素影响时考虑正交化，行业差异大时先进行行业分组，数据含异常值时可采用分位数回归。

股票数据获取

在交易函数编写之前，我们需要在init函数当中利用定时器来获取股票数据，然后通过
因子策略三步走来进行

import numpy as np
import pandas as pd
from statsmodels import regression
import statsmodels.api as smdef init(context):scheduler.run_monthly(rebalance, 1)def rebalance(context, bar_dict):# 首先过滤掉不想要的股票stocks = filter_paused(all_instruments(type='CS').order_book_id)stocks = filter_st(stocks)stocks = filter_new(stocks)# 查询想要的指标fundamental_df=get_fundamentals(query(fundamentals.eoq_derivative_indicator.pb_ratio,fundamentals.eoq_derivative_indicator.market_cap).filter(fundamentals.income_statement.stockcode.in_(stocks))).T.dropna()# 预处理操作：1. 3sigma 2. standard 3. neutral
no_extreme = filter_3sigma(fundamental_df['pb_ratio'])
pb_ratio_standard = standard(no_extreme)
pb_ratio_neutral = neutral(pb_ratio_standard，fundamental_df["market_cap"])# 基于因子对池子做筛选
q = pb_ratio_neutral.quantile(0.2)
storck_list = pb_ratio_neutral[pb_ratio_neutral <= q].index
context.storck_list = storck_list# 拿到手里有的
context.last_main_symbol = context.portfolio.positions
# 删掉不在当前因子选中的池子中的股票
context.delete = set(context.last_main_symbol).difference(context.storck_list)
if len(context.delete) != 0:print('调仓')for stock in context.delete:order_target_percent(stock, 0)for stock in context.storck_list:order_target_percent(stock, 1 / len(context.storck_list))# 去极值操作
def filter_3sigma(series, n=3):mean = series.mean()std = series.std()max_range = mean + n * stdmin_range = mean - n * stdreturn np.clip(series, min_range, max_range)# 标准化操作
def standard(series):mean = series.mean()std = series.std()return (series - mean) / std# 中性化操作
def neutral(factor, market_cap):y = factorx = market_capresult = sm.OLS(y.astype(float), x.astype(float)).fit()return result.resid# 判断是否停牌
def filter_paused(stock_list):return [stock for stock in stock_list if not is_suspended(stock)]# 判断是否ST股
def filter_st(stock_list):return [stock for stock in stock_list if not is_st_stock(stock)]# 判断是否是新股
def filter_new(stock_list):return [stock for stock in stock_list if instruments(stock).days_from_listed() >= 180]

（选出被低估的股票并定期调仓）

因子分析

在量化投资与数据分析领域，优质因子需兼具统计学有效性、逻辑合理性与实战可行性：从统计学看，好因子需与目标变量（如收益率）的相关性显著（p值<0.05），信息比率（IR）通常高于1（IR>2则更优），且因子值与目标变量呈现稳定单调性（如按因子分10组后收益率梯度递增）；从逻辑维度讲，因子需符合金融理论（如市净率低对应估值便宜的价值回归逻辑），且驱动机制可解释（如研发投入占比高预示未来盈利增长潜力），而非无经济意义的数据巧合（如股票代码末位数字与收益率的相关性）。简言之，好因子既要通过统计检验证明预测能力，又需具备清晰的经济逻辑支撑，避免成为脱离现实的“伪因子”。

常见的因子分析方法

常见的因子分析方法包括：因子的IC分析法和因子的收益率分析

因子的IC分析

因子的IC分析就是要计算IC值，在投资领域，因子的IC（信息系数）是衡量因子对投资组合未来收益预测能力的一个指标。它反映了因子值与未来收益之间的相关性。具体来说，IC值是通过计算因子值与未来收益之间的相关系数来得到的。
在这里插入图片描述
判断因子与收益率之间的关系（斯皮尔曼相关系数，取值（-1，1）），并进行可视化分析的展示，通过Alphalens工具包对因子做处理

具体使用案例可参考：
github：github文档
使用文档：官方使用文档

这里我们打开米筐的投资研究，可以在其中找到一个与jupyter notebook相似的界面
在这里插入图片描述

import numpy as np
import pandas as pd
from rqdatac import get_price, get_factor, get_industry
from rqdatac import init
import rqdatac# 初始化 rqdatac
rqdatac.init("手机号", "密码")def get_factor_and_close_data(factor_name, industry_code, start_date, end_date, market='cn'):"""获取因子数据和收盘价数据:param factor_name: 因子名称:param industry_code: 行业代码:param start_date: 开始日期:param end_date: 结束日期:param market: 市场代码，默认为 'cn':return: 因子数据和收盘价数据的 DataFrame"""# 获取指定行业的股票列表order_book_ids = get_industry(industry_code, market=market)if not order_book_ids:raise ValueError(f"No stocks found for industry code {industry_code}")order_book_ids = list(order_book_ids.keys())# 获取因子数据factor_data = get_factor(order_book_ids, start_date, end_date, [factor_name])factor_data = factor_data[factor_name]# 获取收盘价数据close_data = get_price(order_book_ids, start_date, end_date, fields="close", expect_df=True)return factor_data, close_datadef calculate_ic(factor_data, close_data, rank_ic=False):"""计算因子的 IC 值:param factor_data: 因子数据的 DataFrame:param close_data: 收盘价数据的 DataFrame:param rank_ic: 是否使用因子值的排名计算 IC，True 表示使用排名，False 表示直接使用因子值:return: IC 值的 Series"""# 计算下一期的收益率close_data_shifted = close_data.shift(-1)returns = close_data_shifted.pct_change()returns = returns.dropna()  # 删除空值# 初始化 IC 值列表ic_values = []# 遍历每个时间点for date in factor_data.index:# 获取当前时间点的因子数据和下一期的收益率factor_values = factor_data.loc[date]return_values = returns.loc[date]# 检查数据的完整性factor_values = factor_values.dropna()return_values = return_values.dropna()# 如果使用排名计算 IC，则对因子值和收益率进行排名if rank_ic:factor_values = factor_values.rank()return_values = return_values.rank()# 检查数据的长度是否一致if len(factor_values) != len(return_values):raise ValueError("The lengths of factor_values and return_values are not equal")# 计算 IC 值ic_value = factor_values.corr(return_values)ic_values.append(ic_value)# 将 IC 值列表转换为 Seriesic_series = pd.Series(ic_values, index=factor_data.index)return ic_series# 示例使用
if __name__ == "__main__":# 设置参数factor_name = "momentum"  # 替换为实际的因子名称industry_code = "C82"  # 替换为实际的行业代码start_date = "2022-01-01"end_date = "2022-12-31"# 获取数据factor_data, close_data = get_factor_and_close_data(factor_name, industry_code, start_date, end_date)# 计算 IC 值（使用因子值计算）ic_values = calculate_ic(factor_data, close_data, rank_ic=False)print("IC 值（使用因子值计算）：")print(ic_values)# 计算 IC 值（使用因子值的排名计算）ic_values_rank = calculate_ic(factor_data, close_data, rank_ic=True)print("IC 值（使用因子值的排名计算）：")print(ic_values_rank)

以上是能在米筐上使用的计算IC值的代码，但是因为笔者没有license，所以也就用不了rqdatac，我们就举个例子演示一下好了：

import pandas as pd
from scipy.stats import pearsonr
import matplotlib.pyplot as plt
import alphalens as al
from scipy.stats import norm# 设置matplotlib字体为SimHei，以支持中文显示
plt.rcParams['font.sans-serif'] = ['SimHei']  # 设置字体为SimHei
plt.rcParams['axes.unicode_minus'] = False  # 正确显示负号# 创建DataFrame
data = {'日期': ['2024-01-01', '2024-01-01', '2024-01-01', '2024-01-02', '2024-01-02', '2024-01-02', '2024-01-03', '2024-01-03', '2024-01-03'],'股票': ['A', 'B', 'C', 'A', 'B', 'C', 'A', 'B', 'C'],'因子值': [0.5, 0.8, 0.2, 0.6, 0.7, 0.3, 0.4, 0.9, 0.1],'未来收益': [0.03, 0.05, -0.01, 0.04, 0.06, -0.02, 0.02, 0.07, -0.03]
}
df = pd.DataFrame(data)# 按日期分组计算IC值
# 修正DeprecationWarning
ic_values = df.groupby('日期').apply(lambda x: pearsonr(x['因子值'], x['未来收益'])[0], include_groups=False)# 将IC值转换为DataFrame
ic_df = pd.DataFrame(ic_values, columns=['IC'])# 修正alphalens.plotting.plot_ic_qq函数
def custom_plot_ic_qq(ic, theoretical_dist=norm, ax=None):if ax is None:fig, ax = plt.subplots()# 计算样本分位数和理论分位数sample_quantiles = ic.quantile([0.01, 0.05, 0.1, 0.25, 0.5, 0.75, 0.9, 0.95, 0.99])theoretical_quantiles = theoretical_dist.ppf([0.01, 0.05, 0.1, 0.25, 0.5, 0.75, 0.9, 0.95, 0.99])# 绘制QQ图ax.scatter(theoretical_quantiles, sample_quantiles, color='blue')ax.plot(theoretical_quantiles, theoretical_quantiles, color='red', linestyle='--')ax.set_title('IC值的QQ图')ax.set_xlabel('理论分位数')ax.set_ylabel('样本分位数')ax.grid(True)return ax# 绘制IC值的QQ图
custom_plot_ic_qq(ic_df['IC'], theoretical_dist=norm)
plt.show()

在这里插入图片描述

经典因子选股

因子收益率分析

因子收益率就是计算因子在收益率当中的权重

• 估值因子：市盈率、市净率、账面市值比、股息率、现金收益率

• 成长因子：净资产收益率及变动、总资产收益率及变动、主营收入增长率、毛利率及变动、净利率及变动

• 价值因子：1、3、6个月收益率、1、3、6个月换手率及变动

• 预期因子：机构覆盖数量、评级调整……

1. 导入必要的库

import numpy as np  # 用于数值计算和数组操作
from jqdata import *  # 聚宽平台API，提供股票数据和交易功能
import pandas as pd  # 用于数据处理和分析
from datetime import datetime, timedelta  # 用于日期计算

numpy：提供高效的多维数组和数学函数，用于因子计算和收益率分析。
jqdata：聚宽量化平台的核心API，用于获取基本面数据和执行交易操作。
pandas：处理表格数据，支持数据清洗、分析和可视化。
datetime：处理日期时间，计算历史数据区间。

2. 初始化函数 `init`

def init(context):context.stocks = index_components('沪深300')  # 获取沪深300成分股context.lastrank = []  # 存储上次选股结果（未实际使用）context.factor_returns = {}  # 存储各因子的历史收益率scheduler.run_monthly(calculate_factor_returns, 1)  # 每月1日计算因子收益率scheduler.run_monthly(rebalance, 2)  # 每月2日调仓（确保因子已计算）

功能：初始化策略参数并设置定时任务。
关键操作：
- index_components('沪深300')：确定股票池为沪深300成分股。
- 定时任务：每月第一个交易日计算因子收益率，第二个交易日执行调仓。

3. 因子收益率计算函数 `calculate_factor_returns`

def calculate_factor_returns(context, bar_dict):end_date = context.current_dt.date()start_date = end_date - timedelta(days=180)  # 回溯6个月数据all_dates = get_trade_days(start_date, end_date)  # 获取交易日列表# 初始化6个因子的收益率记录factor_returns = {'diluted_earnings_per_share': [],  # 每股收益'return_on_equity': [],  # 净资产收益率'return_on_invested_capital': [],  # 投资资本回报率'debt_to_asset_ratio': [],  # 资产负债率'pb_ratio': [],  # 市净率'market_cap': []  # 市值}# 遍历每个交易日（留出20天计算未来收益）for i in range(len(all_dates) - 20):trade_date = all_dates[i]future_date = all_dates[i + 20]# 获取当日基本面数据fundamental_data = get_fundamentals(query(...), date=trade_date)# 对每个因子：for factor in factor_returns.keys():if factor not in fundamental_data.columns:continue# 1. 按因子值排序并分成5组sorted_stocks = fundamental_data.sort_values(by=factor)groups = np.array_split(sorted_stocks.index, 5)# 2. 计算每组未来20天的平均收益group_returns = []for group in groups:# 获取未来价格并计算收益率close_prices = get_price(group_stocks, start=trade_date, end=future_date)returns = [(p1 / p0 - 1) for stock in group_stocks if stock in close_prices]group_returns.append(np.mean(returns) if returns else 0)# 3. 因子收益率 = 最高组收益 - 最低组收益if len(group_returns) == 5:factor_returns[factor].append(group_returns[-1] - group_returns[0])# 计算每个因子的平均历史收益率for factor in factor_returns:context.factor_returns[factor] = np.mean(factor_returns[factor] or [0])logger.info(f"因子收益率计算完成: {context.factor_returns}")

核心逻辑：
1. 数据准备：获取6个月内的交易日和基本面数据。
2. 分组回测：对每个交易日，按因子值将股票分为5组，计算每组未来20天的收益。
3. 因子有效性评估：用最高组与最低组的收益差衡量因子选股能力。
4. 动态权重：历史表现越好的因子，在选股时权重越高。

4. 选股函数 `get_stocks`

def get_stocks(context):# 获取当前基本面数据fundamental_df = get_fundamentals(query(...)).T# 应用因子收益率作为权重for factor in fundamental_df.columns:factor_return = context.factor_returns.get(factor, 0)# 负向因子（值越小越好）使用负收益率if factor in ['debt_to_asset_ratio', 'pb_ratio']:factor_return = -factor_return# 因子得分 = 因子值 × 因子收益率if factor_return != 0:fundamental_df[factor] *= factor_return# 计算总分并选择前10只股票fundamental_df['score'] = fundamental_df.sum(axis=1)return fundamental_df.sort_values('score', ascending=False).index[:10]

创新点：
- 自适应权重：根据因子历史收益率动态调整权重，替代传统固定权重。
- 因子方向处理：对资产负债率、市净率等负向因子，使用负收益率实现"值越小得分越高"。
- 综合评分：通过加权求和得到总分，选出综合得分最高的股票。

5. 调仓函数 `rebalance`

def rebalance(context, bar_dict):# 确保因子收益率已计算if not context.factor_returns:calculate_factor_returns(context, bar_dict)# 计算待买卖的股票stocks = set(get_stocks(context))holdings = set(get_holdings(context))to_buy = stocks - holdingsto_sell = holdings - stocks# 执行交易for stock in to_sell:order_target_percent(stock, 0)  # 卖出if to_buy:average_value = context.portfolio.value / len(to_buy)for stock in to_buy:order_target_value(stock, average_value)  # 等权买入