4、数据获取与清洗：交易所API对接、历史数据下载、数据格式标准化、缺失值处理
做市策略回测，数据是地基。地基不稳，楼盖得再漂亮也得塌。
我个人习惯把数据获取和清洗看作整个回测系统里最枯燥、但最不能偷懒的环节。你想想看，策略写得再花哨，如果喂进去的数据是脏的，回测结果就是自欺欺人。我在项目中遇到过好几次，因为某根K线的开盘价多了一个小数点，导致整个回测盈亏翻了个个儿。嗯，从那以后，我对数据清洗的敬畏心就拉满了。
4.1 交易所API对接：拿到第一手数据
做市策略对数据时效性要求极高。我们通常直接从交易所拉数据，而不是从第三方聚合平台拿。为什么？因为第三方数据可能有延迟，或者经过了某种「平滑处理」，这对做市策略来说是致命的。
我常用的方式是REST API + WebSocket双通道。REST用来拉历史数据，WebSocket用来接实时行情。这里以币安交易所为例，展示一下最基础的K线数据获取。
import requests
import pandas as pd
def fetch_binance_klines(symbol='BTCUSDT', interval='1m', limit=1000):
url = 'https://api.binance.com/api/v3/klines'
params = {
'symbol': symbol,
'interval': interval,
'limit': limit
}
resp = requests.get(url, params=params)
data = resp.json()
# 币安返回的数据结构： [时间, 开, 高, 低, 收, 量, ...]
df = pd.DataFrame(data, columns=[
'timestamp', 'open', 'high', 'low', 'close', 'volume',
'close_time', 'quote_asset_volume', 'number_of_trades',
'taker_buy_base_vol', 'taker_buy_quote_vol', 'ignore'
])
return df
# 调用示例
df_klines = fetch_binance_klines('BTCUSDT', '1m', 500)
print(df_klines.head())
4.2 历史数据下载：别只拉最近1000根
做市策略回测通常需要几个月甚至一年的数据。但交易所REST接口一次最多返回1000根K线。怎么办？循环拉取，按时间戳拼接。
我习惯用「分页+时间戳」的方式。每次拉取时，把上一批数据的最后一条时间戳作为下一次请求的startTime。这样就能无缝拼接出完整的历史序列。
def fetch_historical_klines(symbol, interval, start_time, end_time):
all_data = []
current_start = start_time
while current_start < end_time:
url = 'https://api.binance.com/api/v3/klines'
params = {
'symbol': symbol,
'interval': interval,
'startTime': current_start,
'limit': 1000
}
resp = requests.get(url, params=params)
data = resp.json()
if not data:
break
all_data.extend(data)
# 更新起始时间：取最后一条的收盘时间 + 1ms
current_start = data[-1][6] + 1
# 礼貌性等待
time.sleep(0.3)
return all_data
4.3 数据格式标准化：统一你的数据字典
不同交易所返回的数据字段名千奇百怪。币安叫open，火币叫open_price，OKX叫o。如果不做标准化，你的策略代码会变成一团乱麻。
我定义了一套统一的数据格式，所有交易所的数据进来后都映射成这套格式。说白了，就是给数据「换衣服」。
| 标准字段名 | 含义 | 币安原始字段 | 数据类型 |
|---|---|---|---|
| timestamp | 开盘时间（毫秒） | 0 | int64 |
| open | 开盘价 | 1 | float64 |
| high | 最高价 | 2 | float64 |
| low | 最低价 | 3 | float64 |
| close | 收盘价 | 4 | float64 |
| volume | 成交量 | 5 | float64 |
def standardize_klines(raw_df, exchange='binance'):
df = raw_df.copy()
if exchange == 'binance':
df = df.iloc[:, :6] # 只取前6列
df.columns = ['timestamp', 'open', 'high', 'low', 'close', 'volume']
# 统一转为数值类型
for col in ['open', 'high', 'low', 'close', 'volume']:
df[col] = pd.to_numeric(df[col], errors='coerce')
# 时间戳转为datetime，方便后续处理
df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
return df
4.4 缺失值处理：别让NaN毁了你的回测
数据下载过程中，难免会遇到缺失值。比如交易所某个时间段宕机了，或者网络波动导致某根K线没拉下来。如果不处理，回测时策略遇到NaN会直接报错，或者产生诡异的计算结果。
我一般分三步走：
- 检查缺失：用df.isnull().sum()快速定位哪些列、哪些时间段有缺失。
- 判断缺失类型：是单点缺失，还是连续缺失？单点缺失可以用插值，连续缺失超过5根K线，我建议直接丢弃那一段。
- 填充策略：做市策略对价格敏感，我通常用前向填充（ffill）或线性插值。千万别用均值填充，那会引入未来信息。
def handle_missing_values(df, max_gap=5):
# 先检查缺失
missing_count = df.isnull().sum()
print(f"缺失统计：\n{missing_count}")
# 找到连续缺失的区间
df['gap_flag'] = df['close'].isna().astype(int)
df['gap_group'] = (df['gap_flag'] != df['gap_flag'].shift()).cumsum()
# 对连续缺失超过max_gap的区间，直接删除
gap_sizes = df[df['gap_flag']==1].groupby('gap_group').size()
large_gaps = gap_sizes[gap_sizes > max_gap].index
for g in large_gaps:
df.loc[df['gap_group'] == g, ['open','high','low','close','volume']] = None
# 删除剩余的大段缺失
df = df.dropna(subset=['open','high','low','close','volume']).reset_index(drop=True)
# 对零星缺失做前向填充
df = df.ffill()
return df
核心原则：做市策略回测中，宁可少一段数据，也不要「编造」数据。我曾经因为用插值填充了一段交易所宕机期间的数据，结果回测收益曲线异常平滑，实盘直接被打脸。真实的市场停顿，就是策略的真空期，别试图美化它。
4.5 数据清洗流程总览
我把整个数据获取与清洗的流程画成了一张图。你跟着这个流程走，基本不会出大问题。
这张图里，我特意把「数据校验」单独拎出来。很多人做完缺失值处理就直接存盘了，其实还差一步：检查价格有没有负数、时间戳有没有跳跃、成交量有没有异常大或异常小。这些脏数据，光靠缺失值检测是抓不到的。
好了，数据清洗这块就聊到这儿。记住一句话：数据干净了，策略才能跑得稳。别急着写策略逻辑，先把数据管道打磨好。这是我在无数次踩坑后换来的教训。
公众号:蓝海资料掘金营，微信deep3321