港股实时行情 API 接口选型与接入实战

import os
import requests
API_BASE = "https://api.deeptick.cc"
API_TOKEN = os.getenv("DEEPTICK_API_TOKEN", "your_api_key")
# DeepTick 用deeptoken 请求头鉴权,其余字段按文档要求
headers = {
"accept": "application/json",
"deeptoken": API_TOKEN, "content-type": "application/json; charset=utf-8"
}

class DeepTickClient:
def __init__(self, base_url=API_BASE, custom_headers=None):
self.base_url = base_url
self.headers = headers.copy()
if custom_headers:
self.headers.update(custom_headers)
def _get(self, path, params=None):
"""统一封装的GET 请求,集中处理超时与异常"""
try:
resp = requests.get(
f"{self.base_url}{path}",
headers=self.headers,
params=params,
timeout=10,
)
resp.raise_for_status()
return resp.json()
except requests.exceptions.RequestException as e:
print(f"请求失败: {e}")
return None

def get_quote(self, symbol="00700.HK"):
实时报价:GET /api/v1/quote?symbols=00700.HK 
:param symbol: 标的代码,格式为代码.市场,如'00700.HK' 为腾讯控股
:return: 单条报价字典,接口data 为数组,这里取第一条 
""" 
resp = self._get("/api/v1/quote", {"symbols": symbol}) 
if resp and resp.get("code") == 0 and resp.get("data"): 
        return resp["data"][0] 
print(f"获取报价失败: {resp.get('msg') if resp else '未知错误'}") 
return None




# 使用示例 import datetime
client = DeepTickClient()
quote = client.get_quote("00700.HK")
if quote:
print(f"标的代码: {quote['s']}")
print(f"最新价: {quote['ld']} 前收: {quote.get('p')}")
print(f"开/高/低: {quote.get('o')} / {quote.get('h')} / {quote.get('l')}")
print(f"成交量: {quote.get('v')} 成交额: {quote.get('tu')}")
print(f"涨跌额: {quote.get('ch')} 涨跌幅: {quote.get('chp')}")
# ts 是交易状态（0 正常 / 1 停牌 / 2 退市 / 3 熔断），不是时间戳
if quote.get("ts") is not None:
print(f"交易状态: {['正常', '停牌', '退市', '熔断'][quote['ts']]}")
# t 是毫秒级行情时间戳
if quote.get("t"):
print(f"行情时间: {datetime.datetime.fromtimestamp(quote['t'] / 1000)}"）

def get_orderbook(self, symbol="00700.HK"):
"""
盘口查询：GET /api/v1/orderbook?symbol=00700.HK
:param symbol: 单个标的，如 '00700.HK'
:return: {s, b, a, t}，买盘 b 价格由高到低，卖盘 a 由低到高
"""
resp = self._get("/api/v1/orderbook", {"symbol": symbol})
if resp and resp.get("code") == 0 and resp.get("data") is not None:
return resp["data"]
print(f"获取盘口失败: {resp.get('msg') if resp else '未知错误'}")
return None


# 使用示例
book = client.get_orderbook("00700.HK")
if book:
    print(f"标的代码: {book['s']}")
    print("=== 卖盘（价格由低到高）===")
    for ask in book["a"]:
        print(f"价格: {ask['p']}  挂单量: {ask['v']}")
    print("=== 买盘（价格由高到低）===")
    for bid in book["b"]:
        print(f"价格: {bid['p']}  挂单量: {bid['v']}")

    ask_volume = sum(float(ask["v"]) for ask in book["a"])
    bid_volume = sum(float(bid["v"]) for bid in book["b"])
    if ask_volume:
        print(f"买卖盘挂单量比(买/卖): {bid_volume / ask_volume:.2%}")

    def get_candles(self, symbol="00700.HK", interval="5m", limit=50, et=None):
        """
        K 线查询：GET /api/v1/candles
        :param interval: 1m / 5m / 15m / 30m / 60m / 4h / 1d / 1w / 1M，默认 5m
        :param limit: 返回条数，1~500，默认 40
        :param et: 截止时间 Unix 毫秒，缺省取最新
        :return: 时间升序的 K 线列表 [{t,o,h,l,c,v,tu}, ...]
        """
        params = {"symbols": symbol, "interval": interval, "limit": limit}
        if et:
            params["et"] = et
        resp = self._get("/api/v1/candles", params)
        if resp and resp.get("code") == 0 and resp.get("data"):
            return resp["data"][0].get("d", [])
        print(f"获取 K 线失败: {resp.get('msg') if resp else '未知错误'}")
        return None


# 使用示例
import pandas as pd
import matplotlib.pyplot as plt

candles = client.get_candles("00700.HK", interval="5m", limit=50)
if candles:
    df = pd.DataFrame(candles)
    df["t"] = pd.to_datetime(df["t"], unit="ms")
    df.set_index("t", inplace=True)
    for col in ["o", "h", "l", "c", "v"]:
        df[col] = pd.to_numeric(df[col])

    print(df.head())
    # 计算 SMA 简单移动平均
    df["sma_5"] = df["c"].rolling(window=5).mean()
    df["sma_20"] = df["c"].rolling(window=20).mean()

    plt.figure(figsize=(12, 6))
    plt.plot(df.index, df["c"], label="收盘价")
    plt.plot(df.index, df["sma_5"], label="SMA5")
    plt.plot(df.index, df["sma_20"], label="SMA20")
    plt.title("腾讯控股 00700.HK（5m）K 线与均线")
    plt.legend()
    plt.show()

import asyncio
import json
import os
import websockets

WS_URL = "wss://api.deeptick.cc/ws/v1"


class DeepTickWSClient:
    """按 Topic 订阅实时行情，断线自动重连"""

    def __init__(self, token=None, url=WS_URL):
        self.url = url
        self.headers = {"deeptoken": token or os.getenv("DEEPTICK_API_TOKEN", "your_api_key")}
        self.ws = None

    async def subscribe(self):
        # quote 与 depth 必须分开发，一条 subscribe 只能同一种 type
        for topic in ("quote.HK.00700", "depth.HK.00700"):
            await self.ws.send(json.dumps({
                "type": "subscribe",
                "topics": [topic],
                "request_id": f"sub-{topic}"
            }))

    async def handle(self, msg):
        if msg.get("type") == "ack":
            print(f"订阅回执: 成功 {len(msg.get('accepted', []))} 条, 失败 {msg.get('rejected', [])}")
            return
        print(f"【推送】{msg.get('topic')} -> {json.dumps(msg.get('data'), ensure_ascii=False)}")

    async def run(self):
        while True:
            try:
                async with websockets.connect(self.url, additional_headers=self.headers) as ws:
                    self.ws = ws
                    print("WebSocket 连接已建立")
                    welcome = await ws.recv()  # 服务端首帧 welcome
                    print(f"welcome: {welcome[:60]}")
                    await self.subscribe()
                    async for raw in ws:
                        await self.handle(json.loads(raw))
            except Exception as e:
                # 出现异常不退出，5 秒后重连并重新订阅
                print(f"连接异常，5 秒后重连: {e}")
                await asyncio.sleep(5)


if __name__ == "__main__":
    asyncio.run(DeepTickWSClient().run())