4、数据获取与清洗：K线数据（OHLCV）的获取与存储，处理缺失值与异常值，时间序列数据的重采样与对齐，数据本地化（CSV/SQLite）
做量化交易，数据就是你的弹药库。弹药不行，再好的策略也白搭。
我个人习惯把数据获取和清洗称为「脏活累活」，但恰恰是这步决定了策略的生死。你想想看，如果K线数据里藏着几个异常值，回测时看着收益曲线漂亮，实盘一跑直接亏到怀疑人生——我在项目中遇到过不止一次这种惨案。
今天咱们就把这块彻底聊透。从交易所拉数据，到清洗、对齐、存本地，一条龙讲清楚。
4.1 K线数据的获取
K线数据，也叫OHLCV——Open（开盘价）、High（最高价）、Low（最低价）、Close（收盘价）、Volume（成交量）。这是最基础也是最核心的数据。
大部分交易所都提供REST API，比如币安的接口：
import requests
import pandas as pd
def fetch_klines(symbol='BTCUSDT', interval='1h', 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()
# 解析成DataFrame
df = pd.DataFrame(data, columns=[
'timestamp', 'open', 'high', 'low', 'close', 'volume',
'close_time', 'quote_asset_volume', 'number_of_trades',
'taker_buy_base_asset_volume', 'taker_buy_quote_asset_volume', 'ignore'
])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
df = df[['open', 'high', 'low', 'close', 'volume']].astype(float)
return df
# 获取最近1000根1小时K线
btc_1h = fetch_klines('BTCUSDT', '1h', 1000)
print(btc_1h.head())
4.2 数据存储：CSV还是SQLite？
数据拿到手，得存起来。两种主流方案：CSV和SQLite。我两个都用，看场景。
| 方案 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| CSV | 简单、可读性强、Excel能打开 | 大文件慢、不支持并发 | 小数据集、快速验证 |
| SQLite | 查询快、支持SQL、支持索引 | 需要写SQL、稍微复杂 | 大数据集、频繁查询 |
我个人习惯：调试阶段用CSV，正式跑策略用SQLite。为什么呢？因为CSV改起来方便，我经常在Excel里手动检查数据。但一旦数据量超过几十万行，CSV就卡得不行了。
存CSV很简单：
df.to_csv('btc_1h.csv')
存SQLite也不复杂：
import sqlite3
conn = sqlite3.connect('crypto_data.db')
df.to_sql('btcusdt_1h', conn, if_exists='replace', index=True)
conn.close()
4.3 处理缺失值与异常值
真实数据永远不干净。交易所偶尔会漏数据，网络波动也可能导致请求失败。缺失值和异常值必须处理。
4.3.1 缺失值处理
K线数据缺失通常表现为某根K线的时间戳跳过了。比如1小时K线，正常应该是整点，但某天13:00的数据没了。
# 检查缺失值
print(df.isnull().sum())
# 方法1：向前填充（用上一根K线的值）
df.fillna(method='ffill', inplace=True)
# 方法2：线性插值
df.interpolate(method='linear', inplace=True)
# 方法3：直接删除（如果缺失不多）
df.dropna(inplace=True)
我一般用向前填充。为什么？因为K线数据有连续性，上一根的价格通常和下一根接近。线性插值在剧烈波动时反而会引入偏差——我曾经吃过这个亏。
4.3.2 异常值处理
异常值更隐蔽。比如某根K线的最高价突然比前后高10倍，这明显是数据错误。
# 用Z-score检测异常值
from scipy import stats
import numpy as np
z_scores = np.abs(stats.zscore(df['close']))
outliers = df[z_scores > 3] # 超过3个标准差
print(f'发现 {len(outliers)} 个异常值')
# 处理：用前后均值替换
for idx in outliers.index:
prev_val = df.loc[:idx].iloc[-2]['close']
next_val = df.loc[idx:].iloc[1]['close']
df.loc[idx, 'close'] = (prev_val + next_val) / 2
4.4 时间序列数据的重采样与对齐
不同策略需要不同周期的K线。你可能从交易所拿到1分钟数据，但策略需要15分钟K线。这时候就要重采样。
# 将1分钟数据重采样为15分钟
df_15min = df.resample('15min').agg({
'open': 'first',
'high': 'max',
'low': 'min',
'close': 'last',
'volume': 'sum'
})
对齐问题更常见。比如你要同时交易BTC和ETH，它们的K线时间戳可能不完全对齐——交易所的时钟有微小差异。
# 对齐两个时间序列
btc = fetch_klines('BTCUSDT', '1h')
eth = fetch_klines('ETHUSDT', '1h')
# 合并并向前填充
combined = pd.merge(btc, eth, left_index=True, right_index=True,
suffixes=('_btc', '_eth'), how='outer')
combined.fillna(method='ffill', inplace=True)
combined.dropna(inplace=True)
嗯，这里要注意：对齐时用outer join，确保不丢数据。然后向前填充，因为缺失的K线用上一根的值是合理的。
4.5 数据本地化：构建你的数据仓库
每次跑策略都去交易所拉数据？太慢了，而且容易被限流。我建议建立本地数据仓库。
我的做法是这样的：
- 写一个定时脚本，每天凌晨拉取最新数据
- 存到SQLite，按交易对和周期分表
- 策略直接从本地读数据
def update_local_db(symbol='BTCUSDT', interval='1h'):
conn = sqlite3.connect('crypto_data.db')
# 获取数据库中最新时间戳
try:
last_ts = pd.read_sql(f'SELECT MAX(timestamp) FROM {symbol}_{interval}', conn).iloc[0,0]
except:
last_ts = None
# 拉取新数据
if last_ts:
start_time = int(pd.Timestamp(last_ts).timestamp() * 1000) + 1
new_data = fetch_klines(symbol, interval, limit=1000, start_time=start_time)
else:
new_data = fetch_klines(symbol, interval, limit=1000)
# 追加到数据库
new_data.to_sql(f'{symbol}_{interval}', conn, if_exists='append', index=True)
conn.close()
print(f'更新完成，新增 {len(new_data)} 条数据')
4.6 本章知识体系
下面这张图总结了数据获取与清洗的完整流程，我画了很久才理清楚：
这张图把整个流程串起来了。从交易所拉数据，存到本地，清洗掉脏数据，重采样到目标周期，对齐多个交易对，最后构建本地数据仓库。每一步都踩过坑，每一步都有优化空间。
数据这块搞定了，后面写策略心里才有底。记住一句话：数据质量决定策略上限。别急着跑回测，先把数据整干净。
公众号:蓝海资料掘金营，微信deep3321