4、滑点数据采集：交易所API数据获取、订单簿数据解析、Tick级数据与K线数据、数据清洗与对齐

数据采集，说白了就是滑点分析的「原材料」环节。你想想看，再厉害的分析模型，喂进去的是脏数据，出来的结果肯定也是垃圾。我在这个坑里栽过不止一次，所以今天咱们好好聊聊怎么把数据这关把住。

4.1 交易所API数据获取

我个人习惯把API获取分成两类：REST和WebSocket。REST适合拉历史数据，WebSocket适合接实时流。别混着用，否则你会被限流搞疯。

下面是我常用的一个REST请求模板，加了重试和限流逻辑：

import requests
import time
from typing import Dict, List

class ExchangeAPI:
    def __init__(self, base_url: str, rate_limit: int = 10):
        self.base_url = base_url
        self.rate_limit = rate_limit  # 每秒请求数
        self.last_request_time = 0
    
    def _rate_limiter(self):
        """简单的速率限制"""
        elapsed = time.time() - self.last_request_time
        if elapsed < 1.0 / self.rate_limit:
            time.sleep(1.0 / self.rate_limit - elapsed)
        self.last_request_time = time.time()
    
    def get_orderbook(self, symbol: str, depth: int = 10) -> Dict:
        """获取订单簿数据"""
        self._rate_limiter()
        url = f"{self.base_url}/api/v3/depth"
        params = {"symbol": symbol, "limit": depth}
        
        for retry in range(3):  # 重试3次
            try:
                resp = requests.get(url, params=params, timeout=5)
                resp.raise_for_status()
                return resp.json()
            except Exception as e:
                print(f"请求失败，第{retry+1}次重试: {e}")
                time.sleep(1)
        return None

4.2 订单簿数据解析

订单簿数据长什么样？说白了就是一堆买单和卖单的集合。交易所返回的格式通常是这样的：

{
    "bids": [
        ["45000.00", "1.5"],  // [价格, 数量]
        ["44999.50", "2.3"],
        ...
    ],
    "asks": [
        ["45001.00", "0.8"],
        ["45001.50", "1.2"],
        ...
    ],
    "timestamp": 1698765432000
}

这里有个坑要注意：价格和数量都是字符串，不是浮点数。为什么？因为浮点数有精度问题。我建议你直接用Decimal来处理：

from decimal import Decimal

def parse_orderbook(raw_data: dict) -> dict:
    """解析订单簿，返回Decimal类型"""
    bids = [(Decimal(price), Decimal(qty)) 
            for price, qty in raw_data["bids"]]
    asks = [(Decimal(price), Decimal(qty)) 
            for price, qty in raw_data["asks"]]
    
    return {
        "bids": bids,
        "asks": asks,
        "timestamp": raw_data["timestamp"]
    }

4.3 Tick级数据与K线数据

Tick数据和K线数据，这两个是滑点分析的核心数据源。我简单说说区别：

• Tick数据：每一笔成交的记录，包含价格、数量、时间戳。数据量大，但最精细。
• K线数据：聚合后的数据，比如1分钟K线包含开盘、最高、最低、收盘、成交量。数据量小，但丢失了细节。

做滑点分析，我个人更推荐用Tick数据。为什么？因为滑点发生在每一笔交易中，K线数据会把这些细节「抹平」了。你想想看，一根1分钟K线里可能有100笔成交，每笔的滑点都不一样，你用K线的均价去算，误差能不大吗？

下面是我常用的Tick数据采集代码：

import websocket
import json

def on_message(ws, message):
    """处理Tick数据"""
    data = json.loads(message)
    tick = {
        "symbol": data["s"],      # 交易对
        "price": float(data["p"]), # 成交价格
        "qty": float(data["q"]),   # 成交数量
        "time": int(data["T"]),    # 成交时间戳(毫秒)
        "side": "buy" if data["m"] else "sell"  # 买卖方向
    }
    # 这里可以写入数据库或队列
    save_tick(tick)

def start_tick_stream(symbol: str):
    """启动Tick数据流"""
    ws_url = f"wss://stream.binance.com:9443/ws/{symbol.lower()}@trade"
    ws = websocket.WebSocketApp(ws_url, on_message=on_message)
    ws.run_forever()

4.4 数据清洗与对齐

数据清洗，说白了就是「去伪存真」。我从交易所拿到的数据，经常会有这些问题：

• 重复数据：同一个Tick被推送了两次
• 缺失数据：网络波动导致某几秒的数据没收到
• 异常数据：价格突然跳变，比如从45000跳到450000
• 时间戳错乱：数据到达顺序和实际发生顺序不一致

我常用的清洗流程是这样的：

def clean_ticks(ticks: list) -> list:
    """清洗Tick数据"""
    # 1. 去重：基于时间戳和价格去重
    seen = set()
    unique_ticks = []
    for tick in ticks:
        key = (tick["time"], tick["price"], tick["qty"])
        if key not in seen:
            seen.add(key)
            unique_ticks.append(tick)
    
    # 2. 排序：按时间戳排序
    unique_ticks.sort(key=lambda x: x["time"])
    
    # 3. 异常检测：剔除价格偏离超过3个标准差的数据
    prices = [t["price"] for t in unique_ticks]
    mean = sum(prices) / len(prices)
    std = (sum((p - mean)**2 for p in prices) / len(prices))**0.5
    
    cleaned = []
    for tick in unique_ticks:
        if abs(tick["price"] - mean) < 3 * std:
            cleaned.append(tick)
    
    return cleaned

数据对齐呢？就是把不同来源的数据对齐到同一个时间轴上。比如你有币安的Tick数据和OKX的Tick数据，它们的时钟可能差了几毫秒。怎么对齐？

下面是我画的一张数据采集流程图，帮你理清整体思路：

最后说一句，数据对齐的时候，要注意交易所的时钟同步问题。有些交易所的服务器时间可能差了几十毫秒。我建议你每次采集数据前，先校准一下本地时钟，用NTP同步到毫秒级。

好了，数据采集这块就聊到这儿。记住一句话：数据质量决定了分析的上限。把数据搞干净了，后面的滑点分析才能站得住脚。