📚 课程目录第4章数据采集与清洗

4. 数据采集与清洗

供应商财务数据的获取与清洗是构建风险评分模型的基础。本章聚焦于如何通过 Python 编程，从 Bloomberg、Wind 等公开金融数据库高效采集供应商的财务指标，并对原始数据进行标准化清洗，确保数据质量满足后续分析与建模要求。

4.1 数据采集架构概览

在实际项目中，数据采集通常采用“API 接口 + 本地缓存”的混合模式。以下架构图展示了从数据库到分析环境的完整数据流。

4.2 环境准备与依赖安装

在开始采集之前，需要安装必要的 Python 库。以下为推荐的环境配置清单：

库名称	用途	安装命令
pandas	数据清洗与结构化存储	`pip install pandas`
blpapi	Bloomberg API 接口	`pip install blpapi`
WindPy	Wind 金融终端接口	`pip install WindPy`
requests	HTTP 请求（备用数据源）	`pip install requests`
python-dotenv	管理与配置	`pip install python-dotenv`

💡 提示： 如果使用 Bloomberg Terminal，请确保已安装 Bloomberg 官方提供的 blpapi SDK，并配置好环境变量 BLPAPI_ROOT。Wind 用户需先启动 Wind 客户端，再调用 w.start() 建立连接。

4.3 从 Bloomberg 采集财务数据

Bloomberg 提供丰富的财务字段，常用指标包括总资产、总负债、营业收入、净利润、经营性现金流等。以下示例演示如何通过 blpapi 批量获取指定供应商的年度财务数据。

import blpapi
import pandas as pd
from datetime import datetime

# 配置 Bloomberg 会话
session_options = blpapi.SessionOptions()
session_options.setServerHost("localhost")
session_options.setServerPort(8194)
session = blpapi.Session(session_options)
session.start()

# 打开服务
session.openService("//blp/refdata")
service = session.getService("//blp/refdata")

# 定义请求
request = service.createRequest("HistoricalDataRequest")
request.getElement("securities").appendValue("AAPL US Equity")  # 供应商代码
request.getElement("fields").appendValue("BS_TOTAL_ASSETS")     # 总资产
request.getElement("fields").appendValue("BS_TOTAL_LIABILITIES")# 总负债
request.getElement("fields").appendValue("IS_SALES_REV_TURN")   # 营业收入
request.getElement("fields").appendValue("CF_CASH_FROM_OPS")    # 经营性现金流

# 设置时间范围
request.getElement("startDate").setValue("2020-01-01")
request.getElement("endDate").setValue("2023-12-31")
request.getElement("periodicitySelection").setValue("YEARLY")

# 发送请求并解析
session.sendRequest(request)
data_records = []
while True:
    event = session.nextEvent()
    for msg in event:
        if msg.messageType() == "HistoricalDataResponse":
            security_data = msg.getElement("securityData")
            field_data = security_data.getElement("fieldData")
            for i in range(field_data.numValues()):
                point = field_data.getValueAsElement(i)
                record = {
                    "date": point.getElement("date").getValueAsString(),
                    "total_assets": point.getElement("BS_TOTAL_ASSETS").getValue(),
                    "total_liabilities": point.getElement("BS_TOTAL_LIABILITIES").getValue(),
                    "revenue": point.getElement("IS_SALES_REV_TURN").getValue(),
                    "operating_cf": point.getElement("CF_CASH_FROM_OPS").getValue()
                }
                data_records.append(record)
    if event.eventType() == blpapi.Event.RESPONSE:
        break

df_bbg = pd.DataFrame(data_records)
print(df_bbg.head())

4.4 从 Wind 采集财务数据

Wind 终端在国内金融市场使用广泛，其 Python 接口 WindPy 提供了类似 Bloomberg 的字段查询能力。以下代码展示如何获取供应商的资产负债与利润表数据。

from WindPy import w
import pandas as pd

# 启动 Wind 连接
w.start()

# 定义供应商代码与指标
codes = ["000001.SZ", "600036.SH"]  # 示例供应商股票代码
indicators = ["total_assets", "total_liabilities", "oper_revenue", "net_profit_is"]

# 获取年度数据（2020-2023）
df_wind = w.wsd(
    codes,
    indicators,
    "2020-01-01",
    "2023-12-31",
    "Period=Y;Fill=Previous"
).Data

