📚 课程目录第2章财务数据整合

2. 财务数据整合

任务目标

构建一个最小可用的基本面因子库：读取公司财务报表原始数据（利润表、资产负债表、股本数据），计算PE、PB、ROE三个常用因子，并解决不同报告期（年报/季报）与行情日期的对齐问题，最终输出按股票、交易日索引的因子面板。

数据与依赖

使用 pandas 与 numpy。输入数据为三张CSV（模拟结构，路径可替换）：

income.csv：股票代码、报告期、归母净利润（单位：元）
balance.csv：股票代码、报告期、归母净资产（单位：元）
shares.csv：股票代码、报告期、总股本（单位：股）
prices.csv：股票代码、交易日期、收盘价（单位：元）

报告期格式为 YYYYMMDD（如20231231代表年报，20240930代表三季报）。财务数据采用“报告期发布日滞后3个月”的规则对齐：即报告期为T的财务数据，从T+3个月后的首个交易日开始生效，避免未来函数。

核心实现

import pandas as pd
import numpy as np

def load_data():
    income = pd.read_csv('income.csv', parse_dates=['报告期'])
    balance = pd.read_csv('balance.csv', parse_dates=['报告期'])
    shares = pd.read_csv('shares.csv', parse_dates=['报告期'])
    prices = pd.read_csv('prices.csv', parse_dates=['交易日期'])
    return income, balance, shares, prices

def merge_financials(income, balance, shares):
    """合并三张财务表，保留最新报告期数据（去重）"""
    df = income.merge(balance, on=['股票代码', '报告期'], how='outer')
    df = df.merge(shares, on=['股票代码', '报告期'], how='outer')
    # 按股票+报告期去重，保留最后一条（若同一报告期多行）
    df = df.sort_values(['股票代码', '报告期']).drop_duplicates(
        subset=['股票代码', '报告期'], keep='last')
    return df

def calc_factors(fin_df):
    """计算PE、PB、ROE（基于报告期数据，不含价格）"""
    fin_df['ROE'] = fin_df['归母净利润'] / fin_df['归母净资产']
    # PE、PB需要价格，此处先计算每股指标
    fin_df['EPS'] = fin_df['归母净利润'] / fin_df['总股本']
    fin_df['BPS'] = fin_df['归母净资产'] / fin_df['总股本']
    return fin_df

def align_to_trading_dates(fin_df, prices):
    """将财务因子对齐到交易日：报告期+3个月后生效"""
    # 计算生效日期（报告期 + 3个月）
    fin_df['生效日期'] = fin_df['报告期'] + pd.DateOffset(months=3)
    
    # 对每个交易日，取该日之前（含）最近已生效的财务数据
    result = []
    for code in prices['股票代码'].unique():
        price_code = prices[prices['股票代码'] == code].sort_values('交易日期')
        fin_code = fin_df[fin_df['股票代码'] == code].sort_values('生效日期')
        if fin_code.empty:
            continue
        # 使用merge_asof：取每个交易日前最近已生效的财务数据
        merged = pd.merge_asof(
            price_code, fin_code,
            left_on='交易日期', right_on='生效日期',
            direction='backward'
        )
        result.append(merged)
    
    if not result:
        return pd.DataFrame()
    return pd.concat(result, ignore_index=True)

def build_factor_library():
    income, balance, shares, prices = load_data()
    fin_df = merge_financials(income, balance, shares)
    fin_df = calc_factors(fin_df)
    
    aligned = align_to_trading_dates(fin_df, prices)
    
    # 计算最终因子值
    aligned['PE'] = aligned['收盘价'] / aligned['EPS']
    aligned['PB'] = aligned['收盘价'] / aligned['BPS']
    
    # 清理无穷值
    factor_cols = ['PE', 'PB', 'ROE']
    aligned[factor_cols] = aligned[factor_cols].replace([np.inf, -np.inf], np.nan)
    
    # 输出因子面板：日期 × 股票
    panel = aligned.pivot_table(
        index='交易日期', columns='股票代码', values=factor_cols
    )
    # 展平多级列：如 ('PE', '000001')
    panel.columns = [f'{factor}_{code}' for factor, code in panel.columns]
    return panel

if __name__ == '__main__':
    panel = build_factor_library()
    print(panel.head())
    print(f'\n因子库形状: {panel.shape}')
    print(f'缺失值比例: {panel.isna().mean().mean():.2%}')

运行与验证

运行方式：将上述代码保存为 factor_library.py，与四个CSV文件置于同一目录，执行 python factor_library.py。

验证方法：

检查输出面板的索引为交易日序列，列名为 因子_股票代码 格式。
手工验证第一行数据：取某股票最早交易日，找到其前最近生效的财务报告期，用该报告期的净利润/净资产/股本计算PE/PB/ROE，与输出值对比。
确认对齐逻辑：2023年年报（报告期20231231）的生效日为2024-03-31，因此2024-03-31之前的交易日不应出现该年报数据。
预期输出示例（仅演示结构，非实测值）：

             PE_000001  PB_000001  ROE_000001  PE_000002  ...
交易日期                                                      
2024-01-02   12.35      1.52       0.123       NaN        ...
2024-01-03   12.40      1.53       0.123       NaN        ...
...

若某股票在当日无已生效财务数据，对应位置为NaN，属正常现象。

常见问题

报告期重复：若同一股票同一报告期出现多行，代码按最后一行保留；实际使用中应确保源数据已去重。
停牌缺失：若股票在交易日无价格记录，则不会出现在该日面板中；如需保留可先对价格做向前填充。
因子极值：PE、PB可能因净利润或净资产接近0产生极端值，代码已替换为NaN；实际使用可再做缩尾处理。