A 股 Level-2 行情数据 API 实战指南

毫秒级延迟、十档盘口、逐笔成交——深入解读 Level-2 行情数据的技术架构与接入实践

在量化交易领域，数据是策略的灵魂。如果说 Level-1 行情让你看到市场的“表象”，那么 Level-2 行情则让你洞察市场的“脉搏”——每一笔委托的博弈、每一档价位的深度、每一次成交的主动方向。本文将系统介绍 A 股 Level-2 行情数据 API 的技术特点、接入方案及实战应用，帮助开发者在量化交易的道路上构建坚实的数据底座。

一、Level-2 行情：超越传统行情的微观视角

1.1 什么是 Level-2 行情数据？

Level-2 行情是目前国内证券市场上最完整、颗粒度最精细的交易信息数据。与传统的 Level-1 行情（展示最优五档买卖盘口）相比，Level-2 提供了更为丰富的数据维度，主要包括以下几类：

• 盘口快照（3 秒更新）：提供十档买卖盘口，能够揭示市场深度，让投资者看清买一到买十、卖一到卖十的全部委托价格和委托数量。
• 委托队列（3 秒更新）：展示买一和卖一位置的前 50 笔委托明细，可以观察大单挂单是否集中在队列头部，识别潜在的支撑或压力位。
• 逐笔成交（实时，毫秒级）：记录每一笔成交的价格、数量、成交时间以及主动买卖方向，是分析主力资金动向的核心数据。
• 逐笔委托（实时，毫秒级）：记录每一笔委托的挂单和撤单信息，可以追踪资金的挂单意图和撤单行为。
• 分钟 K 线（1 分钟）：包含成交笔数的高级 K 线，比普通 K 线提供更多交易活跃度信息。

数据量方面，A 股 Level-2 行情每日增量约 30-45GB，历史数据可达 10TB 级别。这意味着，处理 Level-2 数据不仅需要高效的 API 接入方案，更需要强大的数据存储与计算能力。

1.2 Level-2 数据的核心应用场景

Level-2 数据在量化交易中拥有广泛的应用场景：

• 订单流分析：通过逐笔成交数据识别主力资金的主动买卖方向，判断买卖力量对比。
• 市场微观结构研究：分析买卖价差、订单簿斜率、委托不平衡等指标，理解市场的动态博弈过程。
• 高频信号触发：基于盘口变化在毫秒级时间内生成交易信号，捕捉短线交易机会。
• 算法交易执行：根据十档盘口信息优化拆单策略，降低大额订单对市场的冲击成本。
• T+0 日内交易：利用盘口深度信息捕捉价差机会，提高资金利用效率。

二、Level-2 行情数据 API 选型指南

2.1 主流数据源

根据市场调研，目前主流的 Level-2 数据源可分为几类，各有特点：

券商官方 API：延迟通常在 100 毫秒以内，数据覆盖全市场。使用此类 API 通常需要在该券商开立账户并满足一定的资产门槛，适用于实盘交易场景。

同花顺 iFinD：延迟在 100-200 毫秒，覆盖全市场 A 股及港股数据。采用年费制，数据质量稳定，功能丰富，适合专业投资机构使用。

Tushare Pro：延迟约 500 毫秒，覆盖 A 股和港股市场。采用按调用次数计费的模式，有免费额度可供开发者试用，是个人开发者的入门首选。

iTick API：提供毫秒级实时数据，支持多市场接入。提供免费套餐，适合量化研究人员进行策略开发和测试。

对于个人开发者，Tushare Pro 和 iTick 是性价比较高的选择；对于机构用户，同花顺、Wind 等专业服务商提供更全面的数据支持和合规保障。

2.2 数据接入方式

Level-2 数据接入主要采用两种协议：

RESTful API 适用于历史数据查询和低频数据获取。开发者通过 HTTP 请求获取特定时间范围、特定股票的行情数据，操作简单，易于集成。例如，获取历史分钟 K 线数据时，只需构造一个包含股票代码、时间周期、起止日期的 GET 请求即可。

WebSocket 连接 适用于实时行情推送。通过建立长连接，服务器可以主动将实时数据推送给客户端，延迟极低。使用 WebSocket 时，通常需要先进行身份认证，然后订阅感兴趣的股票代码，之后便可持续接收推送的逐笔成交、盘口快照等实时数据。

