4、行情数据源接入实战（下）：历史Tick数据获取、日线/分钟线数据批量下载、数据存储策略（Parquet/CSV）

好，咱们接着上回聊。上节课我们把实时行情的WebSocket接入了，数据在内存里跑得挺欢。但做量化交易，光有实时数据是不够的——回测需要历史数据，策略分析需要批量数据，模型训练更是离不开干净、规整的数据集。

今天这一讲，我就带大家把「历史数据」这块硬骨头啃下来。说白了，就是三个问题：怎么拿到历史Tick？怎么批量下载日线和分钟线？存成什么格式最划算？

核心观点：历史数据获取的核心不在于「能不能拿到」，而在于「拿得全、拿得快、存得好」。我见过太多团队在数据清洗和存储上浪费了80%的时间。

历史数据接入实战 历史Tick数据获取 逐笔成交、逐笔委托 日线/分钟线批量下载 多周期、多标的并行 数据存储策略 Parquet vs CSV 交易所原始接口 数据商API封装 多线程并行下载 断点续传机制 列式存储优势 压缩与分区策略 目标：构建高效、可靠的历史数据管道

一、历史Tick数据获取——逐笔还原的硬功夫

先说说Tick数据。很多人以为Tick就是「每秒一条的快照」，其实不对。真正的Tick数据是逐笔成交和逐笔委托，每一笔交易都记录在案。可转债市场因为流动性不如股票，Tick数据反而更有价值——你能看到每一笔资金的进出痕迹。

我个人习惯用聚宽和Tushare的接口来拿历史Tick。但要注意，这两个平台的Tick数据覆盖范围不一样。聚宽对可转债的支持更全，Tushare的字段更丰富。我一般两个都拉，交叉验证。

小技巧：拉Tick数据时，别一次性拉太多。我建议按「天」为单位，一天一天拉。为什么？因为Tick数据量太大了——一只可转债一天的Tick可能有几千条，全市场几百只转债，一天就是几十万条。一次性拉一个月，内存直接爆掉。

来看一段实际代码。这是我常用的Tick获取函数：

import tushare as ts
import pandas as pd
from datetime import datetime, timedelta

def fetch_tick_data(bond_code, date):
    """
    获取单只可转债某一天的Tick数据
    bond_code: '123456' 格式
    date: '2024-01-15' 格式
    """
    pro = ts.pro_api('your_token_here')
    
    # 注意：可转债代码需要加上后缀
    ts_code = f'{bond_code}.SZ' if bond_code.startswith('12') else f'{bond_code}.SH'
    
    df = pro.tick_data(
        ts_code=ts_code,
        trade_date=date.replace('-', ''),
        start_time='093000',
        end_time='150000'
    )
    
    if df.empty:
        print(f'警告：{bond_code} 在 {date} 无数据，可能是非交易日')
        return None
    
    # 我习惯加一列，标记数据来源
    df['source'] = 'tushare'
    df['bond_code'] = bond_code
    
    return df

# 使用示例
tick_df = fetch_tick_data('123456', '2024-01-15')
print(tick_df.head())

这里有个坑，我曾经踩过——可转债的代码后缀。深交所的可转债代码以12开头，后缀是.SZ；上交所的以11开头，后缀是.SH。如果你搞反了，接口返回的数据会是空的。我当时排查了半天，最后发现是代码写错了。

注意：不同数据商的Tick数据字段名可能不一样。比如「成交价」在Tushare里叫price，在聚宽里叫last_price。建议统一做一层映射，避免后续处理时混乱。

二、日线/分钟线数据批量下载——多线程才是王道

日线和分钟线相对Tick来说，数据量小很多。但如果你要下载全市场几百只可转债的3年数据，单线程跑的话，够你喝一壶的。我记得有一次，我用单线程下载全市场日线，跑了整整一个下午。

后来我改成了多线程并行下载，速度提升了10倍不止。核心思路很简单：把可转债列表分成几组，每组开一个线程去拉。

来看代码：

import concurrent.futures
import tushare as ts

def download_batch(bond_list, start_date, end_date, freq='D'):
    """
    批量下载多只可转债的K线数据
    freq: 'D' 日线, '30min' 30分钟线, '5min' 5分钟线
    """
    pro = ts.pro_api('your_token_here')
    results = []
    
    def fetch_single(bond):
        try:
            ts_code = f'{bond}.SZ' if bond.startswith('12') else f'{bond}.SH'
            df = pro.cb_bond_daily(
                ts_code=ts_code,
                start_date=start_date,
                end_date=end_date,
                freq=freq
            )
            if df is not None and not df.empty:
                df['bond_code'] = bond
                return df
        except Exception as e:
            print(f'下载 {bond} 失败: {e}')
            return None
    
    # 用线程池并行下载，我一般开8个线程
    with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:
        futures = {executor.submit(fetch_single, bond): bond for bond in bond_list}
        for future in concurrent.futures.as_completed(futures):
            result = future.result()
            if result is not None:
                results.append(result)
    
    if results:
        return pd.concat(results, ignore_index=True)
    return pd.DataFrame()

