在進行A股投資分析、量化研究或構建自己的股票數據平台時，最基礎的一步就是獲取上市公司的基本面數據和股價信息。
本文將介紹如何通過 Infoway API的股票行情 API 查詢 A 股上市公司的公司基本面（如總股本、每股收益、每股淨資產等），併為後續獲取實時行情數據打下基礎。

一、A股基本面數據查詢

Infoway 提供了統一的基礎數據接口 /common/basic/symbols/info，可用於查詢美股、港股、A股等多市場上市公司的基礎信息。
對於A股用户，只需指定參數 type=STOCK_CN，即可獲取滬深兩市的公司基本面數據。

1.1 接口地址

https://data.infoway.io/common/basic/symbols/info

1.2 請求頭參數

1.3 請求參數説明

調用該接口時，需要在請求參數中指定目標股票的代碼以及市場類型。

1.4 基本面請求示例

以下示例展示瞭如何查詢兩隻A股股票的基本面信息（平安銀行與萬科A）：

GET https://data.infoway.io/common/basic/symbols/info?type=STOCK_CN&symbols=000001.SZ,000002.SZ
Header:
apiKey: your_api_key_here
# 申請API KEY: www.infoway.io

1.5 返回結果説明

API返回的數據為一個JSON對象，其中包含股票的基礎屬性、財務指標、以及所處板塊信息。

{
  "ret": 200,
  "msg": "success",
  "data": [
    {
      "symbol": "000001.SZ",
      "market": "CN",
      "name_cn": "平安銀行",
      "exchange": "SZSE",
      "currency": "CNY",
      "lot_size": 100,
      "total_shares": 19405918198,
      "circulating_shares": 19405762053,
      "eps": "2.29",
      "bps": "22.47",
      "dividend_yield": "0.96",
      "board": "SZMainConnect"
    },
    {
      "symbol": "000002.SZ",
      "market": "CN",
      "name_cn": "萬科A",
      "exchange": "SZSE",
      "currency": "CNY",
      "lot_size": 100,
      "total_shares": 11930709471,
      "circulating_shares": 9724196533,
      "eps": "-4.15",
      "bps": "16.49",
      "dividend_yield": "0",
      "board": "SZMainConnect"
    }
  ]
}

1.6 字段解析與常見用途

1.7 快速實戰：Python調用示例

如果你正在使用Python進行數據採集或量化分析，可以用以下幾行代碼完成A股基本面數據的拉取：

import requests
 
url = "https://data.infoway.io/common/basic/symbols/info"
headers = {"apiKey": "your_api_key_here"}
params = {
    "type": "STOCK_CN",
    "symbols": "000001.SZ,000002.SZ"
}
 
# 申請API KEY: www.infoway.io
 
resp = requests.get(url, headers=headers, params=params)
data = resp.json()
print(data)

二、查詢A股實時K線數據（Candlestick）

在瞭解了A股上市公司的基本面信息之後，我們通常還需要獲取股價的時間序列數據，例如分時走勢、日K線、周K線等，用於繪製走勢圖、技術分析或回測策略。

Infoway 提供了統一的 K線查詢接口，可覆蓋股票、加密貨幣、外匯、能源、期貨、貴金屬等多個市場。
對於A股行情，使用的接口路徑是：
`https://data.infoway.io/stock/v2/batch_kline
`
該接口支持一次查詢多隻股票的實時或歷史K線，返回的結果中包含開盤價、收盤價、最高價、最低價、成交量等核心字段。

2.1 接口基本信息

2.2 請求參數説明

K線接口的請求參數通過 JSON 格式 傳入 Request Body，可靈活控制K線週期與數量。

2.3 K線請求示例

下面示例展示如何通過Python請求A股的日K線數據（例如平安銀行與萬科A最近100天的K線）：

import requests
 
url = "https://data.infoway.io/stock/v2/batch_kline"
headers = {"apiKey": "your_api_key_here"}
# 申請API KEY: www.infoway.io
 
payload = {
    "klineType": 8,          # 日K
    "klineNum": 100,         # 查詢最近100根K線
    "codes": "000001.SZ,000002.SZ"
}
 
