第3章数据获取

第三章：数据获取——盘口数据的源头活水

做量化交易，尤其是盘口失衡策略，最怕什么？

怕数据不准，怕数据延迟，怕数据断流。

我刚开始做高频策略那会儿，就吃过这个亏。回测跑得漂漂亮亮，实盘一上线，直接傻眼——数据源不对齐，买卖盘口差了0.5秒，策略信号全乱套。嗯，从那以后，我对数据获取这块就特别较真。

3.1 数据源的选择：Tushare vs JoinQuant

目前国内用得最多的两个数据源，就是Tushare和JoinQuant。说白了，它们各有各的脾气。

对比维度	Tushare	JoinQuant
数据覆盖	全市场，含Level-2	全市场，含Level-2
实时性	延迟约1-3秒	延迟约0.5-1秒
免费额度	积分制，基础免费	注册即用，有限制
API风格	RESTful + WebSocket	Python SDK + 本地回测
文档质量	详细，但更新慢	清晰，社区活跃

我个人习惯用Tushare做历史数据回测，用JoinQuant做实盘模拟。为什么？因为Tushare的历史数据更全，而JoinQuant的实时推送更稳。

3.2 实时盘口数据获取实战

先看Tushare怎么拿Level-2盘口数据。注意，这里需要你有Level-2权限。

import tushare as ts
import pandas as pd
from datetime import datetime

# 初始化
pro = ts.pro_api('your_token_here')

# 获取实时盘口数据
def get_tick_data(code, date):
    """
    获取个股的Tick级盘口数据
    code: 股票代码，如 '000001.SZ'
    date: 日期，如 '2024-01-15'
    """
    df = pro.tick(
        ts_code=code,
        trade_date=date,
        fields='ts_code,trade_time,price,vol,amount,buy1,buy2,buy3,sell1,sell2,sell3'
    )
    return df

# 调用示例
df_tick = get_tick_data('000001.SZ', '20240115')
print(df_tick.head())

我的小技巧： 获取数据时，别一股脑全拉。先拉最近5分钟的数据，看看格式对不对，再全量拉取。我曾经一次性拉了全天的数据，结果发现字段名变了，白忙活一场。

JoinQuant的写法稍微不同，它更贴近策略开发场景：

from jqdata import *
import datetime

# 获取实时行情
def get_realtime_ticks(security, end_date, count=100):
    """
    获取实时Tick数据
    security: 股票代码，如 '000001.XSHE'
    end_date: 截止时间
    count: 获取条数
    """
    ticks = get_ticks(
        security,
        end_date=end_date,
        count=count,
        fields=['time', 'current', 'volume', 'amount', 
                'a1_p', 'a1_v', 'b1_p', 'b1_v']
    )
    return ticks

# 使用示例
now = datetime.datetime.now()
ticks = get_realtime_ticks('000001.XSHE', now, 50)
print(ticks.tail())

3.3 数据清洗与对齐——最容易翻车的环节

数据拿到了，但能用吗？

我告诉你，原始数据里全是坑。比如时间戳不对齐、买卖盘口顺序颠倒、空值乱入。你想想看，如果买一价和卖一价对不上，你的盘口失衡信号就是垃圾。

避坑指南： 我曾经遇到过Tushare返回的buy1和sell1字段，在某些股票上顺序是反的。后来发现是数据源的问题，从那以后我每次都会做一次字段校验。

我的清洗流程是这样的：

时间对齐：把不同数据源的时间戳统一到毫秒级
字段校验：检查买卖盘口是否合理（买一价 < 卖一价）
空值处理：遇到空值，用前一条数据填充，或者直接丢弃
异常过滤：价格波动超过10%的，标记为异常

def clean_tick_data(df):
    """
    清洗Tick数据
    """
    # 1. 时间对齐：统一为毫秒时间戳
    df['timestamp'] = pd.to_datetime(df['trade_time']).astype('int64') // 10**6
    
    # 2. 字段校验：买一必须小于卖一
    mask = (df['buy1'] < df['sell1'])
    df = df[mask]
    
    # 3. 空值处理：前向填充
    df = df.fillna(method='ffill')
    
    # 4. 异常过滤：价格波动超过10%的剔除
    df['price_change'] = df['price'].pct_change()
    df = df[df['price_change'].abs() < 0.1]
    
    return df

# 执行清洗
df_clean = clean_tick_data(df_tick)
print(f"清洗前: {len(df_tick)} 条, 清洗后: {len(df_clean)} 条")

3.4 Tick级数据存储——别小看这一步

数据清洗完了，存哪里？

我建议用InfluxDB或者ClickHouse这类时序数据库。为什么？因为Tick数据量太大了，一天几百万条，MySQL根本扛不住。

如果你只是做研究，用Parquet格式存本地也行：

# 存储为Parquet格式
df_clean.to_parquet('tick_data_20240115.parquet', compression='snappy')

# 读取时
df = pd.read_parquet('tick_data_20240115.parquet')
print(f"数据量: {len(df)} 条, 时间范围: {df['timestamp'].min()} - {df['timestamp'].max()}")

  核心要点： 存储时一定要按日期分区，并且建立时间索引。不然查询的时候，你会等到怀疑人生。

3.5 本章知识体系总览

下面这张图，是我做数据获取时的完整流程。你看一遍，心里就有数了。

这张图把整个流程串起来了。从数据源到获取，再到清洗、存储，最后喂给策略。每一步都环环相扣，少一个环节都不行。

我的建议： 刚开始做的时候，别追求完美。先跑通一个最小闭环——拿一只股票的数据，清洗干净，存下来，算一个简单的盘口失衡指标。等流程跑顺了，再扩展到全市场。

数据获取这块，说白了就是「脏活累活」。但恰恰是这些脏活累活，决定了你的策略能不能在实盘中活下来。我见过太多人，策略写得天花乱坠，结果数据源一换，全盘崩溃。

嗯，这一章的内容就到这儿。数据拿到了，清洗好了，存起来了。下一步，就是怎么从这些数据里，挖出真正的盘口失衡信号。

📚 课程目录

第三章：数据获取——盘口数据的源头活水

3.1 数据源的选择：Tushare vs JoinQuant

3.2 实时盘口数据获取实战

3.3 数据清洗与对齐——最容易翻车的环节

3.4 Tick级数据存储——别小看这一步

3.5 本章知识体系总览