3、数据层封装：统一数据接口设计，支持多数据源无缝切换

做量化回测，最头疼的是什么？

我个人觉得，不是策略逻辑写不出来，而是数据源换来换去，代码改得想骂人。今天用CSV，明天切数据库，后天又要接API——每次切换，业务代码里全是if-else，维护成本直线飙升。

嗯，这一章我们就来解决这个问题。

3.1 为什么需要统一数据接口？

先说说我踩过的坑。几年前做一套回测系统，一开始只支持CSV。后来团队想接入MySQL，我天真地以为加个读取函数就行。结果呢？

  • 策略代码里到处是 pd.read_csv() 和 conn.execute()
  • 换数据源要改十几个文件
  • 测试数据和生产数据混在一起，出了bug定位半天

说白了，问题出在「数据获取」和「业务逻辑」耦合太紧。你想想看，策略只需要知道「给我某只股票某段时间的行情」，它才不关心数据是从文件读的还是从API拉的。

核心原则：面向接口编程，而非面向实现编程。数据层对上层暴露统一的抽象，底层实现可以随意替换。

3.2 统一接口设计：从抽象类开始

我的习惯是，先定义一个抽象基类。它不干具体活，只规定「你要提供什么能力」。

from abc import ABC, abstractmethod
import pandas as pd
from datetime import datetime

class DataSource(ABC):
    """数据源抽象基类"""
    
    @abstractmethod
    def get_bars(self, 
                 symbol: str, 
                 start: datetime, 
                 end: datetime, 
                 freq: str = '1d') -> pd.DataFrame:
        """
        获取K线数据
        :param symbol: 股票代码
        :param start: 开始时间
        :param end: 结束时间
        :param freq: 频率，如 '1d', '1h', '5m'
        :return: DataFrame，包含 open, high, low, close, volume
        """
        pass
    
    @abstractmethod
    def get_tick(self, 
                 symbol: str, 
                 date: datetime) -> pd.DataFrame:
        """获取Tick级数据"""
        pass
    
    @abstractmethod
    def get_instruments(self) -> pd.DataFrame:
        """获取所有可交易标的列表"""
        pass

看到没？三个方法，覆盖了回测最常用的数据需求。具体怎么实现，那是子类的事。

3.3 实现一：CSV数据源

CSV是最简单的，也是我早期最常用的。但要注意，不是随便读个文件就完事。

class CsvDataSource(DataSource):
    def __init__(self, data_dir: str):
        self.data_dir = data_dir
        # 缓存已加载的数据，避免重复IO
        self._cache = {}
    
    def get_bars(self, symbol, start, end, freq='1d'):
        cache_key = f"{symbol}_{freq}"
        if cache_key not in self._cache:
            file_path = f"{self.data_dir}/{symbol}_{freq}.csv"
            df = pd.read_csv(file_path, parse_dates=['datetime'])
            df.set_index('datetime', inplace=True)
            self._cache[cache_key] = df
        
        df = self._cache[cache_key]
        return df.loc[start:end].copy()

避坑指南：我曾经因为没做数据缓存，回测时反复读取同一个CSV文件，结果IO成了瓶颈。后来加了内存缓存，速度提升了10倍。记住：数据层一定要考虑性能。

3.4 实现二：数据库数据源

当数据量大了，CSV就不够用了。数据库查询灵活，还能做增量更新。

class DatabaseDataSource(DataSource):
    def __init__(self, connection_string: str):
        from sqlalchemy import create_engine
        self.engine = create_engine(connection_string)
    
    def get_bars(self, symbol, start, end, freq='1d'):
        query = """
        SELECT datetime, open, high, low, close, volume
        FROM bars
        WHERE symbol = :symbol
          AND freq = :freq
          AND datetime BETWEEN :start AND :end
        ORDER BY datetime
        """
        df = pd.read_sql(query, 
                         self.engine, 
                         params={
                             'symbol': symbol,
                             'freq': freq,
                             'start': start,
                             'end': end
                         },
                         parse_dates=['datetime'])
        df.set_index('datetime', inplace=True)
        return df

这里我用的是SQLAlchemy，好处是底层数据库随便换——MySQL、PostgreSQL、SQLite都行。你只需要改个连接字符串。

3.5 实现三：API数据源

实盘交易时，数据通常来自券商或数据服务商的API。网络请求不稳定，所以要做好容错。

