datapush-sdk

datapush A 股行情 SDK（Python）：REST 历史 + WebSocket 实时 + 本地按需缓存。 零第三方依赖（只用标准库），pip install 后即可用 命令行 与 Python 两种方式使用，前缀都是 datapush。

pip install datapush-sdk
datapush config set-key sk-xxx     # 保存 API Key（也可用环境变量 DATAPUSH_API_KEY） https://datapush.vip/ 获取
datapush kline 000001              # 一行拿到最近 30 个交易日日K（自动缓存）

import datapush
c = datapush.Client()              # Key 自动读；缓存默认开启
page = c.kline("000001", days=250)  # 一行拿数据；第二次调用命中缓存
for bar in page:
    print(bar.time, bar.close)

1. 安装与配置

pip install datapush-sdk            # 运行时零依赖
pip install "datapush-sdk[pandas]"  # 可选：需要 to_pandas()

API Key 的解析优先级（高 → 低）：显式参数 → 环境变量 DATAPUSH_API_KEY → ~/.datapush/config.json。 推荐用 CLI 保存一次即可：datapush config set-key sk-xxx（也可 datapush config show 查看，Key 会打码）。

服务地址固定为 https://api.datapush.vip：不需要、也不支持通过 CLI 参数或配置文件修改 （config set base_url 会被拒绝，配置文件里的旧 base_url 键不再生效）。

验证安装

datapush --version        # datapush-sdk 0.1.5
datapush doctor           # 一键体检：版本 / CLI 是否在 PATH / Key / 网络 / 缓存
python -c "import datapush; print(datapush.__version__)"

datapush 命令找不到怎么办？

pip 会把控制台脚本装到解释器的 Scripts（Windows）或 bin（macOS/Linux）目录，该目录可能不在 PATH， 于是 datapush --version 会提示「不是内部或外部命令」。三种修法（任选其一）：

1. 不改 PATH，直接用模块入口（最省事，任何环境都能用）：

   python -m datapush kline 000001
   python -m datapush doctor
   

2. 把脚本目录加进 PATH（一次对所有 pip 安装的 CLI 生效）：

   • Windows PowerShell（只改用户级，不碰系统 PATH；重开终端生效）：

     [Environment]::SetEnvironmentVariable("PATH", [Environment]::GetEnvironmentVariable("PATH","User") + ";<Scripts目录>", "User")
     

   • macOS / Linux（写入 ~/.zshrc 或 ~/.bashrc 后重开终端）：

     export PATH="$PATH:$(python -c 'import sysconfig;print(sysconfig.get_path("scripts"))')"
     

   <Scripts目录> 可用 python -c "import sysconfig;print(sysconfig.get_path('scripts'))" 查出。

3. 用 pipx 安装 CLI（自带 PATH 管理、彼此隔离）：

   python -m pip install pipx && pipx install datapush-sdk
   

拿不准就用哪种？跑 datapush doctor（或 python -m datapush doctor）——它会按你这台机器的实际路径 直接打印该执行的那条命令。

2. 本地按需缓存（默认开启，用户无感）

• 只存不复权原始价（adj=none）+ 复权因子表 ⇒ 历史永不失效、只需追加；
• 复权在本地算：kind="qfq"|"hfq"|"raw"，除权后自动正确、无需重下历史；
• 增量续拉：用游标 latest_time 只拉新增（当日 forming bar 同键覆盖写，天然幂等）；
• 范围自动补拉：这次要 30 天、下次要 250 天，会自动把缺的那段补上；
• 断网可读：Client(offline=True) 或 --offline 只读本地缓存；
• 复权因子表按「当日分界」刷新：当天 09:15 之后取过的，当天不再刷新；09:15 之前（含昨天）取的 ⇒ 当天刷新一次（除权事件到此时已定型）。可配：Client(segments_cutoff="09:30") 改分界、 Client(segments_ttl=秒数) 退回固定 TTL、c.xdxr(code, refresh=True) 强制刷新；
• 缓存位置：~/.datapush/cache/datapush.db（SQLite 单文件）；可用 --cache-dir / DATAPUSH_CACHE_DIR / 配置文件 cache_dir 指定；DATAPUSH_NO_CACHE=1 或 --no-cache 彻底关闭。

datapush cache info                 # 路径 / 大小 / 标的数 / 根数 / 时间范围
datapush cache path                 # 只打印缓存文件路径（脚本友好）
datapush cache prune --days 90      # 只保留最近 90 天
datapush cache clear --code 000001  # 清掉某只票
datapush update 000001 --period 1m  # 手动增量（平时查询时自动做）

K 线以外都不落盘：股票列表 / 基础信息 / 交易日 / 板块 / 量化因子（目录与数组） 全部直连不缓存 （流量很小）：datapush codes（全市场代码，约 6 页 / 约 1MB）、datapush stock 000001。 量化因子当日行盘中会变、历史行盘后才定型 ⇒ 需要实时因子请用 watch(types=["factor"])（WS 推送全字段）， 而不是反复打 REST（服务端也建议盘中少查）。

3. Python 用法速查

import datapush
c = datapush.Client()                       # 也可 Client(api_key=..., cache=False, offline=True)

