因子分析框架详解（AlphaMiner）

由bqd002m创建，最终由bqd002m更新于2026-03-23 14:03 被浏览 4 用户

本文档基于 因子分析框架 中的 AlphaMiner 类，逐步骤、逐细节地介绍整个因子分析流程。

目录

1. 因子配置参数说明
2. 框架总览与初始化流程
3. 步骤一：因子数据获取与预处理
4. 步骤二：因子分组与个股收益率对齐
5. 步骤三：分组收益率与累计收益率计算
6. 步骤四：综合收益与 IC 指标计算
7. 步骤五：图表展示
8. 关键设计细节汇总

一、因子配置参数说明

每次运行分析前，需要先定义一个 dict 类型的参数对象，例如：

alpha_test = {
    "alpha_class":    "test",
    "alpha_name":     "turn_60",
    "alpha_name_chinese": "60日换手率均值",
    "alpha_sql": """
        SELECT date, instrument, -1 * m_avg(turn,60) AS factor
        FROM cn_stock_bar1d
        ORDER BY date, instrument
    """,
    "alpha_desc":   " ",
    "group_num":    10,
    "instruments":  "全市场",
    "benchmark":    "中证500",
    "data_process": True,
    "is_bigvip":    False,
    "is_featured":  False,
}

各参数详细说明

关于 alpha_sql 的规范

• 必须输出 date、instrument、factor 三列
• instrument 格式为 XXXXXX.SH 或 XXXXXX.SZ
• factor 列为因子值，可通过 SQL 内置函数（如 m_avg、m_lag）计算
• 示例中的 -1 * m_avg(turn,60) 表示对 60 日换手率均值取反（做空高换手、做多低换手）

二、框架总览与初始化流程

类结构

AlphaMiner
│
├── __init__(params)       # 主流程，按顺序调用以下方法
│
├── get_factor_data()      # 步骤一：获取并预处理因子数据
├── get_group_data()       # 步骤二：分组 + 对齐个股收益率
├── get_group_cumret()     # 步骤三：分组收益率 & 累计收益率
├── get_whole_perf()       # 步骤四a：全周期综合绩效
├── get_yearly_perf()      # 步骤四b：年度分年绩效
├── get_all_ic()           # 步骤四c：IC 时序计算
│
└── render()               # 步骤五：图表渲染与展示

初始化流程

def __init__(self, params, version="sql"):
    t0 = time.time()
    self.params = params
    self.params['alpha_id'] = 'alpha_' + params['alpha_class'] + '_' + params['alpha_name']
    self.sd = '2018-04-25'   # 固定起始日期
    self.ed = datetime.now().date().strftime("%Y-%m-%d")  # 今天作为结束日期

关键细节：

• alpha_id 由 alpha_class + alpha_name 自动拼接，作为因子的全局唯一标识
• 回测区间固定从 2018-04-25 开始（A 股新规实施后，数据稳定期），到运行当日截止
• 整个流程按顺序串行执行，每步完成后打印耗时，便于性能调优

三、步骤一：因子数据获取与预处理

方法：get_factor_data(data_process, pool_class, alpha_sql)

这是整个框架最核心的数据清洗步骤，通过一个嵌套 CTE 的 SQL 完成以下四个子步骤：

子步骤 1：执行因子 SQL（data_alpha）

用户传入的 alpha_sql 被嵌套为 CTE data_alpha，直接执行，得到 (date, instrument, factor) 三列原始因子值。

子步骤 2：过滤异常值（data_alpha_origin）

data_alpha_origin AS (
    SELECT *
    FROM data_alpha
    QUALIFY COLUMNS(*) IS NOT NULL
        AND factor != 'Infinity'
        AND factor != '-Infinity'
)

• 过滤掉任意列含 NULL 的记录
• 过滤掉因子值为正无穷或负无穷的记录（常见于除法类因子）

子步骤 3：因子预处理（data_alpha_process）

