QOS.HK QOS.HK
⚡
立即尝试API 报价 API 文档 示例代码 SDK GitHub 博客 注册KEY 找回KEY

QOS 行情 API - SDK使用示例 - 提供实时行情API、K线数据API - 外汇API、美股API、港股API、加密货币数据API - 金融高频数据接口 - 免费试用

订阅 QOS API，您即可 实时获取最新的股票、外汇、期货、加密货币数据，助力交易决策更快一步。我们的 实时行情数据 采用 WebSocket 高速推送，超低延迟，确保市场信息的精准与高效。QOS API 专为 交易所、开发者、量化团队、金融科技公司 及 专业机构 量身打造，提供 全面的金融市场 Tick 数据解决方案，覆盖 外汇、港股、美股、商品期货、加密货币 等多个领域。

选择您喜欢的编程语言查看SDK示例代码，如果需要更多请点击右下角联系客服。

Python SDK安装: pip install qos-api

GO SDK安装: go get github.com/qos-max/qos-quote-api-go-sdk/qosapi 或者 go mod tidy

Python SDK:https://pypi.org/project/qos-api/

Python SDK Github:https://github.com/qos-max/qos-quote-api-python-sdk/

GO SDK:https://pkg.go.dev/github.com/qos-max/qos-quote-api-go-sdk

GO SDK Github:https://github.com/qos-max/qos-quote-api-go-sdk

Python-SDK使用示例
Go-SDK使用示例
HTTP-获取实时快照
HTTP-获取K线
WS-订阅实时行情
from qos_api import QOSClient
# 注册key: https://qos.hk
client = QOSClient(api_key="您的API_KEY")
# 获取实时行情
snapshot = client.get_snapshot(["US:TSLA", "HK:09988"])
print(snapshot)

from qos_api import QOSClient
from qos_api.constants import KLineType

# 注册key: https://qos.hk
client = QOSClient(api_key="您的API_KEY")

# 获取日K线
kline = client.get_kline(
    codes=["SH:600519"],
    ktype=KLineType.DAY.value,
    count=10
)
print(kline)
			
import asyncio
from qos_api import QOSClient

async def handle_snapshot(snapshot):
    print(f"行情更新: {snapshot}")

async def pull_snapshot_periodically(client):
    """每10秒主动请求一次快照"""
    try:
        while True:
            data = await client.request_snapshot(["US:AAPL"])
            print(f"手动拉取快照数据: {data}")
            await asyncio.sleep(10)
    except asyncio.CancelledError:
        print("快照任务已取消")
        raise

async def main():
    # 注册key: https://qos.hk
    client = QOSClient(api_key="您的API_KEY")
    client.register_callback("S", handle_snapshot)
    await client.connect_ws()
    await client.subscribe_snapshot(["US:AAPL", "HK:700"])

    # 启动后台任务
    snapshot_task = asyncio.create_task(pull_snapshot_periodically(client))

    try:
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        print("收到退出信号，正在清理...")
        snapshot_task.cancel()
        await asyncio.gather(snapshot_task, return_exceptions=True)
        await client.close()
        print("已优雅退出")

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        # 再次捕获，防止 asyncio.run 抛出错误
        print("主程序强制退出")

			
package main

