第二章：数据采集与清洗

说实话，做量化交易最怕什么？不是策略亏钱，而是数据有问题。我见过太多人花几个月写策略，最后发现是数据源有bug，白白浪费时间。这一章，咱们就把数据这块地基打牢。

2.1 交易所API对接

先聊聊API对接。你想想看，交易所就是你的数据供应商，API就是水管。水管接不好，后面全是脏数据。

我个人习惯用CCXT这个库，它封装了上百家交易所的API。为什么推荐它？因为不用每家交易所单独写对接代码，省心。

import ccxt
import pandas as pd
from datetime import datetime

# 初始化交易所
exchange = ccxt.binance({
    'apiKey': '你的API_KEY',
    'secret': '你的SECRET_KEY',
    'enableRateLimit': True,  # 限速，别被交易所封了
    'options': {'defaultType': 'spot'}  # 现货交易
})

# 获取当前BTC/USDT的ticker
ticker = exchange.fetch_ticker('BTC/USDT')
print(f"当前价格: {ticker['last']}")

注意：API Key千万别硬编码在代码里。我曾经见过有人把Key传到GitHub上，结果被机器人扫到，账户里的币全被盗了。用环境变量或者配置文件来管理。

2.2 历史数据获取

历史数据是策略回测的基础。没有历史数据，你拿什么验证策略？

获取历史数据时，有几个坑要避开：

  • 时间戳问题：交易所返回的时间戳通常是毫秒级的Unix时间戳，记得转成可读格式
  • 数据完整性：有些交易所会返回空值或缺失数据，需要做校验
  • 限频问题：别一次性拉太多数据，会被封IP
def fetch_historical_data(symbol, timeframe='1h', limit=1000):
    """
    获取历史K线数据
    """
    since = exchange.parse8601('2024-01-01T00:00:00Z')
    all_ohlcv = []
    
    while True:
        ohlcv = exchange.fetch_ohlcv(
            symbol, 
            timeframe=timeframe,
            since=since,
            limit=limit
        )
        
        if len(ohlcv) == 0:
            break
            
        all_ohlcv.extend(ohlcv)
        since = ohlcv[-1][0] + 1  # 避免重复数据
        
        # 加个延时，别太猛
        exchange.sleep(1000)
    
    # 转成DataFrame
    df = pd.DataFrame(
        all_ohlcv, 
        columns=['timestamp', 'open', 'high', 'low', 'close', 'volume']
    )
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df.set_index('timestamp', inplace=True)
    
    return df

# 获取BTC 1小时K线
btc_data = fetch_historical_data('BTC/USDT', '1h', 500)
print(btc_data.head())

小技巧：数据获取建议分批进行，每批500-1000根K线，中间sleep 1-2秒。这样既不会触发限频，也能保证数据完整。

2.3 数据清洗与预处理

数据拿到手了，但别急着用。原始数据就像刚从菜市场买回来的菜，得洗一洗、切一切才能下锅。

我在项目中遇到过最头疼的问题：某交易所的API偶尔会返回重复的K线数据，导致回测结果完全失真。所以数据清洗这步，绝对不能省。

常见的数据问题

问题类型 表现 处理方法
缺失值 某根K线的价格或成交量为NaN 前向填充或插值
重复数据 同一时间戳出现多条记录 去重，保留第一条
异常值 价格突然跳变100倍 用Z-score或IQR检测并剔除
时间戳错乱 数据顺序颠倒 按时间戳重新排序 
def clean_ohlcv_data(df):
    """
    清洗K线数据
    """
    # 1. 去重
    df = df[~df.index.duplicated(keep='first')]
    
    # 2. 排序
    df.sort_index(inplace=True)
    
    # 3. 处理缺失值
    df.fillna(method='ffill', inplace=True)
    
    # 4. 检测异常值（用Z-score）
    from scipy import stats
    z_scores = stats.zscore(df['close'])
    outliers = abs(z_scores) > 3
    df = df[~outliers]
    
    # 5. 检查数据连续性
    time_diff = df.index.to_series().diff()
    expected_diff = pd.Timedelta(hours=1)  # 1小时K线
    
    gaps = time_diff[time_diff > expected_diff]
    if len(gaps) > 0:
        print(f"发现 {len(gaps)} 处数据缺失")
        # 这里可以补全缺失的时间段
    
    return df

