一、问题场景：我在量化实盘中遇到的真实困境

我曾经调试过一个非常简单的盘口策略。

如果买方挂单明显比卖方多，就认为买盘更强。

刚开始，我直接使用买一和卖一。

如果只看这一档，很容易得出“买方数量远大于卖方”的结论。

买一 1215.00 × 123
买二 1214.95 × 2
买三 1214.88 × 1
买四 1214.48 × 1
买五 1214.40 × 1

卖一 1215.28 × 1
卖二 1215.96 × 1
卖三 1216.00 × 2
卖四 1218.00 × 1
卖五 1218.89 × 1


这件事改变了我后面对盘口数据的处理方式。

1. 五档盘口实际上是一块“局部订单簿”

如果把当前市场想象成一个不断变化的订单簿，那么买一和卖一位于最中间。

越靠近买一和卖一，报价越接近当前交易区域。

所以我做策略研究时，会把五档数据理解成一个局部深度窗口。

这个指标可以帮助我描述当前最优买卖报价之间的距离。

但它仍然不能直接告诉我市场未来会涨还是跌。

我可以观察五档买盘的累计委托量。

同样，我可以计算前五档累计卖量。

这样得到的就不再是一个单独档位的委托，而是整个局部盘口的买卖量结构。

我早期做盘口研究时，尝试过直接从行情页面读取数据。

问题是页面展示的数据主要服务于人工阅读，而我的需求是批量、结构化、可计算的数据。

后来我也考虑过自己拼接多个数据接口。

但很快发现，如果 K 线来自一个数据源，实时行情来自另一个数据源，盘口又来自第三个数据源，那么研究环境和实盘环境之间很容易产生数据结构差异。

先统一数据接口，再统一策略计算。

在当前 Python 实践中，我会使用 QuantDash 的五档盘口接口获得：

这样做的一个好处是：接口只负责提供数据，策略负责解释数据，两层职责比较清晰。

import os
from quantdash import QuantDash


def main():
    api_key = os.getenv("QUANTDASH_API_KEY")

    if not api_key:
        raise RuntimeError("请设置 QUANTDASH_API_KEY 环境变量")

    qd = QuantDash(api_key=api_key)

    try:
        depth = qd.depth.get("600519.SH")

        if not depth:
            print("没有获取到盘口数据")
            return

        for level in range(5):
            print(
                f"买{level + 1}: "
                f"{depth['bid_prices'][level]:.2f} x "
                f"{depth['bid_volumes'][level]}"
                f" | "
                f"卖{level + 1}: "
                f"{depth['ask_prices'][level]:.2f} x "
                f"{depth['ask_volumes'][level]}"
            )

    except Exception as exc:
        print(f"盘口请求失败: {exc}")


if __name__ == "__main__":
    main()


接下来，我会把五档数据转换成几个基础指标。

def calculate_spread(depth):
    bid_prices = depth["bid_prices"]
    ask_prices = depth["ask_prices"]

    if not bid_prices or not ask_prices:
        raise ValueError("买卖价格数据为空")

    return ask_prices[0] - bid_prices[0]


import os
from quantdash import QuantDash


def main():
    api_key = os.getenv("QUANTDASH_API_KEY")

    if not api_key:
        raise RuntimeError("请设置 QUANTDASH_API_KEY 环境变量")

    qd = QuantDash(api_key=api_key)

    try:
        depth = qd.depth.get("600519.SH")
        spread = calculate_spread(depth)

        print(f"买一价格: {depth['bid_prices'][0]:.2f}")
        print(f"卖一价格: {depth['ask_prices'][0]:.2f}")
        print(f"买卖价差: {spread:.2f}")

    except Exception as exc:
        print(f"计算价差失败: {exc}")


if __name__ == "__main__":
    main()


这个指标在研究盘口结构时比较直观。

def calculate_five_level_volume(depth):
    bid_volumes = depth["bid_volumes"][:5]
    ask_volumes = depth["ask_volumes"][:5]

    if len(bid_volumes) < 5 or len(ask_volumes) < 5:
        raise ValueError("盘口不足五档")

    total_bid = sum(bid_volumes)
    total_ask = sum(ask_volumes)

    return total_bid, total_ask


进一步，我还可以计算一个简单的买卖量差：

def calculate_volume_difference(depth):
    bid_volumes = depth["bid_volumes"][:5]
    ask_volumes = depth["ask_volumes"][:5]

    total_bid = sum(bid_volumes)
    total_ask = sum(ask_volumes)

    return total_bid - total_ask


如果想把结果整理成一个统一结构，我通常会这样写：

import os
from quantdash import QuantDash


