第三章：数据读取实战——CSV/Excel大文件分块读取、API接口数据抓取、数据库SQL查询

做固收量化，第一步就是跟数据打交道。

说实话，我见过太多人把精力花在模型上，结果数据读进来就是错的。债券数据量大、字段多、来源杂，搞不好一个分号没对齐，整条收益率曲线就歪了。今天咱们就把数据读取这件事聊透。

3.1 大文件分块读取：别让内存爆炸

债券市场的数据，动辄就是几十年的日频行情。一个CSV文件几百兆甚至几个G，直接pd.read_csv()一梭子下去，内存直接飙红。我早期做国债期货回测时就吃过这个亏——笔记本直接卡死，强制关机后代码没保存，心态崩了。

其实解决方案很简单：分块读取。

核心思路：每次只读一部分数据，处理完再读下一块。就像吃自助餐，一次拿一盘，别想一次端走整个餐台。

3.1.1 chunksize参数

pandas的read_csv自带分块功能。你只需要指定chunksize，它就会返回一个迭代器。

import pandas as pd

# 每次读取5000行
chunk_iter = pd.read_csv('bond_daily.csv', chunksize=5000)

for i, chunk in enumerate(chunk_iter):
    # 对每一块做处理
    print(f'正在处理第{i+1}块，共{len(chunk)}行')
    # 比如：过滤出国债数据
    gov_bonds = chunk[chunk['bond_type'] == '国债']
    # 存到数据库或合并到结果集
    gov_bonds.to_csv(f'gov_bond_chunk_{i}.csv', index=False)

嗯，这里要注意：chunksize不是越大越好。我一般设成10000到50000行，具体看你的内存大小。如果你用的是8G内存的老机器，建议从5000开始试。

3.1.2 按需读取列和行

很多时候，你根本不需要所有列。比如我只想要日期、代码、收益率这三列，那就别把其他几十列都读进来。

# 只读需要的列
df = pd.read_csv('bond_data.csv', usecols=['date', 'code', 'ytm'])

# 或者只读前100万行
df = pd.read_csv('bond_data.csv', nrows=1000000)

我曾经帮一个团队优化数据加载流程，他们每次读一个2G的CSV，耗时3分钟。我让他们只读需要的5列，时间直接降到15秒。你想想看，这中间省了多少时间？

3.1.3 Excel大文件怎么办？

Excel文件比CSV更麻烦。它本质上是压缩包，没法直接分块。我的经验是：能转CSV就转CSV。如果实在要读Excel，用openpyxl配合read_only=True模式。

from openpyxl import load_workbook

wb = load_workbook('large_bond.xlsx', read_only=True)
ws = wb.active

for row in ws.iter_rows(min_row=2, values_only=True):
    # 逐行处理，不加载全部到内存
    date, code, ytm = row[0], row[1], row[2]
    # 你的处理逻辑

避坑指南：我曾经用pd.read_excel()读一个50MB的Excel，结果内存占用飙到4GB。后来换成openpyxl的只读模式，内存占用不到100MB。Excel这玩意儿，看着小，读起来吓人。

3.2 API接口数据抓取：Wind与Tushare实战

做固收量化，Wind是绕不开的。但Wind的API说实话，文档写得让人头大。Tushare相对友好，但免费版有频率限制。我两个都用，看场景切换。

3.2.1 Wind数据抓取

Wind的Python接口叫WindPy。用之前记得先打开Wind客户端，不然连不上。

from WindPy import w
w.start()

# 获取国债收益率曲线
# 代码：'yield_curve'，日期范围：2024-01-01到2024-12-31
data = w.wsd("yield_curve", "ytm_b", "2024-01-01", "2024-12-31", 
             "curveType=TREASURY;term=10Y")
df = pd.DataFrame(data.Data, index=data.Fields, columns=data.Times).T

我个人习惯把Wind的请求封装成一个函数，加上重试机制。因为Wind偶尔会抽风，返回空数据。

import time

def get_wind_data(code, fields, start, end, retries=3):
    for i in range(retries):
        try:
            data = w.wsd(code, fields, start, end)
            if data.ErrorCode != 0:
                raise Exception(f'Wind返回错误码：{data.ErrorCode}')
            return data
        except Exception as e:
            print(f'第{i+1}次尝试失败：{e}')
            time.sleep(2)
    return None

小技巧：Wind的w.wsd是获取日频数据，w.wsi是分钟级数据。做日内策略时用wsi，但注意频率别太高，Wind有调用次数限制。

3.2.2 Tushare数据抓取

Tushare的债券数据相对少一些，但胜在免费、稳定。注册后拿到token就能用。