import (
	"fmt"
	"github.com/qos-max/qos-quote-api-go-sdk/qosapi"
	"log"
	"time"
)
//官网:https://qos.hk
//API文档:https://qos.hk/api.html
//注册API KEY:https://qos.hk
func main() {
	// 替换为你的API Key
	apiKey := "your-api-key"
	client := qosapi.NewClient(apiKey)

	// 示例1: 获取基础信息
	codes := []string{
		"US:AAPL,TSLA",
		"HK:00700,09988",
		"SH:600519,600518",
		"SZ:000001,002594",
	}

	info, err := client.GetInstrumentInfo(codes)
	if err != nil {
		log.Fatalf("Failed to get instrument info: %v", err)
	}
	fmt.Println("Instrument Info:")
	for _, item := range info {
		fmt.Printf("%s: %s (%s)\n", item.Code, item.NameCN, item.NameEN)
	}

	// 示例2: 获取行情快照
	snapshots, err := client.GetSnapshot(codes)
	if err != nil {
		log.Fatalf("Failed to get snapshots: %v", err)
	}
	fmt.Println("\nSnapshots:")
	for _, s := range snapshots {
		fmt.Printf("%s: %s (High: %s, Low: %s)\n", s.Code, s.LastPrice, s.High, s.Low)
	}

	// 示例3: 获取盘口深度
	depths, err := client.GetDepth(codes)
	if err != nil {
		log.Fatalf("Failed to get depths: %v", err)
	}
	fmt.Println("\nDepths:")
	for _, d := range depths {
		fmt.Printf("%s: Bids %d, Asks %d\n", d.Code, len(d.Bids), len(d.Asks))
	}

	// 示例4: 获取逐笔成交
	trades, err := client.GetTrade(codes, 5)
	if err != nil {
		log.Fatalf("Failed to get trades: %v", err)
	}
	fmt.Println("\nTrades:")
	for _, t := range trades {
		fmt.Printf("%s: Price %s, Volume %s, Direction %d\n", t.Code, t.Price, t.Volume, t.Direction)
	}

	// 示例5: 获取K线
	klineReqs := []qosapi.KLineRequest{
		{
			Codes:     "US:AAPL,TSLA",
			Count:     2,
			Adjust:    0,
			KLineType: qosapi.KLineTypeDay,
		},
		{
			Codes:     "CF:BTCUSDT,ETHUSDT",
			Count:     2,
			Adjust:    0,
			KLineType: qosapi.KLineTypeDay,
		},
	}

	klineData, err := client.GetKLine(klineReqs)
	if err != nil {
		log.Fatalf("Failed to get KLine: %v", err)
	}
	fmt.Println("\nKLine:")
	for _, klines := range klineData {
		for _, k := range klines {
			fmt.Printf("%s: O:%s C:%s H:%s L:%s V:%s\n",
				k.Code, k.Open, k.Close, k.High, k.Low, k.Volume)
		}
	}

	// 示例6: 获取历史K线
	historyReqs := []qosapi.KLineRequest{
		{
			Codes:     "US:AAPL",
			Count:     2,
			Adjust:    0,
			KLineType: qosapi.KLineTypeDay,
			EndTime:   time.Now().Unix(),
		},
	}

	historyData, err := client.GetHistoryKLine(historyReqs)
	if err != nil {
		log.Fatalf("Failed to get history KLine: %v", err)
	}
	fmt.Println("\nHistory KLine:")
	for _, klines := range historyData {
		for _, k := range klines {
			fmt.Printf("%s: O:%s C:%s H:%s L:%s V:%s\n",
				k.Code, k.Open, k.Close, k.High, k.Low, k.Volume)
		}
	}
}

package main

