如何在Python中更快调用API获取印度股市数据？

你的场景属于典型的IO密集型任务，以下几个优化方向能大幅提升处理数千只股票的效率：

1. 复用HTTP连接 #

每次requests.get都会新建TCP连接，握手过程会消耗额外时间。用requests.Session复用连接池，能减少重复的TCP握手开销：

import requests

# 全局创建一个Session，所有请求复用同一个连接池
session = requests.Session()

def get_prices(stock):
    start_unix = 1669794745
    end_unix = start_unix + 1800
    interval = 1
    # 用f-string拼接URL更简洁
    url = f'https://priceapi.moneycontrol.com/techCharts/indianMarket/stock/history?symbol={stock}&resolution={interval}&from={start_unix}&to={end_unix}'
    
    with session.get(url) as resp:
        resp.raise_for_status()  # 主动捕获HTTP错误
        return resp.json()['c']  # 返回结果，不要在函数内print

2. 使用线程池替代手动管理线程 #

手动创建数千个线程会产生大量调度开销，线程池可以限制并发数、复用线程，同时代码更简洁：

from concurrent.futures import ThreadPoolExecutor

stocks = ['ACC','ADANIENT','ADANIGREEN','ADANIPORTS','ADANITRANS','AMBUJACEM','ASIANPAINT','ATGL','BAJAJ-AUTO','BAJAJHLDNG']

# max_workers设置并发数，建议先测试20-50，根据API限流和机器性能调整
with ThreadPoolExecutor(max_workers=30) as executor:
    # 批量提交任务并获取所有结果
    results = list(executor.map(get_prices, stocks))

# 统一处理结果
for stock, price in zip(stocks, results):
    print(f"{stock}: {price}")

3. 改用异步请求（高并发场景最优解） #

异步框架aiohttp不需要线程切换，能更高效地处理大量IO请求，适合数千只股票的场景：

import aiohttp
import asyncio

async def get_prices_async(session, stock):
    start_unix = 1669794745
    end_unix = start_unix + 1800
    interval = 1
    url = f'https://priceapi.moneycontrol.com/techCharts/indianMarket/stock/history?symbol={stock}&resolution={interval}&from={start_unix}&to={end_unix}'
    
    async with session.get(url) as resp:
        resp.raise_for_status()
        data = await resp.json()
        return (stock, data['c'])

async def main():
    stocks = ['ACC','ADANIENT','ADANIGREEN','ADANIPORTS','ADANITRANS','AMBUJACEM','ASIANPAINT','ATGL','BAJAJ-AUTO','BAJAJHLDNG']
    
    # limit控制并发连接数，避免触发API限流
    connector = aiohttp.TCPConnector(limit=50)
    async with aiohttp.ClientSession(connector=connector) as session:
        # 创建所有异步任务
        tasks = [get_prices_async(session, s) for s in stocks]
        # 批量执行任务，return_exceptions=True允许个别任务失败不中断整体
        results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # 过滤并处理结果
    for res in results:
        if isinstance(res, Exception):
            print(f"请求失败: {str(res)}")
        else:
            stock, price = res
            print(f"{stock}: {price}")

# Jupyter Notebook中直接用await运行异步函数
await main()

额外优化建议 #

• 控制并发数：不要盲目调大并发数，否则可能被API服务商限流，建议先从20开始测试，逐步找到最优值。
• 减少实时IO：不要在请求函数内print，先把所有结果存到内存，最后统一写入文件或打印，降低IO开销。
• 添加重试机制：对失败的请求添加重试（比如用tenacity库），应对网络波动导致的请求失败。
• 检查API批量接口：查看Moneycontrol是否支持批量查询多只股票的接口，若有，能大幅减少请求次数，效率提升最明显。

内容的提问来源于stack exchange，提问作者Samit