第二章 数据获取：Level-2行情数据实战

做量化交易的朋友都知道一句话：数据是交易的血液。没有高质量的数据，再牛的策略也是空中楼阁。今天我就带大家亲手抓取Level-2行情数据，把盘口快照和逐笔成交这些核心数据玩明白。

2.1 数据源选择与接口对比

我个人习惯用万得、聚宽或者通联数据。但说实话，对于个人开发者，聚宽和通联的免费额度够用了。我在项目中遇到过一个问题：某次用某平台免费接口，结果盘中突然限流，策略直接断粮。嗯，从那以后我都是备两个数据源。

数据源	Level-2支持	免费额度	延迟
聚宽	沪深Level-2	每日100万条	约500ms
通联数据	沪深Level-2	每日50万条	约300ms
万得	全市场Level-2	付费	约100ms

2.2 盘口快照数据解析

盘口快照，说白了就是某一时刻的买卖挂单情况。Level-2给你的是十档行情，比普通行情的五档多了一倍信息量。你想想看，多出来的五档能看出什么？能看出主力挂单的意图！

我曾经在分析某只股票时，发现买五档突然挂出大单，但价格没变。这就是典型的托单行为——主力在下面撑着，不让股价跌。这种信号普通五档行情根本看不到。

import pandas as pd
import numpy as np

# 模拟盘口快照数据
def get_level2_snapshot(stock_code):
    """
    获取Level-2盘口快照
    实际使用时替换为API调用
    """
    # 十档买卖盘
    bid_prices = [10.00, 9.99, 9.98, 9.97, 9.96, 9.95, 9.94, 9.93, 9.92, 9.91]
    bid_volumes = [1000, 2000, 1500, 800, 1200, 600, 900, 1100, 700, 500]
    ask_prices = [10.01, 10.02, 10.03, 10.04, 10.05, 10.06, 10.07, 10.08, 10.09, 10.10]
    ask_volumes = [800, 1500, 1000, 1200, 900, 700, 1100, 600, 800, 400]
    
    snapshot = {
        'time': pd.Timestamp.now(),
        'stock': stock_code,
        'bid_prices': bid_prices,
        'bid_volumes': bid_volumes,
        'ask_prices': ask_prices,
        'ask_volumes': ask_volumes,
        'total_bid_vol': sum(bid_volumes),
        'total_ask_vol': sum(ask_volumes)
    }
    return snapshot

# 计算买卖压力比
snapshot = get_level2_snapshot('000001')
pressure_ratio = snapshot['total_bid_vol'] / snapshot['total_ask_vol']
print(f"买卖压力比: {pressure_ratio:.2f}")
# 大于1说明买方强势，小于1说明卖方强势

2.3 逐笔成交数据清洗

逐笔成交才是Level-2的精髓。每一笔真实的成交记录，包括成交时间、价格、数量、买卖方向。为什么说它重要？因为你可以通过逐笔数据还原出主力的真实交易行为。

举个例子：某只股票突然拉涨，普通行情只告诉你涨了2%。但逐笔数据能告诉你——这2%是几笔大单拉起来的，还是无数小单堆上去的。前者是主力行为，后者可能是散户情绪。这区别大了去了。

def clean_tick_data(raw_ticks):
    """
    清洗逐笔成交数据
    我在项目中遇到过数据错乱的问题，所以加了严格的校验
    """
    df = pd.DataFrame(raw_ticks)
    
    # 1. 去除空值
    df = df.dropna(subset=['price', 'volume', 'time'])
    
    # 2. 价格合理性校验
    # 涨跌幅超过10%的视为异常数据
    df = df[df['price'].between(df['price'].mean() * 0.9, 
                                 df['price'].mean() * 1.1)]
    
    # 3. 成交量校验
    # 单笔成交量不能为0或负数
    df = df[df['volume'] > 0]
    
    # 4. 时间排序
    df = df.sort_values('time')
    
    # 5. 标记买卖方向
    # 主动买：成交价 >= 卖一价
    # 主动卖：成交价 <= 买一价
    df['trade_type'] = np.where(
        df['price'] >= df['ask1'], '主动买',
        np.where(df['price'] <= df['bid1'], '主动卖', '中性')
    )
    
    return df

# 模拟原始数据
raw_data = [
    {'time': '09:30:01.123', 'price': 10.00, 'volume': 1000, 'ask1': 10.01, 'bid1': 9.99},
    {'time': '09:30:01.456', 'price': 10.01, 'volume': 2000, 'ask1': 10.01, 'bid1': 9.99},
    {'time': '09:30:01.789', 'price': 9.98, 'volume': 1500, 'ask1': 10.00, 'bid1': 9.98},
]

cleaned = clean_tick_data(raw_data)
print(cleaned[['time', 'price', 'volume', 'trade_type']])

2.4 数据存储与性能优化

Level-2数据量很大，一天下来一只股票可能有几万笔成交。如果你同时监控几百只股票，数据量就是百万级。我建议用Parquet格式存储，比CSV快10倍以上。

# 高效存储示例
def save_level2_data(df, file_path):
    """
    使用Parquet格式存储Level-2数据
    压缩率高，读写速度快
    """
    # 按日期分区
    df['date'] = df['time'].dt.date
    df.to_parquet(
        file_path,
        partition_cols=['date'],
        compression='snappy'
    )
    print(f"数据已保存至: {file_path}")

# 读取时只加载需要的日期
def load_level2_data(file_path, start_date, end_date):
    df = pd.read_parquet(
        file_path,
        filters=[('date', '>=', start_date), 
                 ('date', '<=', end_date)]
    )
    return df

2.5 实战：构建主动买卖盘分析框架

好了，数据拿到了，也清洗干净了。接下来我们做一个实战案例——计算主动买卖盘净额。这是判断主力资金流向的核心指标。

def calculate_active_trade_flow(cleaned_df):
    """
    计算主动买卖盘净额
    正值表示主动买盘强，负值表示主动卖盘强
    """
    # 按分钟聚合
    df = cleaned_df.copy()
    df['minute'] = df['time'].dt.floor('1min')
    
    # 分别计算主动买和主动卖的成交量
    buy_volume = df[df['trade_type'] == '主动买'].groupby('minute')['volume'].sum()
    sell_volume = df[df['trade_type'] == '主动卖'].groupby('minute')['volume'].sum()
    
    # 计算净额
    flow = pd.DataFrame({
        '主动买量': buy_volume,
        '主动卖量': sell_volume
    }).fillna(0)
    
    flow['净额'] = flow['主动买量'] - flow['主动卖量']
    flow['买卖比'] = flow['主动买量'] / (flow['主动卖量'] + 1)  # 避免除零
    
    return flow

# 实战应用
flow_data = calculate_active_trade_flow(cleaned)
print(flow_data.tail())

2.6 本章知识体系

下面这张图是我自己总结的Level-2数据获取与处理流程，你照着这个框架走，基本不会出错。

这张图把整个数据获取流程串起来了。从数据源选择开始，到盘口快照和逐笔成交两条线并行处理，最后汇聚到主动买卖盘分析。你写代码的时候，就按这个框架来，不会乱。

好了，数据获取这块就讲到这里。记住一句话：数据质量决定策略上限。花80%的时间在数据处理上，剩下20%做策略，这才是量化交易的正确姿势。