搞懂股票行情API：如何高效获取实时与逐笔成交数据？

在股票市场中，行情数据接口（Stock Market Data API）是连接交易者与市场的核心桥梁。不同类型的 API 提供不同层次的市场信息，从历史数据到实时逐笔成交，应用场景各不相同。以下是几种常见的股票行情 API 类型及其用途：

1. 实时行情接口（Tick 数据 / 逐笔成交数据）

实时行情接口，也常被称为 逐笔成交接口（Tick Data API），提供市场中每一笔成交的详细数据，包括成交时间、成交价、成交量、买卖方向等。
这种接口能够让用户实时追踪市场的微观变化，捕捉价格波动的细节，是 量化交易、高频交易 和 智能算法策略 的关键数据来源。

（我之前在 CSDN 上写过一篇关于 Tick 数据的详细介绍，感兴趣的读者可以前往阅读 )

2. 历史行情接口（Historical Market Data API）

历史行情接口提供指定时间段内的股票价格数据，如开盘价、收盘价、最高价、最低价和成交量等。这类数据主要用于：

• 技术分析（如均线、K线图、波动率分析）
• 策略回测与验证
• 长期趋势研究与模型训练

对于想要评估投资策略表现或训练机器学习模型的开发者而言，历史行情接口是必不可少的基础。

3. 其他常见的金融数据接口

除了股票行情数据外，金融市场还存在多种类型的 API，用于不同的数据维度，例如：

• 财务报表数据接口：获取公司财报、利润表、资产负债表、现金流量表等；
• 宏观经济数据接口：包括GDP、CPI、利率、就业率等宏观指标；
• 跨市场实时行情接口：涵盖 A股、港股、美股、外汇、黄金、贵金属 等不同市场的即时报价数据。

4. 股票实时行情接口使用教程

无论你是做量化交易，还是开发交易平台，实时行情接口是最常用的，尤其是能够跨市场的API，下面以Infoway API为例，介绍如何查询行情。

4.1 K线的获取方法

import requests
 
# 申请API KEY: www.infoway.io
api_url = 'https://data.infoway.io/stock/batch_kline/1/10/AAPL.US'
 
# 设置请求头
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}")

注意K线的请求格式是这样的：

https://data.infoway.io/stock/batch_kline/{klineType}/{klineNum}/{codes}

klineType 是K线的周期，1代表1分钟，2代表5分钟，更多周期选择请查看官方文档。

klineNum 是需要的K线数量，比如10代表返回10根K线，一次支持最大值为500

codes 是股票代码，比如苹果股票：AAPL.US

下面是返回的数据示例：

{
  "s": "AMZN.US",    //产品代码
  "respList": [
    {
      "t": "1752868800",  //秒时间戳（UTC+8）
      "h": "226.200",    //最高价
      "o": "226.120",    //开盘价
      "l": "226.110",    //最低价
      "c": "226.110",    //收盘价
      "v": "2284098",    //成交量
      "vw": "516502787.475",    //成交额
      "pc": "0.00%",    //涨跌幅
      "pca": "-0.010"    //涨跌额
    }
  ]
}

4.2 成交明细与盘口

成交明细和盘口的请求方式和上面介绍的K线类似，请求地址如下：

# 股票成交明细查询接口：
data.infoway.io/stock/batch_trade/{codes}
 
# 加密货币成交明细查询接口：
data.infoway.io/crypto/batch_trade/{codes}
 
# 外汇商品期货成交明细查询接口：
data.infoway.io/common/batch_trade/{codes}
 
# 股票盘口查询接口：
data.infoway.io/stock/batch_depth/{codes}
 
# 加密货币盘口查询接口：
data.infoway.io/crypto/batch_depth/{codes}
 
# 外汇商品期货盘口查询接口：
data.infoway.io/common/batch_depth/{codes}
 
# API Key申请：
www.infoway.io

5. 股票实时行情接口的WebSocket使用方法

上面介绍的是HTTP请求方式，如果你对延迟要求比较高，推荐使用Websocket订阅的方法，你可以使用下面的代码来建立股票K线、成交、盘口的数据流。代码已包含心跳和自动重连机制。如果你需要Java的代码可以查看这份文档。

import json
import time
import schedule
import threading
import websocket
from loguru import logger
 