2.3 沪深交易所数据差异

需要特别注意的是，上交所和深交所的 Level-2 数据结构存在差异，开发者在设计数据处理逻辑时需进行差异化处理：

• 深交所：逐笔成交数据包含完整的挂撤单信息。当遇到市价单时，其价格会被标记为 0，需要在数据处理时进行特殊处理。
• 上交所：市价单信息只存储于逐笔成交表中，逐笔委托表内无市价单记录。这意味着如果仅分析逐笔委托数据，可能会遗漏市价单的信息。

三、Level-2 数据存储与处理实践

3.1 海量数据的存储挑战

Level-2 数据每天新增 30GB 以上，传统的关系型数据库难以应对。生产环境通常采用时序数据库与列式存储相结合的解决方案。

以 DolphinDB 为例，可以通过分区存储来优化查询性能。分区策略通常采用“按交易日期分区 + 按股票代码哈希分区”的组合方式。按日期分区可以方便地进行历史数据归档和清理，按股票代码哈希分区则可以实现多股票并行查询，提升计算效率。

创建行情快照表时，可以将交易日期和股票代码作为分区键，并将数据按股票代码和交易时间进行排序。这种设计能够带来多方面的优势：多个分区可以同时参与计算，实现并行查询；列式存储显著降低了存储成本，提高了压缩率；按股票和时间精准定位数据，过滤效率极高。

3.2 高频到低频的数据降频

量化策略往往需要在不同时间尺度上运行。从 Level-2 高频数据生成分钟级数据是一个典型的 ETL 流程。具体实现时，可以先将逐笔成交数据的时间字段转换为 Datetime 类型并设置为索引，然后使用重采样函数按分钟聚合。重采样过程中需要分别计算每个分钟区间的开盘价、最高价、最低价、收盘价、总成交量和总成交笔数，最后将这些结果合并为一个完整的分钟 K 线数据框。

3.3 基于 Level-2 的因子计算示例

基于盘口快照可以计算订单簿不平衡因子，这是一个常用的微观结构指标。计算逻辑如下：首先分别计算前十个档位的买量总和与卖量总和，然后计算不平衡率，即买量与卖量的差值除以两者之和。当买量显著大于卖量时，不平衡率为正值，预示价格可能上涨；反之则为负值，预示价格可能下跌。

更进一步，可以计算加权不平衡率，给予更靠近盘口的档位更高的权重。例如，按档位顺序设置递减的权重系数，对每个档位的买量和卖量计算局部不平衡率，再乘以该档位的权重，最后将所有档位的加权结果归一化。这种加权方法能够更加敏感地反映盘口最近价位的委托压力变化。

四、实战：构建实时量化交易系统

4.1 系统架构设计

一个完整的 Level-2 量化交易系统通常采用微服务架构，主要包含以下几个层次：

数据采集层：通过 WebSocket 连接接收 Level-2 实时行情，将原始数据推送到 Kafka 消息队列，然后由数据清洗服务进行异常过滤和格式标准化。

计算引擎层：这一层包含多个并行运行的计算服务。因子计算服务负责实时计算订单簿不平衡、资金流向等微观结构指标；信号生成服务基于这些因子产生买卖信号；风险监控服务实时评估策略敞口和账户风险。

执行层：接收信号层的交易指令后，订单管理服务负责生成具体的委托单，经过风控拦截器的检查后，通过券商交易 API 提交到交易所。

4.2 实时行情接入代码示例

以下提供 iTick API 的完整接入示例，包括 REST API 和 WebSocket 两种方式，涵盖 A 股、港股、美股等市场。

环境准备

首先安装必要的 Python 库：

pip install requests websocket-client pandas

REST API 示例：适合历史数据查询和低频获取

1. 获取 A 股实时报价

import requests
import json

# 配置信息
API_TOKEN = "your_token_here"  # 替换为你的实际 Token
BASE_URL = "https://api.itick.org"

headers = {
    "accept": "application/json",
    "token": API_TOKEN
}

