1.概述

在计算完因子数据之后，进行下一步的模型训练之前，通常需要对因子数据进行预处理，以及中性化处理。其中预处理比较简单，一般就是3倍MAD截断，zscore标准化，缺失值填充为0。中性化稍微复杂一些，本文将从市值中性化开始介绍如何进行市值中性化，下一篇将介绍如何进行行业中性化。

2.市值中性化

2.1 市值中性化的必要性与逻辑

市值中性化是因子中性化处理中最常见且重要的一种，其核心目的是剔除因子值中由于市值（Size）因素引起的系统性影响，使得因子能够更纯粹地反映其自身的信息，从而提升因子的解释力与有效性。

在实际研究中，许多因子（如换手率、波动率、账面市值比等）往往与市值存在显著相关性。例如：

小市值股票通常换手率较高、波动率较大；
大市值公司财务更稳定、信息披露更充分。

如果不对这些因子进行市值中性化处理，其因子值往往会集中在小市值股票上，导致投资组合隐含较强的市值暴露（Size Exposure）。这种情况下，即便因子组合取得了较好的收益，也可能主要来源于市值因子（Size）的表现，而非该因子本身的有效性。

换言之，未经中性化处理的因子，其历史收益可能掺杂了市值因子的 Alpha，从而掩盖了因子自身真实的预测能力。

通过对因子进行市值中性化处理，可以有效控制市值因素的干扰，提取出因子在剔除市值影响后的“纯净”收益，使我们能够更准确地评估其独立的选股能力。

2.2 市值中性化的计算

市值中性化通常通过线性回归残差法实现。其基本思想是使用市值对因子值进行回归，将回归残差作为中性化后的因子值，剔除市值对因子的线性影响。

回归模型设定：

设：

$X_i$ 表示第 $i$ 只股票的原始因子值；
$M_i$ 表示第 $i$ 只股票的市值（通常取对数市值 $\log(\text{MarketCap})$ ）；
$\hat{X}_i$ 为对市值回归后的拟合值；
$\varepsilon_i$ 为残差，即中性化后的因子值。

回归模型如下：

$X_i$ = $\alpha$ + $\beta$ × log（ $M_i$ ) + $\varepsilon_i$

其中：

$\alpha$ 为截距项；
$\beta$ 为市值的回归系数；
$\varepsilon_i$ 为残差项，即市值中性化后的因子值。

中性化后的因子值：

$X_i^{\text{neutral}} = \varepsilon_i = X_i - (\alpha + \beta \cdot \log(M_i))$

该残差 $X^{\text{neutral}}_i$ 即为剔除市值影响后的中性因子值，可用于后续的选股、打分或多因子组合建模。

2.3 代码实例

首先我们先看看市值中性化的代码：

def neutralize_by_mktcap(self, df_factor, df_cap):
    """
    向量化方式对因子做市值中性化（面板数据），返回添加残差列。
    
    参数：
        df_factor: 包含 ts_code, trade_date, 因子值 的 DataFrame
        df_cap:    包含 ts_code, trade_date, total_mv 的市值 DataFrame
        
    返回：
        DataFrame: 包含 ts_code, trade_date, 因子中性化后列（如 "factor_neu"）
    """
    import numpy as np
    import pandas as pd
    import statsmodels.api as sm

    # 合并市值数据
    df = (
        df_factor
        .merge(df_cap, on=["ts_code", "trade_date"], how="inner")
        .copy()
    )

    # 构造 log 市值
    df["log_mkt_cap"] = np.log(df["total_mv"].replace(0, np.nan))

    # 自变量列
    exog_cols = ["log_mkt_cap"]

    # 定义单期中性化逻辑
    def neutralize_group(g):
        X = g[exog_cols]
        y = g[self.factor_name]
        valid = pd.concat([X, y], axis=1).dropna()

        if valid.empty:
            g[self.factor_name + "_neu"] = np.nan
            return g[["ts_code", "trade_date", self.factor_name + "_neu"]]

        X_valid = sm.add_constant(valid[exog_cols], has_constant='add').astype(float)
        y_valid = valid[self.factor_name].astype(float)
        try:
            model = sm.OLS(y_valid, X_valid)
            res = model.fit()
            g[self.factor_name + "_neu"] = res.resid
        except Exception as e:
            print(f"中性化失败: {e}, 日期: {g['trade_date'].iloc[0]}")
            g[self.factor_name + "_neu"] = np.nan

        return g[["ts_code", "trade_date", self.factor_name + "_neu"]]

    # 分组中性化
    df_resid = df.groupby("trade_date", group_keys=False).apply(neutralize_group)
    return df_resid

接下来，我们将用上文中提到的换手率来演示对比增加了市值中性化之后对收益的影响。问了下deepseek，我们直接用20天的平均换手率，然后在cursor里面直接提问，cursor基于原有的框架快速帮我们生成了因子的实现，计算逻辑如下：

 def compute_factor(self, df):
        df = df.sort_values(['ts_code', 'trade_date'])
        df = self._validate_data(df)
        if df.empty:
            print("警告：数据清洗后为空，请检查源数据质量")
            return pd.DataFrame(columns=['ts_code', 'trade_date', self.factor_name])

        def calculate_avg_for_stock(group):
            avg = group[self.data_field].rolling(window=self.window, min_periods=self.window).mean()
            # 对平均换手率取log
            group[self.factor_name] = np.log(avg)
            return group

        df = df.groupby('ts_code').apply(calculate_avg_for_stock).reset_index(drop=True)
        df = df.dropna(subset=[self.factor_name])
        return df[['ts_code', 'trade_date', self.factor_name]]

一开始不进行市值中性化，我们看下因子回测的情况，发现最高组有很多空值。

怀疑可能是20天平均换手率分布不均匀的问题，我们绘制某一天的因子直方图如下：

最大值都被预处理中的3倍MAD截断压缩到了右侧。这种情况可以考虑跟市值因子一样处理，取对数，在计算完的因子值加上对数处理，如下：

    def compute_factor(self, df):
        df = df.sort_values(['ts_code', 'trade_date'])
        df = self._validate_data(df)
        if df.empty:
            print("警告：数据清洗后为空，请检查源数据质量")
            return pd.DataFrame(columns=['ts_code', 'trade_date', self.factor_name])

        def calculate_avg_for_stock(group):
            avg = group[self.data_field].rolling(window=self.window, min_periods=self.window).mean()
            # 对平均换手率取log
            group[self.factor_name] = np.log(avg)
            return group

        df = df.groupby('ts_code').apply(calculate_avg_for_stock).reset_index(drop=True)
        df = df.dropna(subset=[self.factor_name])
        return df[['ts_code', 'trade_date', self.factor_name]]

再次绘制直方图，看下是否有改善。

从图中可以看出，取完对数，分布情况改善很多，再运行一次回测。

很明显是个负向因子，平均换手率越高表现越不好。

接下来加入市值中性化，查看回测情况。

从结果上看，加了市值中性化之后，IC的绝对值有所改善。

4. 结语

本文介绍了如何进行市值中性化，然后选择了20日平均换手作为例子，但整体回测来看，市值中性化之后对该因子的影响有限。同时在遇到分组失败时，需要查看下因子的分布情况，尽量使因子分布更加合理。本期就到这，下一篇给大家分享下如何进行行业中性化，我们下期见！