当前位置：首页 > news >正文

对接印度股票数据实战 - Python实现完整指南

news 来源：原创 2025/5/7 8:48:56

对接印度股票数据实战 - Python实现完整指南

一、API接口概述

根据提供的JSON文档，印度股票数据接口主要包含以下功能：

功能模块	接口描述	核心参数示例
市场列表	获取指定国家的股票列表	countryId=14（印度国家ID）
股票查询	根据ID或名称查询个股	id=7310
K线数据	获取不同时间粒度的K线	interval=PT15M（15分钟）
实时行情	WebSocket推送实时数据	wss协议
涨跌排行榜	获取涨幅/跌幅排行	type=1（涨幅榜）
IPO新股日历	获取新股上市信息	countryId=14

二、Python对接实战

1. 环境准备

# 安装依赖
pip install requests websocket-client pandas

2. 基础配置

import requests
import json
import pandas as pd
from datetime import datetimeBASE_URL = "https://api.stocktv.top/stock"
API_KEY = "MY4b781f618e3f43c4b055f25fa61941ad"  # 替换为实际keyHEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) PythonStockClient/1.0","Accept-Encoding": "gzip"
}

3. 核心接口封装

def get_market_list(country_id=14, page=1, page_size=10):"""获取印度股票市场列表"""params = {"countryId": country_id,"page": page,"pageSize": page_size,"key": API_KEY}try:response = requests.get(f"{BASE_URL}/stocks",params=params,headers=HEADERS,timeout=10)response.raise_for_status()# 转换数据为DataFramedata = response.json()['data']['records']df = pd.DataFrame(data)# 处理时间戳df['time'] = pd.to_datetime(df['time'], unit='s') return df[['id', 'symbol', 'name', 'last', 'chgPct', 'volume', 'time']]except requests.exceptions.RequestException as e:print(f"请求失败: {str(e)}")return Nonedef get_kline(pid, interval="PT15M"):"""获取股票K线数据"""params = {"pid": pid,"interval": interval,"key": API_KEY}response = requests.get(f"{BASE_URL}/kline",params=params,headers=HEADERS)if response.status_code == 200:kline_data = response.json()['data']df = pd.DataFrame(kline_data)df['time'] = pd.to_datetime(df['time'], unit='ms')return dfreturn Nonedef search_stock(search_term):"""多条件搜索股票"""params = {"id": search_term if search_term.isdigit() else "","name": search_term if not search_term.isdigit() else "","key": API_KEY}response = requests.get(f"{BASE_URL}/queryStocks",params=params,headers=HEADERS)return response.json()['data'] if response.ok else []

4. WebSocket实时数据

import websocket
import threading
import timedef on_message(ws, message):"""处理实时行情推送"""data = json.loads(message)print(f"\n实时行情更新 [{data['symbol']}]:")print(f"最新价: {data['last_numeric']}")print(f"涨跌幅: {data['pcp']}%")print(f"成交量: {data['turnover_numeric']}")def on_error(ws, error):print(f"WebSocket错误: {str(error)}")def on_close(ws, close_status_code, close_msg):print("WebSocket连接关闭")def ws_heartbeat(ws):"""保持心跳连接"""while True:time.sleep(30)ws.send(json.dumps({"action": "ping"}))def start_websocket():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)# 启动心跳线程threading.Thread(target=ws_heartbeat, args=(ws,)).start()ws.run_forever()

三、使用示例

1. 获取Nifty50指数

nifty50 = get_market_list().query('name.str.contains("Nifty 50")')
print(nifty50[['symbol', 'last', 'chgPct']])

2. 分析TATA Motors的K线

tata_motors_kline = get_kline(7310)
tata_motors_kline.set_index('time')['close'].plot(title='TATA Motors 15分钟K线'
)

3. 实时监控（需要单独线程运行）

# start_websocket()

四、开发注意事项

参数验证

def validate_country_id(country_id):valid_ids = [14,42,36]  # 印度、马来西亚、印尼等if country_id not in valid_ids:raise ValueError("非法的国家ID")

错误重试机制

from tenacity import retry, stop_after_attempt, wait_exponential@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1))
def safe_api_call(url, params):return requests.get(url, params=params, timeout=5)

数据缓存策略

from cachetools import cached, TTLCachestock_cache = TTLCache(maxsize=100, ttl=300)@cached(stock_cache)
def get_cached_stock_data(pid):return get_kline(pid)