3. 数据源接入：WebSocket实时行情、REST API历史数据、多交易所时钟同步

做高频交易，数据就是你的眼睛。

眼睛瞎了，再好的策略也白搭。我见过太多团队，策略逻辑写得漂漂亮亮，结果一上线就亏钱。查到最后，发现是数据源出了问题——要么延迟高，要么数据对不上，要么时间戳乱跳。

这一章，咱们就聊聊怎么把数据源这件事做扎实。

3.1 WebSocket实时行情：别让数据在路上堵车

实时行情，说白了就是交易所把订单簿的变化推给你。

我个人习惯用WebSocket，而不是轮询REST。为什么？你想想看，轮询就像你每隔一秒去问交易所“有变化吗？”，而WebSocket是交易所主动告诉你“变了”。高频场景下，差一毫秒可能就是几万块的盈亏。

我在项目中遇到过一个问题：某次行情波动剧烈，WebSocket连接突然断了。我们的重连逻辑是等5秒再连，结果这5秒里错过了关键行情，策略直接爆仓。后来我改成了指数退避重连——第一次等1秒，第二次2秒，第三次4秒，最多到30秒。同时保留一个备用连接，主连接断了立刻切过去。

import asyncio
import websockets
import json

class WebSocketClient:
    def __init__(self, url, max_retries=5):
        self.url = url
        self.max_retries = max_retries
        self.retry_delay = 1  # 初始重连延迟1秒

    async def connect(self):
        for attempt in range(self.max_retries):
            try:
                async with websockets.connect(self.url) as ws:
                    print(f"连接成功: {self.url}")
                    self.retry_delay = 1  # 重置延迟
                    await self.listen(ws)
            except Exception as e:
                print(f"连接失败，第{attempt+1}次重试: {e}")
                await asyncio.sleep(self.retry_delay)
                self.retry_delay = min(self.retry_delay * 2, 30)  # 指数退避

    async def listen(self, ws):
        async for message in ws:
            data = json.loads(message)
            # 处理行情数据
            self.process_tick(data)

    def process_tick(self, data):
        # 这里写你的行情处理逻辑
        pass

3.2 REST API历史数据：回测的基石

实时行情是给实盘用的，历史数据是给回测用的。

没有高质量的历史数据，你的策略就是空中楼阁。我曾经用某免费数据源做回测，收益曲线漂亮得不行，结果实盘一跑就亏。后来发现，那个数据源把很多无效成交都过滤掉了，导致回测里的滑点估计严重偏低。

所以，我建议你用交易所官方的REST API拉历史数据。虽然慢一点，但数据质量有保障。

import requests
import time
import pandas as pd

class HistoricalDataFetcher:
    def __init__(self, base_url, api_key=None):
        self.base_url = base_url
        self.api_key = api_key
        self.session = requests.Session()
        if api_key:
            self.session.headers.update({'X-MBX-APIKEY': api_key})

    def fetch_klines(self, symbol, interval, start_time, end_time):
        """
        拉取K线数据
        :param symbol: 交易对，如 'BTCUSDT'
        :param interval: 时间间隔，如 '1m', '5m'
        :param start_time: 开始时间戳（毫秒）
        :param end_time: 结束时间戳（毫秒）
        """
        url = f"{self.base_url}/api/v3/klines"
        params = {
            'symbol': symbol,
            'interval': interval,
            'startTime': start_time,
            'endTime': end_time,
            'limit': 1000  # 每次最多1000条
        }

        all_data = []
        while start_time < end_time:
            params['startTime'] = start_time
            response = self.session.get(url, params=params)
            if response.status_code != 200:
                print(f"请求失败: {response.status_code}")
                time.sleep(1)
                continue

            data = response.json()
            if not data:
                break

            all_data.extend(data)
            # 更新start_time为最后一条数据的时间+1毫秒
            start_time = data[-1][0] + 1
            time.sleep(0.5)  # 控制请求频率

        # 转换为DataFrame
        df = pd.DataFrame(all_data, columns=[
            'timestamp', 'open', 'high', 'low', 'close', 'volume',
            'close_time', 'quote_asset_volume', 'number_of_trades',
            'taker_buy_base_asset_volume', 'taker_buy_quote_asset_volume', 'ignore'
        ])
        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
        return df

3.3 多交易所时钟同步：时间就是金钱

做跨市场套利，最怕什么？

最怕两个交易所的时间对不上。

比如你在A交易所看到BTC价格是50000，在B交易所看到是50001。你觉得有1美元的价差，可以套利。结果一执行，发现A交易所的价格已经变了——因为A交易所的时钟比B交易所慢了100毫秒。你看到的价差，其实是过时的数据。

嗯，这里要注意：每个交易所的服务器时间都不一样。有的用NTP同步，有的用自己的时间服务器。偏差可能在几十到几百毫秒。

我个人的做法是：

1. 获取交易所时间戳：每次连接时，先调交易所的时间API，记录下本地时间与交易所时间的差值。
2. 动态校准：每5分钟重新校准一次，因为时钟漂移是持续的。
3. 打时间戳：收到行情数据后，用本地时间+校准值，算出交易所的“真实时间”。

import time
import requests

class ClockSync:
    def __init__(self, exchange_name, time_api_url):
        self.exchange_name = exchange_name
        self.time_api_url = time_api_url
        self.offset = 0  # 本地时间与交易所时间的差值
        self.last_sync = 0

    def sync(self):
        """同步时钟，计算偏移量"""
        local_before = int(time.time() * 1000)
        try:
            response = requests.get(self.time_api_url, timeout=2)
            exchange_time = response.json()['serverTime']  # 假设返回格式
            local_after = int(time.time() * 1000)
            # 估算网络延迟
            rtt = local_after - local_before
            # 交易所时间的估计值 = 本地时间 + 偏移量
            # 偏移量 = 交易所时间 - (本地时间 + rtt/2)
            self.offset = exchange_time - (local_before + rtt // 2)
            self.last_sync = local_after
            print(f"{self.exchange_name} 时钟同步完成，偏移量: {self.offset}ms")
        except Exception as e:
            print(f"时钟同步失败: {e}")

    def get_exchange_time(self):
        """获取校准后的交易所时间"""
        local_now = int(time.time() * 1000)
        # 如果距离上次同步超过5分钟，重新同步
        if local_now - self.last_sync > 300000:
            self.sync()
        return local_now + self.offset

3.4 整体架构：一张图看懂数据流

说了这么多，咱们用一张图把整个数据接入的流程串起来。

这张图展示了数据从交易所到策略引擎的完整流程。你看，每个交易所的数据先经过时钟同步模块校准时间，然后进入数据聚合层合并，最后才交给策略引擎。这样能保证你看到的数据是“对齐”的，不会因为时间偏差做出错误决策。

3.5 实战建议：从小处着手

如果你刚开始搭建这套系统，我建议你先从一个交易所开始。

把WebSocket实时行情跑通，把REST API历史数据拉下来，把时钟同步做好。等一个交易所的数据流稳定了，再加第二个、第三个。别一上来就想搞三四个交易所，容易翻车。

另外，数据存储也很重要。实时行情数据量很大，我一般用InfluxDB这种时序数据库来存。历史数据则用Parquet格式存到本地，方便回测时快速读取。

好了，数据源接入这块就聊到这儿。记住一句话：数据质量决定策略上限。把数据源搞扎实了，后面的策略开发才能事半功倍。