# 使用 Python 获取A股五档盘口数据分析流动性与主力行为

大多数量化入门教程只讲 K 线——日 K、周 K、均线、MACD、布林带。但如果你只看 K 线，你看到的只是"结果"：价格已经走到这里了。

盘口数据（Order Book / Level 2）提供的是"过程"：此刻有多少人想买、多少人想卖、分别挂在什么价位。它比 K 线多了一个维度——**委托结构**。

AlphaFeed 提供 A 股五档盘口接口（5-level depth），可以实时获取买一到买五、卖一到卖五的价格和挂单量。这篇文章会从实际出发，展示如何获取、解读和利用这些数据。

## 1. 获取五档盘口数据

```python
from alphafeed import AlphaFeed

af = AlphaFeed()

depth = af.depth.get("600519.SH")

print("=== 卖盘（Ask）===")
for i in range(len(depth.ask_prices) - 1, -1, -1):
    print(f"  卖{i+1}: {depth.ask_prices[i]:.2f}  挂单量: {depth.ask_volumes[i]}")

print(f"\n--- 最新价 ---")

print("\n=== 买盘（Bid）===")
for i in range(len(depth.bid_prices)):
    print(f"  买{i+1}: {depth.bid_prices[i]:.2f}  挂单量: {depth.bid_volumes[i]}")
```

返回的数据结构：

| 字段 | 含义 |
| --- | --- |
| `bid_prices` | 买一到买五的价格（从高到低） |
| `bid_volumes` | 对应的挂单量 |
| `ask_prices` | 卖一到卖五的价格（从低到高） |
| `ask_volumes` | 对应的挂单量 |
| `symbol` | 标的代码 |
| `timestamp` | 数据时间戳 |

## 2. 买卖价差（Bid-Ask Spread）

买卖价差是流动性最直接的指标：价差越小，流动性越好，交易成本越低。

```python
from alphafeed import AlphaFeed

af = AlphaFeed()

def calc_spread(symbol: str) -> dict:
    depth = af.depth.get(symbol)

    bid1 = depth.bid_prices[0]
    ask1 = depth.ask_prices[0]
    mid_price = (bid1 + ask1) / 2
    spread = ask1 - bid1
    spread_bps = spread / mid_price * 10000

    return {
        "symbol": symbol,
        "bid1": bid1,
        "ask1": ask1,
        "spread": round(spread, 4),
        "spread_bps": round(spread_bps, 2),
        "mid_price": round(mid_price, 4),
    }

symbols = ["600519.SH", "000001.SZ", "601318.SH", "300750.SZ", "000858.SZ"]

for sym in symbols:
    info = calc_spread(sym)
    print(f"{info['symbol']}: 买一={info['bid1']:.2f} 卖一={info['ask1']:.2f} "
          f"价差={info['spread']:.4f} ({info['spread_bps']:.1f} bps)")
```

`spread_bps` 是用基点（万分之一）来衡量价差。一般来说：

| 价差范围 | 流动性评价 |
| --- | --- |
| < 5 bps | 非常好，大盘蓝筹水平 |
| 5–20 bps | 良好，中盘活跃股 |
| 20–50 bps | 一般，小盘或不活跃 |
| > 50 bps | 差，交易成本显著 |

如果你做策略回测时假设"无滑点成交"，但标的实际价差是 30 bps，那你的回测收益就会虚高。盘口数据可以帮你估算真实的交易成本。

## 3. 挂单量不对称：谁更急？

五档盘口还有一个关键信息：买盘和卖盘的挂单量对比。

```python
from alphafeed import AlphaFeed

af = AlphaFeed()

def order_imbalance(symbol: str) -> dict:
    depth = af.depth.get(symbol)

    total_bid = sum(depth.bid_volumes)
    total_ask = sum(depth.ask_volumes)
    total = total_bid + total_ask

    imbalance = (total_bid - total_ask) / total if total > 0 else 0

    return {
        "symbol": symbol,
        "total_bid_vol": total_bid,
        "total_ask_vol": total_ask,
        "imbalance": round(imbalance, 4),
        "interpretation": "买盘强" if imbalance > 0.2 else ("卖盘强" if imbalance < -0.2 else "相对均衡"),
    }

symbols = ["600519.SH", "000001.SZ", "601318.SH", "300750.SZ", "002594.SZ"]

for sym in symbols:
    info = order_imbalance(sym)
    print(f"{info['symbol']}: 买盘={info['total_bid_vol']:>10,} "
          f"卖盘={info['total_ask_vol']:>10,} "
          f"不平衡度={info['imbalance']:+.4f} → {info['interpretation']}")
```

