使用StockTV API對接印度金融市場數據全指南：K線、實時行情與IPO新股

一、印度金融市場數據特點

印度作為全球增長最快的主要經濟體之一，其金融市場具有以下顯著特征：

雙交易所體系：國家證券交易所(NSE)和孟買證券交易所(BSE)
高流動性品種：Nifty 50指數成分股、銀行股等
獨特交易機制：T+2結算制度，上午9:15至下午3:30交易時間(IST)
豐富IPO市場：2023年印度IPO數量位居全球前列

二、環境配置與基礎對接

1. API密鑰獲取與配置

# 配置StockTV API
API_KEY = "your_api_key_here"  # 通過官網或客服獲取
BASE_URL = "https://api.stocktv.top"# 印度市場特定參數
INDIA_COUNTRY_ID = 14  # 印度國家代碼
NSE_EXCHANGE_ID = 46   # NSE交易所代碼
BSE_EXCHANGE_ID = 74   # BSE交易所代碼

2. 安裝必要庫

pip install requests websocket-client pandas plotly

三、印度K線數據專業對接

1. 多周期K線獲取接口

import pandas as pddef get_india_kline(symbol, exchange, interval="15m"):"""獲取印度股票K線數據:param symbol: 股票代碼(如RELIANCE):param exchange: 交易所(NSE/BSE):param interval: 時間間隔(1m/5m/15m/1h/1d)"""url = f"{BASE_URL}/stock/kline"params = {"symbol": symbol,"exchange": exchange,"interval": interval,"countryId": INDIA_COUNTRY_ID,"key": API_KEY}response = requests.get(url, params=params)data = response.json()# 轉換為Pandas DataFramedf = pd.DataFrame(data['data'])df['time'] = pd.to_datetime(df['time'], unit='ms')  # 轉換印度時區(IST)df['time'] = df['time'].dt.tz_localize('UTC').dt.tz_convert('Asia/Kolkata')return df# 獲取Reliance Industries的15分鐘K線(NSE)
reliance_kline = get_india_kline("RELIANCE", "NSE", "15m")

2. 專業級K線可視化

import plotly.graph_objects as godef plot_advanced_kline(df):fig = go.Figure(data=[go.Candlestick(x=df['time'],open=df['open'],high=df['high'],low=df['low'],close=df['close'],increasing_line_color='green',decreasing_line_color='red')])fig.update_layout(title='印度股票K線圖',xaxis_title='印度標準時間(IST)',yaxis_title='價格(INR)',xaxis_rangeslider_visible=False,template="plotly_dark")# 添加成交量柱狀圖fig.add_trace(go.Bar(x=df['time'],y=df['volume'],name='成交量',marker_color='rgba(100, 100, 255, 0.6)',yaxis='y2'))fig.update_layout(yaxis2=dict(title='成交量',overlaying='y',side='right'))fig.show()plot_advanced_kline(reliance_kline)

四、印度市場實時數據對接

1. WebSocket實時行情訂閱

import websocket
import json
import threadingclass IndiaMarketData:def __init__(self):self.symbol_map = {}  # 存儲symbol到股票名稱的映射def on_message(self, ws, message):data = json.loads(message)# 處理實時行情更新if data.get('type') == 'stock':symbol = data['symbol']print(f"實時行情 {self.symbol_map.get(symbol, symbol)}: "f"最新價 {data['last']} 成交量 {data['volume']}")# 處理指數更新elif data.get('type') == 'index':print(f"指數更新 {data['name']}: {data['last']} ({data['chgPct']}%)")def subscribe_symbols(self, ws):# 訂閱Nifty 50成分股(示例)nifty_stocks = ["RELIANCE", "TCS", "HDFCBANK", "INFY"]for symbol in nifty_stocks:self.symbol_map[symbol] = get_stock_name(symbol)# 訂閱請求subscribe_msg = {"action": "subscribe","countryId": INDIA_COUNTRY_ID,"symbols": nifty_stocks,"indices": ["NSEI"]  # Nifty 50指數}ws.send(json.dumps(subscribe_msg))def start(self):ws = websocket.WebSocketApp(f"wss://ws-api.stocktv.top/connect?key={API_KEY}",on_message=self.on_message,on_open=lambda ws: self.subscribe_symbols(ws))# 啟動WebSocket連接wst = threading.Thread(target=ws.run_forever)wst.start()# 輔助函數：獲取股票名稱
def get_stock_name(symbol):url = f"{BASE_URL}/stock/queryStocks"params = {"symbol": symbol,"countryId": INDIA_COUNTRY_ID,"key": API_KEY}response = requests.get(url, params=params)return response.json()['data'][0]['name']# 啟動實時數據服務
india_data = IndiaMarketData()
india_data.start()