# 批量下载示例
bonds = ['123456', '113044', '127018']  # 实际使用时从列表读取
df_daily = download_batch(bonds, '20230101', '20240101', freq='D')
print(f'共下载 {len(df_daily)} 条日线数据')

你可能会问：为什么是8个线程？其实没有硬性规定。我测试过，8个线程对于大多数数据商接口来说，既不会触发限流，又能充分利用带宽。如果你用16个线程，反而可能因为接口限流导致大量请求失败。

避坑指南：我曾经遇到过一个问题——下载到一半，程序崩了，前面下载的数据全丢了。后来我加了一个「断点续传」机制：每下载完一只可转债，就把它的代码记录到一个已完成列表里。下次启动时，先检查哪些已经下载过，跳过它们。这样即使中途崩溃，也不会重复劳动。

三、数据存储策略——Parquet vs CSV，我选谁？

数据拿到了，存成什么格式？这是个大问题。很多人习惯用CSV，因为简单、通用。但我要说：对于量化交易场景，Parquet几乎是碾压CSV的存在。

为什么？我列个对比表你就明白了：

对比维度 CSV Parquet
存储空间 大（文本存储，无压缩） 小（列式压缩，通常只有CSV的20%-30%）
读取速度 慢（全量读取，无法列裁剪） 快（列式存储，只读需要的列）
数据类型 弱类型（所有字段都是字符串） 强类型（保留原始数据类型）
压缩率 需手动压缩（如gzip） 内置压缩（snappy、gzip等）
兼容性 几乎所有工具都支持 主流大数据工具支持（Spark、Pandas等）
分区支持 不支持 原生支持（按日期、代码分区）

我个人习惯的存储策略是这样的：

  • 原始Tick数据：存Parquet，按日期分区。因为Tick数据量大，查询时通常按日期筛选，分区后查询效率极高。
  • 日线/分钟线：也存Parquet，但按「可转债代码+周期」分区。比如 data/123456/D/ 和 data/123456/30min/。
  • 中间处理结果：偶尔用CSV，方便人工查看和调试。

来看一段存储代码：

import pandas as pd
import os

def save_to_parquet(df, base_path, bond_code, freq, date):
    """
    将数据存储为Parquet格式，按日期分区
    """
    # 构建分区路径：base_path/bond_code/freq/date/
    save_path = os.path.join(base_path, bond_code, freq, date)
    os.makedirs(save_path, exist_ok=True)
    
    file_path = os.path.join(save_path, f'{bond_code}_{freq}_{date}.parquet')
    
    # 使用snappy压缩，速度和压缩率的平衡点
    df.to_parquet(
        file_path,
        compression='snappy',
        index=False
    )
    
    print(f'数据已保存：{file_path}')
    return file_path

def load_from_parquet(base_path, bond_code, freq, start_date, end_date):
    """
    从Parquet分区中读取指定时间范围的数据
    """
    import glob
    
    # 构建日期范围
    dates = pd.date_range(start_date, end_date, freq='D')
    all_data = []
    
    for date in dates:
        date_str = date.strftime('%Y%m%d')
        pattern = os.path.join(base_path, bond_code, freq, date_str, '*.parquet')
        files = glob.glob(pattern)
        
        for file in files:
            df = pd.read_parquet(file)
            all_data.append(df)
    
    if all_data:
        return pd.concat(all_data, ignore_index=True)
    return pd.DataFrame()

# 使用示例
save_to_parquet(tick_df, './data', '123456', 'tick', '20240115')
loaded_df = load_from_parquet('./data', '123456', 'tick', '20240115', '20240115')
print(f'读取到 {len(loaded_df)} 条Tick数据')

关于压缩：Parquet支持多种压缩算法。我推荐用snappy，它压缩率不是最高的，但解压速度极快。对于量化交易来说，读取速度比存储空间更重要。你想想看，回测时每次都要加载几GB的数据，如果解压慢，那体验就很糟糕了。

四、实战中的几个关键决策

最后，我想分享几个我在项目中积累的经验，希望能帮你少走弯路。

第一，数据校验不能省。 我遇到过数据商接口返回的数据有缺失——某一天的分钟线少了下午的数据。后来我加了一个校验步骤：下载完成后，检查数据的时间戳是否连续，是否有跳空。如果发现异常，重新下载。

第二，存储路径要有规范。 我见过团队里不同人用不同的命名规则，最后数据目录乱成一锅粥。我建议统一用 数据源/标的代码/周期/日期/文件名 的结构。这样不管是手动查找还是程序读取，都很清晰。

第三，定期做数据备份。 硬盘会坏，云存储会丢数据。我每个月会把整个数据目录打包，上传到另一个存储服务。虽然麻烦，但心里踏实。

总结一下：历史数据接入的核心就三件事——拿得到、拿得快、存得好。Tick数据按天拉，日线分钟线多线程并行，存储用Parquet按日期分区。这套方案我在多个项目中验证过，稳定运行了两年多，没出过问题。

好了，这一讲的内容就到这里。数据源接入的实战部分基本讲完了，接下来我们会进入更核心的策略开发环节。嗯，到时候再聊。

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

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