第三章：数据源对接与统一接口封装

做量化交易，数据就是你的弹药库。弹药不行，枪法再好也白搭。

我见过太多人，策略逻辑写得漂漂亮亮，一上实盘就崩。为什么？数据源换了，接口不统一，字段名对不上。今天我们就来解决这个最基础也最要命的问题。

3.1 常见数据源：各有各的脾气

国内量化圈，主流数据源就那么几个。我用过的有Tushare、Wind、聚宽、米筐。每个我都踩过坑，今天一并告诉你。

Tushare

免费开源，社区活跃。适合个人研究。但有个问题——它的数据格式经常小改。我记得有一次升级，字段名从trade_date改成了cal_date，我整个回测框架全崩。嗯，从那以后我加了一层适配器。

Wind

机构标配，数据全，质量高。但贵，而且接口是C++风格的。我个人习惯用Python调用WindPy，但要注意——Wind的Tick数据有时会丢包，尤其是开盘那几分钟。

聚宽 & 米筐

这两家都是云端平台，数据封装得比较好。但如果你要本地跑策略，就得自己拉数据。聚宽的API返回的是DataFrame，米筐的返回的是dict。你想想看，同一个策略，换个数据源就得改一堆代码，多烦人。

数据源	优势	坑点
Tushare	免费、社区好	字段名不稳定
Wind	数据全、质量高	贵、Tick丢包
聚宽	封装好、易用	本地化麻烦
米筐	数据规范	格式不统一

3.2 统一数据接口设计

说白了，我们要做一个「数据中间人」。不管背后是哪个数据源，对外暴露的接口都一样。

Bar数据接口

Bar就是K线。日线、小时线、分钟线。我设计的接口长这样：

class BarData:
    def __init__(self, symbol, open, high, low, close, volume, timestamp):
        self.symbol = symbol
        self.open = open
        self.high = high
        self.low = low
        self.close = close
        self.volume = volume
        self.timestamp = timestamp

    def to_dict(self):
        return {
            'symbol': self.symbol,
            'open': self.open,
            'high': self.high,
            'low': self.low,
            'close': self.close,
            'volume': self.volume,
            'timestamp': self.timestamp
        }

为什么要统一成类？因为你可以加方法。比如to_dict()，方便序列化。比如validate()，做数据校验。

Tick数据接口

Tick数据更细，每一笔成交都记录。我遇到过最头疼的事——不同数据源的Tick时间戳格式不一样。有的用毫秒，有的用微秒，有的直接用字符串。

class TickData:
    def __init__(self, symbol, price, volume, timestamp, bid_price=None, ask_price=None):
        self.symbol = symbol
        self.price = price
        self.volume = volume
        self.timestamp = timestamp  # 统一为毫秒时间戳
        self.bid_price = bid_price
        self.ask_price = ask_price

    def is_valid(self):
        # 我曾经遇到price为0的Tick，直接导致策略下单
        return self.price > 0 and self.volume > 0

3.3 数据缓存与本地化存储

每次跑策略都去拉数据？太慢了。而且万一网络断了呢？

我的做法是：第一次拉取后，存到本地。下次优先读本地，本地没有再去拉。

存储格式选择

• CSV：简单，但慢。适合小数据量。
• Parquet：快，压缩率高。我推荐这个。
• SQLite：适合需要查询的场景。

我个人习惯用Parquet。为什么？因为回测时经常要读大量历史数据，Parquet的列式存储能快10倍以上。

import pandas as pd
import os

class DataCache:
    def __init__(self, cache_dir='./data_cache'):
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)

    def get(self, symbol, start_date, end_date):
        file_path = f"{self.cache_dir}/{symbol}_{start_date}_{end_date}.parquet"
        if os.path.exists(file_path):
            return pd.read_parquet(file_path)
        return None

    def save(self, symbol, start_date, end_date, data):
        file_path = f"{self.cache_dir}/{symbol}_{start_date}_{end_date}.parquet"
        data.to_parquet(file_path)

3.4 数据质量校验与清洗

数据质量是量化交易的命门。我见过最离谱的事——某数据源某天的收盘价全是0，策略直接满仓买入。幸好是仿真交易，不然亏大了。

校验规则

1. 空值检查：有没有NaN？有没有0？
2. 单调性检查：时间戳是不是递增的？
3. 范围检查：价格是不是在合理范围内？比如A股不可能有10000元的收盘价。
4. 跳空检查：相邻两根K线的价格差是不是太大？

def validate_bar_data(df):
    errors = []
    # 空值检查
    if df.isnull().any().any():
        errors.append("存在空值")
    # 时间戳单调性
    if not df['timestamp'].is_monotonic_increasing:
        errors.append("时间戳不单调")
    # 价格范围
    if (df['close'] < 0).any():
        errors.append("收盘价小于0")
    # 跳空检查（超过10%视为异常）
    pct_change = df['close'].pct_change()
    if (pct_change.abs() > 0.1).any():
        errors.append("存在超过10%的跳空")
    return errors

清洗策略

发现问题后怎么办？我的原则是：

• 能修复的修复（比如时间戳排序）
• 不能修复的丢弃（比如价格异常）
• 丢弃比例超过5%的，整个数据源标记为不可用

3.5 整体架构图

下面这张图，是我在实际项目中总结出来的数据层架构。你看一眼就明白了。

你看，从数据源到策略层，中间隔了三层。每一层解决一个问题。这样设计的好处是——换数据源时，只需要改适配器层，上层代码完全不用动。

我曾经帮一个团队重构数据层，他们原来每个策略都直接调Tushare。后来要切Wind，改了整整两周。用了我这套架构后，切数据源只需要一天。

嗯，今天就讲到这里。数据源对接这件事，看似简单，但细节决定成败。下一章我们聊聊策略回测框架的搭建，到时候你会感谢今天做的这些基础工作。

---

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