def get_stock_quote(region, code):
    """
    获取股票实时报价
    region: SH(上海), SZ(深圳), HK(香港), US(美股)
    code: 股票代码，如 600519, 000001, 700, AAPL
    """
    url = f"{BASE_URL}/stock/quote"
    params = {
        "region": region,
        "code": code
    }

    try:
        response = requests.get(url, params=params, headers=headers)
        response.raise_for_status()
        data = response.json()

        if data.get("code") == 0:
            quote = data["data"]
            print(f"股票: {quote['s']}")
            print(f"最新价: {quote['ld']}")
            print(f"涨跌幅: {quote['chp']}%")
            print(f"成交量: {quote['v']}")
            print(f"最高价: {quote['h']}")
            print(f"最低价: {quote['l']}")
            return quote
        else:
            print(f"API 错误: {data.get('msg')}")
            return None
    except Exception as e:
        print(f"请求失败: {str(e)}")
        return None

# 调用示例：获取贵州茅台实时行情
get_stock_quote("SH", "600519")

# 获取腾讯控股实时行情
get_stock_quote("HK", "700")

# 获取苹果公司实时行情
get_stock_quote("US", "AAPL")

1. 获取历史 K 线数据

def get_stock_kline(region, code, ktype, limit=100):
    """
    获取历史 K 线数据
    ktype: 1-1分钟, 2-5分钟, 3-15分钟, 4-30分钟, 5-60分钟, 8-日K, 9-周K, 10-月K
    limit: 返回的 K 线条数
    """
    url = f"{BASE_URL}/stock/kline"
    params = {
        "region": region,
        "code": code,
        "kType": ktype,
        "limit": limit
    }

    try:
        response = requests.get(url, params=params, headers=headers)
        response.raise_for_status()
        data = response.json()

        if data.get("code") == 0:
            klines = data["data"]
            print(f"获取到 {len(klines)} 条 K 线数据")

            # 转换为 DataFrame 便于分析
            import pandas as pd
            df = pd.DataFrame(klines)
            df.columns = ["timestamp", "open", "high", "low", "close", "volume", "turnover"]
            df["datetime"] = pd.to_datetime(df["timestamp"], unit="ms")

            return df
        else:
            print(f"API 错误: {data.get('msg')}")
            return None
    except Exception as e:
        print(f"请求失败: {str(e)}")
        return None

# 调用示例：获取贵州茅台最近 100 根日 K 线
df = get_stock_kline("SH", "600519", ktype=8, limit=100)
if df is not None:
    print(df.head())

1. 获取 Level-2 十档盘口数据

def get_stock_depth(region, code):
    """获取股票十档盘口数据"""
    url = f"{BASE_URL}/stock/depth"
    params = {
        "region": region,
        "code": code
    }

    try:
        response = requests.get(url, params=params, headers=headers)
        response.raise_for_status()
        data = response.json()

        if data.get("code") == 0:
            depth = data["data"]
            print("买盘（前五档）:")
            for i, bid in enumerate(depth.get("b", [])[:5]):
                print(f"  买{i+1}: 价格 {bid['p']} 数量 {bid['v']}")

            print("卖盘（前五档）:")
            for i, ask in enumerate(depth.get("a", [])[:5]):
                print(f"  卖{i+1}: 价格 {ask['p']} 数量 {ask['v']}")
            return depth
        else:
            print(f"API 错误: {data.get('msg')}")
            return None
    except Exception as e:
        print(f"请求失败: {str(e)}")
        return None

# 调用示例
get_stock_depth("SH", "600519")

WebSocket 实时推送：适合高频交易场景

WebSocket 是获取实时行情的最佳方式，延迟可控制在毫秒级。以下代码实现了完整的连接管理、心跳保活和自动重连机制。

import websocket
import json
import threading
import time

# WebSocket 配置
WS_URL = "wss://api.itick.org/stock"  # 股票市场，期货用 wss://api.itick.org/future
API_TOKEN = "your_token_here"

# 订阅配置
SUBSCRIBE_SYMBOLS = "600519$SH,000001$SZ"  # 多个标的用逗号分隔，格式：代码$市场
DATA_TYPES = "quote,tick,depth"  # 订阅类型：quote报价、tick成交、depth盘口

