第4章数据获取与处理

4. 数据获取与处理：用ccxt搞定交易所行情

做市商系统里，数据就是你的眼睛。没有数据，报价就是瞎蒙。

我个人习惯把数据获取分成三个层次：Ticker（实时快照）、K线（历史切片）、深度数据（订单簿）。今天咱们重点讲前两个，深度数据后面单独开一章。

4.1 为什么选ccxt？

市面上有很多交易所的API SDK，但ccxt是唯一一个让我觉得「嗯，这玩意儿靠谱」的库。它统一了100多家交易所的接口，写法几乎一样。

你想想看，如果每个交易所都单独写一套对接代码，光维护就够你喝一壶的。我在项目中遇到过最夸张的情况——同时对接6家交易所，每家API风格都不一样。后来全换成ccxt，代码量直接砍掉70%。

核心优势：

统一接口：Binance、OKX、Bybit、Coinbase……写法一致
自动处理：签名、频率限制、重连机制
活跃维护：几乎每周更新，跟得上交易所变化

4.2 安装与初始化

安装很简单，一行命令搞定：

pip install ccxt

初始化交易所对象时，我建议把API密钥放在环境变量里，别硬编码。曾经有个同事把密钥直接写在代码里提交到GitHub……嗯，后果你懂的。

import ccxt
import os

# 初始化交易所
exchange = ccxt.binance({
    'apiKey': os.getenv('BINANCE_API_KEY'),
    'secret': os.getenv('BINANCE_SECRET'),
    'enableRateLimit': True,  # 自动限速，很重要
    'options': {
        'defaultType': 'spot',  # 现货市场
    }
})

# 测试连接
try:
    exchange.load_markets()
    print(f"已连接 {exchange.name}，支持 {len(exchange.markets)} 个交易对")
except Exception as e:
    print(f"连接失败: {e}")

我的小习惯：enableRateLimit一定要开。不开的话，交易所分分钟封你IP。我刚开始做的时候吃过这个亏，被Binance封了3次才长记性。

4.3 获取Ticker数据

Ticker就是当前市场的快照——最新价、24小时涨跌、成交量这些。做市商系统里，Ticker用来判断当前市场状态。

def get_ticker(symbol='BTC/USDT'):
    """获取实时Ticker"""
    ticker = exchange.fetch_ticker(symbol)
    
    # 提取关键字段
    data = {
        'symbol': ticker['symbol'],
        'last': ticker['last'],           # 最新成交价
        'bid': ticker['bid'],             # 买一价
        'ask': ticker['ask'],             # 卖一价
        'baseVolume': ticker['baseVolume'], # 24h成交量
        'quoteVolume': ticker['quoteVolume'], # 24h成交额
        'percentage': ticker['percentage'],   # 24h涨跌幅
        'timestamp': ticker['timestamp'],
    }
    return data

# 调用
ticker_data = get_ticker('BTC/USDT')
print(f"BTC最新价: {ticker_data['last']} USDT")

这里有个坑要注意——不同交易所返回的字段名可能不一样。ccxt虽然做了统一，但偶尔会有遗漏。我建议拿到数据后先打印看看结构，别直接信文档。

4.4 获取K线数据

K线是做策略的基础。做市商系统里，K线用来计算波动率、均线、布林带这些指标。

def get_ohlcv(symbol='BTC/USDT', timeframe='1h', limit=100):
    """
    获取K线数据
    :param symbol: 交易对
    :param timeframe: 时间周期 (1m, 5m, 15m, 1h, 4h, 1d)
    :param limit: 返回条数
    """
    ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
    
    # 转换成结构化数据
    candles = []
    for candle in ohlcv:
        candles.append({
            'timestamp': candle[0],   # 开盘时间
            'open': candle[1],        # 开盘价
            'high': candle[2],        # 最高价
            'low': candle[3],         # 最低价
            'close': candle[4],       # 收盘价
            'volume': candle[5],      # 成交量
        })
    return candles

# 获取最近100根1小时K线
btc_1h = get_ohlcv('BTC/USDT', '1h', 100)
print(f"获取到 {len(btc_1h)} 根K线")
print(f"最新收盘价: {btc_1h[-1]['close']}")

注意：不同交易所对limit的限制不一样。Binance最多1000根，OKX最多300根。如果你需要更多历史数据，得用循环分批拉取。我一般一次拉500根，够用又不触发限速。

4.5 数据清洗——别让脏数据毁了你的策略

