如何在Python中实现截面Fama-Macbeth回归及解决库调用报错？

问题背景 #

我是Fama-Macbeth回归领域的新手，对该回归的逻辑及量化实现存在困惑，虽已查阅多篇相关资料，但仍需专业帮助。

我的数据为月度个股数据（存储于DataFrame），核心字段样例如下：

注：完整数据包含更多财务指标，RIe Pct Return为个股收益率，Total Return Index为指数收益率。

我理解Fama-Macbeth回归分为两步：

• 第一步：个股收益率对指数收益率回归，得到β值
• 第二步：在每个时间点t执行截面回归

希望获取Python实现该两步回归的模块化代码，支持切换月度/年度等时间频率，结果可存入DataFrame。

现有代码报错 #

此前尝试使用finance-byu库实现，代码如下：

from finance_byu.fama_macbeth import fama_macbeth, fm_summary

merged_df['date_x']=pd.to_datetime(merged_df['date_x'],format='mixed')

result=fama_macbeth(merged_df,pd.to_datetime(merged_df['date_x'],format='mixed'),'Total return index','RIe Pct Return')
print(fm_summary(result))

出现如下AssertionError：

50 def _assertions(data,t,yvar,xvar,intercept,n_jobs=-999,backend=-999,memmap=-999,parallel=-999):
52     assert isinstance(data,pd.core.frame.DataFrame), 'Invalid input for `data`.'
--> 53     assert isinstance(t,str), 'Invalid input for `t`.'
54     assert isinstance(xvar,list), 'Invalid input for `xvar`.'
55     assert isinstance(intercept,bool), 'Invalid input for `intercept`.'

AssertionError: Invalid input for t.

报错原因与修正方案 #

报错原因 #

finance-byu库的fama_macbeth函数要求t参数为字符串类型的日期列名，但代码中传入了datetime序列，不符合参数要求；同时xvar参数需要传入特征列的列表，而非单个字符串。

修正后的finance-byu代码 #

import pandas as pd
from finance_byu.fama_macbeth import fama_macbeth, fm_summary

# 转换日期列
merged_df['date_x'] = pd.to_datetime(merged_df['date_x'], format='mixed')

# 调用函数：t传日期列名，xvar传特征列表
result = fama_macbeth(
    data=merged_df,
    t='date_x',  # 改为日期列的列名字符串
    yvar='RIe Pct Return',
    xvar=['Total Return Index'],  # 改为列表格式
    intercept=True  # 显式指定是否加截距，默认True可省略
)

print(fm_summary(result))

手动实现模块化Fama-Macbeth回归代码 #

如果不想依赖第三方库，可手动实现支持时间频率切换的代码：

import pandas as pd
import numpy as np
from linearmodels import PanelOLS

def fama_macbeth_manual(df, y_col, x_cols, date_col, freq='M'):
    """
    手动实现Fama-Macbeth回归，支持时间频率切换
    参数：
        df: 包含个股数据的DataFrame
        y_col: 因变量列名（个股收益率）
        x_cols: 自变量列名列表（如指数收益率、财务指标）
        date_col: 日期列名
        freq: 时间频率，'M'=月度，'Y'=年度，'Q'=季度，参考pandas offset别名
    返回：
        fm_results: 包含回归结果的DataFrame
        beta_time_series: 各时间点截面回归系数的时间序列
    """
    # 1. 按指定频率生成时间分组键
    df[date_col] = pd.to_datetime(df[date_col])
    df['time_group'] = df[date_col].dt.to_period(freq)
    
    # 2. 遍历每个时间点执行截面回归
    beta_time_series = []
    for time_period, group in df.groupby('time_group'):
        # 跳过样本量不足的时间点
        if len(group) < len(x_cols) + 1:
            continue
        
        # 构建截面回归模型
        formula = f'{y_col} ~ 1 + {" + ".join(x_cols)}'
        model = PanelOLS.from_formula(formula, data=group)
        res = model.fit(cov_type='clustered', cluster_entity=True)
        
        # 保存当前时间点的系数
        beta_dict = {'time_period': time_period}
        beta_dict.update({coef: res.params[coef] for coef in res.params.index})
        beta_time_series.append(beta_dict)
    
    beta_time_series = pd.DataFrame(beta_time_series).set_index('time_period')
    
    # 3. 计算系数的均值、标准差与t统计量
    fm_results = pd.DataFrame({
        '系数均值': beta_time_series.mean(),
        '系数标准差': beta_time_series.std(),
        't值': beta_time_series.mean() / beta_time_series.std(),
        '观测数': beta_time_series.count()
    })
    
    return fm_results, beta_time_series

# 调用示例
fm_results, beta_ts = fama_macbeth_manual(
    df=merged_df,
    y_col='RIe Pct Return',
    x_cols=['Total Return Index', 'Mcap'],  # 可添加更多财务指标
    date_col='date_x',
    freq='M'  # 切换为'Y'则按年度执行
)

# 输出结果
print("Fama-Macbeth回归结果：")
print(fm_results)
print("\n各时间点截面回归系数：")
print(beta_ts)

结果说明 #

• fm_results包含最终的Fama-Macbeth回归系数均值、标准差、t值及观测数，可直接用于分析因子显著性
• beta_time_series存储了每个时间点的截面回归系数，可用于观察因子系数的时间变化趋势

内容的提问来源于stack exchange，提问作者Mostafa Bouzari