第三章：数据获取——从交易所API到本地存储

做量化挖矿策略，第一步就是搞数据。没有数据，你连策略长什么样都看不到。

我个人习惯把数据获取分成三步：拉取、清洗、存储。每一步都有坑，我踩过不少，今天一并告诉你。

3.1 从交易所API获取历史K线数据

大部分交易所都提供RESTful API。以币安为例，它的K线接口长这样：

import requests
import pandas as pd
from datetime import datetime

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=[
        'open_time', '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['open_time'] = pd.to_datetime(df['open_time'], unit='ms')
    df['close_time'] = pd.to_datetime(df['close_time'], unit='ms')
    
    # 只保留核心字段
    df = df[['open_time', 'open', 'high', 'low', 'close', 'volume']]
    
    # 数值列转float
    for col in ['open', 'high', 'low', 'close', 'volume']:
        df[col] = df[col].astype(float)
    
    return df

# 测试一下
df_1h = fetch_klines('BTCUSDT', '1h', 500)
print(df_1h.head())

这里有个细节：limit最大是1000。你想拿更多数据怎么办？循环请求，每次用上次的结束时间作为下次的开始时间。

3.2 数据清洗与预处理

拿到的数据，说白了就是原始矿料。你得筛一遍才能用。

常见问题有三个：

• 缺失值：某根K线数据没返回
• 异常值：价格突然跳变，比如从10000跳到0.01
• 重复数据：同一个时间戳出现两次

我的清洗流程是这样的：

def clean_klines(df):
    # 1. 去重
    df = df.drop_duplicates(subset='open_time')
    
    # 2. 排序
    df = df.sort_values('open_time')
    
    # 3. 检查缺失时间戳
    # 生成完整的时间序列
    full_range = pd.date_range(
        start=df['open_time'].min(),
        end=df['open_time'].max(),
        freq='1h'
    )
    missing_times = full_range.difference(df['open_time'])
    if len(missing_times) > 0:
        print(f'发现{len(missing_times)}个缺失时间戳')
        # 用前向填充
        df = df.set_index('open_time').reindex(full_range, method='ffill').reset_index()
        df.rename(columns={'index': 'open_time'}, inplace=True)
    
    # 4. 异常值检测
    # 价格不能为0或负数
    for col in ['open', 'high', 'low', 'close']:
        df = df[df[col] > 0]
    
    # 5. 重置索引
    df = df.reset_index(drop=True)
    
    return df

df_clean = clean_klines(df_1h)
print(f'清洗前{len(df_1h)}行，清洗后{len(df_clean)}行')

3.3 数据存储：CSV vs HDF5

数据清洗完了，得存起来。两种主流方案：CSV和HDF5。

特性	CSV	HDF5
可读性	高，记事本就能打开	低，需要专用工具
存储大小	大，文本格式	小，二进制压缩
读写速度	慢，尤其大文件	快，支持索引
跨平台	极好	好，但依赖库
适合场景	小数据量、调试	大数据量、生产环境

我个人建议：调试用CSV，正式跑用HDF5。

CSV存储很简单：

df_clean.to_csv('btcusdt_1h.csv', index=False)
# 读取
df = pd.read_csv('btcusdt_1h.csv', parse_dates=['open_time'])

HDF5稍微复杂一点，但性能好很多：

# 存储
df_clean.to_hdf('crypto_data.h5', key='btcusdt_1h', mode='a', complevel=9)

# 读取
df = pd.read_hdf('crypto_data.h5', key='btcusdt_1h')

# 支持条件查询
df_filter = pd.read_hdf('crypto_data.h5', key='btcusdt_1h',
                         where='open_time > "2024-01-01"')

3.4 完整的数据获取流程

把上面三步串起来，就是一个完整的数据管道：

def data_pipeline(symbol='BTCUSDT', interval='1h', start_date='2023-01-01', end_date='2024-12-31'):
    # 1. 计算需要多少根K线
    total_hours = int((pd.to_datetime(end_date) - pd.to_datetime(start_date)).total_seconds() / 3600)
    
    # 2. 分批获取
    all_data = []
    current_start = pd.to_datetime(start_date)
    
    while current_start < pd.to_datetime(end_date):
        # 每次最多拿1000根
        limit = min(1000, total_hours)
        df_batch = fetch_klines(symbol, interval, limit)
        all_data.append(df_batch)
        
        # 更新开始时间
        current_start = df_batch['open_time'].iloc[-1] + pd.Timedelta(hours=1)
        total_hours -= limit
        
        # 礼貌性等待
        time.sleep(0.1)
    
    # 3. 合并
    df = pd.concat(all_data, ignore_index=True)
    
    # 4. 清洗
    df = clean_klines(df)
    
    # 5. 存储
    df.to_hdf(f'{symbol}_{interval}.h5', key='data', mode='w', complevel=9)
    
    return df

# 执行
df_final = data_pipeline('BTCUSDT', '1h', '2023-01-01', '2024-06-30')
print(f'共获取{len(df_final)}根K线')

3.5 本章知识体系

下面这张图，把数据获取的整个流程串起来了：

嗯，到这里数据获取这块就讲完了。记住一句话：数据是策略的命根子。API拉取、清洗、存储，每一步都马虎不得。我见过太多人策略写得漂亮，结果数据源有问题，回测结果全是假的——那还不如不跑。

下一章，我们会用这些数据来做真正的回测。到时候你就知道，今天花时间把数据搞干净，绝对值。

---

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