顺盈数据 Python SDK 安装与使用教程

Shunplus Python SDK 用于访问 Shunplus 数据 API，适合量化研究、数据分析、批处理任务和服务端拉数。SDK 已内置 API 地址，用户只需要准备 API Token，然后通过 shun_api() 或 Client() 调用数据接口。

SDK 的核心体验：

• 默认支持 pandas DataFrame，适合分析脚本和 Notebook。
• 支持结构化 TableResult，适合服务端、分页抓取和保留游标。
• 自动处理 Token 权益限制、单页条数、请求节奏和少量临时网络失败。
• 自动规范化常见 A 股代码格式。

安装

SDK 支持 Python 3.8 及以上版本。

pip install shunplus

如果项目使用 uv：

uv add shunplus

安装完成后可以直接使用 DataFrame 能力，不需要额外安装 pandas。

配置 Token

推荐在本机或服务器环境变量中配置 Token：

export SHUNPLUS_API_TOKEN="你的 API Token"

也可以在 Python 代码里保存一次 Token，后续 Client() 和 shun_api() 会自动读取：

from shunplus import set_token

set_token("你的 API Token")

生产环境建议优先使用环境变量或密钥管理服务，不要把 Token 写进代码仓库。

5 分钟跑通第一段代码

如果你主要做数据分析，推荐从 shun_api() 开始。它默认返回 pandas DataFrame。

from shunplus import shun_api

api = shun_api()

df = api.daily(
    ts_code="301662.SZ",
    start_date="20260501",
    end_date="20260510",
    fields=["ts", "open", "high", "low", "close", "volume"],
)

print(df.head())

daily() 是兼容型日线接口，适合迁移已有的 Tushare 风格脚本。返回字段仍以 Shunplus K 线接口为准，例如时间列为 ts。

两种使用入口

服务端或批处理任务建议使用上下文管理器，自动关闭底层连接池：

from shunplus import Client

with Client() as client:
    result = client.kline(period="day", symbol="SZ301662", limit=20)
    print(result.fields)
    print(result.to_dicts()[:3])

如果只想临时拿 DataFrame：

from shunplus import Client

with Client() as client:
    df = client.df.symbols(exchange="SZ", limit=20)
    print(df.head())

常用数据查询

查询基础标的

from shunplus import Client

with Client() as client:
    rows = client.symbols(exchange="SZ", limit=10, format="dict")
    print(rows)

常见 exchange 可传 SZ、SH、BJ、HK。

查询最近 K 线

如果只查最近一段 K 线，通常只传 period + symbol + limit 即可，SDK 会自动补合适的时间范围。

from shunplus import shun_api

api = shun_api()

df = api.kline(
    period="day",
    symbol="301662.SZ",
    limit=20,
    fields=["ts", "open", "high", "low", "close", "volume"],
)

print(df.tail())

period 支持 1m、5m、15m、30m、60m、120m、day、month、year。

查询分钟线

from shunplus import shun_api

api = shun_api()

df = api.stk_mins(
    ts_code="301662.SZ",
    freq="5min",
    start_date="2026-05-10 09:30:00",
    end_date="2026-05-10 15:00:00",
    fields=["ts", "open", "close", "volume"],
)

stk_mins() 是兼容型分钟线接口，freq 支持 1min、5min、15min、30min、60min、120min。

查询个股资讯与公告

from shunplus import Client

with Client() as client:
    news = client.stock_news(
        symbol="06999.HK",
        source="xueqiu",
        start_time="2026-05-01",
        end_time="2026-05-10",
        limit=20,
        format="dict",
    )

    announcements = client.announcements(
        symbol="601318.SH",
        start_time="2026-05-01",
        end_time="2026-05-10",
        limit=20,
        format="dict",
    )

print(news[:2])
print(announcements[:2])

stock_news.source 支持 xueqiu 和 futu。

返回格式

表格类接口都支持 format 参数：

from shunplus import Client

with Client() as client:
    table = client.symbols(exchange="SZ", limit=10, format="table")
    rows = client.symbols(exchange="SZ", limit=10, format="dict")
    df = client.symbols(exchange="SZ", limit=10, format="dataframe")
    raw = client.symbols(exchange="SZ", limit=10, format="raw")