resp = requests.post(url, headers=headers, json=payload)
data = resp.json()
print(data)

2.4 返回示例

接口返回的JSON數據結構如下，每個標的s對應一個K線數組（respList）：

{
  "ret": 200,
  "msg": "success",
  "data": [
    {
      "s": "000001.SZ",
      "respList": [
        {
          "t": "1751372340",
          "h": "12.85",
          "o": "12.70",
          "l": "12.60",
          "c": "12.75",
          "v": "3429000",
          "pc": "0.32%",
          "pca": "0.04"
        },
        {
          "t": "1751285940",
          "h": "12.80",
          "o": "12.75",
          "l": "12.50",
          "c": "12.65",
          "v": "4152000",
          "pc": "-0.16%",
          "pca": "-0.02"
        }
      ]
    }
  ]
}

2.5 字段説明

三、A股實時盤口查詢教程

在量化交易或行情展示系統中，除了K線數據之外，實時買賣盤口（Order Book / Depth） 是非常關鍵的一類行情數據。
盤口信息反映了市場上買賣雙方的即時掛單情況，可用於判斷市場流動性、短期供需關係以及價格壓力區間。

Infoway 提供統一的盤口數據接口，覆蓋股票、加密貨幣、外匯、能源、期貨等市場。
對於A股產品，我們使用以下路徑：
`https://data.infoway.io/stock/batch_depth/{codes}
`
該接口可同時查詢多隻股票的實時買盤與賣盤前五檔至十檔掛單數據。

3.1 接口基本信息

3.2 請求參數説明

3.3 請求示例

以下是一個獲取平安銀行（000001.SZ）和寧德時代（300750.SZ）盤口數據的HTTP請求示例：

import requests
 
url = "https://data.infoway.io/stock/batch_depth/000001.SZ,300750.SZ"
headers = {"apiKey": "your_api_key_here"}
# 申請API KEY: www.infoway.io
 
resp = requests.get(url, headers=headers)
data = resp.json()
print(data)

3.4 返回示例

{
  "ret": 200,
  "msg": "success",
  "data": [
    {
      "s": "002594.SZ",
      "t": 1747378802012,
      "a": [
        ["389.17", "389.45", "389.50", "389.57", "389.69"],
        ["60", "6", "1", "1", "8"]
      ],
      "b": [
        ["389.16", "389.09", "389.05", "389.02", "389.01"],
        ["6", "1", "1", "1", "10"]
      ]
    }
  ]
}

3.5 盤口返回字段説明

其中：

• a[0] 為賣盤價格數組（從低到高）
• a[1] 為對應的掛單數量數組（單位：股）
• b[0] 為買盤價格數組（從高到低）
• b[1] 為對應的掛單數量數組（單位：股）

四、WebSocket獲取A股實時行情

通過 WebSocket 實時訂閲 A股的逐筆成交、盤口（Depth）、K線（Kline） 等數據。
該接口支持毫秒級推送延遲，適合高頻交易、行情監控、量化分析等實時場景。

4.1 訂閲地址

`wss://data.infoway.io/ws?business=stock&apikey=YourAPIKey
`

4.2 訂閲指令説明

WebSocket 建立連接後，通過發送指定格式的 JSON 請求來訂閲不同類型的數據。

4.3 Python代碼示例

import json
import time
import schedule
import threading
import websocket
from loguru import logger
 
# 申請API KEY: www.infoway.io
 
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()

4.4 推送數據格式

成交明細（code = 10000）

{
  "s": "600519.SH",
  "t": 1747382976831,
  "p": "1498.50",
  "v": "100",
  "bs": "B"
}

盤口數據（code = 10003）

{
  "s": "600519.SH",
  "t": 1747382976831,
  "a": [["1498.6","1498.7","1498.8"],["200","300","400"]],
  "b": [["1498.5","1498.4","1498.3"],["100","150","180"]]
}

K線數據（code = 10006）

{
  "s": "600519.SH",
  "t": 1747382976831,
  "o": "1495.0",
  "h": "1500.0",
  "l": "1490.0",
  "c": "1498.5",
  "v": "123400"
}

更多接入方法請查看：Github