云原生架构下港股高可用行情系统构建 ——API 限频场景实践

import websocket
import json
import time
import threading
from typing import Dict, Optional

# 配置项（适配腾讯云环境）
API_KEY: str = "你的AllTick API密钥"
WS_URL: str = "wss://api.alltick.co/v1/hk/stock/realtime"
POLL_INTERVAL: int = 5  # 备用链路拉取间隔（秒）
CACHE: Dict[str, Dict] = {}  # 内存缓存：key=标的代码，value=最新行情数据

def update_cache(symbol: str, data: Dict) -> None:
    """更新内存缓存，保障数据时序最新"""
    global CACHE
    CACHE[symbol] = data

def get_latest_data(symbol: str) -> Optional[Dict]:
    """获取缓存数据，供业务系统调用"""
    return CACHE.get(symbol)

# 主链路：WebSocket订阅推送
def on_message(ws, message: str) -> None:
    """处理实时推送数据，更新缓存"""
    try:
        data = json.loads(message)
        if data.get("symbol") and data.get("price"):
            update_cache(data["symbol"], data)
            print(f"【实时行情】{data['symbol']} 价格：{data['price']} 时间：{data['timestamp']}")
    except Exception as e:
        print(f"主链路数据解析异常：{str(e)}")

def on_open(ws) -> None:
    """连接成功，订阅目标标的（可批量扩展）"""
    subscribe_payload = {
        "action": "subscribe",
        "symbols": ["00700.HK", "09988.HK"],  # 支持多标的批量订阅
        "api_key": API_KEY
    }
    ws.send(json.dumps(subscribe_payload))
    print("主链路连接成功，已订阅核心标的")

# 备用链路：低频率定时拉取
def backup_poll_task() -> None:
    """定时拉取行情数据，主链路异常时兜底补全"""
    import requests
    while True:
        try:
            for symbol in ["00700.HK", "09988.HK"]:
                if not get_latest_data(symbol):
                    # 调用AllTick API拉取最新行情
                    resp = requests.get(
                        url="https://api.alltick.co/v1/hk/stock/quote",
                        params={"symbol": symbol, "api_key": API_KEY},
                        timeout=10
                    )
                    resp_data = resp.json()
                    if resp_data.get("code") == 0 and resp_data.get("data"):
                        update_cache(symbol, resp_data["data"])
                        print(f"【备用链路补全】{symbol} 数据已更新")
        except Exception as e:
            print(f"备用链路拉取异常：{str(e)}")
        time.sleep(POLL_INTERVAL)

# 启动行情服务
if __name__ == "__main__":
    # 启动备用链路（守护线程，异常自动重启）
    threading.Thread(target=backup_poll_task, daemon=True).start()
    # 启动主链路（断线自动重连）
    ws_app = websocket.WebSocketApp(
        url=WS_URL,
        on_open=on_open,
        on_message=on_message
    )
    ws_app.run_forever(ping_interval=10, ping_timeout=5)