4、数据采集与连接：连接Oracle/MySQL数据库、读取Excel/CSV文件、API接口数据获取

数据采集，说白了就是监管报送的「第一公里」。

我见过太多项目，模型再漂亮、逻辑再严谨，结果数据源连不上或者读出来是乱码——全白搭。所以这一章，咱们把最基础也最关键的几种数据接入方式，一次性讲透。

4.1 连接Oracle数据库

Oracle在金融行业里，可以说是「老大哥」了。我最早做监管报送时，对接的就是某国有大行的Oracle RAC集群。嗯，这里要注意——连接Oracle，驱动版本必须和数据库版本匹配，否则连上了也会报奇怪的错。

我个人习惯用 cx_Oracle 这个库，稳定且文档齐全。下面是一个标准的连接示例：

import cx_Oracle

# 连接字符串格式：用户名/密码@主机:端口/服务名
dsn = cx_Oracle.makedsn("192.168.1.100", 1521, service_name="orclpdb1")
conn = cx_Oracle.connect(user="report_user", password="P@ssw0rd", dsn=dsn)

cursor = conn.cursor()
cursor.execute("SELECT * FROM v$version")
for row in cursor:
    print(row)

cursor.close()
conn.close()

4.2 连接MySQL数据库

MySQL在中小金融机构里用得很多，轻量、免费、社区活跃。我个人推荐用 pymysql 或者 mysql-connector-python。

你想想看，监管报送的数据量通常不会特别大（几百万行以内），MySQL完全扛得住。但要注意——连接池！每次查询都新建连接，性能会很难看。

import pymysql

conn = pymysql.connect(
    host="192.168.1.101",
    port=3306,
    user="report_user",
    password="P@ssw0rd",
    database="report_db",
    charset="utf8mb4"
)

with conn.cursor() as cursor:
    sql = "SELECT report_date, count(*) FROM daily_report GROUP BY report_date"
    cursor.execute(sql)
    results = cursor.fetchall()
    for row in results:
        print(f"日期: {row[0]}, 记录数: {row[1]}")

conn.close()

4.3 读取Excel/CSV文件

说实话，Excel是监管报送里「最让人头疼」的数据源。为什么？因为Excel太灵活了——合并单元格、空行、日期格式乱、甚至有人把数字存成文本。

我建议用 pandas 来读取，它处理这些「脏数据」的能力最强。

4.3.1 读取CSV文件

import pandas as pd

df = pd.read_csv(
    "监管报表_2024Q1.csv",
    encoding="utf-8",
    dtype={"机构代码": str},  # 防止前导0被丢弃
    parse_dates=["报告日期"]
)

print(df.head())
print(df.info())

4.3.2 读取Excel文件

import pandas as pd

# 读取Excel，指定sheet和列范围
df = pd.read_excel(
    "监管数据模板.xlsx",
    sheet_name="Sheet1",
    header=0,
    usecols="A:F",
    dtype=str
)

# 处理空值
df.fillna("", inplace=True)

4.4 API接口数据获取

现在越来越多的监管系统提供API接口，比如人行、银保监会的报送平台。说白了，就是通过HTTP请求把数据推过去或者拉下来。

我个人习惯用 requests 库，简单直接。但要注意——监管API通常有严格的认证和限流机制。

import requests
import json

url = "https://api.cbirc.gov.cn/v1/report/submit"
headers = {
    "Authorization": "Bearer " + access_token,
    "Content-Type": "application/json"
}

data = {
    "report_id": "RPT20240301",
    "institution_code": "BANK001",
    "data": df.to_dict(orient="records")
}

response = requests.post(url, headers=headers, json=data, timeout=30)

if response.status_code == 200:
    result = response.json()
    print("报送成功，流水号:", result.get("serial_no"))
else:
    print("报送失败:", response.text)

4.5 知识体系总览

下面这张图，把本章的核心逻辑串起来了。你一看就明白：数据采集的三种方式，以及它们各自的关键技术点。

4.6 实战建议

最后，我分享几个实战中的小经验：

• 先验证，再加载：不管从哪种数据源读取，先做一次「数据质量检查」。比如检查必填字段是否为空、日期格式是否正确。我习惯写一个 validate_data() 函数，在正式入库前调用。
• 日志记录不能少：每次数据采集，都要记录「什么时间、从哪、读了多少条、成功/失败」。出了问题，日志就是你的救命稻草。
• 异常处理要细：不要只写一个 except Exception。数据库连接失败、文件不存在、API超时——每种异常单独处理，这样排查问题快得多。

---