2. ESG数据获取与预处理:常见ESG数据源介绍
做ESG量化选股,第一步就是搞数据。这活儿看着简单,实际上坑特别多。我刚开始做ESG策略的时候,光找数据就折腾了两周——有的数据源要付费,有的接口文档写得像天书,还有的数据格式乱得让人想摔键盘。
今天咱们就把这块彻底捋清楚。从数据源到清洗,从对齐到频率处理,一条龙讲透。
2.1 常见ESG数据源介绍
先说说市面上主流的ESG数据供应商。我按自己的使用经验排个序:
| 数据源 | 覆盖范围 | 更新频率 | 获取难度 | 我的评价 |
|---|---|---|---|---|
| MSCI ESG Ratings | 全球8500+公司 | 月度/季度 | 需付费,API较规范 | 行业标杆,但贵 |
| Sustainalytics | 全球12000+公司 | 月度 | 需付费,有API | 风险维度做得好 |
| CDP | 全球20000+公司 | 年度 | 部分免费,需申请 | 气候数据最权威 |
| Refinitiv | 全球10000+公司 | 月度 | 需付费,Eikon平台 | 数据维度最全 |
| Bloomberg ESG | 全球11000+公司 | 月度 | 需付费,终端访问 | 和财务数据整合好 |
2.2 数据获取方式
数据源选好了,怎么拿到手?三种主流方式:API、爬虫、数据库直连。
2.2.1 API方式(推荐)
API是最正规的获取方式。MSCI和Sustainalytics都提供RESTful API。我个人习惯用Python的requests库来调。
import requests
import pandas as pd
import json
# MSCI ESG API示例(伪代码,实际需要API Key)
def get_msci_esg_data(ticker, api_key):
url = f"https://api.msci.com/esg/v1/ratings/{ticker}"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
# 提取关键字段
result = {
"ticker": ticker,
"esg_rating": data.get("rating"),
"environment_score": data.get("environment_score"),
"social_score": data.get("social_score"),
"governance_score": data.get("governance_score"),
"date": data.get("as_of_date")
}
return result
else:
print(f"请求失败: {response.status_code}")
return None
# 批量获取
tickers = ["AAPL", "MSFT", "GOOGL"]
api_key = "your_api_key_here"
results = []
for ticker in tickers:
data = get_msci_esg_data(ticker, api_key)
if data:
results.append(data)
df = pd.DataFrame(results)
print(df.head())
2.2.2 爬虫方式(备选)
有些数据源没有API,或者你想抓取公开披露的信息。这时候就得用爬虫。但说实话,我不太建议大规模爬ESG数据——很多网站有反爬机制,而且法律风险你得自己掂量。
import requests
from bs4 import BeautifulSoup
import time
# CDP公开数据爬取示例
def scrape_cdp_data(company_name):
url = f"https://www.cdp.net/en/responses?query={company_name}"
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
response = requests.get(url, headers=headers)
soup = BeautifulSoup(response.content, "html.parser")
# 解析页面内容(具体选择器根据实际页面调整)
# 这里只是示意
data_points = soup.find_all("div", class_="response-item")
results = []
for item in data_points:
# 提取数据
pass
time.sleep(2) # 礼貌爬取
return results
2.2.3 数据库直连
如果你在机构里做,大概率有数据库权限。Bloomberg终端、Refinitiv Eikon都支持SQL查询。这种方式最稳定,但需要IT权限。
# 假设连接Refinitiv数据库
import pyodbc
conn = pyodbc.connect(
"DRIVER={ODBC Driver 17 for SQL Server};"
"SERVER=your_server;"
"DATABASE=ESG_Data;"
"UID=your_username;"
"PWD=your_password"
)
query = """
SELECT
ticker,
esg_score,
environment_pillar,
social_pillar,
governance_pillar,
as_of_date
FROM esg_ratings
WHERE as_of_date >= '2023-01-01'
AND ticker IN ('AAPL', 'MSFT', 'GOOGL')
"""
df = pd.read_sql(query, conn)
conn.close()
2.3 数据清洗与标准化
数据拿到手了,但你会发现——不同数据源的评分标准完全不一样。MSCI用AAA到CCC,Sustainalytics用0-100分,CDP用A到F。这就得做标准化。
我一般分三步走:
- 缺失值处理——ESG数据缺失很常见,尤其是中小盘股
- 评分映射——把字母评级转成数值
- 归一化——统一到0-1区间
import pandas as pd
import numpy as np
from sklearn.preprocessing import MinMaxScaler
# 示例数据
data = {
"ticker": ["AAPL", "MSFT", "GOOGL", "AMZN", "META"],
"msci_rating": ["AA", "AAA", "A", "BBB", "BB"],
"sustainalytics_score": [15, 10, 20, 25, 35],
"cdp_grade": ["A", "A-", "B", "B-", "C"]
}
df = pd.DataFrame(data)
# 第一步:字母评级转数值
rating_map = {
"AAA": 10, "AA": 9, "A": 8,
"BBB": 7, "BB": 6, "B": 5,
"CCC": 4, "CC": 3, "C": 2, "D": 1
}
grade_map = {
"A": 10, "A-": 9, "B": 8, "B-": 7,
"C": 6, "C-": 5, "D": 4, "D-": 3, "F": 1
}
df["msci_numeric"] = df["msci_rating"].map(rating_map)
df["cdp_numeric"] = df["cdp_grade"].map(grade_map)
# 第二步:处理缺失值(用中位数填充)
df["msci_numeric"].fillna(df["msci_numeric"].median(), inplace=True)
df["cdp_numeric"].fillna(df["cdp_numeric"].median(), inplace=True)
# 第三步:归一化到0-1
scaler = MinMaxScaler()
df[["msci_norm", "sustain_norm", "cdp_norm"]] = scaler.fit_transform(
df[["msci_numeric", "sustainalytics_score", "cdp_numeric"]]
)
# 第四步:合成综合ESG得分(等权重)
df["esg_composite"] = (
df["msci_norm"] + df["sustain_norm"] + df["cdp_norm"]
) / 3
print(df[["ticker", "esg_composite"]])
2.4 数据对齐与频率处理
这是最头疼的部分。MSCI月度更新,CDP年度更新,财务数据日频。怎么对齐?
我的做法是:以最低频率为准,向上填充。
import pandas as pd
# 模拟数据
dates = pd.date_range("2023-01-01", "2023-12-31", freq="D")
tickers = ["AAPL", "MSFT"]
# 日频价格数据
price_data = pd.DataFrame(
np.random.randn(len(dates), len(tickers)),
index=dates,
columns=tickers
)
# 月频ESG数据(假设每月1号更新)
esg_dates = pd.date_range("2023-01-01", "2023-12-01", freq="MS")
esg_data = pd.DataFrame(
np.random.uniform(0, 1, (len(esg_dates), len(tickers))),
index=esg_dates,
columns=tickers
)
# 对齐:ESG数据向前填充到日频
esg_aligned = esg_data.reindex(price_data.index, method="ffill")
# 合并
combined = pd.concat([price_data, esg_aligned.add_suffix("_esg")], axis=1)
print(combined.head(10))
还有一种情况:不同数据源的报告日期不同。MSCI可能3月出报告,Sustainalytics可能4月。这时候我建议统一对齐到季度末,避免时间错配。
# 对齐到季度末
def align_to_quarter(df, date_col):
df["quarter"] = pd.to_datetime(df[date_col]).dt.to_period("Q")
df["quarter_end"] = df["quarter"].dt.end_time
return df.groupby(["ticker", "quarter_end"]).last().reset_index()
# 应用
msci_aligned = align_to_quarter(msci_data, "report_date")
sustain_aligned = align_to_quarter(sustain_data, "report_date")
# 合并
merged = pd.merge(
msci_aligned, sustain_aligned,
on=["ticker", "quarter_end"],
how="outer"
)
知识体系总览
下面这张图把本章的核心逻辑串起来了。从数据源到最终可用的特征矩阵,每一步都有坑,每一步也都有解法。
嗯,数据这块就讲这么多。说白了,ESG数据预处理没有银弹,每个环节都得根据实际情况调整。但只要你把数据源、获取方式、清洗标准化、对齐频率这四个环节都走通了,后面的策略构建就会顺畅很多。