第2章数据采集与存储

2. 数据采集与存储：API接口调用、CSV/Excel批量导入、数据库建表规范

好，咱们进入第二章。数据采集与存储，说白了就是解决「数据从哪来、怎么存」的问题。ESG评分数据有个特点——来源杂、格式乱、更新频率不统一。有的数据商给你API，有的只给Excel附件，还有的让你去爬网页。我刚开始做ESG量化的时候，光数据清洗就占了60%的时间。后来我学乖了，先把采集和存储的规范定死，后面能省一大半力气。

2.1 API接口调用：别光想着调通，要考虑容错

调用API拿ESG数据，听起来简单，但坑不少。我见过太多人写个requests.get()就跑，结果半夜数据断了没人知道。

核心原则：幂等、重试、限速、日志。

举个例子，调用MSCI的ESG评级API。我个人习惯这样写：

import requests
import time
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def fetch_msci_esg(company_id, api_key):
    url = f"https://api.msci.com/esg/v1/ratings/{company_id}"
    headers = {"Authorization": f"Bearer {api_key}"}
    resp = requests.get(url, headers=headers, timeout=10)
    resp.raise_for_status()
    return resp.json()

# 调用时注意限速
for cid in company_list:
    data = fetch_msci_esg(cid, API_KEY)
    # 存数据...
    time.sleep(1.5)  # 别把人家API打崩了

为什么要加重试？因为ESG数据商的API稳定性参差不齐。我曾经遇到过一个数据商，每天下午3点准时超时，加了重试机制后，问题自动解决了。

小技巧：API返回的JSON里，字段名往往带大写或下划线。我建议你拿到后立刻统一转成小写蛇形命名，后面处理起来省心。

2.2 CSV/Excel批量导入：别双击打开，用代码处理

很多ESG数据还是以Excel附件形式提供的。你想想看，一个季度一次，一次几十个文件，手动打开再另存为？那太原始了。

我的做法是写一个批量导入脚本。核心就三点：

文件发现：自动扫描指定目录下的所有.xlsx/.csv文件
格式校验：检查列名、数据类型是否匹配预期
增量导入：只导入新数据，避免重复

import pandas as pd
import os
from pathlib import Path

def batch_import_esg_data(data_dir):
    all_dfs = []
    for f in Path(data_dir).glob("*.xlsx"):
        print(f"正在处理: {f.name}")
        df = pd.read_excel(f, sheet_name="ESG数据")
        # 统一列名
        df.columns = [col.lower().replace(" ", "_") for col in df.columns]
        # 校验必填字段
        required_cols = ["company_id", "esg_score", "e_score", "s_score", "g_score"]
        if not all(col in df.columns for col in required_cols):
            print(f"警告: {f.name} 缺少必要列，跳过")
            continue
        all_dfs.append(df)
    return pd.concat(all_dfs, ignore_index=True)

注意：Excel文件里经常有合并单元格、空行、或者隐藏行。导入前最好先做一次「展平」处理。我吃过这个亏——有一次导入后数据对不上，查了半天发现是Excel里有个隐藏行，里面是旧数据。

2.3 数据库建表规范：命名、类型、索引

数据拿到手了，存哪？我强烈建议用关系型数据库（PostgreSQL或MySQL）来存ESG原始数据。别一上来就扔Hadoop，没那么大量。

建表规范，我总结了三条铁律：

表名用复数名词：比如esg_ratings、company_master
字段名用小写蛇形：比如esg_score、report_date
时间字段统一用UTC：别问为什么，问就是时区坑太多

来看一个实际建表语句：

CREATE TABLE esg_ratings (
    id BIGSERIAL PRIMARY KEY,
    company_id VARCHAR(20) NOT NULL,
    rating_date DATE NOT NULL,
    esg_score NUMERIC(5,2),
    e_score NUMERIC(5,2),
    s_score NUMERIC(5,2),
    g_score NUMERIC(5,2),
    data_source VARCHAR(50),
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW(),
    UNIQUE(company_id, rating_date, data_source)
);

-- 索引：按公司和日期查询
CREATE INDEX idx_esg_company_date ON esg_ratings(company_id, rating_date);

这里有个细节——我加了UNIQUE约束。为什么？因为同一家公司同一天的ESG评分，理论上只应该有一条记录。如果数据商重复推送，这个约束能帮你挡住脏数据。

实战经验：ESG数据经常有「修正值」。比如某公司2023年的评分，2024年又修正了。我的做法是加一个version字段，每次修正就插入新版本，保留历史轨迹。这样回溯分析时，你能知道当时用的是哪个版本的数据。

2.4 知识体系总览

下面这张图，是我自己梳理的「ESG数据采集与存储」核心逻辑。你看一眼，基本就明白整个流程了。

这张图里，我特意把「建表规范」单独拉出来放在左侧。因为在实际项目中，很多人只关注怎么把数据拿进来，却忽略了存储的规范性。结果就是——数据越存越乱，最后谁也看不懂。

2.5 避坑指南

最后分享几个我踩过的坑：

Excel日期格式：Excel里的日期有时候是数字（序列值），有时候是字符串。导入时一定要用pd.to_datetime()统一转换。
API限流：有些ESG数据商的API限流策略很隐蔽，不是返回429，而是直接返回空数据。我建议你每次调用后都检查一下返回的数据量是否合理。
数据库连接池：批量导入时别每次新建连接，用连接池。不然数据库会被你搞崩溃。

我的习惯：每次导入完成后，跑一个简单的数据质量检查脚本——比如检查是否有空值、是否有重复记录、日期范围是否合理。发现问题立刻报警，别等到分析时才发现数据有问题。

好了，数据采集和存储这块就聊到这。记住一个原则：源头规范，后面省心。下一节咱们开始讲数据清洗，那才是真正考验耐心的地方。

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

📚 课程目录

2. 数据采集与存储：API接口调用、CSV/Excel批量导入、数据库建表规范

2.1 API接口调用：别光想着调通，要考虑容错

2.2 CSV/Excel批量导入：别双击打开，用代码处理

2.3 数据库建表规范：命名、类型、索引

2.4 知识体系总览

2.5 避坑指南