# 历史
c.codes()                                   # 全市场代码（实时）
c.stocks(all=True)                          # 全市场列表（实时）；c.stocks(page=1, page_size=20) 单页
c.stock("000001")                           # 基础信息 12 字段（实时）
c.kline("000001", days=250)                 # 日K（前复权）
c.kline("000001", days=250, kind="hfq")     # 后复权（本地因子表算）
c.kline("000001", start="2026-01-01", end="2026-09-26", kind="raw")
c.minute("000001", days=1)                  # 1 分钟 K 线（当日）
c.fiveminute("000001", days=5)              # 5 分钟 K 线
c.kline("000001", since="2026-09-26 14:59") # 增量（闭区间，含该根）
c.xdxr("000001")                            # 复权因子表（当天 09:15 后取过就不再刷新；除权后自动重算）
c.factors("000001", size=30, fields="close,ma5")   # 量化因子（高级用户；直连不缓存）
c.factor_catalog()                          # 因子目录（直连不缓存）

# 实时（WebSocket）
q = c.quote("000001")                       # 一次快照（WS 首帧；REST 没有该接口）
for e in c.watch(["000001", "600108"], types=["quote", "k1"], seconds=30):
    if e.type == "quote":
        print(e.quote.price)

# 导出与缓存
c.kline("000001", days=250).to_csv("000001.csv")
c.kline("000001", days=250).to_dicts()      # list[dict]
c.to_pandas(c.kline("000001", days=250))    # 需要 pandas
c.cache_info(); c.update("000001", period="day"); c.close()

4. CLI 速查

datapush kline 000001 [--days 250 | --size 100 --offset 0 | --start 2026-01-01 --end 2026-09-26 | --since "..."]
                     [--period day|1m|5m] [--kind qfq|hfq|raw] [--limit 60] [--csv out.csv] [--json] [--no-refresh]
datapush minute 000001 --days 1
datapush fiveminute 000001 --days 5
datapush xdxr 000001 [--refresh]
datapush stocks [--page 1 --page-size 20] [--all]
datapush codes [--market sh|sz|bj] [--prefixed]        # 默认一行一个，可直接管道
datapush stock 000001
datapush factors 000001 --size 30 [--fields close,ma5] [--daily-kline 0]
datapush quote 000001 [--types quote|depth]
datapush watch 000001 600108 [--types quote,k1] [--seconds 30]
datapush update 000001 --period 1m
datapush cache info|path|clear|prune
datapush config set-key sk-xxx | config show | config path | config set cache_dir D:/datapush-cache
datapush doctor                                        # 一键排障
datapush get /api/v1/stocks/000001/xdxr --param k=v     # 通用逃生舱（调试新接口）

通用选项可写在命令前或命令后：--json（原始 JSON，便于管道）、--csv FILE、--api-key、 --timeout、--cache-dir、--no-cache、--offline。 退出码：0 成功 / 1 业务或网络错误 / 2 参数错误。

5. 本地复权（与官方文档公式一致）

adj=none 取原始价 + xdxr() 取因子表 ⇒ 一次乘法：

段 = segments 里第一个 from ≤ bar.date（都不满足 ⇒ from 为空的兜底段）
前复权价 = raw价 × seg.qfq      后复权价 = raw价 × seg.hfq
换手率   = cjl / ltgb × seg.share_ratio × 100        自检：seg.hfq × P == seg.qfq（P = 兜底段 qfq）

SDK 已内置：datapush.adjust_bars(bars, table, kind=...)、datapush.pick_segment(table, date)、 datapush.verify_segments(table)。细节见官方文档「复权因子表（本地复权计算）」。

6. 常见问题

• import datapush 不是我装的这个包？ 尝试使用 import datapush_sdk as datapush。

• 需要 Key 但没有？ 先 datapush config set-key sk-xxx（注册后在控制台复制）；datapush doctor 可自查。

• 返回 402 / 429？ 402 = 积分不足；429 = 分钟接口限流（10 QPS/用户，SDK 已按 Retry-After 自动退避重试）。

• 能离线用吗？ 能：缓存过的标的用 offline=True/--offline 可读；实时接口必须联网。

• WS 单连接能订阅多少？ 最多 50 个 code（SDK 会自动分批订阅）；普通用户最多 10 个连接。

• 怎么确认 SDK 拿到的数据和线上一致？ 官方仓库带零依赖对拍脚本 tools/verify_live.py： A 路径用 urllib 直连线上（WS 也是独立实现，不复用 SDK 代码），B 路径走 SDK，逐接口比对 （K 线按时间戳位级比对；当日 forming bar 单独说明）。Key 只从环境变量读、不写文件、输出打码：

  set DATAPUSH_API_KEY=sk-xxx     # PowerShell：$env:DATAPUSH_API_KEY="sk-xxx"
  python tools/verify_live.py     # 全量（REST + WS）；--gate 只跑门禁三项；--json r.json 落结果
  

7. 兼容性与许可

• Python 3.8+（Windows / macOS / Linux）。3.8 是下限的真正原因：本地缓存用到 SQLite 的 UPSERT 语法（ON CONFLICT … DO UPDATE，需 SQLite ≥ 3.24；官方 Windows 包自 Python 3.8 起自带 ≥ 3.31）。 更老的 Python 也能用，只要 Client(cache=False) / DATAPUSH_NO_CACHE=1 关掉缓存 （SQLite 太老时 SDK 会直接提示这句）。
• 代码层面未使用任何 3.9+ 语法/特性：仓库自带 python tools/compat_check.py，会对 3.9 / 3.8 / 3.7 三档语法做校验（并扫描 match、removeprefix、zoneinfo、tomllib 等新版特性）。
• 运行时零第三方依赖（仅标准库）；py.typed 已附带，支持类型检查。
• License: MIT。接口与字段含义以官方文档为准（https://api.datapush.vip/dp/docs）。