跳至主要内容

快速開始

本指南將協助您透過 API 存取行情數據，內容涵蓋：安裝 Webull SDK、申請 API 金鑰，以及查詢歷史和即時行情數據的方法

1. 安裝 Webull Client SDK​

系統需求​

需安裝 Python 3.8 至 3.11 版本

安裝 SDK​

Install via pip

pip3 install --upgrade webull_openapi_python_sdk

2. 產生 API 金鑰並進行驗證​

每次呼叫 API 都需要根據 App Key 與以 App Secret 生成的簽名進行認證 客戶端必須在 HTTP 請求標頭中分別加入名為 x-app-key 與 x-signature 的 App Key 及簽名值 如需取得 App Key 和 App Secret，請參考：申請應用程式

注意

為符合香港的安全及監管要求，除了 App Key 及簽名驗證外，OpenAPI 還需要進行 Token 驗證 有關 Token 的建立與驗證方法，請參考：Token 建立與驗證

3. 透過 SDK 請求行情資料​

安裝 SDK 並取得 API 金鑰後，即可使用行情資料 API。以下範例說明如何請求 K 線圖資料。其他類型的資料，請參考行情資料 API 參考文件。

請求歷史資料範例：以 K 線圖資料為例​

from webull.data.common.category import Category
from webull.data.common.timespan import Timespan
from webull.core.client import ApiClient
from webull.data.data_client import DataClient

optional_api_endpoint = "<api_endpoint>"
your_app_key = "<your_app_key>"
your_app_secret = "<your_app_secret>"
region_id = "<region_id>"
api_client = ApiClient(your_app_key, your_app_secret, region_id)
api_client.add_endpoint(region_id, optional_api_endpoint)

if __name__ == '__main__':
    data_client = DataClient(api_client)

    res = data_client.market_data.get_history_bar('AAPL', Category.US_STOCK.name, Timespan.M1.name)
    if res.status_code == 200:
        print('get_history_bar:', res.json())
        
    res = data_client.market_data.get_batch_history_bar(['AAPL', 'TSLA'], Category.US_STOCK.name, Timespan.M1.name, 1)
    if res.status_code == 200:
        print('get_batch_history_bar:', res.json())

以下範例說明如何使用 SDK 取得即時報價推播。如果您不打算使用 SDK，請參考資料串流 API。

取得即時報價推播範例​

from webull.data.common.category import Category
from webull.data.common.subscribe_type import SubscribeType
from webull.data.data_streaming_client import DataStreamingClient


your_app_key = "<your_app_key>"
your_app_secret = "<your_app_secret>"
optional_api_endpoint = "api.sandbox.webull.hk"
optional_quotes_endpoint = "data-api.sandbox.webull.hk"
region_id = "<region_id>"

session_id = "demo_session_1"
data_streaming_client = DataStreamingClient(your_app_key, your_app_secret, region_id, session_id,
                                    http_host=optional_api_endpoint,
                                    mqtt_host=optional_quotes_endpoint)

if __name__ == '__main__':
    def my_connect_success_func(client, api_client,session_id):
        print("connect success with session_id:%s" % client.get_session_id())

        symbols = ['AAPL']
        sub_types = [SubscribeType.QUOTE.name, SubscribeType.SNAPSHOT.name, SubscribeType.TICK.name]
        client.subscribe(symbols, Category.US_STOCK.name, sub_types)


    def my_quotes_message_func(client, topic, quotes):
        print("receive message: topic:%s, quotes:%s" % (topic, quotes))


    def my_subscribe_success_func(client, api_client,session_id):
        print("subscribe success with session_id:%s" % client.get_session_id())


    # set connect success callback func
    data_streaming_client.on_connect_success = my_connect_success_func
    # set quotes receiving callback func
    data_streaming_client.on_quotes_message = my_quotes_message_func
    # set subscribe success callback func
    data_streaming_client.on_subscribe_success = my_subscribe_success_func
    # the sync mode, blocking in current thread
    data_streaming_client.connect_and_loop_forever()