**Order Imbalance（委托不平衡度）**的计算公式：

```
imbalance = (total_bid - total_ask) / (total_bid + total_ask)
```

- 值接近 +1：买盘远大于卖盘，多头挂单积极
- 值接近 -1：卖盘远大于买盘，抛压明显
- 值接近 0：买卖力量基本均衡

需要注意的是：挂单量不等于真实意图。大额挂单可能是"诱多/诱空"，挂了又撤。但在统计意义上，持续的委托不平衡仍然有一定的价格预测价值。

## 4. 盘口厚度：大单能不能吃得动

如果你有 100 万要买入某只股票，盘口上卖一到卖五的总挂单量够不够"吃"？如果不够，你的买入就会把价格推上去，产生冲击成本。

```python
from alphafeed import AlphaFeed

af = AlphaFeed()

def estimate_impact(symbol: str, buy_amount: float) -> dict:
    """估算买入指定金额时的价格冲击"""
    depth = af.depth.get(symbol)

    filled_amount = 0
    filled_volume = 0
    levels_consumed = 0
    last_fill_price = 0

    for i in range(len(depth.ask_prices)):
        price = depth.ask_prices[i]
        vol = depth.ask_volumes[i]
        level_amount = price * vol

        if filled_amount + level_amount >= buy_amount:
            remaining = buy_amount - filled_amount
            partial_vol = remaining / price
            filled_volume += partial_vol
            filled_amount += remaining
            last_fill_price = price
            levels_consumed = i + 1
            break
        else:
            filled_amount += level_amount
            filled_volume += vol
            last_fill_price = price
            levels_consumed = i + 1

    avg_price = buy_amount / filled_volume if filled_volume > 0 else 0
    ask1_price = depth.ask_prices[0]
    impact_bps = (avg_price / ask1_price - 1) * 10000 if ask1_price > 0 else 0

    return {
        "symbol": symbol,
        "buy_amount": buy_amount,
        "avg_fill_price": round(avg_price, 4),
        "ask1_price": ask1_price,
        "impact_bps": round(impact_bps, 2),
        "levels_consumed": levels_consumed,
        "fully_filled": filled_amount >= buy_amount * 0.99,
    }

symbols = ["600519.SH", "000001.SZ", "300750.SZ"]

for sym in symbols:
    result = estimate_impact(sym, buy_amount=1_000_000)
    status = "✓ 可填满" if result["fully_filled"] else "✗ 五档不够"
    print(f"{result['symbol']}: "
          f"卖一={result['ask1_price']:.2f} "
          f"均价={result['avg_fill_price']:.2f} "
          f"冲击={result['impact_bps']:.1f}bps "
          f"吃掉{result['levels_consumed']}档 {status}")
```

这个分析对大资金量化尤其重要。如果五档盘口总共只能容纳 50 万的买入量，而你的策略需要一次买入 200 万，那冲击成本会显著侵蚀你的策略收益。

## 5. 持续采集盘口快照

单次盘口数据只是一个瞬间切片。要做有意义的分析，你需要在盘中持续采集：

```python
import time
import json
from datetime import datetime
from alphafeed import AlphaFeed

af = AlphaFeed()

def collect_depth_snapshots(symbol: str, interval: int = 10, duration_minutes: int = 5):
    """每隔 interval 秒采集一次盘口快照"""
    snapshots = []
    end_time = time.time() + duration_minutes * 60

    print(f"开始采集 {symbol} 盘口数据，间隔 {interval}s，持续 {duration_minutes} 分钟")

    while time.time() < end_time:
        try:
            depth = af.depth.get(symbol)
            snapshot = {
                "time": datetime.now().isoformat(),
                "bid_prices": depth.bid_prices,
                "bid_volumes": depth.bid_volumes,
                "ask_prices": depth.ask_prices,
                "ask_volumes": depth.ask_volumes,
            }
            snapshots.append(snapshot)

            total_bid = sum(depth.bid_volumes)
            total_ask = sum(depth.ask_volumes)
            imbalance = (total_bid - total_ask) / (total_bid + total_ask) if (total_bid + total_ask) > 0 else 0

            print(f"[{snapshot['time'][-8:]}] "
                  f"bid1={depth.bid_prices[0]:.2f} ask1={depth.ask_prices[0]:.2f} "
                  f"imbalance={imbalance:+.3f}")

        except Exception as e:
            print(f"采集异常: {e}")

        time.sleep(interval)

    filename = f"depth_{symbol.replace('.', '_')}_{datetime.now().strftime('%Y%m%d_%H%M')}.json"
    with open(filename, "w") as f:
        json.dump(snapshots, f, ensure_ascii=False, indent=2)

    print(f"\n采集完成，共 {len(snapshots)} 条，已保存到 {filename}")
    return snapshots

snapshots = collect_depth_snapshots("600519.SH", interval=10, duration_minutes=5)
```