2. 實時數據存儲方案

from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from datetime import datetimeBase = declarative_base()class RealTimeData(Base):__tablename__ = 'india_realtime_data'id = Column(Integer, primary_key=True)symbol = Column(String(20))exchange = Column(String(10))last_price = Column(Float)volume = Column(Integer)timestamp = Column(DateTime)created_at = Column(DateTime, default=datetime.utcnow)# 初始化數據庫連接
engine = create_engine('sqlite:///india_market.db')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)def save_realtime_data(data):session = Session()try:record = RealTimeData(symbol=data['symbol'],exchange=data.get('exchange', 'NSE'),last_price=data['last'],volume=data['volume'],timestamp=datetime.fromtimestamp(data['timestamp']))session.add(record)session.commit()except Exception as e:print(f"保存數據失敗: {e}")session.rollback()finally:session.close()# 在on_message回調中調用
# save_realtime_data(data)

五、印度IPO新股數據深度對接

1. 獲取IPO日歷與詳情

def get_india_ipo_list(status="upcoming"):"""獲取印度IPO列表:param status: upcoming(即將上市)/recent(近期上市)"""url = f"{BASE_URL}/stock/getIpo"params = {"countryId": INDIA_COUNTRY_ID,"status": status,"key": API_KEY}response = requests.get(url, params=params)return response.json()# 獲取即將上市的IPO
upcoming_ipos = get_india_ipo_list("upcoming")
print("即將上市的IPO:")
for ipo in upcoming_ipos['data'][:5]:print(f"{ipo['company']} ({ipo['symbol']}) - 發行價: ?{ipo['ipoPrice']}")# 獲取近期上市的IPO表現
recent_ipos = get_india_ipo_list("recent")
print("\n近期IPO表現:")
for ipo in recent_ipos['data'][:5]:change = (ipo['last'] - ipo['ipoPrice']) / ipo['ipoPrice'] * 100print(f"{ipo['company']}: 發行價 ?{ipo['ipoPrice']} → 當前 ?{ipo['last']} ({change:.2f}%)")

2. IPO數據分析與可視化

import plotly.express as pxdef analyze_ipo_performance():# 獲取過去6個月的IPO數據ipos = get_india_ipo_list("recent")['data']df = pd.DataFrame(ipos)# 計算首日/首周漲跌幅df['listing_gain'] = (df['listingPrice'] - df['ipoPrice']) / df['ipoPrice'] * 100df['current_gain'] = (df['last'] - df['ipoPrice']) / df['ipoPrice'] * 100# 繪制散點圖fig = px.scatter(df, x='listing_gain', y='current_gain', hover_data=['company', 'symbol'],title="印度IPO表現分析",labels={'listing_gain':'首日漲幅(%)', 'current_gain':'當前漲幅(%)'})# 添加參考線fig.add_hline(y=0, line_dash="dash")fig.add_vline(x=0, line_dash="dash")fig.show()return dfipo_analysis = analyze_ipo_performance()

六、生產環境最佳實踐

1. 錯誤處理與重試機制

from tenacity import retry, stop_after_attempt, wait_exponential
import logginglogging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10),before_sleep=lambda retry_state: logger.warning(f"重試 {retry_state.attempt_number} 次，原因: {retry_state.outcome.exception()}"))
def safe_api_call(url, params):try:response = requests.get(url, params=params, timeout=10)response.raise_for_status()return response.json()except requests.exceptions.RequestException as e:logger.error(f"API請求失敗: {e}")raise

2. 性能優化方案

import redis
from functools import lru_cache# 初始化Redis連接
r = redis.Redis(host='localhost', port=6379, db=0)@lru_cache(maxsize=100)
def get_stock_info(symbol):"""緩存股票基本信息"""cache_key = f"stock:{symbol}:info"cached = r.get(cache_key)if cached:return json.loads(cached)url = f"{BASE_URL}/stock/queryStocks"params = {"symbol": symbol,"countryId": INDIA_COUNTRY_ID,"key": API_KEY}data = safe_api_call(url, params)r.setex(cache_key, 3600, json.dumps(data))  # 緩存1小時return data# 批量獲取K線數據優化
def batch_get_kline(symbols, interval):"""批量獲取K線數據，減少API調用次數"""results = {}with ThreadPoolExecutor(max_workers=5) as executor:future_to_symbol = {executor.submit(get_india_kline, sym, "NSE", interval): sym for sym in symbols}for future in as_completed(future_to_symbol):symbol = future_to_symbol[future]try:results[symbol] = future.result()except Exception as e:logger.error(f"獲取{symbol}數據失敗: {e}")return results