在量化交易和金融數據分析領域,獲取實時、高質量的股票行情數據是關鍵。亞太地區作為全球經濟的重要引擎,日本、韓國、新加坡和印尼等市場的股市行情備受關注。本文將介紹一個高效的亞太股市數據 API,支持日本股市行情、韓國股市行情、新加坡股市行情和印尼股市行情的實時數據查詢。該 API 提供實時價格、K 線數據、歷史數據查詢、實時 tick、實時報價、低延時報價 api、盤口數據和逐筆成交等功能,特別適合量化交易開發者使用。我們將重點通過 Python 代碼示例來演示如何實現這些功能。
API 概述
這個亞太股市數據 API 專注於日本(JP)、韓國(KR)、新加坡(SG)和印尼(ID)等市場的股票行情數據。數據源直接對接交易所,確保低延遲和高可用性。API 支持 RESTful 接口和 WebSocket 推送,覆蓋實時 tick、實時報價、盤口數據、逐筆成交以及歷史 K 線數據查詢。無論是構建投資儀表盤、進行量化交易策略回測,還是監控市場波動,這個 API 都能提供可靠的支持。
主要優勢:
- 實時數據:毫秒級更新,支持實時價格和逐筆成交。
- 歷史數據查詢:批量獲取多股 K 線數據,包括開盤價、最高價、最低價、收盤價和成交量(OHLCV)。
- 多市場覆蓋:專注於亞太地區,輕鬆切換日本股市行情、韓國股市行情、新加坡股市行情和印尼股市行情。
- 低延時報價 API:適合高頻交易場景。
- 易集成:提供 Python、Java、Go 等多種語言示例,這裏重點用 Python 實現。
在使用前,需要獲取 API token,並注意速率限制和訂閲計劃。
核心功能詳解
1. 實時成交(Tick 數據)
這個功能提供股票的逐筆成交數據,包括最新價、成交數量和時間戳。適用於監控實時 tick 和逐筆成交,支持低延時報價 api。
- 請求路徑:GET /stock/tick?region={region}&code={code}
-
參數:
- region:市場代碼(如 JP 日本、KR 韓國、SG 新加坡、ID 印尼)。
- code:股票代碼。
- 響應:包含產品代碼、最新價、時間戳和成交數量。
2. 批量歷史 K 線查詢
支持多股同時查詢歷史 K 線數據,週期從分鐘線到月線。適合歷史數據查詢和 K 線數據分析。
- 請求路徑:GET /stock/klines?region={region}&codes={codes}&kType={kType}&limit={limit}&et={et}
-
參數:
- region:市場代碼。
- codes:多個股票代碼,用逗號分隔。
- kType:K 線類型(1:分鐘 K,8:日 K 等)。
- limit:K 線數量。
- et:截止時間戳(可選)。
- 響應:OHLCV 數據數組,包括成交額。
3. WebSocket 實時行情推送
通過 WebSocket 實現實時報價、盤口數據和逐筆成交的推送。適合需要持續監控的應用,如量化交易系統。
- 連接:wss://api.itick.org/stock,攜帶 token header。
- 訂閲:發送 JSON 指令訂閲股票和類型(tick、quote、depth)。
- 心跳:每 30 秒發送 ping 保持連接。
Python 代碼示例
下面通過 Python 代碼演示如何使用這些功能。假設你已經安裝了requests和websocket-client庫(pip install requests websocket-client)。
示例 1:獲取實時成交(Tick 數據)
import requests
# API基礎URL和token
BASE_URL = "https://api.itick.org"
TOKEN = "your_token" # 替換為你的API token
def get_tick_data(region, code):
url = f"{BASE_URL}/stock/tick?region={region}&code={code}"
headers = {
"accept": "application/json",
"token": TOKEN
}
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
if data['code'] == 0:
tick = data['data']
print(f"股票 {tick['s']} 最新價: {tick['ld']}, 成交量: {tick['v']}, 時間: {tick['t']}")
else:
print("請求失敗:", data['msg'])
else:
print("HTTP錯誤:", response.status_code)
# 示例:查詢日本股市行情(例如Toyota股票,代碼7203)
get_tick_data("JP", "7203")
# 示例:查詢韓國股市行情(例如Samsung,代碼005930)
get_tick_data("KR", "005930")
# 示例:查詢新加坡股市行情(例如DBS銀行,代碼D05)
get_tick_data("SG", "D05")
# 示例:查詢印尼股市行情(例如Bank Central Asia,代碼BBCA)
get_tick_data("ID", "BBCA")
這個示例展示瞭如何獲取實時 tick 數據,支持日本股市行情、韓國股市行情等市場的低延時報價 api。
示例 2:批量歷史 K 線查詢
import requests
def get_klines(region, codes, k_type, limit):
url = f"{BASE_URL}/stock/klines?region={region}&codes={codes}&kType={k_type}&limit={limit}"
headers = {
"accept": "application/json",
"token": TOKEN
}
response = requests.get(url, headers=headers)
if response.status_code == 200:
data = response.json()
if data['code'] == 0:
for code, klines in data['data'].items():
print(f"股票 {code} 的K線數據:")
for k in klines:
print(f"時間: {k['t']}, 開: {k['o']}, 高: {k['h']}, 低: {k['l']}, 收: {k['c']}, 量: {k['v']}")
else:
print("請求失敗:", data['msg'])
else:
print("HTTP錯誤:", response.status_code)
# 示例:批量查詢韓國股市行情和新加坡股市行情的日K線(5條)
get_klines("KR", "005930,000660", 8, 5) # Samsung和SK Hynix
get_klines("SG", "D05,O39", 8, 5) # DBS和OCBC
這個功能便於歷史數據查詢和 K 線數據分析,支持多股批量操作。
示例 3:WebSocket 實時行情推送
import websocket
import json
import threading
import time
WS_URL = "wss://api.itick.org/stock"
def on_message(ws, message):
data = json.loads(message)
if data.get("code") == 1 and data.get("msg") == "Connected Successfully":
print("連接成功")
elif data.get("resAc") == "auth" and data.get("code") == 1:
print("認證成功")
subscribe(ws)
elif data.get("resAc") == "subscribe" and data.get("code") == 1:
print("訂閲成功")
elif data.get("data"):
market_data = data["data"]
data_type = market_data.get("type")
symbol = market_data.get("s")
print(f"{data_type} 數據 for {symbol}: {market_data}")
def on_error(ws, error):
print("錯誤:", error)
def on_close(ws, close_status_code, close_msg):
print("連接關閉")
def on_open(ws):
print("WebSocket連接打開")
def subscribe(ws):
subscribe_msg = {
"ac": "subscribe",
"params": "7203$JP,005930$KR,D05$SG,BBCA$ID", # 日本、韓國、新加坡、印尼股票
"types": "tick,quote,depth" # 實時tick、報價、盤口數據
}
ws.send(json.dumps(subscribe_msg))
print("訂閲消息已發送")
def send_ping(ws):
while True:
time.sleep(30)
ping_msg = {
"ac": "ping",
"params": str(int(time.time() * 1000))
}
ws.send(json.dumps(ping_msg))
print("Ping 已發送")
if __name__ == "__main__":
ws = websocket.WebSocketApp(
WS_URL,
header={"token": TOKEN},
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close
)
ping_thread = threading.Thread(target=send_ping, args=(ws,))
ping_thread.daemon = True
ping_thread.start()
ws.run_forever()
這個 WebSocket 示例實現了實時報價、盤口數據和逐筆成交的推送,適用於量化交易場景。
結語
通過這個亞太股市數據 API,你可以輕鬆獲取日本股市行情、韓國股市行情、新加坡股市行情和印尼股市行情的股票行情數據。Python 實現簡單高效,適合初學者和專業開發者。注意在實際使用中,遵守 API 的速率限制,並處理異常情況。如果你是量化交易愛好者,這個 API 將大大提升你的數據獲取效率。
温馨提示:本文僅供參考,不構成任何投資建議。市場有風險,投資需謹慎
GitHub:https://github.com/itick-org/