data_alpha_process AS (
    SELECT
        date, instrument, factor,
        clip(factor, c_avg(factor) - 3*c_std(factor), c_avg(factor) + 3*c_std(factor)) AS clipped_factor,
        c_normalize(clipped_factor) AS normalized_factor,
        c_neutralize(normalized_factor, sw2021_level1, LOG(total_market_cap)) AS neutralized_factor,
    FROM data_alpha_origin
    JOIN cn_stock_factors_base USING (date, instrument)
    WHERE {pool_filter}
      AND amount > 0         -- 排除停牌（成交额为零）
      AND st_status = 0      -- 排除 ST 股
      AND trading_days > 252 -- 排除上市不足一年的次新股
      AND (instrument LIKE '%SH' OR instrument LIKE '%SZ')  -- 只保留沪深两市
    QUALIFY COLUMNS(*) IS NOT NULL
    ORDER BY date, instrument
)

三步预处理的逻辑如下：

① 去极值（Winsorize / Clip）

clipped_factor = clip(factor, mean - 3σ, mean + 3σ)

• 每个截面日（同一 date 下）计算均值 c_avg 和标准差 c_std
• 将超出 [均值 - 3σ, 均值 + 3σ] 范围的因子值截断到边界
• 目的：消除极端离群值对后续分析的干扰

② 截面标准化（Z-score Normalize）

normalized_factor = c_normalize(clipped_factor)

• 对每个截面日的去极值后因子做 Z-score 标准化：(x - mean) / std
• 使不同时期、不同量级的因子值具有可比性

③ 行业 + 市值中性化（Neutralize）

neutralized_factor = c_neutralize(normalized_factor, sw2021_level1, LOG(total_market_cap))

• 以申万2021一级行业分类（sw2021_level1）和对数市值（LOG(total_market_cap)）为控制变量
• 通过截面回归，剥离因子中与行业、市值相关的部分，保留纯因子暴露
• 这是量化因子分析的标准流程，确保因子不是在暗中做行业轮动或大小盘轮动

子步骤 4：股票池过滤

pool_dict = {
    "中证500":  "is_zz500 = 1",
    "中证1000": "is_zz1000 = 1",
    "沪深300":  "is_hs300 = 1",
    "全市场":   "1=1",
}

• 通过 cn_stock_factors_base 表中的指数成分股字段来过滤
• 全市场 使用 1=1 不做任何过滤

是否预处理的分支

data_process_sql = (
    "SELECT date, instrument, neutralized_factor AS factor FROM data_alpha_process ..."
    if data_process == True
    else "SELECT * FROM data_alpha"
)

• data_process=True（默认）：最终取中性化后的 neutralized_factor
• data_process=False：直接使用原始 factor，不做任何处理

数据查询

df = dai.query(sql, filters={'date': [self.sd, self.ed]}).df()

• 使用 dai 框架执行 SQL，并通过 filters 参数在数据库层面过滤日期范围，提升效率
• 返回 pd.DataFrame，列为 (date, instrument, factor)

四、步骤二：因子分组与个股收益率对齐

方法：get_group_data()

将因子数据与未来收益率对齐，并按因子值大小划分分组。

子步骤 1：获取每日收益率

SELECT
    date, instrument,
    (m_lead(open, 2) / m_lead(open, 1) - 1) AS daily_ret
FROM cn_stock_bar1d
WHERE date BETWEEN DATE '{sd}' - INTERVAL 10 DAY AND '{ed}'
ORDER BY date, instrument

关键细节：T+2 开盘价收益率

• m_lead(open, 1)：下下个交易日的开盘价（T+1 开盘价）
• m_lead(open, 2)：再下一个交易日的开盘价（T+2 开盘价）
• daily_ret = m_lead(open,2) / m_lead(open,1) - 1：即在 T+1 日开盘买入，T+2 日开盘卖出 的收益率
• 这样设计是为了模拟实盘中：今日（T 日）收盘后得到因子信号，T+1 日开盘建仓，持有一天后 T+2 日开盘平仓
• 日期范围向前额外延伸 10 个自然日（- INTERVAL 10 DAY），确保有足够的 m_lead 数据不会产生空值

子步骤 2：因子分组函数

def cut(df, group_num=10):
    df = df.drop_duplicates('factor')       # 去掉相同因子值的重复记录
    df['group'] = pd.qcut(df['factor'], q=group_num, labels=False, duplicates='drop')
    df = df.dropna(subset=['group'], how='any')
    df['group'] = df['group'].apply(int).apply(str)  # 转为字符串 "0"~"9"
    return df

• pd.qcut：等分位数分组，将因子值从小到大排序后均匀切分为 group_num 组
• 第 0 组（"0"）：因子值最小的股票（空头组合）
• 第 9 组（"9"）：因子值最大的股票（多头组合）
• drop_duplicates('factor')：避免完全相同的因子值导致分位数边界冲突
• duplicates='drop'：进一步处理边界重叠问题