import (
	"fmt"
	"github.com/qos-max/qos-quote-api-go-sdk/qosapi"
	"log"
	"time"
)
//官网:https://qos.hk
//API文档:https://qos.hk/api.html
//注册API KEY:https://qos.hk
func main() {
	// 替换为你的API Key
	apiKey := "your-api-key"
	client := qosapi.NewClient(apiKey)

	// 示例1: 获取基础信息
	codes := []string{
		"US:AAPL,TSLA",
		"HK:00700,09988",
		"SH:600519,600518",
		"SZ:000001,002594",
	}

	info, err := client.GetInstrumentInfo(codes)
	if err != nil {
		log.Fatalf("Failed to get instrument info: %v", err)
	}
	fmt.Println("Instrument Info:")
	for _, item := range info {
		fmt.Printf("%s: %s (%s)\n", item.Code, item.NameCN, item.NameEN)
	}

	// 示例2: 获取行情快照
	snapshots, err := client.GetSnapshot(codes)
	if err != nil {
		log.Fatalf("Failed to get snapshots: %v", err)
	}
	fmt.Println("\nSnapshots:")
	for _, s := range snapshots {
		fmt.Printf("%s: %s (High: %s, Low: %s)\n", s.Code, s.LastPrice, s.High, s.Low)
	}

	// 示例3: 获取盘口深度
	depths, err := client.GetDepth(codes)
	if err != nil {
		log.Fatalf("Failed to get depths: %v", err)
	}
	fmt.Println("\nDepths:")
	for _, d := range depths {
		fmt.Printf("%s: Bids %d, Asks %d\n", d.Code, len(d.Bids), len(d.Asks))
	}

	// 示例4: 获取逐笔成交
	trades, err := client.GetTrade(codes, 5)
	if err != nil {
		log.Fatalf("Failed to get trades: %v", err)
	}
	fmt.Println("\nTrades:")
	for _, t := range trades {
		fmt.Printf("%s: Price %s, Volume %s, Direction %d\n", t.Code, t.Price, t.Volume, t.Direction)
	}

	// 示例5: 获取K线
	klineReqs := []qosapi.KLineRequest{
		{
			Codes:     "US:AAPL,TSLA",
			Count:     2,
			Adjust:    0,
			KLineType: qosapi.KLineTypeDay,
		},
		{
			Codes:     "CF:BTCUSDT,ETHUSDT",
			Count:     2,
			Adjust:    0,
			KLineType: qosapi.KLineTypeDay,
		},
	}

	klineData, err := client.GetKLine(klineReqs)
	if err != nil {
		log.Fatalf("Failed to get KLine: %v", err)
	}
	fmt.Println("\nKLine:")
	for _, klines := range klineData {
		for _, k := range klines {
			fmt.Printf("%s: O:%s C:%s H:%s L:%s V:%s\n",
				k.Code, k.Open, k.Close, k.High, k.Low, k.Volume)
		}
	}

	// 示例6: 获取历史K线
	historyReqs := []qosapi.KLineRequest{
		{
			Codes:     "US:AAPL",
			Count:     2,
			Adjust:    0,
			KLineType: qosapi.KLineTypeDay,
			EndTime:   time.Now().Unix(),
		},
	}

	historyData, err := client.GetHistoryKLine(historyReqs)
	if err != nil {
		log.Fatalf("Failed to get history KLine: %v", err)
	}
	fmt.Println("\nHistory KLine:")
	for _, klines := range historyData {
		for _, k := range klines {
			fmt.Printf("%s: O:%s C:%s H:%s L:%s V:%s\n",
				k.Code, k.Open, k.Close, k.High, k.Low, k.Volume)
		}
	}
}

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/qos-max/qos-quote-api-go-sdk/qosapi"
)

