贵金属 API 避坑：黄金/白银行情接口常见陷阱（数据漂移、断点、延迟）

import requests

API_TOKEN = "your_token_here"
headers = {"accept": "application/json", "token": API_TOKEN}
url = "https://api.itick.org/forex/quotes?region=GB&codes=XAUUSD,XAGUSD"

resp = requests.get(url, headers=headers, timeout=3)
if resp.status_code == 200:
    data = resp.json()
    gold = data["data"]["XAUUSD"]
    silver = data["data"]["XAGUSD"]
    print(f"黄金: {gold['ld']} @ {gold['t']}  白银: {silver['ld']} @ {silver['t']}")

import websocket
import json
import threading
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

API_TOKEN = "your_api_token_here"
# WebSocket 端点：免费套餐使用 wss://api-free.itick.org/forex
WS_URL = "wss://api.itick.org/forex"  # 付费套餐

class GoldDataMonitor:
    def __init__(self, token):
        self.token = token
        self.ws = None
        self.keep_running = True
        self.subscribed = False
        self.last_price = None
        self.last_timestamp = None
        self.data_gap_detected = False
        # 配置数据新鲜度阈值（秒）
        self.freshness_threshold = 30

    def on_message(self, ws, message):
        try:
            data = json.loads(message)
            # 连接成功确认
            if data.get("code") == 1 and data.get("resAc") == "auth":
                logger.info("认证成功，开始订阅数据...")
                self.subscribe_data(ws)
            # 订阅成功确认
            elif data.get("code") == 1 and data.get("resAc") == "subscribe":
                logger.info("订阅成功，接收行情数据...")
                self.subscribed = True
            # 行情数据推送
            elif data.get("code") == 1 and "data" in data:
                tick = data["data"]
                # 提取关键字段
                symbol = tick.get("s")       # 品种代码（GC/SI）
                price = tick.get("ld")       # 最新价
                timestamp_ms = tick.get("t")  # 交易所成交时间戳
                msg_type = tick.get("type")   # 数据类型：tick/quote

                # 数据新锐度检查：如果最新数据的 timestamp 明显落后于当前系统时间
                if timestamp_ms:
                    now_ms = int(time.time() * 1000)
                    latency = now_ms - timestamp_ms
                    if latency > self.freshness_threshold * 1000:
                        logger.warning(f"[断点告警] 数据延迟 {latency//1000}s，超出阈值，可能处于断点区域")
                        self.data_gap_detected = True
                    else:
                        self.data_gap_detected = False
                        self.last_price = price
                        self.last_timestamp = timestamp_ms
                        logger.info(f"{symbol}: {price}, 延迟 {latency}ms")

        except json.JSONDecodeError:
            logger.error(f"消息解析失败: {message}")
        except Exception as e:
            logger.error(f"处理消息异常: {e}")

    def on_error(self, ws, error):
        logger.error(f"WebSocket 错误: {error}")
        self.data_gap_detected = True

    def on_close(self, ws, close_status_code, close_msg):
        logger.warning(f"WebSocket 断开，状态码 {close_status_code}，正在重连...")
        self.subscribed = False
        self.data_gap_detected = True
        self.reconnect()

    def on_open(self, ws):
        logger.info("WebSocket 连接已建立，进行认证...")
        # 认证已在连接时通过 header 中的 token 完成，此处留空即可

    def subscribe_data(self, ws):
        # 订阅黄金和白银实时行情
        subscribe_msg = {
            "ac": "subscribe",
            "params": "XAUUSD$GB,XAGUSD$GB",    # 黄金 XAUUSD（GB 市场）、白银 XAGUSD（GB 市场）
            "types": "quote"             # quote: 报价数据， tick: 逐笔成交
        }
        ws.send(json.dumps(subscribe_msg))

    def send_heartbeat(self):
        """发送心跳维持连接"""
        while self.keep_running and self.ws:
            try:
                time.sleep(30)
                if self.ws and self.ws.sock and self.ws.sock.connected:
                    heartbeat_msg = {"ac": "ping"}
                    self.ws.send(json.dumps(heartbeat_msg))
                    logger.debug("发送心跳消息")
            except Exception as e:
                logger.error(f"发送心跳异常: {e}")

    def reconnect(self):
        """带指数退避的断线重连"""
        retry_delay = 2
        max_delay = 60

        while self.keep_running:
            try:
                logger.info(f"尝试重连，等待 {retry_delay} 秒...")
                time.sleep(retry_delay)
                self.start()
                break
            except Exception as e:
                logger.error(f"重连失败: {e}")
                retry_delay = min(retry_delay * 2, max_delay)

    def start(self):
        websocket.enableTrace(False)
        self.ws = websocket.WebSocketApp(
            WS_URL,
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close,
            header={"token": self.token}
        )
        # 启动心跳线程
        heartbeat_thread = threading.Thread(target=self.send_heartbeat, daemon=True)
        heartbeat_thread.start()
        # 运行 WebSocket（阻塞）
        self.ws.run_forever()