TableResult 常用属性和方法：

股票代码、时间与字段

SDK 会自动规范化这些常见股票代码：

纯数字代码不做市场推断，例如 301662、06999 不能直接传入。

时间参数支持 YYYYMMDD、YYYY-MM-DD、YYYY-MM-DD HH:MM:SS、datetime.date 和无时区的 datetime.datetime。

rows = client.kline(
    period="day",
    symbol="SZ301662",
    start_time="2026-05-01T00:00:00+08:00",
    end_time="2026-05-10T23:59:59+08:00",
    fields=["ts", "close", "volume"],
    format="dict",
)

fields 可以传列表，也可以传逗号分隔字符串，用来只保留关心的列。

分页拉取

大多数列表型接口都支持分页。小批量请求可以用 fetch_all() 一次性合并拉取，注意，使用 fetch_all() 会将拉取后的结果放置于内存中一次性合并返回，不适用于大量数据场景：

from shunplus import Client

with Client() as client:
    rows = client.fetch_all(
        "symbols",
        exchange="SZ",
        limit=1000,
        format="dict",
    )

print(len(rows))

大结果量建议用 iter_pages() 边拉边处理：

from shunplus import Client

with Client() as client:
    for page in client.iter_pages(
        "stock_news",
        symbol="SZ301662",
        source="xueqiu",
        limit=100,
    ):
        for row in page.iter_dicts():
            print(row["title"])

异常、重试与限流

SDK 自定义异常都继承自 ShunplusError。接口返回 HTTP 错误或非成功业务码时，SDK 会抛出 ApiError 或它的子类，并在异常对象上保留排障信息：

常见异常类型：

自动重试规则

在遇到可恢复的临时问题时，SDK 默认最多额外尝试 1 次。

会自动重试的情况：

• 网络连接错误。
• 请求超时。
• HTTP 500、502、503、504。

网络连接错误或超时在重试后仍失败时，会透出底层 httpx 异常；HTTP 5xx 在重试后仍失败时会抛出 ServerError。

不会自动重试的情况：

• HTTP 401、403：认证或权限问题。
• HTTP 400、422：参数问题。
• HTTP 429：服务端已经明确限流。
• 服务端返回非成功业务码但 HTTP 状态码为 200 的业务错误。

可以按需调整或关闭重试：

from shunplus import Client

client = Client(max_retries=2, retry_backoff=1.0)
client_without_retry = Client(max_retries=0)

本地限流规则

默认情况下，SDK 会在首次查询数据前调用 entitlements() 获取当前 Token 权益，并在客户端侧提前做一层限流，减少主动打满套餐的失败。服务端仍然是最终校验方；如果权益查询本身失败，SDK 会跳过本地限流配置，让后续请求直接由服务端校验。

SDK 会使用服务端购买的套餐权益来控制限流，主要包括每分钟请求次数的限流以及并发上限的控制：

如果你希望完全自己控制请求节奏，可以关闭自动权益配置：

from shunplus import Client

client = Client(auto_configure_limits=False)

from shunplus import Client, RateLimitError, ShunplusError, ValidationError

with Client() as client:
    try:
        rows = client.kline(period="day", symbol="SZ301662", limit=20, format="dict")
    except RateLimitError as exc:
        print("调用频率超限：", exc)
        print("建议等待秒数：", exc.retry_after)
    except ValidationError as exc:
        print("参数错误：", exc)
    except ShunplusError as exc:
        print("请求失败：", exc)

方法速查

常见问题

提示缺少 Token 怎么办？
确认已经设置 SHUNPLUS_API_TOKEN，或在代码中调用 set_token("你的 API Token")。

为什么传 301662 会报错？
纯数字代码无法判断市场。请传 301662.SZ 或 SZ301662 这类无歧义格式。

为什么我传的 limit 没有完全生效？
SDK 会根据当前 Token 权益自动调整单页条数，服务端也会按套餐限制进行最终校验。

数据量很大时应该怎么拉？
优先使用 iter_pages() 分页处理，避免一次性把所有数据放进内存。