沪深A股实时行情API接入指南

接口基本信息

请求地址：
https://data.infoway.io/stock/batch_kline/{klineType}/{klineNum}/{codes}
查询方式：REST, WebSocket
行情类型：实时
市场：A股，港股，美股，外汇，期货，虚拟币
接入文档：https://docs.infoway.io/
API Key申请：https://infoway.io/

适用对象

• 开发者：专注于构建交易分析工具与市场趋势研判工具的开发者群体。
• 量化团队：致力于量化交易策略研发与算法开发的专业团队。
• 金融科技公司：提供金融科技解决方案及配套服务的企业机构。
• 专业机构：对精准、及时的金融市场数据有持续需求的专业型机构。

获取A股实时K线数据

提供实时的A股股票K线数据，包括开盘、最高、最低、收盘价格等信息，帮助用户捕捉市场走势。返回示例如下：

{
  "t": "1751958000",
  "h": "326.880",
  "o": "326.880",
  "l": "326.880",
  "c": "326.880",
  "v": "1410",
  "vw": "460900.80",
  "pc": "0.00%",
  "pca": "0.000"
}

A股逐笔成交数据（实时）

逐笔Tick数据接口，查询A股上市公司的最新成交明细，确保获取市场的最新交易信息。返回示例如下：

{
  "s": "002594.SZ",
  "t": 1751958000999,
  "p": "326.88",
  "v": "1410",
  "vw": "460900.80",
  "td": 1
}

五档盘口数据（实时）

实时提供五档买卖盘数据，帮助用户了解市场深度和流动性：

{
  "s": "002594.SZ",
  "t": 1751958003335,
  "a": [
    [
      "326.89",
      "326.90",
      "326.91",
      "326.92",
      "326.93"
    ],
    [
      "8",
      "30",
      "10",
      "1",
      "22"
    ]
  ],
  "b": [
    [
      "326.88",
      "326.87",
      "326.86",
      "326.85",
      "326.84"
    ],
    [
      "1452",
      "77",
      "188",
      "37",
      "21"
    ]
  ]
}

WebSocket订阅

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

    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)

        # 使用线程定期检查连接状态
        threading.Thread(target=lambda: schedule.every(10).seconds.do(check_connection), daemon=True).start()

    def is_connected(self):
        """检查WebSocket连接状态"""
        return self.session and self.session.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")

        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)

            # 启动定时心跳任务
            threading.Thread(target=lambda: schedule.every(30).seconds.do(self.ping), daemon=True).start()

        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}")

    def on_error(self, ws, error):
        """错误处理的回调"""
        logger.error(f"WebSocket error: {str(error)}")

    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()

本作品采用《CC 协议》，转载必须注明作者和本文链接