第4章数据获取与清洗

4. 数据获取与清洗：交易所API对接

做量化交易，数据就是你的弹药。弹药不行，枪法再好也白搭。这一章我重点聊聊怎么从交易所拿到干净的数据，以及那些坑我是怎么踩过来的。

4.1 交易所API对接（以Binance为例）

先说说API对接。我个人习惯用Binance的API做演示，因为它的文档最规范，接口也最全。你想想看，如果连Binance都搞不定，其他小交易所就更别说了。

对接API其实就三步：注册、获取Key、写代码。但这里有个细节——权限控制。我在项目中遇到过有人直接把API Key写死在代码里，结果代码泄露，资产被转走。嗯，这是个血泪教训。

⚠️ 警告：API Key一定要设置IP白名单，而且只开交易权限，别开提现权限。这是底线。

下面是一个简单的Binance REST API对接示例：

import requests
import hashlib
import hmac
import time

# 配置
API_KEY = '你的API_KEY'
SECRET_KEY = '你的SECRET_KEY'
BASE_URL = 'https://api.binance.com'

# 获取K线数据
def get_klines(symbol='BTCUSDT', interval='1m', limit=100):
    endpoint = '/api/v3/klines'
    params = {
        'symbol': symbol,
        'interval': interval,
        'limit': limit
    }
    response = requests.get(BASE_URL + endpoint, params=params)
    return response.json()

# 带签名的请求（用于交易）
def signed_request(endpoint, params={}):
    params['timestamp'] = int(time.time() * 1000)
    query_string = '&'.join([f'{k}={v}' for k, v in params.items()])
    signature = hmac.new(
        SECRET_KEY.encode('utf-8'),
        query_string.encode('utf-8'),
        hashlib.sha256
    ).hexdigest()
    params['signature'] = signature
    headers = {'X-MBX-APIKEY': API_KEY}
    response = requests.get(BASE_URL + endpoint, params=params, headers=headers)
    return response.json()

# 测试
data = get_klines('BTCUSDT', '1m', 10)
print(data[:2])  # 打印前两条数据

这段代码看着简单，但实际跑起来你会发现——嗯，Binance的限频策略很严格。1秒最多20次请求，超了直接封IP。我建议你加个请求队列，或者用WebSocket代替REST。

4.2 高频数据采集技巧

高频数据采集，说白了就是跟时间赛跑。1分钟K线还好说，如果是Tick级数据，每秒可能上千条。这时候REST API就扛不住了。

我的做法是：WebSocket + 本地缓存。WebSocket负责实时推送，本地缓存负责防丢数据。具体来说：

用WebSocket订阅深度流和K线流
数据落地到内存队列，再异步写入磁盘
每隔5分钟做一次数据校验，防止断线导致数据缺失

我曾经在实盘中发现，WebSocket偶尔会断线重连，中间会丢失几秒数据。别小看这几秒，在高频策略里可能就是几万块的盈亏。后来我加了个心跳检测，每3秒检查一次连接状态，断了就自动重连并补数据。

💡 技巧：WebSocket重连后，建议从交易所重新拉取最近100条K线做数据对齐，确保本地数据跟交易所一致。

下面是一个WebSocket订阅的示例：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    # 处理K线数据
    if 'k' in data:
        kline = data['k']
        print(f"时间: {kline['t']}, 开盘: {kline['o']}, 收盘: {kline['c']}")

def on_error(ws, error):
    print(f"WebSocket错误: {error}")

def on_close(ws, close_status_code, close_msg):
    print("WebSocket已关闭，准备重连...")

def on_open(ws):
    # 订阅BTCUSDT的1分钟K线
    subscribe_msg = {
        "method": "SUBSCRIBE",
        "params": ["btcusdt@kline_1m"],
        "id": 1
    }
    ws.send(json.dumps(subscribe_msg))

# 启动WebSocket
ws = websocket.WebSocketApp(
    "wss://stream.binance.com:9443/ws",
    on_open=on_open,
    on_message=on_message,
    on_error=on_error,
    on_close=on_close
)
ws.run_forever()

4.3 数据清洗与对齐

数据拿到手了，但别急着用。交易所返回的数据格式五花八门，时间戳可能是毫秒、秒甚至微秒。你想想看，如果时间不对齐，策略信号全乱套。

数据清洗我一般做这几件事：

时间戳统一：全部转成UTC时间，精确到毫秒
字段重命名：把open_time、close_price这种改成统一的命名规范
排序：按时间升序排列，确保数据是连续的
去重：WebSocket可能重复推送，需要去重

举个例子，Binance返回的K线数据长这样：

