第一章：数据获取与预处理——打好ESG轮动的地基

各位同学，欢迎来到《ESG因子在行业轮动中的应用实战》的第一章。

做量化投资，尤其是ESG因子相关的策略，最怕什么？我个人觉得，最怕数据还没整明白，就开始跑模型。你想想看，ESG评级数据、财务数据、行情数据，这三个东西来源不同、频率不同、格式不同，硬凑在一起，那结果肯定是一团浆糊。

所以，这一章咱们就踏踏实实地把数据这块地基打牢。说白了，就是教会你怎么用Python把这些数据对齐、清洗、标准化，最终构建出一个可以直接用于分析的面板数据结构。

1.1 数据获取：三路数据源的汇合

做ESG行业轮动，我们需要三路数据：

• ESG评级数据：来自商道融绿、华证、中财绿金院等机构。我习惯用华证ESG评级，因为它的覆盖度广，更新也及时。
• 财务数据：主要是ROE、营收增速、资产负债率等。这些可以从Wind、Tushare或者AkShare获取。
• 行情数据：日频的收盘价、成交量，用于计算行业指数收益。

这里有个坑，我必须要提醒你。ESG评级通常是季度或半年度更新，而财务数据是季度更新，行情数据是日频。频率不一致，是第一个要解决的问题。

下面是我常用的数据获取代码示例：

import akshare as ak
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# 1. 获取华证ESG评级数据（示例）
def get_esg_data():
    # 注意：实际使用时需要替换为真实的API接口
    esg_df = ak.stock_hsgt_esg(symbol="全部A股")
    esg_df['报告期'] = pd.to_datetime(esg_df['报告期'])
    return esg_df

# 2. 获取财务数据（以ROE为例）
def get_financial_data():
    fin_df = ak.stock_yjbb_em(date="20240930")
    fin_df = fin_df[['股票代码', '净资产收益率']]
    fin_df.columns = ['code', 'roe']
    return fin_df

# 3. 获取行业指数行情
def get_market_data():
    # 这里用申万一级行业指数为例
    index_df = ak.stock_zh_index_daily(symbol="sw801010")
    index_df['date'] = pd.to_datetime(index_df['date'])
    return index_df

1.2 数据对齐：让三个时钟同步

数据拿回来了，接下来就是对账。我一般会按照“季度”这个粒度来对齐。为什么是季度？因为ESG评级和财务数据都是季度更新，行情数据可以取季度末的收盘价。

具体做法是：

1. 把ESG评级数据按“股票代码+报告期”做唯一索引
2. 把财务数据也按“股票代码+报告期”做唯一索引
3. 把行情数据按“股票代码+日期”做索引，然后只保留每个季度最后一个交易日的数据

嗯，这里要注意一个细节：ESG评级的报告期可能和财务报告期不完全一致。比如有些ESG评级是6月底发布，但覆盖的是3月底的数据。我建议统一使用“数据覆盖的截止日期”作为对齐基准。

# 数据对齐示例
def align_data(esg_df, fin_df, market_df):
    # 1. 提取季度末日期
    market_df['quarter'] = market_df['date'].dt.to_period('Q')
    quarter_end = market_df.groupby('quarter')['date'].max().reset_index()
    quarter_end.columns = ['quarter', 'end_date']
    
    # 2. 合并行情与季度末日期
    market_aligned = pd.merge(market_df, quarter_end, on=['quarter', 'date'], how='inner')
    
    # 3. 合并ESG与财务数据
    esg_fin = pd.merge(esg_df, fin_df, on=['code', '报告期'], how='outer')
    
    # 4. 最终对齐
    final_data = pd.merge(esg_fin, market_aligned, 
                          left_on=['code', '报告期'], 
                          right_on=['code', 'end_date'], 
                          how='inner')
    return final_data

1.3 缺失值处理：别让NaN毁了你的策略

数据对齐后，你会发现有很多缺失值。尤其是ESG评级数据，小市值公司经常没有评级。怎么办？

我个人的处理原则是：

• ESG评级缺失：如果某只股票连续两个季度都没有ESG评级，直接剔除。如果只是偶尔缺失，用前向填充（forward fill）。
• 财务数据缺失：用行业均值填充。比如某公司的ROE缺失，就用同行业其他公司的平均ROE来补。
• 行情数据缺失：如果停牌导致缺失，用复牌后的第一个交易日数据填充。如果退市了，直接剔除。

# 缺失值处理
def handle_missing(data):
    # 1. ESG评级前向填充
    data['esg_score'] = data.groupby('code')['esg_score'].transform(
        lambda x: x.fillna(method='ffill')
    )
    
    # 2. 财务数据用行业均值填充
    data['roe'] = data.groupby('industry')['roe'].transform(
        lambda x: x.fillna(x.mean())
    )
    
    # 3. 剔除ESG评级连续缺失超过2期的股票
    data['esg_missing_count'] = data.groupby('code')['esg_score'].transform(
        lambda x: x.isna().rolling(2).sum()
    )
    data = data[data['esg_missing_count'] < 2]
    
    return data.drop(columns=['esg_missing_count'])

1.4 标准化：让不同量纲的数据可比较

ESG评级可能是A、B、C这样的等级，ROE是百分比，行情收益率是小数。量纲不同，没法直接放在一起做因子分析。

我常用的标准化方法有两种：

方法	适用场景	公式
Z-score标准化	连续型数据（如ROE、收益率）	(x - mean) / std
排序标准化	等级型数据（如ESG评级）	rank(x) / N

对于ESG评级，我建议先把它映射成数值。比如A=5，B=4，C=3，D=2，E=1。然后再做排序标准化，这样能保留评级之间的相对顺序关系。

# 标准化处理
def standardize_factors(data):
    # 1. ESG评级映射为数值
    esg_map = {'A': 5, 'B': 4, 'C': 3, 'D': 2, 'E': 1}
    data['esg_numeric'] = data['esg_rating'].map(esg_map)
    
    # 2. 排序标准化（按行业分组）
    data['esg_rank'] = data.groupby('industry')['esg_numeric'].rank(pct=True)
    
    # 3. Z-score标准化（按行业分组）
    data['roe_zscore'] = data.groupby('industry')['roe'].transform(
        lambda x: (x - x.mean()) / x.std()
    )
    
    return data

1.5 构建面板数据结构：最终形态

经过以上步骤，我们最终得到的数据结构应该是这样的：

code	date	industry	esg_rank	roe_zscore	return
000001	2024-03-31	银行	0.85	1.23	0.05
000002	2024-03-31	房地产	0.42	-0.67	-0.02
...	...	...	...	...	...

这就是标准的面板数据结构。每一行代表一只股票在某个时间点的截面数据，所有时间点的数据堆叠在一起，就形成了面板数据。

有了这个结构，我们就可以做后续的因子分析、行业分组、轮动策略回测了。

好了，这一章的内容就到这里。数据预处理虽然繁琐，但它是整个ESG行业轮动策略的基石。下一章我们会基于这个面板数据，开始构建真正的ESG因子。

---

公众号:蓝海资料掘金营，微信deep3321