import tushare as ts

# 设置token
ts.set_token('你的token')
pro = ts.pro_api()

# 获取债券基础信息
df = pro.bond_basic(bond_type='国债', fields='ts_code,bond_full_name,issue_date,carry_rate')
print(df.head())

Tushare有个坑：免费版每分钟只能调用200次。我写过一个带限速的装饰器，分享给你：

import time
from functools import wraps

def rate_limiter(max_calls=200, period=60):
    def decorator(func):
        calls = []
        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            # 清理过期记录
            calls[:] = [c for c in calls if c > now - period]
            if len(calls) >= max_calls:
                sleep_time = calls[0] + period - now
                print(f'达到调用上限，等待{sleep_time:.1f}秒')
                time.sleep(sleep_time)
            calls.append(now)
            return func(*args, **kwargs)
        return wrapper
    return decorator

@rate_limiter(max_calls=180, period=60)
def fetch_bond_data(code):
    return pro.bond_daily(ts_code=code)

3.3 数据库SQL查询：从菜鸟到老手

数据量大了之后，CSV和API都不够用。数据库才是归宿。我见过很多团队，几千万条债券行情数据还放在CSV里，查询一次要等半天。其实用SQL，几秒钟就搞定。

3.3.1 连接数据库

最常用的是MySQL和PostgreSQL。连接方式大同小异：

import pymysql
import pandas as pd

# 连接MySQL
conn = pymysql.connect(
    host='localhost',
    user='root',
    password='your_password',
    database='bond_db',
    charset='utf8mb4'
)

# 查询并直接返回DataFrame
sql = "SELECT * FROM bond_daily WHERE trade_date >= '2024-01-01' LIMIT 1000"
df = pd.read_sql(sql, conn)
conn.close()

我个人习惯用SQLAlchemy做连接池管理，这样不用每次查询都新建连接：

from sqlalchemy import create_engine

engine = create_engine('mysql+pymysql://root:password@localhost/bond_db?charset=utf8mb4')

# 直接用engine查询
df = pd.read_sql("SELECT * FROM bond_daily WHERE bond_type='国债'", engine)

3.3.2 分页查询大表

如果表里有几千万行，直接SELECT *会死人的。用分页：

page_size = 10000
offset = 0

while True:
    sql = f"SELECT * FROM bond_daily LIMIT {page_size} OFFSET {offset}"
    df = pd.read_sql(sql, engine)
    if len(df) == 0:
        break
    # 处理这一页数据
    print(f'处理第{offset//page_size + 1}页，共{len(df)}行')
    offset += page_size

避坑指南：我曾经用OFFSET分页查一个5000万行的表，越往后越慢。后来改成基于主键的WHERE id > last_id方式，速度提升了几十倍。记住：OFFSET是魔鬼，能不用就别用。

3.3.3 常用SQL技巧

做固收量化，这几个SQL场景你一定会遇到：

场景 SQL示例 说明
按日期聚合 SELECT trade_date, AVG(ytm) FROM bond_daily GROUP BY trade_date 计算每日平均收益率
窗口函数 SELECT *, ROW_NUMBER() OVER (PARTITION BY bond_code ORDER BY trade_date DESC) as rn FROM bond_daily 取每只债券最新一条数据
多表关联 SELECT a.*, b.bond_name FROM bond_daily a LEFT JOIN bond_info b ON a.bond_code = b.bond_code 行情表关联基本信息表

3.4 本章知识体系

说了这么多，咱们用一张图把数据读取的脉络理清楚：

固收量化数据读取知识体系 数据源 CSV / Excel 文件 API 接口 数据库 文件读取技术 • chunksize 分块迭代 • usecols 按需加载列 • openpyxl 只读模式 API 调用技术 • WindPy wsd/wsi • Tushare pro 接口 • 限速与重试机制 SQL 查询技术 • 连接池管理 • 分页查询优化 • 窗口函数应用 清洗后的 DataFrame

这张图把数据读取的三大来源和对应的核心技术点串起来了。你想想看，不管是文件、API还是数据库，最终目标都是生成一个干净的DataFrame，供后续的特征工程使用。

我的建议：刚开始做固收量化，别追求一次性把所有数据都读进来。先拿小数据量跑通流程，再逐步放大。我见过太多人一上来就想读全量数据，结果卡在数据读取这一步，模型连影子都没见到。

好了，数据读取这块就聊到这儿。记住：数据读得好，分析没烦恼。下一章咱们聊聊特征工程，那才是真正出活儿的地方。

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

← 上一章 📖 返回目录 下一章 →