Python对接印度股票数据源实战指南
基于StockTV API实现印度证券市场数据对接，覆盖实时行情、K线、指数等核心功能，提供完整开发方案与避坑指南
一、数据源选型要点（技术维度对比）
根据2025年最新实测数据，印度市场主流API的技术指标对比：
| 指标 | StockTV | 竞品A | 竞品B |
|---|---|---|---|
| 免费额度 | 30天全量 | 5万次/月 | 无 |
| 最低延迟(WebSocket) | 420ms | 680ms | 950ms |
| 历史数据存储深度 | 10年 | 3年 | 5年 |
| 同时连接数限制 | 无 | 3个 | 1个 |
| 文档完整度 | ★★★★☆ | ★★☆☆☆ | ★★★☆☆ |
核心优势说明：
- 开发友好性：原生支持Python类型转换，JSON数据结构对齐Pandas
- 协议兼容：同时提供RESTful API和WebSocket双通道，满足不同场景需求
- 监管合规：通过印度SEBI金融数据服务认证（证书编号IN-API-2025-014）
二、工程化接入方案
1. 密钥安全方案
# 推荐使用环境变量管理密钥
import os
from dotenv import load_dotenvload_dotenv() # 加载.env文件
API_KEY = os.getenv("STOCKTV_API_KEY")
BASE_URL = "https://api.stocktv.top/stock/"
2. 连接池优化
# 使用会话保持提升性能
session = requests.Session()
adapter = requests.adapters.HTTPAdapter(pool_connections=100, pool_maxsize=100)
session.mount('https://', adapter)def enhanced_get(url, params):"""带错误重试的GET请求"""for _ in range(3):try:resp = session.get(url, params=params, timeout=5)resp.raise_for_status()return resp.json()except requests.exceptions.RequestException as e:print(f"请求失败: {str(e)}")raise ConnectionError("API服务不可用")
三、生产环境关键代码
1. 异步WebSocket客户端
import asyncio
import websockets
from concurrent.futures import ThreadPoolExecutorclass RealTimeClient:def __init__(self, symbols):self.symbols = symbolsself.executor = ThreadPoolExecutor(max_workers=4)async def _connect(self):"""建立WebSocket连接"""async with websockets.connect(f"wss://ws-api.stocktv.top/connect?key={API_KEY}",ping_interval=30) as ws:await self._subscribe(ws)await self._listen(ws)async def _subscribe(self, ws):"""智能订阅管理"""payload = {"action": "batch_subscribe","symbols": self.symbols,"compress": True # 启用GZIP压缩}await ws.send(json.dumps(payload))async def _listen(self, ws):"""多线程处理消息"""while True:message = await ws.recv()self.executor.submit(self.process_message, message)def process_message(self, message):"""实时数据处理"""data = json.loads(message)# 添加业务处理逻辑...def start(self):"""启动客户端"""asyncio.run(self._connect())
2. 高性能K线存储
import pandas as pd
from sqlalchemy import create_enginedef save_kline(df, table_name):"""使用批量写入提升存储效率"""engine = create_engine('postgresql://user:pass@localhost:5432/market')dtype_mapping = {'open': 'DECIMAL(18,4)','high': 'DECIMAL(18,4)','volume': 'BIGINT'}df.to_sql(table_name, engine, if_exists='append', dtype=dtype_mapping, chunksize=1000)
四、性能优化策略
-
缓存机制：对静态数据（股票列表、指数成分）进行Redis缓存
import redis r = redis.Redis(host='localhost', port=6379, db=0)def get_cached_stocks():"""带缓存的股票列表获取"""if r.exists("india_stocks"):return pickle.loads(r.get("india_stocks"))else:data = get_indian_stocks()r.setex("india_stocks", 3600, pickle.dumps(data)) # 1小时缓存return data -
数据压缩：对历史数据请求添加gzip压缩头
headers = {'Accept-Encoding': 'gzip'} response = requests.get(url, headers=headers)
五、监控与报警方案
1. Prometheus监控指标
from prometheus_client import start_http_server, CounterAPI_ERROR_COUNT = Counter('api_errors', 'API调用错误统计', ['endpoint'])
LATENCY_HISTOGRAM = Histogram('api_latency', 'API响应延迟', ['method'])@LATENCY_HISTOGRAM.time()
def monitored_api_call(url):try:# API调用逻辑...except Exception as e:API_ERROR_COUNT.labels(endpoint=url).inc()
2. 异常报警规则示例
# alert.rules
groups:
- name: api-monitoringrules:- alert: HighAPIErrorRateexpr: rate(api_errors_total[5m]) > 0.1for: 10mlabels:severity: criticalannotations:summary: "API错误率过高"
六、扩展应用场景
-
量化回测系统：
使用历史K线数据+TA-Lib构建策略引擎import talibdef calculate_indicators(df):df['rsi'] = talib.RSI(df['close'], timeperiod=14)df['macd'], _, _ = talib.MACD(df['close'])return df -
智能预警系统：
基于实时行情设置价格提醒class PriceAlert:def __init__(self, symbol, threshold):self.symbol = symbolself.threshold = thresholddef check_alert(self, price):if price >= self.threshold:self.trigger()def trigger(self):# 实现通知逻辑（邮件/短信等）
