股票行情API是金融科技领域常见的应用之一，但今天介绍的是实时数据接口，它和普通的行情接口的区别主要在于数据的实时性，基本搞量化交易的都会用到。我们对国内外比较知名的实时接口做了详细的对比测评，可以直接看以下结果：

1、Alpha Vantage

Alpha Vantage是YCombinator孵化的项目，在国外量化圈的知名度比较高，它支持多个市场，包括股票、指数、外汇和加密货币等。缺点是只支持HTTP，没有WebSocket订阅。

价格：149.99美金/月，无HTTP请求次数限制

2、infoway

infoway API提供了港股、美股、A股、外汇、期货、加密货币的实时报价与历史数据拉取，它的优点是有免费套餐，而且同时支持HTTP和Websocket，个人和企业都能用。

3、Yahoo Finance API

老牌的股票行情API，数据齐全，但延时非常高。

价格：35美金/月，每个月15万次HTTP请求（没有WS订阅）

4、IEX Cloud

IEX Cloud是一个专注于美国市场的股票行情API，提供HTTP+Websock订阅，但价格太贵。

价格：1500美金/月，不包含历史数据，历史数据要加200刀。

6、Xignite

Xignite是面向B端的产品，它的优点是数据覆盖面广，提供了全球范围内的资产类别，包括股票、债券、期货、外汇等都有。但如果没有美国公司主体是无法申请的。

价格：$12000/年，包含一年的历史数据，5年数据$17000/年。注意是美金，价格很贵。

7、Intrinio

Intrinio也是面向企业客户的，个人或者没有资质无法申请，而且只有纳斯达克的数据，所以比Xignite要便宜。

价格：美股实时数据$2100/年，不包含历史数据。高频数据接口$6000刀一年。

股票API接口如何选？

首先看你的查询方法，获取实时行情可以通过HTTP请求或者WS订阅。这两者的主要区别是频次和延迟。HTTP的请求频次是有限制的，延时会高一点，适合用来拉取一些历史数据、日K线等，而WS是长连接，只要连接没有断开（你可以向服务端发送心跳保持连接），行情价格有更新会直接推送过来，适合用来做实盘交易。而我们上面提到的产品中，并不是每家都支持websocket。所以挑选行情API时最好搞清楚你需要使用哪种查询方式，如果不确定，那无脑选择两种都支持的。

代码示例

下面给出HTTP和WS订阅的代码示例，供各位参考。

HTTP请求

import requests
 
api_url = 'https://data.infoway.io/stock/batch_kline/1/10/002594.SZ%2C00285.HK%2CTSLA.US'
 
# 请先在官网申请免费Token: https://infoway.io
# 详细对接文档：https://docs.infoway.io
 
# 设置请求头
headers = {
    'User-Agent': 'Mozilla/5.0',
    'Accept': 'application/json',
    'apiKey': 'yourApikey'
}
 
# 发送GET请求
response = requests.get(api_url, headers=headers)
 
# 输出结果
print(f"HTTP code: {response.status_code}")
print(f"message: {response.text}")

WebSocket

import asyncio
import json
import websockets
 
WS_URL = "wss://data.infoway.io/ws?business=crypto&apikey=yourApikey"
 
# 请先在官网申请免费Token: https://infoway.io
# 详细对接文档：https://docs.infoway.io
 
async def connect_and_receive():
    async with websockets.connect(WS_URL) as websocket:
        # 发送初始消息
        init_message = {
            "code": 10000,
            "trace": "01213e9d-90a0-426e-a380-ebed633cba7a",
            "data": {"codes": "BTCUSDT"}
        }
        await websocket.send(json.dumps(init_message))
 
        # 设置ping任务
        async def send_ping():
            while True:
                await asyncio.sleep(30)
                ping_message = {
                    "code": 10010,
                    "trace": "01213e9d-90a0-426e-a380-ebed633cba7a"
                }
                await websocket.send(json.dumps(ping_message))
 
        # 启动ping任务协程
        ping_task = asyncio.create_task(send_ping())
        
        try:
            # 持续接收消息
            while True:
                message = await websocket.recv()
                print(f"Message received: {message}")
        except websockets.exceptions.ConnectionClosedOK:
            print("Connection closed normally")
        finally:
            # 取消ping任务
            ping_task.cancel()
 
# 运行主函数
asyncio.run(connect_and_receive())