对接马来西亚、印度、韩国、越南等环球金融数据示例 - IT评测·应用市场-qidao123.com

import requests
import websockets
import asyncio
from typing import Dict, List, Optional, Union
from datetime import datetime
class 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_key
self.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_time
if end_time:
params["endTime"] = end_time
response = 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())

复制代码

from functools import lru_cache
class StockTVClient:
@lru_cache(maxsize=128)
def get_stock_info(self, pid: int):
"""带缓存的股票信息查询"""
# 实现代码...

复制代码

import aiohttp
async 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()

复制代码

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')

复制代码