Python對接StockTV全球金融數據API的封裝實現及使用教程：

import requests
import websockets
import asyncio
from typing import Dict, List, Optional, Union
from datetime import datetimeclass StockTVClient:"""StockTV全球金融數據API客戶端支持股票、外匯、期貨、加密貨幣等市場數據"""BASE_URL = "https://api.stocktv.top"WS_URL = "wss://ws-api.stocktv.top/connect"def __init__(self, api_key: str):"""初始化客戶端:param api_key: API密鑰，需通過官方渠道獲取"""self.api_key = api_keyself.session = requests.Session()self.session.headers.update({"User-Agent": "StockTV-PythonClient/1.0"})def _handle_response(self, response: requests.Response) -> Union[Dict, List]:"""統一處理API響應"""if response.status_code != 200:raise Exception(f"API請求失敗，狀態碼：{response.status_code}，響應：{response.text}")return response.json()# ------------------ 股票市場接口 ------------------def get_stock_markets(self, country_id: int,page: int = 1,page_size: int = 10) -> Dict:"""獲取股票市場列表:param country_id: 國家ID（例如14代表印度）:param page: 頁碼:param page_size: 每頁數量"""endpoint = "/stock/stocks"params = {"countryId": country_id,"page": page,"pageSize": page_size,"key": self.api_key}response = self.session.get(f"{self.BASE_URL}{endpoint}", params=params)return self._handle_response(response)def get_stock_kline(self,pid: int,interval: str = "PT15M",start_time: Optional[int] = None,end_time: Optional[int] = None) -> List[Dict]:"""獲取股票K線數據:param pid: 產品ID:param interval: 時間間隔（PT5M, PT15M, PT1H等）:param start_time: 開始時間戳（可選）:param end_time: 結束時間戳（可選）"""endpoint = "/stock/kline"params = {"pid": pid,"interval": interval,"key": self.api_key}if start_time:params["startTime"] = start_timeif end_time:params["endTime"] = end_timeresponse = self.session.get(f"{self.BASE_URL}{endpoint}", params=params)return self._handle_response(response)# ------------------ 外匯接口 ------------------def get_forex_rates(self, base_currency: str = "USD") -> Dict:"""獲取實時外匯匯率"""endpoint = "/market/currencyList"params = {"key": self.api_key}response = self.session.get(f"{self.BASE_URL}{endpoint}", params=params)data = self._handle_response(response)return data.get("conversions", {}).get(base_currency, {})# ------------------ WebSocket實時數據 ------------------async def websocket_client(self, callback):"""WebSocket實時數據客戶端:param callback: 數據處理回調函數"""url = f"{self.WS_URL}?key={self.api_key}"async with websockets.connect(url) as ws:while True:try:data = await ws.recv()await callback(json.loads(data))# 發送心跳保持連接await asyncio.sleep(30)await ws.send("ping")except Exception as e:print(f"WebSocket錯誤: {str(e)}")break# ================== 使用示例 ==================
if __name__ == "__main__":API_KEY = "YOUR_API_KEY"  # 替換為實際API密鑰# 初始化客戶端client = StockTVClient(API_KEY)# 示例1：獲取印度股票市場列表india_stocks = client.get_stock_markets(country_id=14)print(f"印度股票市場數據：{india_stocks['data']['records'][0]}")# 示例2：獲取股票K線數據kline_data = client.get_stock_kline(pid=7310, interval="PT1H")print(f"最新K線數據：{kline_data['data'][-1]}")# 示例3：WebSocket實時數據async def handle_realtime_data(data):"""處理實時數據回調函數"""print(f"實時更新：{data}")async def main():await client.websocket_client(handle_realtime_data)# 運行WebSocket客戶端asyncio.get_event_loop().run_until_complete(main())

關鍵功能說明：

1. 模塊化設計：

• 使用面向對象封裝，方便擴展和維護
• 分離不同市場接口（股票、外匯等）
• 統一響應處理機制

2. 類型提示：

• 參數和返回值均使用Python類型提示
• 提高代碼可讀性和IDE支持

3. 錯誤處理：

• 統一HTTP響應處理
• WebSocket自動重連機制
• 異常捕獲和提示

4. 高級功能：

• 支持同步HTTP請求和異步WebSocket
• 靈活的時間參數處理（支持時間戳）
• 可配置的分頁參數

5. 最佳實踐：

• 使用requests.Session保持連接池
• 自定義User-Agent標識
• 完善的文檔字符串
• 符合PEP8編碼規范

擴展建議：

1. 緩存機制：

from functools import lru_cacheclass StockTVClient:@lru_cache(maxsize=128)def get_stock_info(self, pid: int):"""帶緩存的股票信息查詢"""# 實現代碼...

2. 異步HTTP請求：

import aiohttpasync def async_get_stock_markets(self, country_id: int):async with aiohttp.ClientSession() as session:async with session.get(url, params=params) as response:return await response.json()

3. 數據轉換工具：

def convert_kline_to_dataframe(kline_data: List) -> pd.DataFrame:"""將K線數據轉換為Pandas DataFrame"""return pd.DataFrame(kline_data,columns=["timestamp", "open", "high", "low", "close", "volume", "vo"]).set_index('timestamp')

使用注意事項：

1. 申請并妥善保管API密鑰
2. 遵守API調用頻率限制
3. 處理時區轉換（所有時間戳為UTC）
4. 使用try/except塊捕獲潛在異常
5. 生產環境建議添加重試機制

完整項目應包含：

• 單元測試
• 日志記錄
• 配置文件管理
• 更完善的類型定義
• API文檔生成（使用Sphinx等）

該實現為開發者提供了可靠的數據接入基礎，可根據具體需求進一步擴展功能模塊。建議配合官方文檔使用，及時關注API更新。