子步骤 3：合并并逐日分组

merge_data = pd.merge(factor_data, daily_ret_data, on=['date','instrument'], how='left')
group_data = merge_data.groupby('date', group_keys=False).apply(cut, group_num=...)

• 以 (date, instrument) 为键，将因子值与 T+2 收益率左连接合并
• 按日期分组，对每个截面日独立执行 cut 分组，确保每天都是等量分组
• 最终 group_data 包含列：date, instrument, factor, daily_ret, group

五、步骤三：分组收益率与累计收益率计算

方法：get_group_cumret()

子步骤 1：获取基准收益率

SELECT date, instrument,
    (close - m_Lag(close,1)) / m_LAG(close, 1) as benchmark_ret
FROM cn_stock_index_bar1d
WHERE instrument = '{benchmark_code}'

基准指数代码映射：

• 使用指数的日度收盘价涨跌幅作为基准收益率

子步骤 2：计算各分组日均收益率

groupret_data = group_data[['date','group','daily_ret']] \
    .groupby(['date','group'], group_keys=False) \
    .apply(lambda x: np.nanmean(x)) \
    .reset_index()

• 对每个 (date, group) 组合，取组内所有股票 daily_ret 的等权平均（忽略 NaN）
• 即每个分组的日收益率 = 该组所有成分股的等权平均日收益率

子步骤 3：Pivot 宽表并添加多空组合与基准

groupret_pivotdata = groupret_data.pivot(index='date', values='g_ret', columns='group')
groupret_pivotdata['ls'] = groupret_pivotdata[str(group_num-1)] - groupret_pivotdata['0']
groupret_pivotdata['bm'] = bm_ret['benchmark_ret']
groupret_pivotdata = groupret_pivotdata.shift(1)   # 关键：shift(1) 时间对齐

• 将长表转为宽表，每列对应一个分组（"0" 到 "9"），行为日期
• ls（Long-Short）= 多头组（第 9 组）- 空头组（第 0 组），即多空对冲组合
• bm：基准收益率列
• shift(1) 的作用：收益率发生在 T+1 ~ T+2 之间，而因子信号在 T 日生成，shift 操作将收益率向后移一天，与因子日期对齐，避免未来函数

子步骤 4：计算累计收益率

groupcumret_pivotdata = groupret_pivotdata.cumsum()

• 对各分组的日收益率做累加求和（cumsum），得到累计收益率曲线
• 注意：这里用的是算术累加（非复利），适合短期/因子分层对比，不是精确的复利净值

六、步骤四：综合收益与 IC 指标计算

（一）全周期综合绩效：get_whole_perf()

分别对多头组（Long）、空头组（Short）、**多空对冲（Long-Short）**三个组合计算以下指标：

1. 基础绩效指标（get_basic_perf）

关键细节： 夏普比率的无风险利率使用 0.035/242，即 3.5% 年化利率折算到日度，这是 A 股常用的基准利率设定。

2. IC 绩效指标（get_ic）

def cal_ic(df):
    return df['daily_ret'].corr(df['factor'], method='spearman')

• 使用 Spearman 秩相关系数计算 IC（Information Coefficient）
• Spearman IC 对排名的相关性，比 Pearson IC 更稳健，不受极端值影响

针对多头/空头组合，只用对应分组的数据计算 IC；多空组合则合并两组数据。

3. 换手率（get_turnover）

def count_repeat(dfs):
    if dfs.name > 0:
        return len(set(dfs['instrument']) & set(dfs['instrument_lag']))
    else:
        return 0

df_ins['turnover'] = 1 - df_ins['repeat_count'] / df_ins['instrument_count']

• 每天计算当日成分股与前一日成分股的交集比例
• 换手率 = 1 - 重合数量 / 当日成分股数量
• 换手率越高，说明因子信号每天调仓幅度越大，交易成本越高
• 多空组合换手率 = 多头组换手率 + 空头组换手率

汇总

三类指标（IC 绩效、基础绩效、换手率）被合并为一张宽表 whole_perf，每行对应一种组合类型（long/short/long_short）。

（二）年度分年绩效：get_yearly_perf()

对多头组合按年度切分，分别计算每年的绩效指标。

