数据获取与处理：交易所API对接、K线数据获取、数据清洗与预处理、数据存储方案

做量化交易，数据就是你的弹药库。没有干净、可靠的数据，再牛的策略也是空中楼阁。这一章，我带你亲手搭建数据管道——从交易所拿到原始数据，到清洗干净存进数据库。嗯，这里面的坑，我踩过不少。

交易所API对接：你的第一道关卡

说白了，API就是你和交易所之间的翻译官。你发一个请求，它给你返回数据。但各家交易所的脾气不一样，对接起来各有门道。

REST API vs WebSocket

我个人习惯把REST API当成「查字典」——你需要什么，主动去查。WebSocket则是「订阅报纸」——交易所主动推给你。做历史回测用REST就够了，实盘交易必须上WebSocket。

常见交易所API特点

交易所	REST限频	WebSocket支持	认证方式
Binance	1200次/分钟	完善	API Key + Secret
OKX	600次/分钟	完善	API Key + Passphrase
Bybit	600次/分钟	完善	API Key + Secret

K线数据获取：从零到一

K线数据是量化策略的基石。获取K线看似简单，但细节决定成败。

获取历史K线

以Binance为例，获取BTC/USDT的日线数据：

import requests
import pandas as pd

def get_klines(symbol, interval, limit=500):
    url = "https://api.binance.com/api/v3/klines"
    params = {
        "symbol": symbol,
        "interval": interval,
        "limit": limit
    }
    response = requests.get(url, params=params)
    data = response.json()
    
    # 解析数据
    klines = []
    for k in data:
        klines.append({
            "timestamp": k[0],
            "open": float(k[1]),
            "high": float(k[2]),
            "low": float(k[3]),
            "close": float(k[4]),
            "volume": float(k[5])
        })
    return pd.DataFrame(klines)

df = get_klines("BTCUSDT", "1d", 1000)
print(df.head())

实时K线订阅

WebSocket订阅实时K线，我建议用异步方式处理：

import asyncio
import websockets
import json

async def subscribe_klines(symbol, interval):
    url = "wss://stream.binance.com:9443/ws"
    subscribe_msg = {
        "method": "SUBSCRIBE",
        "params": [f"{symbol.lower()}@kline_{interval}"],
        "id": 1
    }
    
    async with websockets.connect(url) as ws:
        await ws.send(json.dumps(subscribe_msg))
        
        while True:
            msg = await ws.recv()
            data = json.loads(msg)
            # 处理K线数据
            print(f"最新收盘价: {data['k']['c']}")

# 运行
asyncio.run(subscribe_klines("btcusdt", "1m"))

数据清洗与预处理：脏活累活

你想想看，交易所返回的数据真的可靠吗？我告诉你，不一定。缺失值、异常值、时间戳错乱，这些都是家常便饭。

常见数据问题

• 缺失值：网络波动导致某根K线没获取到
• 异常值：交易所系统故障，返回了离谱的价格
• 重复数据：重试机制导致同一根K线被多次获取
• 时间戳不一致：不同交易所的时间戳格式不同

清洗流程

我一般按这个顺序处理：

def clean_klines(df):
    # 1. 去重
    df = df.drop_duplicates(subset=["timestamp"])
    
    # 2. 排序
    df = df.sort_values("timestamp")
    
    # 3. 处理缺失值
    df = df.set_index("timestamp")
    df = df.asfreq("1min")  # 假设是1分钟K线
    df = df.ffill()  # 前向填充
    
    # 4. 检测异常值
    for col in ["open", "high", "low", "close"]:
        mean = df[col].mean()
        std = df[col].std()
        df[col] = df[col].clip(mean - 3*std, mean + 3*std)
    
    return df.reset_index()

数据存储方案：选对工具

数据存哪里？这取决于你的数据量和查询频率。我试过CSV、SQLite、InfluxDB，各有优劣。

存储方案对比

方案	适用场景	优点	缺点
CSV文件	小规模、一次性分析	简单、通用	查询慢、无索引
SQLite	个人回测、小团队	轻量、支持SQL	并发写入差
InfluxDB	时序数据、高频交易	高性能、自动过期	学习成本高
ClickHouse	海量数据、分析型	极速查询、列式存储	部署复杂

我的推荐

个人做回测，SQLite足够了。实盘交易，我建议用InfluxDB。为什么？因为时序数据库天生适合K线数据，查询速度快，还能自动清理旧数据。

import sqlite3
import pandas as pd

def save_to_sqlite(df, db_path, table_name):
    conn = sqlite3.connect(db_path)
    df.to_sql(table_name, conn, if_exists="append", index=False)
    conn.close()

def query_from_sqlite(db_path, table_name, start_time, end_time):
    conn = sqlite3.connect(db_path)
    query = f"""
    SELECT * FROM {table_name}
    WHERE timestamp BETWEEN {start_time} AND {end_time}
    ORDER BY timestamp
    """
    df = pd.read_sql(query, conn)
    conn.close()
    return df

知识体系总览

下面这张图，概括了本章的核心逻辑：

从交易所API获取数据，经过清洗预处理，最终存入数据库。每一步都有坑，但每一步也都有解法。我个人觉得，数据管道的质量，直接决定了策略的上限。花时间把数据搞干净，比花时间调参数划算得多。