我过去做量化数据工程时，最容易忽视的并不是接口，而是数据分层：实时行情、五档盘口和 K 线经常被放进同一个数据对象，最后导致策略层越来越混乱。本文从一次数据架构重构开始，介绍我如何区分三类行情数据，并使用 Python 建立一个简单的行情数据处理层。

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

我以前维护过一个股票数据模块。

所有策略都从这个方法拿数据。

随着策略增加，问题慢慢出现了。

get_quote()
    ├── 日线
    ├── 分钟线
    ├── 实时价格
    ├── 五档
    └── 其他字段


我当时觉得这样“统一”很好。

但实际上，这是一种错误的统一。

因为不同数据本身的生命周期和语义完全不同。

我最终把行情数据分成三个核心层次：

第一层：时间序列
日K / 分钟K / 日内分时

第二层：当前行情
实时行情

第三层：当前订单簿
五档盘口


这个划分之后，很多之前混乱的问题突然变得简单了。

1. K 线的数据粒度是“时间窗口”

5 分钟 K 则表达一个 5 分钟窗口。

我不会把 K 线对象设计成实时行情对象。

2. 实时行情的数据粒度是“当前状态”

symbol
last_price
prev_close
volume
ext.name
ext.change_pct


T1 → 当前状态
T2 → 当前状态
T3 → 当前状态


如果我不断采集，就可以把多个 snapshot 保存下来。

但单个 snapshot 并不等于完整的历史序列。

3. 五档行情的数据粒度是“当前盘口结构”

bid_prices
bid_volumes
ask_prices
ask_volumes


所以它实际上是一棵简化的订单簿结构。

我后来在设计内部对象时，甚至不再把它简单叫做 quote，而是把它单独归类为：

这样从代码结构上就能减少误用。

我曾经考虑过自己维护三套数据接口：

K线服务
实时行情服务
盘口服务


问题在于，我还需要自己处理：

后来我发现，真正让我浪费时间的不是策略，而是数据接入层。

根据我手头的 SDK 文档，QuantDash 可以通过不同模块分别获取 K 线、实时行情、日内分时、标的信息、除权因子以及五档盘口。

我真正采用它的原因不是“功能多”，而是我可以把底层数据访问和上层策略解耦。

我现在习惯先把 SDK 初始化封装起来。

import os
from quantdash import QuantDash


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

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

    return QuantDash(api_key=api_key)


def main():
    try:
        qd = create_client()
        print("QuantDash 客户端初始化成功")

    except Exception as e:
        print(f"客户端初始化失败: {e}")


if __name__ == "__main__":
    main()


这样做的好处是后面的业务代码不需要反复处理 API Key。

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:
        df = qd.klines.get(
            "600519.SH",
            period="1d",
            count=5,
            to_dataframe=True
        )

        if df.empty:
            print("没有历史行情数据")
            return

        print(
            df[
                [
                    "symbol",
                    "name",
                    "trade_date",
                    "open",
                    "high",
                    "low",
                    "close",
                    "volume"
                ]
            ].to_string(index=False)
        )

    except Exception as e:
        print(f"获取日K失败: {e}")


if __name__ == "__main__":
    main()


这里我会把 DataFrame 直接作为研究层的数据结构。

因为后面做 Pandas 分析非常方便。

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:
        df = qd.quotes.get(
            symbols=[
                "600519.SH",
                "000001.SZ"
            ],
            to_dataframe=True
        )

        if df.empty:
            print("实时行情为空")
            return

        print(
            df[
                [
                    "symbol",
                    "last_price",
                    "prev_close",
                    "volume",
                    "ext.name",
                    "ext.change_pct"
                ]
            ].to_string(index=False)
        )

    except Exception as e:
        print(f"获取实时行情失败: {e}")


if __name__ == "__main__":
    main()


我会把这个 DataFrame 看作“当前行情快照”。

import os
from quantdash import QuantDash


def get_depth(qd, symbol):
    try:
        depth = qd.depth.get(symbol)

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

        return depth

    except Exception as e:
        raise RuntimeError(
            f"获取 {symbol} 五档盘口失败: {e}"
        ) from e


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 = get_depth(qd, "600519.SH")

        print(f"标的: {depth['symbol']}")
        print(f"地区: {depth['region']}")

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

    except Exception as e:
        print(f"程序执行失败: {e}")


if __name__ == "__main__":
    main()


这里我特意没有把盘口结果强行转成和 K 线一样的 DataFrame。

因为我现在更倾向于尊重原始数据的语义。

对于实时行情，可以直接传入多个 symbols。

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)

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

    try:
        df = qd.quotes.get(
            symbols=symbols,
            to_dataframe=True
        )

        if df.empty:
            print("股票池行情为空")
            return

        print(df.to_string(index=False))

    except Exception as e:
        print(f"股票池行情获取失败: {e}")


if __name__ == "__main__":
    main()


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)

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

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

        if not result:
            print("没有返回盘口数据")
            return

        for symbol, depth in result.items():
            print(f"--- {symbol} ---")

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

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


if __name__ == "__main__":
    main()


知识库中的 SDK 文档明确提供了 depth.batch 的批量五档盘口调用方式。

避坑一：数据对象不要“大一统”

这是我重构以后最明显的体会。

如果我要做历史收益率研究，K 线的复权方式必须明确。

forward
backward
forward_additive
backward_additive
none


其中比例复权和差值复权的用途也不同。

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:
        df = qd.klines.get(
            "600519.SH",
            period="1d",
            count=10,
            adjust="forward",
            to_dataframe=True
        )

        if df.empty:
            print("没有获取到复权数据")
            return

        print(df.tail().to_string(index=False))

    except Exception as e:
        print(f"复权K线请求失败: {e}")


if __name__ == "__main__":
    main()


我现在做回测时，会明确记录使用哪一种复权方式，而不会让它成为一个隐藏参数。

避坑三：回测数据和实时数据不要混用语义

我以前最容易犯的错误之一，就是：

回测：
使用 close

实盘：
使用 last_price

然后认为两者就是同一个东西。


实际上，它们的数据形成方式不同。

量化系统里行情数据怎么分层？我把实时行情、五档和 K 线重新设计了一遍

编程语言

数据库

后端

量化数据工程核心在于行情数据分层建模，避免将K线、实时快照与五档盘口混为一谈。本文通过Python实战演示，利用QuantDash SDK分别处理三类数据，实现回测与实盘解耦，解决数据语义混乱与接口统一问题。

数据架构

数据接口

数据结构

数据处理

Python

2026上云采购 | AI焕新·智启新局

lexiang

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

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

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

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

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

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

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

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

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

量化系统里行情数据怎么分层？我把实时行情、五档和 K 线重新设计了一遍

量化系统里行情数据怎么分层？我把实时行情、五档和 K 线重新设计了一遍

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