积累一段时间的盘口快照数据后，你可以做更深入的分析。

## 6. 从盘口快照中提取信号

有了一系列盘口快照后，我们可以计算时间序列上的指标：

```python
import pandas as pd
import json

with open("depth_600519_SH_20250613_1400.json", "r") as f:
    snapshots = json.load(f)

records = []
for snap in snapshots:
    total_bid = sum(snap["bid_volumes"])
    total_ask = sum(snap["ask_volumes"])
    imbalance = (total_bid - total_ask) / (total_bid + total_ask) if (total_bid + total_ask) > 0 else 0
    spread = snap["ask_prices"][0] - snap["bid_prices"][0]
    mid = (snap["ask_prices"][0] + snap["bid_prices"][0]) / 2

    records.append({
        "time": snap["time"],
        "mid_price": mid,
        "spread": spread,
        "imbalance": imbalance,
        "total_bid": total_bid,
        "total_ask": total_ask,
    })

depth_df = pd.DataFrame(records)
depth_df["time"] = pd.to_datetime(depth_df["time"])

depth_df["imbalance_ma"] = depth_df["imbalance"].rolling(6).mean()

depth_df["mid_return"] = depth_df["mid_price"].pct_change()

corr = depth_df["imbalance_ma"].corr(depth_df["mid_return"].shift(-1))
print(f"委托不平衡度与下一期中间价变动的相关性: {corr:.4f}")

print("\n=== 盘口时序数据 ===")
print(depth_df[["time", "mid_price", "spread", "imbalance", "imbalance_ma"]].to_string(index=False))
```

如果 `imbalance_ma` 和 `mid_return.shift(-1)` 之间有正相关性，说明当买盘挂单积累时，价格确实在未来短期内有上涨倾向。这是很多高频/中频策略的基础信号之一。

## 7. 盘口数据能告诉你什么，不能告诉你什么

### 能做的

| 用途 | 说明 |
| --- | --- |
| 流动性评估 | 判断标的是否适合你的资金量级 |
| 交易成本估算 | 用盘口厚度估算实际滑点 |
| 委托不平衡监控 | 观察买卖力量的短期变化 |
| 冲击成本模型 | 为回测加入更真实的成交假设 |
| 异常检测 | 大额挂单突然出现/消失 |

### 不能做或需要谨慎的

| 场景 | 原因 |
| --- | --- |
| 判断"主力"意图 | 挂单可以随时撤掉，"大买单"不等于真实买入意愿 |
| 预测价格方向 | 盘口信号的预测力很弱，且衰减极快，不能替代策略逻辑 |
| 做高频交易 | 五档数据粒度不够，真正的高频需要逐笔委托（Level 3） |
| 忽略时间衰减 | 盘口数据在毫秒级变化，10 秒前的快照可能已经过时 |

## 8. 盘口分析的实际应用场景

### 场景一：策略回测中加入滑点估算

很多回测假设"以收盘价成交"，这在小资金量下没问题。但如果你的策略每天要交易 500 万，你需要知道：这笔交易会产生多少冲击成本？

把第 4 节的 `estimate_impact` 函数嵌入你的回测框架，就可以对每笔交易估算一个更真实的成交价格。

### 场景二：选股时排除流动性不足的标的

在第 08 篇全市场选股的基础上，你可以对候选列表做一轮流动性过滤：

