跳至主要内容

Python SDK

SDK 包裝 /v2，回傳可轉成 pandas、polars 或 Arrow 的表格。需要 Python 3.10 以上。

安裝​

pip install "veinmesh[all] @ https://data.cyberun.cloud/docs/sdk/veinmesh-0.2.0-py3-none-any.whl"

只裝需要的：veinmesh[pandas]、veinmesh[polars]、veinmesh[arrow]、 veinmesh[live] 或 [all]。基本套件只需要 httpx。

import veinmesh

client = veinmesh.Client()  # VEINMESH_API_KEY

Client(key=..., base_url=..., timeout=120, pending_timeout=3600, retries=3)。 pending_timeout 是等待資料準備的上限；None 代表一直等。忙碌回應與網路錯誤 會重試。

用 context manager 關閉連線池：

with veinmesh.Client() as client:
    venues = client.venues()

目錄​

client.venues()
client.products(venue="SHFE")
client.instruments(venue="SHFE", product="SHFE.rb", kind="future")
client.instrument("SHFE.rb2701")   # 規格與資料集
client.datasets(venue="SHFE")
client.calendar("SHFE", "2026-10-19", "2026-10-23", product="SHFE.rb")
client.schemas()

instruments() 會跟完所有分頁，回傳完整清單。instrument() 回傳規格與它的 資料集。

資料​

t = client.data("book.snapshot", ["SHFE.rb2701"], trading_day="2026-10-20")
t = client.data("book.snapshot", "SHFE.rb2701", from_day="2026-10-01", to_day="2026-10-20",
                fields=["last", "volume"], start=None, end=None)

data() 會把過長區間與商品清單拆成 API 接受的請求、跟 next_cursor，回傳 一張先依商品再依時間排序的表。傳 catalog_version 可固定版本。請求視窗與 商品數見限制。

K 線​

daily = client.bars("SHFE.rb.main", "1d", from_day="2025-01-02", to_day="2026-10-20",
                    adjust="diff")
eighteen = client.bars("SHFE.rb2701", 1080, trading_day="2026-10-20")

interval 是秒數（60、1080）或 timedelta，或 1d、1w、1mo、1q、 1y。其他鍵跟 /v2/bars 相同：type、anchor、adjust、transform、 fill、dataset、catalog_version。連續合約接受 diff 與 ratio。

表格​

data() 與 bars() 回傳 Table：

len(t)                 # 列數
t.columns              # 欄位名稱
t[0]                   # 一列，dict
t.to_pandas()          # 小數預設轉 float
t.to_polars()
t.to_arrow()
t.catalog_version

to_pandas(exact=True)（polars 與 Arrow 也有同一選項）保留精確 Decimal， 不轉 float。

即時​

for message in client.stream("SHFE.rb2701"):
    for row in message.rows:
        print(row["ts"], row["last"])

client.stream(..., bars={"interval": 60}) 改收 K 線；message.bar 與 message.complete 描述該根 K 線。start 從當天的 RFC 3339 時間點開始。SDK 會重連並從最後收到的資料列接續。需要 websockets extra。

匯出​

job = client.export("book.snapshot", ["SHFE.rb2701", "SHFE.rb2705"],
                    from_day="2026-09-01", to_day="2026-10-20", format="parquet")
df = job.wait().to_pandas()
files = job.download("./data")

job.wait(poll=10, timeout=None) 輪詢到 done；download() 寫出檔案； to_pandas()、to_polars()、to_arrow() 會下載並合併。client.get_export(id) 可重開工作。

錯誤​

  • APIError 帶 status 與 problem dict；code 是穩定部分。
  • PendingTimeout 代表等了 pending_timeout，資料還在準備。
  • ExportFailed 代表匯出工作以 failed 結束；訊息含工作與 problem。

錯誤碼與重試方式見錯誤。