如何對接韓國和日本股票數據源API 詳情 - json,數據結構,python爬蟲 CryptoRzz 動態日志

摘要：本文將詳細介紹如何通過StockTV API對接韓國和日本股票市場的實時行情數據，涵蓋從API申請、接口調用到數據解析的完整流程，並提供Python代碼示例。

一、StockTV API

StockTV API 提供全球多個國家的金融市場數據，其中韓國和日本市場數據尤為全面。以下是選擇StockTV API的幾大理由：

覆蓋全面：支持韓國KOSPI、KOSDAQ和日本東京證券交易所
實時性強：提供WebSocket實時數據推送
數據豐富：包含股票、指數、期貨等多種金融產品
易於集成：提供多種語言的SDK和詳細文檔

二、準備工作

1. 註冊開發者賬號

訪問StockTV開發者門户，獲取測試key並創建應用。

2. 獲取API密鑰

系統會分配一個唯一的API Key，用於身份驗證。

3. 安裝依賴

pip install requests websocket-client

三、獲取韓國股票數據

1. API接口説明

URL: https://api.stocktv.top/stock/stocks
Method: GET
參數:
- key: API密鑰
- countryId: 國家ID（韓國為43）
- pageSize: 每頁數量
- page: 頁碼

2. Python實現

import requests

def get_korea_stocks(api_key, page_size=20, page=1):
    url = "https://api.stocktv.top/stock/stocks"
    params = {
        "key": api_key,
        "countryId": 43,
        "pageSize": page_size,
        "page": page
    }
    
    response = requests.get(url, params=params)
    response.raise_for_status()
    return response.json()

# 使用示例
api_key = "YOUR_API_KEY"
stocks = get_korea_stocks(api_key)
for stock in stocks['data']['records']:
    print(f"{stock['name']}: {stock['last']}")

四、獲取日本股票數據

1. API接口説明

URL: https://api.stocktv.top/stock/stocks
Method: GET
參數:
- key: API密鑰
- countryId: 國家ID（日本為44）
- pageSize: 每頁數量
- page: 頁碼

2. Python實現

def get_japan_stocks(api_key, page_size=20, page=1):
    url = "https://api.stocktv.top/stock/stocks"
    params = {
        "key": api_key,
        "countryId": 44,
        "pageSize": page_size,
        "page": page
    }
    
    response = requests.get(url, params=params)
    response.raise_for_status()
    return response.json()

# 使用示例
stocks = get_japan_stocks(api_key)
for stock in stocks['data']['records']:
    print(f"{stock['name']}: {stock['last']}")

五、獲取實時行情數據

1. WebSocket接口説明

URL: wss://ws-api.stocktv.top/connect?key=YOUR_API_KEY
數據格式: JSON
字段説明:
- symbol: 股票代碼
- last: 最新價格
- pcp: 漲跌幅

2. Python實現

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    if data.get("type") == "stock":
        print(f"[{data['symbol']}] 價格: {data['last']} 漲跌幅: {data['pcp']}%")

def on_error(ws, error):
    print(f"錯誤: {error}")

def on_close(ws, close_status_code, close_msg):
    print("連接關閉")

def on_open(ws):
    print("連接成功")

# 啓動WebSocket客户端
api_key = "YOUR_API_KEY"
ws_url = f"wss://ws-api.stocktv.top/connect?key={api_key}"

ws = websocket.WebSocketApp(ws_url,
                          on_message=on_message,
                          on_error=on_error,
                          on_close=on_close)
ws.on_open = on_open
ws.run_forever()

六、獲取KOSPI指數數據

1. API接口説明

URL: https://api.stocktv.top/stock/indices
Method: GET
參數:
- key: API密鑰
- countryId: 國家ID（韓國為43）
- flag: 國家代碼（韓國為"KR"）

2. Python實現

def get_kospi_index(api_key):
    url = "https://api.stocktv.top/stock/indices"
    params = {
        "key": api_key,
        "countryId": 43,
        "flag": "KR"
    }
    
    response = requests.get(url, params=params)
    response.raise_for_status()
    return response.json()

# 使用示例
kospi_data = get_kospi_index(api_key)
for index in kospi_data['data']:
    if index['name'] == 'KOSPI':
        print(f"KOSPI 最新價: {index['last']}")

七、高級功能

1. 數據緩存

from functools import lru_cache

@lru_cache(maxsize=100)
def get_cached_korea_stocks(api_key):
    return get_korea_stocks(api_key)

2. 異常重試

from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def get_stocks_with_retry(api_key):
    return get_korea_stocks(api_key)

八、常見問題解答

Q1：如何獲取完整的股票列表？

通過分頁參數遍歷所有數據頁

def get_all_korea_stocks(api_key):
  page = 1
  all_stocks = []
  while True:
      data = get_korea_stocks(api_key, page=page)
      all_stocks.extend(data['data']['records'])
      if len(data['data']['records']) < 20:
          break
      page += 1
  return all_stocks

Q2：WebSocket斷線如何處理？

實現自動重連機制

import time

def run_websocket():
  while True:
      try:
          ws.run_forever()
      except Exception as e:
          print(f"連接異常: {e}, 5秒後重連...")
          time.sleep(5)

Q3：如何提高數據獲取效率？

使用異步請求

import aiohttp
import asyncio

async def async_get_stocks(api_key):
  async with aiohttp.ClientSession() as session:
      url = "https://api.stocktv.top/stock/stocks"
      params = {
          "key": api_key,
          "countryId": 43
      }
      async with session.get(url, params=params) as response:
          return await response.json()