class WebsocketExample:
    def __init__(self):
        self.session = None
        self.ws_url = "wss://data.infoway.io/ws?business=crypto&apikey=yourApikey"
        self.reconnecting = False
        self.is_ws_connected = False  # 添加连接状态标志
 
    def connect_all(self):
        """建立WebSocket连接并启动自动重连机制"""
        try:
            self.connect(self.ws_url)
            self.start_reconnection(self.ws_url)
        except Exception as e:
            logger.error(f"Failed to connect to {self.ws_url}: {str(e)}")
 
    def start_reconnection(self, url):
        """启动定时重连检查"""
        def check_connection():
            if not self.is_connected():
                logger.debug("Reconnection attempt...")
                self.connect(url)
        
        # 使用线程定期检查连接状态
        schedule.every(10).seconds.do(check_connection)
        def run_scheduler():
            while True:
                schedule.run_pending()
                time.sleep(1)
        threading.Thread(target=run_scheduler, daemon=True).start()
 
    def is_connected(self):
        """检查WebSocket连接状态"""
        return self.session and self.is_ws_connected
 
    def connect(self, url):
        """建立WebSocket连接"""
        try:
            if self.is_connected():
                self.session.close()
            
            self.session = websocket.WebSocketApp(
                url,
                on_open=self.on_open,
                on_message=self.on_message,
                on_error=self.on_error,
                on_close=self.on_close
            )
            
            # 启动WebSocket连接（非阻塞模式）
            threading.Thread(target=self.session.run_forever, daemon=True).start()
        except Exception as e:
            logger.error(f"Failed to connect to the server: {str(e)}")
 
    def on_open(self, ws):
        """WebSocket连接建立成功后的回调"""
        logger.info(f"Connection opened")
        self.is_ws_connected = True  # 设置连接状态为True
        
        try:
            # 发送实时成交明细订阅请求
            trade_send_obj = {
                "code": 10000,
                "trace": "01213e9d-90a0-426e-a380-ebed633cba7a",
                "data": {"codes": "BTCUSDT"}
            }
            self.send_message(trade_send_obj)
            
            # 不同请求之间间隔一段时间
            time.sleep(5)
            
            # 发送实时盘口数据订阅请求
            depth_send_obj = {
                "code": 10003,
                "trace": "01213e9d-90a0-426e-a380-ebed633cba7a",
                "data": {"codes": "BTCUSDT"}
            }
            self.send_message(depth_send_obj)
            
            # 不同请求之间间隔一段时间
            time.sleep(5)
            
            # 发送实时K线数据订阅请求
            kline_data = {
                "arr": [
                    {
                        "type": 1,
                        "codes": "BTCUSDT"
                    }
                ]
            }
            kline_send_obj = {
                "code": 10006,
                "trace": "01213e9d-90a0-426e-a380-ebed633cba7a",
                "data": kline_data
            }
            self.send_message(kline_send_obj)
            
            # 启动定时心跳任务
            schedule.every(30).seconds.do(self.ping)
            
        except Exception as e:
            logger.error(f"Error sending initial messages: {str(e)}")
 
    def on_message(self, ws, message):
        """接收消息的回调"""
        try:
            logger.info(f"Message received: {message}")
        except Exception as e:
            logger.error(f"Error processing message: {str(e)}")
 
    def on_close(self, ws, close_status_code, close_msg):
        """连接关闭的回调"""
        logger.info(f"Connection closed: {close_status_code} - {close_msg}")
        self.is_ws_connected = False  # 设置连接状态为False
 
    def on_error(self, ws, error):
        """错误处理的回调"""
        logger.error(f"WebSocket error: {str(error)}")
        self.is_ws_connected = False  # 发生错误时设置连接状态为False
 
    def send_message(self, message_obj):
        """发送消息到WebSocket服务器"""
        if self.is_connected():
            try:
                self.session.send(json.dumps(message_obj))
            except Exception as e:
                logger.error(f"Error sending message: {str(e)}")
        else:
            logger.warning("Cannot send message: Not connected")
 
    def ping(self):
        """发送心跳包"""
        ping_obj = {
            "code": 10010,
            "trace": "01213e9d-90a0-426e-a380-ebed633cba7a"
        }
        self.send_message(ping_obj)
 
# 使用示例
if __name__ == "__main__":
    ws_client = WebsocketExample()
    ws_client.connect_all()
    
    # 保持主线程运行
    try:
        while True:
            schedule.run_pending()
            time.sleep(1)
    except KeyboardInterrupt:
        logger.info("Exiting...")
        if ws_client.is_connected():
            ws_client.session.close()