第三章 数据获取与预处理：Level-2行情数据的清洗与对齐

各位同学，欢迎来到实战课的第三讲。

上一章我们聊了集合竞价和连续竞价的基本规则，说白了就是搞清楚了价差是怎么产生的。但光知道原理没用，你得有数据才能干活。这一章，我们就来搞定数据获取和预处理——这是整个套利策略的地基。

我个人习惯把数据工作分成三步：拿到数据 → 洗干净 → 对齐时间。每一步都有坑，我踩过不少，今天一并分享给你。

3.1 Level-2行情数据：到底长什么样？

先说说Level-2行情。普通行情是3秒快照，Level-2是逐笔成交，频率高得多。对于集合竞价和连续竞价的价差套利，我们至少需要以下字段：

字段名	说明	重要性
时间戳	精确到毫秒的成交时间	★★★★★
股票代码	6位数字代码	★★★★★
成交价格	该笔成交的价位	★★★★★
成交数量	该笔成交的股数	★★★★
买卖方向	主动买还是主动卖	★★★★
交易阶段	集合竞价/连续竞价/收盘	★★★★★

嗯，这里要注意：交易阶段这个字段很多数据源不直接提供。我曾经用过一个第三方数据接口，它把集合竞价和连续竞价的数据混在一起，没有标记。结果我算价差时，把9:25的集合竞价成交和9:30的连续竞价成交混为一谈，算出来的价差完全失真。

3.2 数据获取：从哪儿拿？怎么拿？

我个人常用的数据源有两个：一是券商提供的量化接口（比如中泰XTP、华泰MATIC），二是第三方数据服务商（比如万得、聚宽）。

这里我以聚宽为例，展示一段获取Level-2数据的代码。注意，这只是示例，实际使用时你需要根据接口文档调整参数。

import jqdatasdk as jq
import pandas as pd

# 登录聚宽（需要账号）
jq.auth('your_username', 'your_password')

# 获取某只股票某一天的Level-2逐笔成交数据
stock_code = '000001.XSHE'  # 平安银行
date = '2024-01-15'

# 获取逐笔成交
tick_data = jq.get_ticks(
    security=stock_code,
    start_date=date,
    end_date=date,
    fields=['time', 'price', 'volume', 'direction', 'type']
)

# 看看数据长什么样
print(tick_data.head())

跑完这段代码，你会得到一个DataFrame。每一行是一笔成交。但问题来了——这个数据里没有“交易阶段”字段。怎么办？

我的做法是：根据时间戳手动打标签。9:15到9:25之间的数据标记为“集合竞价”，9:30之后标记为“连续竞价”。

# 添加交易阶段标签
def label_trading_stage(time_str):
    # 假设time_str格式为 '09:25:00.123'
    hour = int(time_str[:2])
    minute = int(time_str[3:5])
    
    if (hour == 9 and minute >= 15 and minute <= 25) or (hour == 9 and minute == 25):
        return 'call_auction'  # 集合竞价
    elif hour >= 9 and hour < 15:
        return 'continuous_auction'  # 连续竞价
    else:
        return 'other'

tick_data['stage'] = tick_data['time'].apply(label_trading_stage)

3.3 数据清洗：脏数据是最大的敌人

数据拿到手，别急着算价差。先洗一洗。我见过太多人直接拿原始数据跑策略，结果被一根异常K线坑得血本无归。

常见的脏数据有这几类：

• 重复数据：同一笔成交被记录两次。检查方法：按时间戳和价格去重。
• 缺失数据：某些时间点没有成交记录。比如集合竞价期间可能只有几笔成交。
• 异常价格：价格明显偏离正常范围。比如某股票正常价10元，突然出现一笔100元的成交。
• 时间错乱：时间戳顺序不对，后面的成交时间比前面的还早。

我曾经遇到过一个数据源，它在9:25:00.000这个时间点重复发送了3笔相同的成交记录。如果不做去重，集合竞价的价格就会被重复计算，价差直接偏掉。

下面是我的清洗流程：

def clean_tick_data(df):
    # 1. 去重：按时间戳和价格去重
    df = df.drop_duplicates(subset=['time', 'price', 'volume'])
    
    # 2. 排序：按时间戳升序排列
    df = df.sort_values('time')
    
    # 3. 过滤异常价格：比如剔除价格超过3倍标准差的数据
    mean_price = df['price'].mean()
    std_price = df['price'].std()
    df = df[(df['price'] > mean_price - 3*std_price) & 
            (df['price'] < mean_price + 3*std_price)]
    
    # 4. 检查时间连续性：如果两笔成交间隔超过5分钟，标记一下
    df['time_diff'] = df['time'].diff().dt.total_seconds()
    df['gap_flag'] = df['time_diff'] > 300  # 5分钟
    
    return df

tick_data_clean = clean_tick_data(tick_data)

3.4 数据对齐：让集合竞价和连续竞价“对上话”

数据洗干净了，接下来要对齐。为什么要对齐？因为集合竞价和连续竞价的时间粒度不一样。

集合竞价：9:15到9:25，总共10分钟，可能只有几十笔成交。连续竞价：9:30之后，每秒可能几十笔。如果你直接拿原始数据算价差，时间点对不上。

我的做法是：统一到1秒的时间轴上。把集合竞价的数据按秒聚合，连续竞价的数据也按秒聚合，然后按时间对齐。

# 按秒聚合
def aggregate_by_second(df):
    # 把时间戳截断到秒
    df['second'] = df['time'].dt.floor('S')
    
    # 按秒计算平均价格和总成交量
    agg_df = df.groupby(['second', 'stage']).agg(
        avg_price=('price', 'mean'),
        total_volume=('volume', 'sum'),
        trade_count=('price', 'count')
    ).reset_index()
    
    return agg_df

tick_agg = aggregate_by_second(tick_data_clean)

# 分离集合竞价和连续竞价
call_df = tick_agg[tick_agg['stage'] == 'call_auction']
cont_df = tick_agg[tick_agg['stage'] == 'continuous_auction']

# 按时间对齐
aligned_df = pd.merge(
    call_df[['second', 'avg_price']].rename(columns={'avg_price': 'call_price'}),
    cont_df[['second', 'avg_price']].rename(columns={'avg_price': 'cont_price'}),
    on='second',
    how='outer'
)

# 计算价差
aligned_df['price_diff'] = aligned_df['cont_price'] - aligned_df['call_price']

你看，这样对齐之后，每一秒都有一个集合竞价价格和一个连续竞价价格，价差就自然算出来了。

3.5 核心逻辑：一张图看懂数据流程

说了这么多，我画了一张流程图，把整个数据获取和预处理的核心逻辑串起来。你一看就明白。

3.6 实战中的几个坑

最后，分享几个我实战中踩过的坑，你遇到了能少走弯路。

• 坑一：集合竞价数据太少。 有些股票在集合竞价期间只有几笔成交，甚至没有成交。这时候算出来的价差没有统计意义。我一般要求集合竞价期间至少有5笔成交，否则跳过这只股票。
• 坑二：时间戳精度不一致。 不同数据源的时间戳精度不一样，有的精确到毫秒，有的精确到秒。对齐前一定要统一精度。
• 坑三：复权问题。 如果你拿的是历史数据，注意除权除息。集合竞价和连续竞价的价格都要用复权后的价格，否则价差会失真。

好了，这一章就到这里。数据准备好了，下一章我们就可以开始真正的价差分析和策略设计了。

---