def on_message(ws, message):
    """处理接收到的消息"""
    try:
        data = json.loads(message)

        # 连接成功提示
        if data.get("code") == 1 and data.get("msg") == "Connected Successfully":
            print("✅ 连接成功，等待认证...")

        # 认证结果处理
        elif data.get("resAc") == "auth":
            if data.get("code") == 1:
                print("✅ 认证通过，开始订阅数据...")
                subscribe(ws)
            else:
                print(f"❌ 认证失败：{data.get('msg')}")
                ws.close()

        # 订阅结果处理
        elif data.get("resAc") == "subscribe":
            if data.get("code") == 1:
                print(f"✅ 订阅成功！标的：{SUBSCRIBE_SYMBOLS}，类型：{DATA_TYPES}")
            else:
                print(f"❌ 订阅失败：{data.get('msg')}")

        # 实时数据处理
        elif data.get("data"):
            market_data = data["data"]
            data_type = market_data.get("type")
            symbol = market_data.get("s")

            if data_type == "quote":
                # 报价数据
                print(f"📊 {symbol} 报价 - 最新价: {market_data.get('ld')}, "
                      f"涨跌幅: {market_data.get('chp')}%")

            elif data_type == "tick":
                # 逐笔成交数据
                print(f"💹 {symbol} 成交 - 价格: {market_data.get('p')}, "
                      f"数量: {market_data.get('v')}, 方向: {market_data.get('side')}")

            elif data_type == "depth":
                # 盘口深度数据
                print(f"📚 {symbol} 盘口 - 买一: {market_data.get('b')[0] if market_data.get('b') else 'N/A'}, "
                      f"卖一: {market_data.get('a')[0] if market_data.get('a') else 'N/A'}")

    except json.JSONDecodeError as e:
        print(f"❌ 数据解析失败：{e}")

def on_error(ws, error):
    """错误处理"""
    print(f"❌ 连接错误：{error}")

def on_close(ws, close_status_code, close_msg):
    """连接关闭时自动重连"""
    print(f"🔌 连接关闭，3秒后自动重连...")
    time.sleep(3)
    start_websocket()

def on_open(ws):
    """连接建立后触发"""
    print("🔗 WebSocket 连接已打开")

def subscribe(ws):
    """发送订阅请求"""
    subscribe_msg = {
        "ac": "subscribe",
        "params": SUBSCRIBE_SYMBOLS,
        "types": DATA_TYPES
    }
    ws.send(json.dumps(subscribe_msg))

def send_ping(ws):
    """每30秒发送心跳包，维持连接"""
    while True:
        time.sleep(30)
        try:
            ping_msg = {
                "ac": "ping",
                "params": str(int(time.time() * 1000))
            }
            ws.send(json.dumps(ping_msg))
            # print("📡 发送心跳包")  # 调试时可取消注释
        except Exception as e:
            print(f"❌ 发送心跳包失败：{e}")

def start_websocket():
    """启动 WebSocket 连接"""
    ws = websocket.WebSocketApp(
        WS_URL,
        header={"token": API_TOKEN},
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )

    # 启动心跳线程
    ping_thread = threading.Thread(target=send_ping, args=(ws,))
    ping_thread.daemon = True
    ping_thread.start()

    # 运行连接
    ws.run_forever()

if __name__ == "__main__":
    print("🚀 启动实时行情接收程序...")
    start_websocket()

一个完整的 WebSocket 行情客户端需要实现连接管理、消息处理和订阅管理等功能。连接建立后，首先发送登录消息进行身份认证。认证成功后，发送订阅消息，指定需要接收行情的股票代码列表。

在消息处理函数中，根据消息类型进行分发处理。对于逐笔成交数据，需要提取股票代码、成交价格、成交数量和主动方向（主动买或主动卖）。基于这些信息可以实时计算净流入资金量：如果是主动买成交，则将该笔成交金额累加到净流入中；如果是主动卖成交，则从净流入中扣除。这个实时计算的净流入指标可以直观反映主力资金的动向。

4.3 风控机制设计

Level-2 数据驱动的量化系统需要建立多层次风控体系：

数据层风控：监控行情数据的丢包率，当丢包率超过预设阈值（如 1%）时，说明数据质量可能存在问题，系统应触发熔断，自动切换到备用数据源。