数据拿到手，别急着用。交易所偶尔会返回异常数据——价格突变、时间戳错乱、成交量异常。不做清洗的话，你的策略可能会被一根假K线带偏。

我曾经遇到过一件事：某交易所的API在系统升级时返回了重复的K线数据，我的做市系统直接按那个数据调整了报价，结果亏了2个BTC……从那以后，数据清洗成了我的铁律。

def clean_ohlcv(candles):
    """清洗K线数据"""
    cleaned = []
    
    for i, candle in enumerate(candles):
        # 1. 检查价格是否合理（防止交易所bug）
        if candle['high'] <= 0 or candle['low'] <= 0:
            print(f"跳过异常K线 #{i}: 价格异常")
            continue
            
        # 2. 检查高低价关系
        if candle['high'] < candle['low']:
            print(f"跳过异常K线 #{i}: 最高价小于最低价")
            continue
            
        # 3. 检查时间戳是否递增（防止重复数据）
        if cleaned and candle['timestamp'] <= cleaned[-1]['timestamp']:
            print(f"跳过异常K线 #{i}: 时间戳重复或倒退")
            continue
            
        # 4. 检查成交量是否异常（防止插针）
        if cleaned and candle['volume'] > cleaned[-1]['volume'] * 10:
            print(f"警告: 成交量异常放大 #{i}")
            # 这里可以标记，但不一定删除
            
        cleaned.append(candle)
    
    return cleaned

# 使用
raw_data = get_ohlcv('BTC/USDT', '1h', 500)
clean_data = clean_ohlcv(raw_data)
print(f"清洗前: {len(raw_data)} 根, 清洗后: {len(clean_data)} 根")

4.6 数据存储——选对工具省一半力气

数据存哪里？这取决于你的使用场景。

存储方式	适用场景	我的评价
CSV文件	本地测试、小规模回测	简单粗暴，但别用于生产
SQLite	单机策略、个人项目	够用，但并发写入不行
InfluxDB	时序数据、高频场景	我的首选，查询快、压缩率高
PostgreSQL	团队协作、复杂查询	功能强大，但配置稍重

我个人习惯用InfluxDB存K线数据。原因很简单——时序数据库天生就是干这个的，写入快、查询快、磁盘占用小。

# 用CSV存储（最简单的方式）
import csv
import pandas as pd

def save_to_csv(data, filename='btc_1h.csv'):
    """保存K线到CSV"""
    df = pd.DataFrame(data)
    df.to_csv(filename, index=False)
    print(f"数据已保存到 {filename}")

def load_from_csv(filename='btc_1h.csv'):
    """从CSV加载K线"""
    df = pd.read_csv(filename)
    return df.to_dict('records')

# 使用
save_to_csv(clean_data, 'btc_1h_2024.csv')
loaded_data = load_from_csv('btc_1h_2024.csv')

我的建议：开发阶段用CSV就够了，方便调试。上线前再切换到InfluxDB或PostgreSQL。别一开始就上重型数据库，容易陷入「过度设计」的坑。

4.7 完整的数据流水线

把上面这些串起来，就是一个完整的数据获取与处理流程。我画了张图帮你理解：

4.8 避坑指南

最后分享几个我踩过的坑，希望能帮你省点学费：

频率限制：别用单线程高频请求。我一般用asyncio异步，或者加个简单的sleep(0.1)。
时间同步：交易所的时间戳是UTC，记得转成你本地时区。我吃过这个亏，策略在凌晨3点莫名其妙开仓。
断线重连：网络波动是常态。写个重试机制，最多重试3次，每次间隔递增。
数据对齐：不同交易所的K线起始时间可能差几秒。做多交易所策略时，一定要对齐时间戳。

曾经有一次：我在生产环境忘了处理交易所的维护公告时间。结果API返回了空数据，系统直接报错停了15分钟。后来我加了个健康检查——如果连续3次获取失败，就切换到备用数据源。

好了，数据获取与处理这块就讲这么多。代码都在上面了，你直接复制就能跑。记住一点：数据质量决定策略上限。花80%的时间在数据上，20%的时间写策略，这个比例不夸张。

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

📚 课程目录

4. 数据获取与处理：用ccxt搞定交易所行情

4.1 为什么选ccxt？

4.2 安装与初始化

4.3 获取Ticker数据

4.4 获取K线数据

4.5 数据清洗——别让脏数据毁了你的策略

4.6 数据存储——选对工具省一半力气

4.7 完整的数据流水线

4.8 避坑指南