if __name__ == "__main__":
    monitor = GoldDataMonitor(API_TOKEN)
    monitor.start()

import requests
import time
import websocket
import json
import threading
from datetime import datetime

API_TOKEN = "your_api_token_here"
BASE_URL = "https://api.itick.org"

def measure_rest_latency():
    """测量 REST API 端到端延迟"""
    url = f"{BASE_URL}/forex/quotes?region=GB&codes=XAUUSD"
    headers = {"accept": "application/json", "token": API_TOKEN}

    t_start_local = time.time()

    try:
        response = requests.get(url, headers=headers, timeout=5)
        t_received_local = time.time()

        if response.status_code == 200:
            data = response.json()
            if data.get("code") == 0:
                gc_data = data["data"]["XAUUSD"]
                # 交易所成交时间戳（毫秒）
                exchange_timestamp_ms = gc_data.get("t")
                if exchange_timestamp_ms:
                    exchange_time = exchange_timestamp_ms / 1000.0

                    # 计算端到端延迟
                    e2e_latency = t_received_local - exchange_time
                    request_latency = t_received_local - t_start_local

                    print(f"[REST] 端到端延迟: {e2e_latency*1000:.1f}ms")
                    print(f"[REST] 请求往返延迟: {request_latency*1000:.1f}ms")
                    return e2e_latency
    except Exception as e:
        print(f"REST 延迟测量失败: {e}")
    return None

# WebSocket 延迟测量（被动接收）
ws_latency_samples = []

def on_ws_message(ws, message):
    try:
        data = json.loads(message)
        if "data" in data and "t" in data["data"]:
            tick = data["data"]
            exchange_timestamp_ms = tick["t"]
            now_ms = time.time() * 1000

            e2e_latency_ms = now_ms - exchange_timestamp_ms

            ws_latency_samples.append(e2e_latency_ms)
            # 每 10 条打印一次统计
            if len(ws_latency_samples) % 10 == 0:
                avg_latency = sum(ws_latency_samples[-100:]) / min(len(ws_latency_samples), 100)
                print(f"[WebSocket] 当前延迟 {e2e_latency_ms:.1f}ms, 平均延迟 {avg_latency:.1f}ms, 样本数 {len(ws_latency_samples)}")
    except:
        pass

def measure_websocket_latency_demo():
    """启动 WebSocket 延迟监控示例"""
    ws_url = "wss://api.itick.org/forex"

    ws = websocket.WebSocketApp(
        ws_url,
        on_message=on_ws_message,
        header={"token": API_TOKEN}
    )

    def run_ws():
        ws.run_forever()

    thread = threading.Thread(target=run_ws, daemon=True)
    thread.start()

    # 运行 30 秒后断开
    time.sleep(30)
    ws.close()

    if ws_latency_samples:
        avg = sum(ws_latency_samples) / len(ws_latency_samples)
        p99 = sorted(ws_latency_samples)[int(len(ws_latency_samples) * 0.99)]
        print(f"\n===== WebSocket 延迟统计 =====")
        print(f"平均延迟: {avg:.1f}ms")
        print(f"P99 延迟: {p99:.1f}ms")
        print(f"最小延迟: {min(ws_latency_samples):.1f}ms")
        print(f"最大延迟: {max(ws_latency_samples):.1f}ms")

if __name__ == "__main__":
    # 测量 REST API 延迟
    measure_rest_latency()

    # 测量 WebSocket 延迟分布
    measure_websocket_latency_demo()