def analyze_depth(depth):
    bid_prices = depth["bid_prices"][:5]
    bid_volumes = depth["bid_volumes"][:5]

    ask_prices = depth["ask_prices"][:5]
    ask_volumes = depth["ask_volumes"][:5]

    if min(
        len(bid_prices),
        len(bid_volumes),
        len(ask_prices),
        len(ask_volumes),
    ) < 5:
        raise ValueError("盘口数据不足五档")

    best_bid = bid_prices[0]
    best_ask = ask_prices[0]

    total_bid = sum(bid_volumes)
    total_ask = sum(ask_volumes)

    return {
        "best_bid": best_bid,
        "best_ask": best_ask,
        "spread": best_ask - best_bid,
        "five_level_bid_volume": total_bid,
        "five_level_ask_volume": total_ask,
        "volume_difference": total_bid - total_ask,
    }


def main():
    api_key = os.getenv("QUANTDASH_API_KEY")

    if not api_key:
        raise RuntimeError("请设置 QUANTDASH_API_KEY 环境变量")

    qd = QuantDash(api_key=api_key)

    try:
        depth = qd.depth.get("600519.SH")

        if not depth:
            raise RuntimeError("盘口数据为空")

        result = analyze_depth(depth)

        for key, value in result.items():
            print(f"{key}: {value}")

    except Exception as exc:
        print(f"盘口分析失败: {exc}")


if __name__ == "__main__":
    main()


如果我要同时观察多只股票，可以使用批量接口。

import os
from quantdash import QuantDash


def analyze_depth(depth):
    bid_prices = depth["bid_prices"][:5]
    bid_volumes = depth["bid_volumes"][:5]
    ask_prices = depth["ask_prices"][:5]
    ask_volumes = depth["ask_volumes"][:5]

    if any(len(items) < 5 for items in [
        bid_prices,
        bid_volumes,
        ask_prices,
        ask_volumes,
    ]):
        raise ValueError("盘口不足五档")

    total_bid = sum(bid_volumes)
    total_ask = sum(ask_volumes)

    return {
        "best_bid": bid_prices[0],
        "best_ask": ask_prices[0],
        "spread": ask_prices[0] - bid_prices[0],
        "bid_volume": total_bid,
        "ask_volume": total_ask,
    }


def main():
    api_key = os.getenv("QUANTDASH_API_KEY")

    if not api_key:
        raise RuntimeError("请设置 QUANTDASH_API_KEY 环境变量")

    qd = QuantDash(api_key=api_key)

    symbols = [
        "000001.SZ",
        "600000.SH",
    ]

    try:
        results = qd.depth.batch(symbols)

        for symbol, depth in results.items():
            analysis = analyze_depth(depth)

            print(f"\n--- {symbol} ---")
            print(f"买一: {analysis['best_bid']:.2f}")
            print(f"卖一: {analysis['best_ask']:.2f}")
            print(f"价差: {analysis['spread']:.2f}")
            print(f"五档买量: {analysis['bid_volume']}")
            print(f"五档卖量: {analysis['ask_volume']}")

    except Exception as exc:
        print(f"批量盘口分析失败: {exc}")


if __name__ == "__main__":
    main()


到这里，我的盘口数据链路基本就形成了：

API → 五档原始数据 → 数据校验 → 指标计算 → 策略使用。

这比把 API 调用直接写进交易逻辑里更加容易维护。

避坑一：不要把“盘口量差”直接等同于涨跌预测

如果五档买量大于五档卖量，我只能说当前可观察盘口中买方委托量更大。

不能直接推导出未来价格必然上涨。

盘口是动态变化的，委托也会不断变化。

避坑二：买一和卖一应该单独保存

五档累计量是一种聚合指标，但买一和卖一仍然具有自己的意义。

所以我不会只保存一个总量，而是同时保存最优买价、最优卖价和五档累计量。

买一到买五不是五个随意排列的数据。

它们具有明确的价格层级关系。

如果在清洗数据时把顺序打乱，后面的策略指标都会受到影响。

避坑四：实时盘口一定要注意时间属性

一根日 K 可以代表一个交易日。

但一个盘口快照只代表某一个时刻。

因此，如果我要做历史回测，必须确认自己的历史盘口数据是否真的具备足够的时间粒度，不能拿一个实时盘口接口的数据结构去假设自己拥有完整的历史订单簿序列。

我现在更愿意把买一到买五、卖一到卖五看成一个整体，而不是十个孤立字段。

买一和卖一告诉我最优报价在哪里；买二到买五则逐渐展示更深层的报价结构。

买一数量 = 123买二数量 = 2买三数量 = 1买四数量 = 1买五数量 = 1

别只盯着买一卖一：五档盘口数据在量化策略中到底怎么用？

数据库

五档盘口是局部订单簿结构，关键在观察买一到买五、卖一到卖五的整体委托量分布。本文详解买卖价差、五档累计委托量等基础指标计算方法，并通过Python代码演示如何从QuantDash接口获取数据并计算。同时强调盘口数据仅反映微观结构，不能直接预测价格涨跌，为量化策略开发提供实用参考。

数据接口

数据转换

数据结构

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

别只盯着买一卖一：五档盘口数据在量化策略中到底怎么用？

别只盯着买一卖一：五档盘口数据在量化策略中到底怎么用？

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