# 转换为 DataFrame
df_wind = pd.DataFrame(df_wind).T
df_wind.columns = ["total_assets", "total_liabilities", "revenue", "net_profit"]
df_wind.index = codes
print(df_wind)

⚠️ 注意： Wind 的字段名称与 Bloomberg 不完全一致。例如，Wind 中“营业收入”对应 oper_revenue，而 Bloomberg 中为 IS_SALES_REV_TURN。建议在采集前建立字段映射表，避免数据错位。

4.5 数据清洗与标准化流程

原始财务数据通常存在缺失值、单位不统一、异常值等问题。以下清洗流程确保数据可用于风险评分模型。

4.6 清洗代码实战

以下代码实现了上述清洗流程的核心逻辑，包括缺失值处理、异常值检测与单位转换。

import pandas as pd
import numpy as np

def clean_financial_data(df: pd.DataFrame) -> pd.DataFrame:
    """
    清洗供应商财务数据
    - 处理缺失值
    - 过滤异常值（基于 Z-score）
    - 统一单位为百万元
    """
    df_clean = df.copy()

    # 1. 缺失值处理：删除缺失率超过 30% 的列
    missing_ratio = df_clean.isnull().mean()
    cols_to_drop = missing_ratio[missing_ratio > 0.3].index
    df_clean.drop(columns=cols_to_drop, inplace=True)

    # 2. 数值列填充：使用行业中位数（假设已有行业分组列 'industry'）
    numeric_cols = df_clean.select_dtypes(include=[np.number]).columns
    for col in numeric_cols:
        if df_clean[col].isnull().sum() > 0:
            median_val = df_clean.groupby('industry')[col].transform('median')
            df_clean[col].fillna(median_val, inplace=True)

    # 3. 异常值过滤：Z-score 值大于 3 视为异常，用中位数替换
    for col in numeric_cols:
        z_scores = np.abs((df_clean[col] - df_clean[col].mean()) / df_clean[col].std())
        outlier_idx = z_scores > 3
        if outlier_idx.any():
            median_val = df_clean[col].median()
            df_clean.loc[outlier_idx, col] = median_val

    # 4. 单位统一：假设原始数据单位为元，转换为百万元
    for col in numeric_cols:
        if col in ['total_assets', 'total_liabilities', 'revenue', 'net_profit']:
            df_clean[col] = df_clean[col] / 1_000_000
            df_clean.rename(columns={col: f"{col}_mil"}, inplace=True)

    return df_clean

# 示例调用
df_raw = pd.DataFrame({
    'supplier': ['A', 'B', 'C'],
    'industry': ['制造业', '制造业', '服务业'],
    'total_assets': [5000000, np.nan, 3000000],
    'revenue': [2000000, 1500000, 2500000],
    'net_profit': [200000, -50000, 300000]
})
df_cleaned = clean_financial_data(df_raw)
print(df_cleaned)

4.7 字段映射与标准化

不同数据源的字段命名差异较大，建议建立统一的字段映射表，便于后续合并与分析。

标准字段名	Bloomberg 字段	Wind 字段	说明
total_assets	BS_TOTAL_ASSETS	total_assets	总资产
total_liabilities	BS_TOTAL_LIABILITIES	total_liabilities	总负债
revenue	IS_SALES_REV_TURN	oper_revenue	营业收入
net_profit	IS_NET_INCOME	net_profit_is	净利润
operating_cf	CF_CASH_FROM_OPS	cash_flows_oper_act	经营性现金流

  📌 最佳实践： 将字段映射关系保存在 YAML 或 JSON 配置文件中，避免硬编码。采集时通过映射表自动转换字段名，提升代码的可维护性与复用性。

4.8 数据质量检查清单

在将清洗后的数据交付给风险评分模块之前，建议逐项检查以下质量指标：

完整性： 每个供应商的财务数据是否覆盖了至少 3 个连续财年？
一致性： 同一供应商在不同数据源中的同一指标是否一致（误差 < 5%）？
时效性： 数据是否包含最近一个完整财年的信息？
合理性： 关键比率（如资产负债率、流动比率）是否在行业正常范围内？
可追溯性： 每条数据是否保留了原始来源标记（Bloomberg / Wind / 手工录入）？

通过以上步骤，你可以构建一个稳健的供应商财务数据采集与清洗管道，为后续的交付表现分析与风险评分提供高质量的数据基础。