第4章分歧数据的获取与清洗

4. 分歧数据的获取与清洗：API调用、网页爬虫、数据对齐与缺失值处理

好，咱们进入实操环节。前面聊了那么多分歧指标的计算逻辑，现在得解决一个现实问题——数据从哪来？怎么把它弄干净？

说实话，ESG评级数据的分歧，很大一部分其实不是真正的意见分歧，而是数据源本身的问题。你想想看，MSCI、Sustainalytics、Refinitiv这几家，数据格式、更新频率、覆盖范围全都不一样。我刚开始做这个课题时，光是把三家数据对齐就花了两周。

这一章，我就把我在项目中踩过的坑、总结的经验，一次性讲清楚。

4.1 数据源的选择与API调用

目前主流的ESG评级数据源，我列个表给你看：

数据源	访问方式	覆盖范围	更新频率	成本
MSCI ESG Ratings	API / 文件下载	全球8000+公司	月度	高
Sustainalytics	API / 终端	全球12000+公司	季度	高
Refinitiv (LSEG)	API (Eikon)	全球9000+公司	年度	中
Bloomberg ESG	终端 / API	全球11000+公司	年度	高
CDP (气候披露)	API / 公开下载	自愿披露公司	年度	低

我个人习惯优先用API。为什么？因为手动下载Excel文件，版本管理太容易出问题。我曾经有个项目，分析师手动下载了不同日期的MSCI数据，结果时间戳对不上，回测结果差了8个点。

下面是一个Refinitiv Eikon API的调用示例，我稍微简化了一下：

import eikon as ek
import pandas as pd

# 设置API密钥
ek.set_app_key('your_api_key_here')

# 定义需要获取的ESG字段
fields = [
    'TR.ESGScore',           # 总ESG评分
    'TR.EnvironmentalScore', # 环境评分
    'TR.SocialScore',        # 社会评分
    'TR.GovernanceScore'     # 治理评分
]

# 获取数据
rics = ['AAPL.O', 'MSFT.O', 'GOOGL.O']  # RIC代码
df, err = ek.get_data(
    instruments=rics,
    fields=fields,
    params={'SDate': '2023-01-01', 'EDate': '2023-12-31'}
)

print(df.head())

注意：不同数据源的API返回格式差异很大。MSCI返回的是JSON嵌套结构，Refinitiv是扁平化的DataFrame。我建议统一转换成宽表格式，方便后续处理。

4.2 网页爬虫：当API不够用时

有些数据源没有公开API，或者API收费太贵。这时候就得靠爬虫了。比如CDP的公开数据、某些评级机构的摘要页面。

嗯，这里要注意——爬虫不是万能的。很多ESG数据页面是动态加载的，需要处理JavaScript渲染。我常用的工具组合是：

Requests + BeautifulSoup：适合静态页面，速度快
Selenium：适合动态页面，但慢
Scrapy：适合大规模爬取，但配置复杂

举个例子，爬取某个评级机构的公司列表页面：

import requests
from bs4 import BeautifulSoup
import time

def scrape_esg_scores(ticker_list):
    base_url = "https://example-esg-data.com/company/"
    results = []
    
    for ticker in ticker_list:
        try:
            url = base_url + ticker
            response = requests.get(url, headers={'User-Agent': 'Mozilla/5.0'})
            soup = BeautifulSoup(response.text, 'html.parser')
            
            # 假设评分在class为'esg-score'的div中
            score = soup.find('div', class_='esg-score').text
            results.append({'ticker': ticker, 'score': score})
            
            # 礼貌性延迟
            time.sleep(1)
            
        except Exception as e:
            print(f"Error fetching {ticker}: {e}")
    
    return pd.DataFrame(results)

避坑指南：我曾经因为爬虫频率太高，被对方服务器封了IP。后来学乖了，每次请求之间至少间隔1秒，并且使用代理池。另外，记得检查robots.txt，别踩法律红线。

4.3 数据对齐：最头疼的一步

数据对齐，说白了就是把不同来源的数据放到同一个时间轴上。为什么难？因为各家评级机构的发布时间不一样。

MSCI是每月更新，Sustainalytics是每季度，Refinitiv是每年。而且它们发布的不是同一时刻——MSCI可能在月初，Sustainalytics可能在月中。