// 官网:https://qos.hk
// API文档:https://qos.hk/api.html
// 注册API KEY:https://qos.hk
func main() {
	// 替换为你的API Key
	apiKey := "your-api-key"
	client := qosapi.NewWSClient(apiKey)

	// 连接到WebSocket服务器
	if err := client.Connect(); err != nil {
		log.Fatalf("Failed to connect: %v", err)
	}
	defer client.Close()

	// 启动心跳
	client.StartHeartbeat(20 * time.Second)

	// 示例1: 订阅实时快照
	codes := []string{"US:AAPL,TSLA", "HK:700"}
	err := client.SubscribeSnapshot(codes, func(s qosapi.WSSnapshot) {
		fmt.Printf("Snapshot - %s: %s (Time: %v)\n", s.Code, s.LastPrice, time.Unix(s.Timestamp, 0))
	})
	if err != nil {
		log.Printf("Failed to subscribe snapshot: %v", err)
	}

	// 示例2: 订阅实时逐笔成交
	err = client.SubscribeTrade(codes, func(t qosapi.WSTrade) {
		fmt.Printf("Trade - %s: %s %s (Dir: %d, Time: %v)\n",
			t.Code, t.Price, t.Volume, t.Direction, time.Unix(t.Timestamp, 0))
	})
	if err != nil {
		log.Printf("Failed to subscribe trade: %v", err)
	}

	// 示例3: 订阅实时盘口
	err = client.SubscribeDepth(codes, func(d qosapi.WSDepth) {
		fmt.Printf("Depth - %s: Bids %d, Asks %d (Time: %v)\n",
			d.Code, len(d.Bids), len(d.Asks), time.Unix(d.Timestamp, 0))
	})
	if err != nil {
		log.Printf("Failed to subscribe depth: %v", err)
	}

	// 示例4: 订阅实时K线
	err = client.SubscribeKLine([]string{"CF:BTCUSDT"}, qosapi.KLineTypeDay, func(k qosapi.WSKLine) {
		fmt.Printf("KLine - %s: O:%s C:%s H:%s L:%s V:%s\n",
			k.Code, k.Open, k.Close, k.High, k.Low, k.Volume)
	})
	if err != nil {
		log.Printf("Failed to subscribe KLine: %v", err)
	}

	// 示例5: 请求实时快照
	go func() {
		time.Sleep(5 * time.Second)
		snapshots, err := client.RequestSnapshot(codes)
		if err != nil {
			log.Printf("Failed to request snapshots: %v", err)
			return
		}
		fmt.Println("\nRequested Snapshots:")
		for _, s := range snapshots {
			fmt.Printf("%s: %s (High: %s, Low: %s)\n", s.Code, s.LastPrice, s.High, s.Low)
		}
	}()

	// 示例6: 请求实时逐笔成交
	go func() {
		time.Sleep(6 * time.Second)
		trades, err := client.RequestTrade(codes, 5)
		if err != nil {
			log.Printf("Failed to request trades: %v", err)
			return
		}
		fmt.Println("\nRequested Trades:")
		for _, t := range trades {
			fmt.Printf("%s: Price %s, Volume %s, Direction %d\n", t.Code, t.Price, t.Volume, t.Direction)
		}
	}()

	// 示例7: 请求实时盘口
	go func() {
		time.Sleep(7 * time.Second)
		depths, err := client.RequestDepth(codes)
		if err != nil {
			log.Printf("Failed to request depths: %v", err)
			return
		}
		fmt.Println("\nRequested Depths:")
		for _, d := range depths {
			fmt.Printf("%s: Bids %d, Asks %d\n", d.Code, len(d.Bids), len(d.Asks))
		}
	}()

	// 示例8: 请求实时K线
	go func() {
		time.Sleep(8 * time.Second)
		klineReqs := []qosapi.KLineRequest{
			{
				Codes:     "US:AAPL,TSLA",
				Count:     2,
				Adjust:    0,
				KLineType: qosapi.KLineTypeDay,
			},
		}

		klineData, err := client.RequestKLine(klineReqs)
		if err != nil {
			log.Printf("Failed to request KLine: %v", err)
			return
		}
		fmt.Println("\nRequested KLine:")
		for _, klines := range klineData {
			for _, k := range klines {
				fmt.Printf("%s: O:%s C:%s H:%s L:%s V:%s\n",
					k.Code, k.Open, k.Close, k.High, k.Low, k.Volume)
			}
		}
	}()

	// 示例9: 请求历史K线
	go func() {
		time.Sleep(9 * time.Second)
		historyReqs := []qosapi.KLineRequest{
			{
				Codes:     "US:AAPL",
				Count:     2,
				Adjust:    0,
				KLineType: qosapi.KLineTypeDay,
				EndTime:   time.Now().Unix(),
			},
		}

		historyData, err := client.RequestHistoryKLine(historyReqs)
		if err != nil {
			log.Printf("Failed to request history KLine: %v", err)
			return
		}
		fmt.Println("\nRequested History KLine:")
		for _, klines := range historyData {
			for _, k := range klines {
				fmt.Printf("%s: O:%s C:%s H:%s L:%s V:%s\n",
					k.Code, k.Open, k.Close, k.High, k.Low, k.Volume)
			}
		}
	}()

	// 示例10: 请求基础信息
	go func() {
		time.Sleep(10 * time.Second)
		info, err := client.RequestInstrumentInfo(codes)
		if err != nil {
			log.Printf("Failed to request instrument info: %v", err)
			return
		}
		fmt.Println("\nRequested Instrument Info:")
		for _, item := range info {
			fmt.Printf("%s: %s (%s)\n", item.Code, item.NameCN, item.NameEN)
		}
	}()

	// 保持连接运行
	select {}
}

联系客服

Email: [email protected]

联系客服

Telegram: @stock_quote_api

联系客服

Telegram Group: @stock_quote_api_hk_us_ashre_api