Infoway API是一套覆盖多个金融市场的行情数据接口，提供低延迟的实时行情数据。产品被广泛用在量化交易系统、AI驱动的股票分析平台、交易所等多个需要实时行情数据的业务场景。今天介绍的是外汇行情API。

Infoway API的外汇行情接口主要有以下特点：

  • 低延迟 （实测延迟低于100ms）
  • 数据流稳定 （SLA 99.6%)
  • 数据整合自大型做市商，数据准确可靠
  • 接入简单、易用

支持哪些货币对

Infoway的外汇API支持大部分主流货币对：

AUDCAD澳元/加元
AUDCHF澳元/瑞士法郎
AUDDKK澳元/丹麦克朗
AUDJPY澳元/日元
AUDNZD澳元/新西兰元
AUDUSD澳元/美元
CADCHF加元/瑞士法郎
CADJPY加元/日元
CADUSD加拿大/美元
CHFJPY瑞士法郎/日元
CHFUSD瑞士法郎/美元
CNYUSD人民币/美元
EURAUD欧元/澳元
EURCAD欧元/加元
EURCHF欧元/瑞士法郎
EURGBP欧元/英镑
EURJPY欧元/日元
EURNZD欧元/新西兰元
EURUSD欧元/美元
GBPAUD英镑/澳元
GBPCAD英镑/加元
GBPCHF英镑/瑞士法郎
GBPJPY英镑/日元
GBPNZD英镑/新西兰元
GBPUSD英镑/美元
JPYUSD日元/美元
NZDCAD新西兰元/加元
NZDJPY新西兰元/日元
NZDUSD新西兰元/美元
SGDUSD新加坡/美元
USDCAD美元/加元
USDCHF美元/瑞士法郎
USDCNH美元/人民币
USDCNY美元/人民币
USDEUR美元/欧元
USDGBP美元/英镑
USDHKD美元/港元
USDJPY美元/日元
USDRUB美元/卢布
USDSGD美元/新加坡元
USDTHB美元/泰铢
USDTWD美元/新台币

使用教程

1. 注册获取Token

首先在我们的官网注册免费账户，我们提供免费试用，注册无需实名认证。点击前往注册页面。

注册后自动获得试用套餐，在后台可以看到Token。查询数据时需要带上你的Token。

2. 了解接口调用方法

获取了Token以后就可以开始尝试查询数据，我们提供了完整的接口文档：

前往接口文档

在Github查看API文档

下面是外汇行情接口的查询代码示例。

代码示例

获取最新一笔交易明细

Python
import requests

url = "https://data.infoway.io/common/batch_trade/USDCNY%2C%20USDCAD"

headers = {
    'User-Agent': 'Mozilla/5.0',
    'Accept': 'application/json',
    'apiKey': 'Your_API_Key'
}

response = requests.get(url, headers=headers)

print(response.text)

获取K线

Python
import requests

url = "https://data.infoway.io/common/batch_kline/1/10/USDCNY%2CUSDCAD"

headers = {
    'User-Agent': 'Mozilla/5.0',
    'Accept': 'application/json',
    'apiKey': 'Your_API_Key'
}

response = requests.get(url, headers=headers)

print(response.text)

获取盘口

Python
import requests

url = "https://data.infoway.io/common/batch_depth/USDCNY%2CUSDCAD"

response = requests.get(url)

print(response.text)

WebSocket订阅

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