year_df['year'] = year_df['date'].apply(lambda x: x.year)
yearly_perf = year_df.groupby(['year']).apply(cal_Performance)

年度绩效指标与全周期基本一致，包含：return_ratio、annual_return_ratio、ex_return_ratio、ex_annual_return_ratio、sharp_ratio、return_volatility、max_drawdown、win_percent、trading_days

额外添加年度 IC：

ic_data['year'] = ic_data['date'].apply(lambda x: x.year)
yearly_ic = ic_data.groupby('year').apply(lambda x: np.nanmean(x['g_ic']))
yearly_perf['ic'] = yearly_ic

（三）IC 时序：get_all_ic()

def cal_ic(df):
    return df['daily_ret'].corr(df['factor'], method='spearman')

group_ic_data = group_data[['date','daily_ret','factor']] \
    .groupby('date', group_keys=False) \
    .apply(lambda x: cal_ic(x)) \
    .reset_index()

group_ic_data = group_ic_data.shift(1)                     # 时间对齐
group_ic_data['ic_cumsum'] = group_ic_data['g_ic'].cumsum()  # 累计 IC
group_ic_data['ic_roll_ma'] = group_ic_data['g_ic'].rolling(22).mean()  # 近22日均值

• 对全部股票（不区分分组）逐日计算当日截面 Spearman IC
• shift(1)：与收益率时间对齐
• ic_cumsum：IC 累积曲线，可直观反映因子的持续有效性，斜率越稳定越好
• ic_roll_ma：22 日滚动平均 IC，约等于一个月，平滑噪声，观察因子近期有效性趋势

七、步骤五：图表展示

方法：render(local_plot=False)

调用 bigcharts 库生成 7 张图表，统一渲染为一个 page 页面：

IC 显著性判断

abs_ic = self.ic['g_ic'].abs()
significant_ic_ratio = abs_ic[abs_ic >= 0.02].shape[0] / abs_ic.shape[0]

• |IC| >= 0.02 被视为具有统计显著性的 IC 值
• significant_ic_ratio 表示所有交易日中，IC 绝对值超过 0.02 的比例
• 该比例越高，说明因子的预测效果越稳定

local_plot 参数

• local_plot=True：在 Jupyter Notebook 中直接 display 渲染，用于本地调试
• local_plot=False：不本地展示，通常用于平台提交流程

八、关键设计细节汇总

时间对齐逻辑

整个框架的时间对齐关系如下：

T 日收盘 → 生成因子信号
T+1 日开盘 → 按信号建仓
T+2 日开盘 → 平仓，计算收益

• daily_ret = m_lead(open,2) / m_lead(open,1) - 1：T+1 开 → T+2 开的收益
• groupret_pivotdata.shift(1)：将收益率向后移一天，与 T 日的因子对齐
• group_ic_data.shift(1)：同上，IC 计算也做相同对齐

因子方向约定

• 因子值越大 → 第 group_num-1 组（最高组）→ 多头
• 因子值越小 → 第 0 组（最低组）→ 空头
• 若因子与收益负相关（如高换手率对应低未来收益），应在 SQL 中乘以 -1 进行方向调整（示例中已对换手率取反）

股票过滤规则

以下类型股票被强制排除，确保分析的可操作性：

年化收益率计算约定

annual_return_ratio = series.sum() * 242 / trading_days

• 使用 242 个交易日/年 作为年化基准（A 股惯例）
• 采用算术年化（非几何复利），适合分析因子多空组合的超额收益

IC 计算范围差异

快速使用示例

# 1. 定义因子参数
alpha_test = {
    "alpha_class": "test",
    "alpha_name": "turn_60",
    "alpha_name_chinese": "60日换手率均值",
    "alpha_sql": """
        SELECT date, instrument, -1 * m_avg(turn,60) AS factor
        FROM cn_stock_bar1d
        ORDER BY date, instrument
    """,
    "alpha_desc": " ",
    "group_num": 10,
    "instruments": "全市场",
    "benchmark": "中证500",
    "data_process": True,
    "is_bigvip": False,
    "is_featured": False,
}

# 2. 运行因子分析
alpha_miner = AlphaMiner(params=alpha_test)

# 3. 展示图表（本地 Jupyter）
alpha_miner.render(local_plot=True)

运行完成后，可通过以下属性直接访问中间结果：

\