[
  [
    1499040000000,      // 开盘时间
    "0.01634790",       // 开盘价
    "0.80000000",       // 最高价
    "0.01575800",       // 最低价
    "0.01577100",       // 收盘价
    "148976.11427815",  // 成交量
    1499644799999,      // 收盘时间
    "2434.19055334",    // 成交额
    308,                // 成交笔数
    "1756.87402397",    // 主动买入成交量
    "28.46694368",      // 主动买入成交额
    "17928899.62484339" // 忽略
  ]
]

我习惯把它转成Pandas DataFrame，方便后续处理：

import pandas as pd

def clean_klines(raw_data):
    df = pd.DataFrame(raw_data, columns=[
        'open_time', 'open', 'high', 'low', 'close', 'volume',
        'close_time', 'quote_volume', 'count', 'taker_buy_volume',
        'taker_buy_quote_volume', 'ignore'
    ])
    # 时间戳转datetime
    df['open_time'] = pd.to_datetime(df['open_time'], unit='ms')
    df['close_time'] = pd.to_datetime(df['close_time'], unit='ms')
    # 字符串转浮点数
    numeric_cols = ['open', 'high', 'low', 'close', 'volume', 'quote_volume']
    df[numeric_cols] = df[numeric_cols].apply(pd.to_numeric, errors='coerce')
    # 删除无用列
    df.drop('ignore', axis=1, inplace=True)
    # 按时间排序
    df.sort_values('open_time', inplace=True)
    # 去重
    df.drop_duplicates(subset='open_time', keep='last', inplace=True)
    return df

4.4 缺失值处理

数据缺失是家常便饭。交易所宕机、网络波动、API限频，都可能导致数据断档。我遇到过最夸张的一次，Binance的某个交易对因为流动性不足，连续3个小时没有成交，K线全是空的。

缺失值处理有几种常见方法：

方法	适用场景	缺点
前向填充	价格变化不大时	会忽略真实波动
线性插值	短期缺失（1-2根K线）	长期插值误差大
删除缺失行	缺失比例小于1%	会破坏时间连续性
从其他交易所补数据	主流币种	不同交易所价差可能较大

我个人最常用的是前向填充+线性插值的组合。具体做法：

如果缺失1根K线，用前一根的价格填充
如果缺失2-5根，用线性插值
如果缺失超过5根，直接标记为异常，策略暂停交易

  🔑 核心原则：宁可错过，不要做错。数据质量不达标时，策略应该自动暂停，而不是强行交易。

代码实现如下：

def handle_missing_data(df, max_gap=5):
    # 检查缺失的时间点
    full_range = pd.date_range(
        start=df['open_time'].min(),
        end=df['open_time'].max(),
        freq='1min'
    )
    missing_times = full_range.difference(df['open_time'])
    
    if len(missing_times) == 0:
        return df
    
    # 重新索引
    df = df.set_index('open_time').reindex(full_range)
    
    # 前向填充
    df.fillna(method='ffill', inplace=True)
    
    # 对连续缺失超过max_gap的做标记
    df['gap_flag'] = df['close'].isna().astype(int)
    df['gap_group'] = df['gap_flag'].diff().ne(0).cumsum()
    gap_counts = df[df['gap_flag'] == 1].groupby('gap_group').size()
    large_gaps = gap_counts[gap_counts > max_gap].index
    
    # 标记异常
    df['data_quality'] = 'good'
    df.loc[df['gap_group'].isin(large_gaps), 'data_quality'] = 'bad'
    
    # 对短缺失做线性插值
    df['close'] = df['close'].interpolate(method='linear', limit=max_gap)
    
    return df.reset_index().rename(columns={'index': 'open_time'})

这里有个细节——标记数据质量。我习惯在数据框里加一列data_quality，这样策略在运行时可以快速判断当前数据是否可靠。如果连续5分钟数据质量都是bad，我会让策略自动清仓。

💡 经验之谈：数据清洗不是一次性工作。我每周都会跑一次数据质量报告，检查缺失率、异常值、时间戳偏差。发现问题及时修复，别等到策略亏钱了才想起来。

最后，我把整个数据获取与清洗的流程画了张图，方便你理解：

这张图把整个流程串起来了。从API对接开始，到数据采集、清洗、对齐、缺失值处理，最后存到本地供策略使用。中间还有个反馈循环——数据质量不好时，要回溯到清洗步骤重新处理。

好了，数据获取与清洗这部分就聊到这儿。记住一句话：数据质量决定策略上限。花80%的时间在数据上，剩下20%写策略，这个比例我觉得很合理。

📚 课程目录

4. 数据获取与清洗：交易所API对接

4.1 交易所API对接（以Binance为例）

4.2 高频数据采集技巧

4.3 数据清洗与对齐

4.4 缺失值处理