第四章：数据源接入与适配器模式

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

我个人觉得，不是策略写不出来，而是数据搞不定。你想想看，交易所的API三天两头改，不同交易所的数据格式千奇百怪，有的用REST，有的走WebSocket，还有的历史数据要自己去爬。我刚开始做数字孪生回测引擎的时候，光对接数据就折腾了两个月。

这一章，我们就来聊聊怎么用适配器模式，把数据源接入这件事做得优雅一点。

4.1 交易所API对接：REST与WebSocket

先说说两种最常见的API方式。

REST API，说白了就是你去问交易所要数据。发个HTTP请求，它给你返回JSON。适合拿历史数据、查询账户信息。优点是简单，缺点是慢——你想想看，每次都要建立连接、等待响应，高频交易根本来不及。

WebSocket，则是交易所主动推数据给你。建立一条长连接，行情一变，数据就过来了。延迟低，适合实时行情。但麻烦的是，你得自己维护心跳、处理断线重连。

我在项目中遇到过一个问题：某个交易所的WebSocket每30秒发一次ping，如果你没及时回pong，它就把你踢了。一开始没注意，结果回测跑到一半数据断了，整个策略分析全废了。

来看一个简单的REST对接示例：

import requests
import json

class BinanceRESTClient:
    def __init__(self, base_url="https://api.binance.com"):
        self.base_url = base_url
    
    def get_klines(self, symbol, interval, limit=500):
        endpoint = f"{self.base_url}/api/v3/klines"
        params = {
            "symbol": symbol,
            "interval": interval,
            "limit": limit
        }
        response = requests.get(endpoint, params=params)
        return response.json()

嗯，这里要注意：REST请求一定要加超时处理。我见过有人没设timeout，结果交易所挂了，整个程序卡死在那里。

4.2 数据适配器接口设计

不同的交易所，数据格式不一样。Binance的K线是[时间, 开, 高, 低, 收, 量...]，OKX的可能是另一种结构。如果每个策略都去适配不同格式，代码就乱套了。

这时候适配器模式就派上用场了。

我的做法是：定义一个统一的数据接口，然后每个交易所写一个适配器，把原始数据转成统一格式。

from abc import ABC, abstractmethod

class DataAdapter(ABC):
    @abstractmethod
    def to_unified_kline(self, raw_data):
        """将原始数据转为统一格式"""
        pass
    
    @abstractmethod
    def to_unified_ticker(self, raw_data):
        pass

class BinanceAdapter(DataAdapter):
    def to_unified_kline(self, raw_data):
        # Binance原始格式: [time, open, high, low, close, volume, ...]
        return {
            "timestamp": raw_data[0],
            "open": float(raw_data[1]),
            "high": float(raw_data[2]),
            "low": float(raw_data[3]),
            "close": float(raw_data[4]),
            "volume": float(raw_data[5])
        }

class OKXAdapter(DataAdapter):
    def to_unified_kline(self, raw_data):
        # OKX原始格式可能不同
        return {
            "timestamp": int(raw_data[0]),
            "open": float(raw_data[1]),
            "high": float(raw_data[2]),
            "low": float(raw_data[3]),
            "close": float(raw_data[4]),
            "volume": float(raw_data[5])
        }

这样设计的好处是：策略层只认统一格式，底层换了交易所，改个适配器就行。

4.3 多数据源统一管理

实际回测中，你可能同时用多个数据源。比如历史数据用本地CSV，实时行情用WebSocket，深度数据用REST。怎么统一管理？

我习惯用「数据源管理器」来做这件事。

class DataSourceManager:
    def __init__(self):
        self._sources = {}
    
    def register_source(self, name, source):
        self._sources[name] = source
    
    def get_data(self, source_name, symbol, **kwargs):
        source = self._sources.get(source_name)
        if not source:
            raise ValueError(f"数据源 {source_name} 未注册")
        return source.fetch(symbol, **kwargs)

你想想看，这样注册机制的好处是什么？

• 新增数据源不用改核心代码
• 可以动态切换数据源
• 方便做数据源优先级管理

4.4 数据缓存与重放机制

回测引擎最怕什么？重复请求数据。

你跑一次策略，请求一次数据。调参数再跑一次，又请求一次。如果数据是从交易所实时拉的，不仅慢，还可能被限流。

所以，缓存机制是必须的。

我常用的缓存策略分三层：

层级	存储方式	适用场景
L1 内存缓存	Python dict / LRU Cache	当前回测会话内的数据复用
L2 本地文件缓存	Parquet / Feather 格式	跨会话的历史数据复用
L3 数据库缓存	InfluxDB / TimescaleDB	大规模数据管理与查询

来看一个简单的内存缓存实现：

from functools import lru_cache

class DataCache:
    def __init__(self, maxsize=128):
        self._cache = {}
        self.maxsize = maxsize
    
    def get(self, key):
        return self._cache.get(key)
    
    def set(self, key, data):
        if len(self._cache) >= self.maxsize:
            # 淘汰最旧的数据
            oldest = next(iter(self._cache))
            del self._cache[oldest]
        self._cache[key] = data

至于数据重放，说白了就是把历史数据按时间顺序「播放」一遍。回测引擎的核心就是这个机制。

我曾经犯过一个错误：重放数据时没有考虑时间戳对齐。结果策略在1分钟K线上跑得好好的，换到5分钟K线就全乱了。后来我加了一个时间戳对齐器，才解决这个问题。

知识体系总览

这一章的内容比较多，我画了一张图帮你理清思路：

好了，这一章的内容就到这里。数据源是整个回测引擎的基石，这块做扎实了，后面的策略开发才能放心。下一章我们聊聊回测引擎的核心——订单执行与撮合逻辑。

---

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