我常用的对齐策略有三种：

最近可用值法：取评级发布后最近的一个交易日作为对齐点。这是最常用的方法。
月末对齐法：把所有评级统一对齐到月末。适合月度回测。
事件驱动对齐法：以评级变更事件为基准，对齐到变更日。适合事件驱动策略。

下面是我写的一个对齐函数，处理MSCI和Sustainalytics的数据：

def align_esg_ratings(msci_df, sust_df, method='nearest'):
    """
    对齐MSCI和Sustainalytics的ESG评级数据
    
    Parameters:
    - msci_df: MSCI数据，包含'date'和'score'列
    - sust_df: Sustainalytics数据，包含'date'和'score'列
    - method: 'nearest' 或 'month_end'
    """
    
    if method == 'nearest':
        # 使用最近可用值法
        # 对每个交易日，取最近一次发布的评级
        all_dates = pd.date_range(
            start=min(msci_df['date'].min(), sust_df['date'].min()),
            end=max(msci_df['date'].max(), sust_df['date'].max()),
            freq='D'
        )
        
        aligned = pd.DataFrame(index=all_dates)
        aligned['msci'] = msci_df.set_index('date')['score'].reindex(
            all_dates, method='ffill'
        )
        aligned['sust'] = sust_df.set_index('date')['score'].reindex(
            all_dates, method='ffill'
        )
        
    elif method == 'month_end':
        # 使用月末对齐法
        msci_monthly = msci_df.set_index('date').resample('M').last()
        sust_monthly = sust_df.set_index('date').resample('M').last()
        aligned = pd.concat([msci_monthly, sust_monthly], axis=1)
        aligned.columns = ['msci', 'sust']
    
    return aligned.dropna()

  关键点：对齐方法的选择直接影响分歧指标的计算。我建议至少尝试两种对齐方法，看看分歧指标是否稳健。如果两种方法算出来的分歧差异很大，说明数据本身就有问题。

4.4 缺失值处理：别让空值毁了你的策略

ESG数据缺失是个普遍问题。小公司、新兴市场、非强制披露行业，缺失率可能高达30%-40%。

我遇到过最夸张的情况——某家印度公司的ESG评分，MSCI有数据，Sustainalytics没有，Refinitiv只有环境评分。三家数据凑不齐一个完整的评分。

缺失值处理，我按优先级排序：

行业均值填充：用同行业、同规模公司的均值填充。这是最常用的方法。
时间序列插值：如果公司有历史数据，用线性插值或前向填充。
模型预测填充：用其他特征（如市值、行业、地区）预测缺失值。适合缺失率较高的情况。
直接删除：如果缺失率超过50%，我建议直接删除这个样本。别心疼。

下面是一个简单的行业均值填充示例：

def fill_missing_by_industry(df, industry_col='industry', score_col='esg_score'):
    """
    按行业均值填充缺失的ESG评分
    """
    # 计算每个行业的均值
    industry_means = df.groupby(industry_col)[score_col].mean()
    
    # 填充缺失值
    df[score_col + '_filled'] = df.apply(
        lambda row: industry_means[row[industry_col]] 
        if pd.isna(row[score_col]) 
        else row[score_col],
        axis=1
    )
    
    return df

警告：千万别用全局均值填充！不同行业的ESG评分差异巨大——科技公司环境评分普遍低，金融公司治理评分普遍高。用全局均值填充，相当于把苹果和橘子混在一起算平均价。

4.5 本章知识体系

我把整个数据获取与清洗的流程画了一张图，方便你理解：

这张图把整个流程串起来了。从数据源到最终的对齐数据集，每一步都有坑。我个人建议，第一次做的时候，每一步都手动检查几个样本，确保逻辑正确。

好了，数据获取与清洗就讲到这里。记住一句话：垃圾进，垃圾出。数据质量决定了你策略的天花板。

公众号:蓝海数据掘金营，微信deep3321

📚 课程目录

4. 分歧数据的获取与清洗：API调用、网页爬虫、数据对齐与缺失值处理

4.1 数据源的选择与API调用

4.2 网页爬虫：当API不够用时

4.3 数据对齐：最头疼的一步

4.4 缺失值处理：别让空值毁了你的策略

4.5 本章知识体系