3. 数据采集与清洗：交易数据、订单簿数据、链上数据的获取与预处理
各位同学，咱们直接进入正题。流动性危机预警，说白了就是跟数据打交道。数据不准，模型再漂亮也是白搭。我见过太多团队，模型跑得飞起，结果一上线就崩——原因往往是数据源没处理好。
今天这一章，我带你手把手搞定三类核心数据：交易数据、订单簿数据、链上数据。这三样东西，是流动性预警模型的「三根柱子」。少一根，模型就站不稳。
3.1 交易数据：最基础的「血液」
交易数据是什么？说白了就是每一笔成交的记录。谁在什么时间、以什么价格、买了多少。这是最原始的市场行为记录。
3.1.1 数据来源与获取
我个人习惯用交易所的API直接拉。比如Binance、OKX、Coinbase这些头部交易所，都提供RESTful API和WebSocket接口。
举个例子，用Python拉取BTC/USDT的分钟级K线数据：
import requests
import pandas as pd
def fetch_klines(symbol='BTCUSDT', interval='1m', limit=1000):
url = f'https://api.binance.com/api/v3/klines'
params = {
'symbol': symbol,
'interval': interval,
'limit': limit
}
resp = requests.get(url, params=params)
data = resp.json()
# 解析成DataFrame
df = pd.DataFrame(data, columns=[
'open_time', 'open', 'high', 'low', 'close', 'volume',
'close_time', 'quote_volume', 'count', 'taker_buy_volume',
'taker_buy_quote_volume', 'ignore'
])
df['open_time'] = pd.to_datetime(df['open_time'], unit='ms')
return df[['open_time', 'open', 'high', 'low', 'close', 'volume']]
# 拉取最近1000条分钟线
btc_data = fetch_klines()
print(btc_data.head())
嗯，这里要注意：API有频率限制。我曾经有个项目，直接开多线程狂拉数据，结果IP被交易所封了24小时。后来学乖了，加了个time.sleep(0.1)。
3.1.2 清洗要点
交易数据常见的坑，我列一下：
- 缺失值：某些时间段的K线可能为空（比如交易所维护）。我建议用前向填充，或者直接丢弃。
- 异常价格：比如突然出现一个0.0001的价格。这通常是数据错误。用3-sigma法则剔除。
- 时间戳对齐：不同交易所的时间可能有偏差。统一转成UTC时间。
我的经验：清洗交易数据时，别只看价格。成交量异常也很关键。比如某根K线成交量突然放大10倍，往往是流动性事件的先兆。
3.2 订单簿数据：市场的「心电图」
交易数据是「过去」，订单簿数据是「现在」。它告诉你当前市场上，有多少人在什么价位等着买或卖。
为什么订单簿对流动性预警重要？因为流动性危机发生前，订单簿会先「变薄」。买单和卖单之间的价差会拉大，深度会变浅。
3.2.1 获取订单簿快照
订单簿数据通常分两种：增量更新和全量快照。我个人建议用全量快照做分析，增量更新太容易出错了。
def fetch_orderbook(symbol='BTCUSDT', limit=100):
url = f'https://api.binance.com/api/v3/depth'
params = {
'symbol': symbol,
'limit': limit
}
resp = requests.get(url, params=params)
data = resp.json()
bids = pd.DataFrame(data['bids'], columns=['price', 'quantity'], dtype=float)
asks = pd.DataFrame(data['asks'], columns=['price', 'quantity'], dtype=float)
return bids, asks
bids, asks = fetch_orderbook()
print(f"买一价: {bids.iloc[0]['price']}, 数量: {bids.iloc[0]['quantity']}")
print(f"卖一价: {asks.iloc[0]['price']}, 数量: {asks.iloc[0]['quantity']}")
3.2.2 关键指标计算
拿到订单簿后，我们要算几个核心指标：
| 指标 | 计算公式 | 含义 |
|---|---|---|
| 买卖价差 | ask_price_1 - bid_price_1 | 价差越大，流动性越差 |
| 市场深度 | sum(quantity) 在±1%范围内 | 深度越浅，越容易被砸穿 |
| 订单簿不平衡 | (bid_volume - ask_volume) / (bid_volume + ask_volume) | 正值表示买方强势 |
避坑指南：我曾经在计算市场深度时，忘了把不同价位的数量按价格加权。结果算出来的深度虚高。记住：深度 = sum(price * quantity)，不是简单加总数量。
3.3 链上数据：DeFi的「透明账本」
链上数据是DeFi特有的。所有交易、转账、合约交互都记录在区块链上。这玩意儿公开透明，但数据量巨大，处理起来很头疼。
3.3.1 获取链上数据
我常用的方式有两种：
- 节点RPC：直接连以太坊节点，用JSON-RPC拉数据。慢，但免费。
- 数据服务商：比如The Graph、Dune Analytics、Alchemy。快，但要花钱。
举个例子，用Web3.py拉取某个地址的USDT转账记录：
from web3 import Web3
# 连接以太坊节点（这里用Infura）
w3 = Web3(Web3.HTTPProvider('https://mainnet.infura.io/v3/YOUR_PROJECT_ID'))
# USDT合约地址
usdt_contract = '0xdAC17F958D2ee523a2206206994597C13D831ec7'
# 获取最新区块号
latest_block = w3.eth.block_number
print(f"当前区块: {latest_block}")
3.3.2 清洗链上数据
链上数据有几个特点：
- 冗余字段多：每个交易都有gas、nonce、input data等。我们只关心from、to、value、block_number。
- 精度问题：ERC20代币的金额通常有18位小数。记得除以10^18。
- 重入攻击：同一个区块内可能有多个交易。按交易索引排序。
核心思路：链上数据清洗的关键是「去冗余、标准化」。把原始数据转成统一的DataFrame格式，方便后续建模。
3.4 数据整合：三流合一
拿到三类数据后，怎么整合？我画了一张图，你看一眼就明白了。
整合时，我建议以时间戳为主键。交易数据是分钟级，订单簿数据可以秒级采样，链上数据按区块时间对齐。说白了，就是把三个表按时间join起来。
3.5 实战案例：从零搭建数据管道
最后，我分享一个我在项目中用过的数据管道框架。它不复杂，但很实用。
class DataPipeline:
def __init__(self, symbols=['BTCUSDT', 'ETHUSDT']):
self.symbols = symbols
self.trade_data = {}
self.orderbook_data = {}
self.onchain_data = {}
def fetch_all(self):
"""拉取所有数据"""
for symbol in self.symbols:
# 交易数据
self.trade_data[symbol] = fetch_klines(symbol)
# 订单簿数据
bids, asks = fetch_orderbook(symbol)
self.orderbook_data[symbol] = {'bids': bids, 'asks': asks}
# 链上数据（示例：只拉USDT）
self.onchain_data['usdt'] = fetch_onchain_transfers()
def clean_all(self):
"""清洗所有数据"""
for symbol in self.symbols:
# 清洗交易数据
df = self.trade_data[symbol]
df = df.dropna()
df = df[(df['close'] > 0) & (df['volume'] > 0)]
self.trade_data[symbol] = df
# 链上数据清洗
df = self.onchain_data['usdt']
df['value'] = df['value'] / 1e18 # 精度转换
self.onchain_data['usdt'] = df
def merge(self):
"""整合数据"""
# 按时间戳合并
merged = self.trade_data['BTCUSDT'].copy()
merged['spread'] = self.orderbook_data['BTCUSDT']['asks'].iloc[0]['price'] - \
self.orderbook_data['BTCUSDT']['bids'].iloc[0]['price']
return merged
# 使用
pipeline = DataPipeline()
pipeline.fetch_all()
pipeline.clean_all()
final_data = pipeline.merge()
print(f"最终数据维度: {final_data.shape}")
避坑指南：我曾经在merge时忘了处理时间戳的时区问题。结果交易数据是UTC，链上数据是本地时间，差了8小时。模型训练出来全是错的。所以，统一时区是第一步。
好了，数据采集与清洗就讲到这里。记住：数据质量决定模型上限。花80%的时间在数据上，20%的时间在模型上，这是值得的。
公众号:蓝海资料掘金营，微信deep3321