信号层风控：为每个交易信号设置置信度阈值。当模型输出的信号置信度低于 0.7 时，即使信号触发，也应禁止开新仓，避免在模糊信号下进行交易。

订单层风控：限制单笔下单量不得超过账户权益的 5%。对于超过限额的大单，可以自动拆分为多笔小单提交，或直接拒绝执行。

账户层风控：监控账户的日最大回撤，当回撤幅度超过 10%时，启动强制平仓流程，并暂停后续所有交易操作，防止亏损进一步扩大。

系统层风控：监控交易 API 的响应延迟，当响应时间超过 2 秒时，说明主交易通道可能存在问题，应立即切换到备用交易通道。

五、Level-2 数据 API 的合规与门槛

5.1 准入门槛

根据多家券商和平台的规定，Level-2 数据接口通常有以下限制：

资产门槛：投资者需要满足账户资产不低于 10 万元人民币的条件，才有资格申请开通 Level-2 行情权限。

权限申请：需要通过券商官方或数据服务商提供的渠道，单独申请开通 Level-2 权限。不同券商的开通流程和要求可能略有差异。

合规要求：Level-2 数据仅限个人或机构内部使用，不得私自转发、二次销售或提供给第三方使用，否则可能面临法律责任。

5.2 数据使用注意事项

数据时效性：Level-2 数据具有很强的时效性，当日实时数据在收盘后通常会被系统自动清空。如果需要进行盘后复盘分析，必须在盘中主动进行数据缓存和存储。

历史数据补全：如果需要查询历史 Level-2 数据，通常需要等到 T+1 日才能下载完整的日终数据文件。这意味着当天的数据无法当天查询历史。

多实例部署：当需要部署多个进程同时接收行情时，每个进程必须独立初始化连接。重复使用同一连接或同时建立过多连接可能触发服务端的端口冲突限制。

六、开发者实践建议

6.1 渐进式开发路径

对于初次接触 Level-2 数据的开发者，建议按以下路径循序渐进：

第一阶段：使用免费数据源（如 Tushare Pro 的免费额度）熟悉 Level-2 的基本数据结构，本地搭建时序数据库（如 DolphinDB 或 InfluxDB）进行数据存储练习。

第二阶段：开发基于分钟 K 线的低频策略，验证数据接入的稳定性和准确性。此阶段主要检验数据管道是否可靠。

第三阶段：引入盘口快照数据，开发订单簿类因子（如订单簿不平衡、深度斜率等），学习如何从盘口数据中提取有价值的信息。

第四阶段：接入逐笔成交数据，实现高频信号和算法交易，深入理解市场微观结构的动态变化。

6.2 技术选型建议

在技术组件选择方面，以下是较为成熟的方案：

• 时序数据库：推荐使用 DolphinDB 或 InfluxDB，两者都支持列式存储和向量化计算，能够高效处理海量时序数据。
• 消息队列：Kafka 是行业标准，具备高吞吐量、数据持久化和良好的生态支持。
• 编程语言：可以采用 Python 进行原型开发和策略研究，关键的性能敏感模块可以使用 C++ 扩展来提升执行效率，兼顾开发效率和运行性能。
• 容器化：使用 Docker 打包应用，通过 Kubernetes 进行编排管理，实现弹性伸缩和便捷部署。
• 监控系统：Prometheus 搭配 Grafana 是成熟的监控方案，可以实时观测系统运行状态、数据延迟、错误率等关键指标。

七、结语

A 股 Level-2 行情数据 API 为量化开发者打开了一扇通往市场微观结构的大门。从十档盘口的深度分析到逐笔成交的资金流向追踪，Level-2 数据承载着比传统行情丰富十倍的信号价值。当然，海量数据的处理能力、稳定的系统架构、严格的风控机制，都是这条路途中必须跨越的挑战。

无论你是个人量化爱好者，还是专业机构的开发者，希望本文能帮助你建立起对 Level-2 数据 API 的完整认知，在量化交易的道路上走得更稳、更远。

参考文档：https://docs.itick.org/rest-api/stocks/stock-depth
GitHub：https://github.com/itick-org/