```python
from alphafeed import AlphaFeed

af = AlphaFeed()

candidate_symbols = ["600519.SH", "000001.SZ", "002594.SZ", "300750.SZ"]

for sym in candidate_symbols:
    depth = af.depth.get(sym)
    spread = depth.ask_prices[0] - depth.bid_prices[0]
    mid = (depth.ask_prices[0] + depth.bid_prices[0]) / 2
    spread_bps = spread / mid * 10000

    ask_depth = sum(p * v for p, v in zip(depth.ask_prices, depth.ask_volumes))

    status = "通过" if spread_bps < 20 and ask_depth > 500_000 else "排除"
    print(f"{sym}: 价差={spread_bps:.1f}bps 卖盘深度={ask_depth:,.0f}元 → {status}")
```

### 场景三：盘中监控大单异动

当某一档突然出现远超正常水平的大额挂单，可能意味着有机构在该价位设定了防线或目标价：

```python
from alphafeed import AlphaFeed

af = AlphaFeed()

def check_large_orders(symbol: str, threshold_multiplier: float = 5.0):
    depth = af.depth.get(symbol)

    all_volumes = depth.bid_volumes + depth.ask_volumes
    avg_vol = sum(all_volumes) / len(all_volumes) if all_volumes else 0

    alerts = []
    for i, (price, vol) in enumerate(zip(depth.bid_prices, depth.bid_volumes)):
        if vol > avg_vol * threshold_multiplier:
            alerts.append(f"买{i+1} {price:.2f} 挂单 {vol:,} 手（平均的 {vol/avg_vol:.1f} 倍）")

    for i, (price, vol) in enumerate(zip(depth.ask_prices, depth.ask_volumes)):
        if vol > avg_vol * threshold_multiplier:
            alerts.append(f"卖{i+1} {price:.2f} 挂单 {vol:,} 手（平均的 {vol/avg_vol:.1f} 倍）")

    if alerts:
        print(f"[{symbol}] 发现大单:")
        for a in alerts:
            print(f"  {a}")
    else:
        print(f"[{symbol}] 盘口正常，无异常大单")

check_large_orders("600519.SH")
check_large_orders("000001.SZ")
```

## 9. 把盘口数据纳入你的量化工具箱

盘口数据不是一个独立的策略，它是你量化工具箱里的一个额外维度。

在之前的文章里，我们已经用 K 线做了趋势策略、动量选股、ETF 轮动。盘口数据可以在这些策略之上叠加一层：

| 已有策略 | 盘口数据的作用 |
| --- | --- |
| 双均线策略 | 买入信号触发时，检查盘口流动性是否足够 |
| 全市场选股 | 对候选列表做流动性排序，优先选盘口厚的 |
| ETF 轮动 | 切换时估算冲击成本，扣减回测收益 |
| 实时监控 | 在价格告警基础上增加委托不平衡告警 |

数据维度越丰富，策略的假设越接近真实交易。

## 结语

五档盘口数据是 K 线之外的另一扇窗。

K 线告诉你"发生了什么"，盘口告诉你"正在发生什么"——此刻市场上的买卖力量分布、流动性深度、大额挂单的位置。这些信息对于估算交易成本、监控市场微观结构、筛选适合交易的标的都有实际价值。

AlphaFeed 的 `af.depth.get()` 让获取这些数据变得很简单——一行代码就能拿到五档买卖挂单。在此基础上，你可以做价差分析、委托不平衡监控、冲击成本估算，以及持续的盘口数据采集。

这些分析不会直接告诉你"该买什么"。但它们会让你的策略假设更贴近真实市场，让你的选股结果更具可操作性。


大多数量化入门教程只讲 K 线——日 K、周 K、均线、MACD、布林带。但如果你只看 K 线，你看到的只是"结果"：价格已经走到这里了。

使用 Python 获取A股五档盘口数据分析流动性与主力行为

编程语言

大数据

后端

Python获取A股五档盘口数据，分析买卖价差、委托不平衡度和冲击成本，揭示市场流动性和主力行为。通过实时采集盘口快照，构建交易成本模型，优化量化策略回测准确性。盘口数据比K线多一个维度，能监控市场微观结构变化，为选股和交易执行提供决策依据。

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

使用 Python 获取A股五档盘口数据分析流动性与主力行为-腾讯云开发者社区-腾讯云

使用 Python 获取A股五档盘口数据分析流动性与主力行为

使用 Python 获取A股五档盘口数据分析流动性与主力行为

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