# 清洗数据
clean_data = clean_ohlcv_data(btc_data)
print(f"清洗前: {len(btc_data)} 条, 清洗后: {len(clean_data)} 条")

核心原则：数据清洗宁可多花时间，也不要放过任何可疑数据。一条脏数据，可能让你的策略回测结果偏差10%以上。

2.4 特征工程入门

特征工程说白了，就是把原始数据变成模型能理解的语言。原始K线只有开高低收和成交量，但模型需要更多信息才能做出判断。

嗯，这里要注意：特征不是越多越好。我见过有人一口气造了200个特征，结果模型过拟合得一塌糊涂。好的特征工程，讲究的是「少而精」。

常用技术指标特征

def create_features(df):
    """
    创建基础特征
    """
    # 复制数据，避免修改原数据
    data = df.copy()
    
    # 1. 收益率
    data['returns'] = data['close'].pct_change()
    
    # 2. 波动率（20周期）
    data['volatility'] = data['returns'].rolling(20).std()
    
    # 3. 成交量变化
    data['volume_ratio'] = data['volume'] / data['volume'].rolling(20).mean()
    
    # 4. 价格位置（当前价格在20周期内的位置）
    data['price_position'] = (
        (data['close'] - data['low'].rolling(20).min()) /
        (data['high'].rolling(20).max() - data['low'].rolling(20).min())
    )
    
    # 5. 简单移动平均线
    data['sma_20'] = data['close'].rolling(20).mean()
    data['sma_50'] = data['close'].rolling(50).mean()
    
    # 6. 价格与均线的偏离度
    data['sma_20_distance'] = (data['close'] - data['sma_20']) / data['sma_20']
    
    # 删除NaN值
    data.dropna(inplace=True)
    
    return data

# 创建特征
feature_data = create_features(clean_data)
print(feature_data[['close', 'returns', 'volatility', 'price_position']].head())

特征选择的原则

  • 相关性不要太高：比如SMA20和SMA50相关性很高，选一个就行
  • 要有业务含义：每个特征都要能解释为什么对预测有帮助
  • 稳定性要好：在不同市场环境下表现一致
  • 计算成本要低：别搞太复杂的特征，影响实盘速度

我的经验：刚开始做特征工程时，先造10-15个核心特征就够了。等模型跑通了，再慢慢加特征。这样能快速验证思路，避免一开始就陷入特征工程的泥潭。

2.5 数据流水线搭建

最后，咱们把整个流程串起来。一个完整的数据流水线应该包含：

  1. 数据获取：定时从交易所拉取最新数据
  2. 数据清洗：去重、补全、异常检测
  3. 特征计算：生成模型需要的特征
  4. 数据存储：保存到本地或数据库
  5. 数据校验：检查数据质量，发现问题及时告警

我曾经因为偷懒，没做数据校验这一步，结果策略跑了一个月才发现数据源出了问题。那一个月的时间全白费了。所以，数据校验一定要自动化，每天跑一遍。

def data_pipeline(symbol, timeframe='1h'):
    """
    完整的数据流水线
    """
    print(f"开始处理 {symbol} {timeframe} 数据...")
    
    # 1. 获取数据
    raw_data = fetch_historical_data(symbol, timeframe)
    
    # 2. 清洗数据
    clean_data = clean_ohlcv_data(raw_data)
    
    # 3. 创建特征
    feature_data = create_features(clean_data)
    
    # 4. 保存数据
    filename = f"{symbol.replace('/', '_')}_{timeframe}.csv"
    feature_data.to_csv(filename)
    print(f"数据已保存到 {filename}")
    
    # 5. 数据校验
    print(f"数据范围: {feature_data.index[0]} 到 {feature_data.index[-1]}")
    print(f"总数据量: {len(feature_data)} 条")
    print(f"缺失值: {feature_data.isnull().sum().sum()}")
    
    return feature_data

# 运行流水线
btc_features = data_pipeline('BTC/USDT', '1h')

记住：数据流水线要设计成可重复执行的。每次运行都产生相同的结果，这样才能保证回测的可复现性。

好了，数据采集与清洗这块就聊到这儿。这些基础工作虽然枯燥，但绝对是整个量化交易系统最关键的环节。数据质量决定了策略的天花板，再牛的模型也救不了垃圾数据。

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

← 上一章 📖 返回目录 下一章 →