量化-因子处理

超额收益：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_func
    run_func(init=init, handle_bar=handle_bar, config=config)

代码流程说明：

1. 先导入平台模块，定义 log_cash 函数，它借助 context 获取账户现金并打印。
2. init 函数里通过 scheduler.run_daily(log_cash) ，安排 log_cash 每天执行。
3. handle_bar 这里虽简单示例没复杂逻辑，但实际策略中可处理实时行情、交易信号等。
4. 最后通过 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 pd

def filter_extreme_mad(series, n):
#n一般是取1.4826
    median = series.quantile(0.5)
    mad = ((series - median).abs().quantile(0.5))
    max_range = median + n * mad
    min_range = median - n * mad
    return 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 np

def filter_extreme_3sigma(series, n=3):
    mean = series.mean()
    std = series.std()
    max_range = mean + n * std
    min_range = mean - n * std
    return 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 sm

def 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 * std
    min_range = mean - n * std
    return 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 = factor
    x = market_cap
    result = 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_data

def 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 值列表转换为 Series
    ic_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 - holdings
    to_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)  # 等权买入

• 交易逻辑：
  1. 对比当前持仓与目标组合，计算需要买入和卖出的股票。
  2. 先卖出不需要的股票，再将资金平均分配到待买入的股票。
  3. 等权分配：每只股票持有相同市值，分散风险。

策略核心优势

1. 动态因子权重：通过历史数据自动学习各因子的有效性，避免主观设定权重。
2. 多因子融合：综合考虑盈利性（EPS、ROE）、估值（PB）、财务健康（负债率）等多个维度。
3. 因子方向自适应：自动处理正向因子（值越高越好）和负向因子（值越低越好）。
4. 定期调仓：每月重新评估因子有效性并调整持仓，适应市场风格变化。

潜在优化方向

1. 增加更多因子：如动量因子、波动率因子等，丰富选股维度。
2. 优化回测区间：根据市场周期动态调整历史数据长度。
3. 加入风险控制：设置最大持仓数量、单只股票最大权重、止损机制等。
4. 交易成本考虑：在收益计算中纳入交易成本，避免过度交易。

这个策略通过数据驱动的方式挖掘因子有效性，相比传统固定权重模型更能适应市场变化，具有较强的实战价值。