class ApiDataSource(DataSource):
    def __init__(self, api_key: str, base_url: str):
        self.api_key = api_key
        self.base_url = base_url
        self.session = requests.Session()
        # 重试机制
        from requests.adapters import HTTPAdapter
        from urllib3.util.retry import Retry
        
        retry = Retry(total=3, backoff_factor=1)
        adapter = HTTPAdapter(max_retries=retry)
        self.session.mount('http://', adapter)
        self.session.mount('https://', adapter)
    
    def get_bars(self, symbol, start, end, freq='1d'):
        params = {
            'symbol': symbol,
            'start': start.isoformat(),
            'end': end.isoformat(),
            'freq': freq,
            'api_key': self.api_key
        }
        resp = self.session.get(
            f"{self.base_url}/bars", 
            params=params,
            timeout=10
        )
        resp.raise_for_status()
        data = resp.json()
        df = pd.DataFrame(data)
        df.set_index('datetime', inplace=True)
        return df

注意：API调用一定要加超时和重试。我曾经因为网络抖动，回测跑到一半卡住了，浪费了3个小时。另外，注意API的调用频率限制，别被封了。

3.6 数据源工厂：一键切换

有了统一的接口，怎么让上层代码无感知地切换？工厂模式是个好选择。

class DataSourceFactory:
    @staticmethod
    def create(source_type: str, **kwargs) -> DataSource:
        if source_type == 'csv':
            return CsvDataSource(kwargs['data_dir'])
        elif source_type == 'database':
            return DatabaseDataSource(kwargs['connection_string'])
        elif source_type == 'api':
            return ApiDataSource(kwargs['api_key'], kwargs['base_url'])
        else:
            raise ValueError(f"Unsupported data source: {source_type}")

# 使用示例
config = {
    'source_type': 'csv',
    'data_dir': './data/historical'
}
# 想切数据库？只需改配置
# config = {
#     'source_type': 'database',
#     'connection_string': 'mysql://user:pass@localhost/quant'
# }

data_source = DataSourceFactory.create(**config)
# 业务代码完全不用改
df = data_source.get_bars('AAPL', start, end)

你看，业务代码只依赖 DataSource 这个抽象，具体是CSV还是数据库，配置说了算。这就是「依赖倒置」原则的实战应用。

3.7 整体架构图

下面这张图，把整个数据层的设计思路串起来了：

数据层统一接口架构 策略引擎 / 回测框架 DataSource（抽象基类） get_bars() | get_tick() | get_instruments() CsvDataSource 读取本地CSV文件 支持内存缓存 DatabaseDataSource SQLAlchemy查询 支持MySQL/PostgreSQL ApiDataSource HTTP请求获取数据 自动重试+超时 DataSourceFactory（工厂类） 通过工厂模式，上层业务无需关心底层数据源的具体实现

3.8 数据一致性：别忘了时间对齐

不同数据源的时间格式可能不一样。CSV里可能是字符串，数据库里可能是datetime，API返回的可能是时间戳。统一接口里，我建议强制转换成统一的格式。

class DataSource(ABC):
    def _normalize_datetime(self, df: pd.DataFrame) -> pd.DataFrame:
        """统一时间格式"""
        if not isinstance(df.index, pd.DatetimeIndex):
            df.index = pd.to_datetime(df.index)
        # 统一时区到UTC
        if df.index.tz is None:
            df.index = df.index.tz_localize('UTC')
        else:
            df.index = df.index.tz_convert('UTC')
        return df

这个私有方法，所有子类都可以复用。数据到了上层，时间格式一定是统一的，策略逻辑不用做额外处理。

3.9 总结一下

数据层封装的核心就三件事：

  1. 定义抽象接口——让上层依赖抽象，不依赖具体实现
  2. 实现多种数据源——CSV、数据库、API各司其职
  3. 工厂模式切换——改一行配置，数据源就换了

这样做的好处，你马上就能感受到。下一章我们开始封装策略层，你会发现数据层和策略层可以完全独立开发、独立测试。这就是架构设计带来的红利。

一句话记住：好的数据层设计，让你换数据源像换衣服一样简单。不好的设计，让你换数据源像换心脏一样痛苦。

蓝海数据掘金营，专注资料整理
← 上一章 📖 返回目录 下一章 →