第三章 数据获取：多资产历史行情数据的获取与清洗（含API对接）

做套利策略，数据就是你的眼睛。眼睛要是花了，策略再牛也白搭。

我刚开始做量化那会儿，就吃过数据的大亏。当时跑回测发现收益曲线漂亮得不像话，结果一查，是某只股票的复权数据没处理好，凭空多出来20%的收益。嗯，从那以后，我对数据清洗这件事就特别较真。

3.1 数据源的选择与评估

市面上数据源不少，但质量参差不齐。我个人习惯把数据源分成三类：

类型	代表	优点	缺点
专业金融终端	Wind、Bloomberg	数据全面、质量高	贵，一年几万到几十万
开源数据平台	AKShare、Tushare	免费、社区活跃	稳定性一般，有调用限制
交易所直连	CTP、FIX	实时、无中间环节	开发成本高，维护麻烦

3.2 API对接实战：以AKShare为例

AKShare是我用得比较多的开源库。它封装了国内主流交易所的数据接口，用起来很方便。咱们直接看代码：

import akshare as ak
import pandas as pd
from datetime import datetime, timedelta

# 获取期货主力连续合约数据
def fetch_futures_data(symbol, start_date, end_date):
    """
    获取期货历史行情
    symbol: 合约代码，如 'RB' 代表螺纹钢
    """
    try:
        df = ak.futures_main_sina(symbol=symbol, 
                                  start_date=start_date,
                                  end_date=end_date)
        return df
    except Exception as e:
        print(f"数据获取失败: {e}")
        return None

# 获取股票日线数据
def fetch_stock_data(stock_code, start_date, end_date):
    """
    获取A股日线数据
    stock_code: 如 '000001' 代表平安银行
    """
    df = ak.stock_zh_a_hist(symbol=stock_code,
                            period="daily",
                            start_date=start_date,
                            end_date=end_date,
                            adjust="qfq")  # 前复权
    return df

# 实战调用
if __name__ == "__main__":
    # 获取螺纹钢和热卷的数据，用于跨品种套利
    rb_data = fetch_futures_data("RB", "20230101", "20231231")
    hc_data = fetch_futures_data("HC", "20230101", "20231231")
    
    print(f"螺纹钢数据量: {len(rb_data)} 条")
    print(f"热卷数据量: {len(hc_data)} 条")

3.3 数据清洗：从原始数据到可用数据

原始数据拿回来，基本不能直接用。你想想看，交易所的数据里什么幺蛾子都有：空值、重复值、异常跳空……

我总结了一套清洗流程，咱们一步步来：

3.3.1 缺失值处理

def handle_missing_data(df):
    """
    处理缺失值
    """
    # 检查缺失情况
    missing_count = df.isnull().sum()
    print(f"缺失值统计:\n{missing_count}")
    
    # 前向填充（用上一个有效值填充）
    df = df.fillna(method='ffill')
    
    # 如果开头还有缺失，用后向填充
    df = df.fillna(method='bfill')
    
    return df

3.3.2 异常值检测

套利交易最怕的就是异常跳空。我一般用两种方法：

• 标准差法： 超过3倍标准差的数据直接剔除
• 百分位法： 低于1%或高于99%分位数的数据标记为异常

def detect_outliers(df, column='close', method='std'):
    """
    异常值检测
    """
    if method == 'std':
        mean = df[column].mean()
        std = df[column].std()
        upper = mean + 3 * std
        lower = mean - 3 * std
    elif method == 'percentile':
        upper = df[column].quantile(0.99)
        lower = df[column].quantile(0.01)
    
    outliers = df[(df[column] > upper) | (df[column] < lower)]
    print(f"检测到 {len(outliers)} 个异常值")
    
    return outliers

3.3.3 时间对齐

做套利，时间对齐是重中之重。不同品种的交易时间可能不一样，比如股指期货和ETF的收盘时间就差15分钟。

def align_time_series(df1, df2, freq='1min'):
    """
    将两个时间序列对齐到相同的时间频率
    """
    # 设置时间索引
    df1.set_index('datetime', inplace=True)
    df2.set_index('datetime', inplace=True)
    
    # 重采样到统一频率
    df1_resampled = df1.resample(freq).last()
    df2_resampled = df2.resample(freq).last()
    
    # 合并，只保留共同的时间段
    aligned = pd.merge(df1_resampled, df2_resampled, 
                       left_index=True, right_index=True,
                       suffixes=('_asset1', '_asset2'))
    
    # 删除任何一方缺失的时间点
    aligned.dropna(inplace=True)
    
    return aligned

3.4 数据存储与版本管理

数据拿到手、洗干净了，怎么存也是个学问。我踩过的坑是：一开始用CSV存，结果数据量大了之后，读取速度慢得让人抓狂。

现在我的方案是分层存储：

层级	存储方式	用途
原始层	Parquet + 压缩	保存原始API数据，方便回溯
清洗层	HDF5	清洗后的数据，用于日常分析
特征层	内存/Redis	计算好的特征，供策略实时调用

3.5 数据质量监控

数据不是洗一次就完事了。每天跑策略之前，我都会跑一遍质量检查：

def data_quality_check(df):
    """
    每日数据质量检查
    """
    issues = []
    
    # 检查数据量
    if len(df) < expected_count:
        issues.append(f"数据量不足: 期望{expected_count}, 实际{len(df)}")
    
    # 检查时间连续性
    time_diff = df.index.to_series().diff()
    if time_diff.max() > pd.Timedelta(minutes=5):
        issues.append("存在时间断点")
    
    # 检查价格合理性
    if df['close'].min() <= 0:
        issues.append("存在非正价格")
    
    if issues:
        print("数据质量问题:")
        for issue in issues:
            print(f"  - {issue}")
        return False
    else:
        print("数据质量检查通过")
        return True

3.6 数据获取与清洗的整体流程

说了这么多，咱们用一张图把整个流程串起来：

这张图就是我每天跑数据的标准流程。说白了，数据工作就是个体力活，但体力活做扎实了，策略才能站得住脚。