行情数据对接与 WebSocket

import requests
import timeit


def get_orderbook():
    """通过 REST 接口获取 orderbook"""
    orderbook = requests.get(
        "https://api.gemini.com/v1/book/btcusd"
    ).json()


# 测试 REST 接口延迟
n = 10
latency = timeit.timeit(
    'get_orderbook()',
    setup='from __main__ import get_orderbook',
    number=n
) / n
print(f'REST 平均延迟: {latency * 1000:.2f} ms')
# 典型输出（美国东海岸服务器）: ~200 ms

# 测量 TCP 和 SSL 握手的耗时
curl -w "TCP: %{time_connect}s, SSL: %{time_appconnect}s\n" \
     -so /dev/null https://www.gemini.com
# TCP handshake: 0.072s
# SSL handshake: 0.119s
# 握手总耗时: ~0.2s —— 占了总延迟的一大半！

import websocket
import threading
import time


def on_message(ws, message):
    """收到服务器消息时调用"""
    print(f'Received: {message}')


def on_open(ws):
    """连接建立时调用"""
    def send_messages():
        for i in range(5):
            time.sleep(0.01)
            ws.send(f"{i}")
            print(f'Sent: {i}')
        time.sleep(1)
        ws.close()
        print("WebSocket closed")

    # 在独立线程中发送消息（不阻塞接收）
    threading.Thread(target=send_messages).start()


ws = websocket.WebSocketApp(
    "ws://echo.websocket.org/",
    on_message=on_message,
    on_open=on_open
)
ws.run_forever()

# 输出:
# Sent: 0
# Sent: 1
# Received: 0
# Sent: 2
# Received: 1
# ...

import ssl
import websocket
import json


# 全局计数器
count = 5


def on_message(ws, message):
    global count
    data = json.loads(message)
    print(f"事件类型: {data['type']}, "
          f"sequence: {data.get('socket_sequence', 'N/A')}")
    count -= 1
    # 接收 5 次消息后关闭连接
    if count == 0:
        ws.close()


ws = websocket.WebSocketApp(
    "wss://api.gemini.com/v1/marketdata/btcusd?top_of_book=true&offers=true",
    on_message=on_message
)
ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})

# 输出（服务器主动推送，客户端没有发送任何请求！）:
# 事件类型: update, sequence: 0
# 事件类型: update, sequence: 1
# 事件类型: update, sequence: 2
# ...

import copy
import json
import ssl
import time
import websocket


class OrderBook:
    """委托账本数据结构"""

    BIDS = 'bid'
    ASKS = 'ask'

    def __init__(self, limit=20):
        self.limit = limit
        self.bids = {}           # {price: amount}
        self.asks = {}
        self.bids_sorted = []    # [(price, amount), ...]
        self.asks_sorted = []

    def insert(self, price, amount, direction):
        """
        插入或更新一条委托数据。

        关键逻辑: amount == 0 表示该价位已清空，需要删除
        """
        target = self.bids if direction == self.BIDS else self.asks
        if amount == 0:
            target.pop(price, None)
        else:
            target[price] = amount

    def sort_and_truncate(self):
        """排序并截取前 limit 条"""
        # bids 从高到低排序
        self.bids_sorted = sorted(
            self.bids.items(), reverse=True
        )[:self.limit]
        # asks 从低到高排序
        self.asks_sorted = sorted(
            self.asks.items()
        )[:self.limit]
        # 回写字典
        self.bids = dict(self.bids_sorted)
        self.asks = dict(self.asks_sorted)

    def get_copy_of_bids_and_asks(self):
        """
        深拷贝返回排序后的数据。

        为什么用深拷贝？
        如果直接返回引用，下一次 sort_and_truncate() 会修改
        已返回的数据，造成潜在的 bug。
        """
        return (
            copy.deepcopy(self.bids_sorted),
            copy.deepcopy(self.asks_sorted)
        )


class Crawler:
    """WebSocket 行情爬虫"""

    def __init__(self, symbol, output_file):
        self.orderbook = OrderBook(limit=10)
        self.output_file = output_file

        # 注意：on_message 是类方法，需要 lambda 包装传递 self
        self.ws = websocket.WebSocketApp(
            f'wss://api.gemini.com/v1/marketdata/{symbol}',
            on_message=lambda ws, msg: self.on_message(msg)
        )
        self.ws.run_forever(sslopt={'cert_reqs': ssl.CERT_NONE})

    def on_message(self, message):
        """处理收到的行情更新"""
        data = json.loads(message)

        # 处理每个变更事件
        for event in data['events']:
            price = float(event['price'])
            amount = float(event['remaining'])  # 注意：用 remaining 而非 delta
            direction = event['side']
            self.orderbook.insert(price, amount, direction)

        # 排序和截取
        self.orderbook.sort_and_truncate()

        # 持久化到文件
        with open(self.output_file, 'a+') as f:
            bids, asks = self.orderbook.get_copy_of_bids_and_asks()
            output = {
                'bids': bids,
                'asks': asks,
                'ts': int(time.time() * 1000)
            }
            f.write(json.dumps(output) + '\n')


if __name__ == '__main__':
    crawler = Crawler(symbol='BTCUSD', output_file='BTCUSD.txt')

{
  "bids": [[11398.73, 0.963], [11398.72, 0.989], [11397.32, 1.0]],
  "asks": [[11407.42, 1.308], [11407.92, 1.0], [11409.48, 2.0]],
  "ts": 1562558996535
}