4、数据获取与存储：K线数据获取、数据清洗与格式化、使用SQLite/CSV存储数据

做量化交易，数据就是你的弹药库。没有干净、可靠的数据，再牛的策略也是空中楼阁。这一章，我们就来聊聊怎么把交易所的K线数据搞到手，再把它收拾得服服帖帖，最后存进咱们的仓库里。

4.1 K线数据获取：从交易所拉数据

我个人习惯用CCXT这个库，它把各大交易所的API都封装好了。你写一套代码，就能对接币安、OKX、火币等等。省心。

先看一个最简单的例子，拉取BTC/USDT的日线数据：

import ccxt
import pandas as pd
from datetime import datetime

# 初始化交易所对象
exchange = ccxt.binance()

# 获取K线数据
symbol = 'BTC/USDT'
timeframe = '1d'  # 日线
limit = 100       # 最近100根

ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)

# 转成DataFrame
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')

print(df.head())

嗯，这里要注意。CCXT返回的时间戳是毫秒级的，得转成我们看得懂的日期格式。我在项目中遇到过有人忘了这步，结果时间轴全乱了，回测数据对不上，排查了半天。

4.2 数据清洗与格式化：把脏数据洗干净

交易所返回的数据，说实话，没那么干净。偶尔会有空值、重复数据，甚至时间戳错乱。你想想看，如果策略基于这些脏数据下单，后果不堪设想。

我一般会做这几步清洗：

1. 去重：按时间戳去重，保留第一条。
2. 填充空值：用前向填充法，或者直接删除。
3. 排序：确保时间戳是升序的。
4. 检查异常值：比如价格突然变成0或者负数。

看代码：

def clean_ohlcv(df):
    # 1. 去重
    df = df.drop_duplicates(subset='timestamp', keep='first')

    # 2. 排序
    df = df.sort_values('timestamp').reset_index(drop=True)

    # 3. 检查空值
    if df.isnull().values.any():
        # 前向填充
        df = df.fillna(method='ffill')

    # 4. 检查异常价格
    df = df[(df['close'] > 0) & (df['high'] >= df['low'])]

    return df

df_clean = clean_ohlcv(df)

我曾经在清洗这一步栽过跟头。有一次，币安的某个交易对突然返回了一根成交量是0的K线，我的策略没过滤掉，结果回测里出现了诡异的滑点。从那以后，我加上了成交量必须大于0的检查。

4.3 数据存储：SQLite vs CSV

数据洗干净了，得找个地方存起来。我个人偏好两种方式：

• CSV：轻量、方便，适合小规模数据或快速调试。
• SQLite：支持SQL查询，适合中等规模数据，不需要额外装数据库。

先看CSV存储：

df_clean.to_csv('btc_usdt_1d.csv', index=False)

简单粗暴。但CSV有个问题——你没法按时间范围快速查询。每次都得全量读进来再过滤。

所以，我建议用SQLite。它就是个单文件数据库，但支持SQL，查询效率高得多。

import sqlite3

# 连接数据库（文件不存在会自动创建）
conn = sqlite3.connect('crypto_data.db')
cursor = conn.cursor()

# 创建表
cursor.execute('''
    CREATE TABLE IF NOT EXISTS kline_1d (
        timestamp TEXT PRIMARY KEY,
        open REAL,
        high REAL,
        low REAL,
        close REAL,
        volume REAL
    )
''')

# 写入数据
df_clean.to_sql('kline_1d', conn, if_exists='append', index=False)

# 查询示例
query = "SELECT * FROM kline_1d WHERE timestamp > '2024-01-01'"
result = pd.read_sql(query, conn)

conn.close()

4.4 本章知识体系

下面这张图，帮你理清数据获取与存储的完整流程：

说白了，数据获取与存储就是量化交易的「地基」。地基不稳，上面盖的楼再漂亮也得塌。我个人建议，哪怕你只是做回测，也把数据存到SQLite里。养成好习惯，后面实盘部署时能省不少事。

好了，数据已经到手，也存好了。接下来，我们就可以基于这些数据，开始构建策略了。