第3章实时数据获取

3、实时数据获取：使用WebSocket获取实时行情数据、数据清洗与预处理

做VWAP策略，最核心的一环就是数据。你想想看，策略再牛，数据要是慢了半拍，或者数据本身就有问题，那结果就是白忙活。我个人习惯，在实时交易中，坚决不用轮询（Polling）去拉行情，延迟太高了。必须上WebSocket，这才是实时该有的样子。

3.1 为什么非得用WebSocket？

传统的HTTP请求，是你问一句，服务器答一句。这在查历史数据时没问题，但实时行情每秒都在变，你总不能每秒都去问一次吧？效率太低，而且容易漏掉中间的数据。

WebSocket不一样。它建立一条长连接，服务器有数据就主动推给你。说白了，就像你开了个水龙头，水（数据）自己流过来，你不用一遍遍去拧开关。

核心优势：

低延迟：建立连接后，数据推送几乎无延迟。我在项目中遇到过，用HTTP轮询时，行情峰值延迟能到200ms，换成WebSocket后直接降到10ms以内。
双向通信：你不仅可以收数据，还可以随时向服务器发送订阅请求，动态调整关注的品种。
节省带宽：没有HTTP请求头那些冗余信息，纯数据帧传输，效率高得多。

3.2 实战：用Python连接WebSocket行情

嗯，这里要注意。不同的交易所，WebSocket的协议细节略有不同，但核心逻辑是一样的。我以币安（Binance）的现货行情为例，给你演示一下完整的流程。

首先，安装必要的库：

pip install websocket-client

然后，写一个连接和接收数据的类：

import json
import websocket
import threading

class RealTimeDataFeed:
    def __init__(self, symbol="btcusdt"):
        self.symbol = symbol.lower()
        # 币安的WebSocket地址，订阅深度数据
        self.url = f"wss://stream.binance.com:9443/ws/{self.symbol}@depth20@100ms"
        self.ws = None
        self.data_queue = []  # 用来暂存原始数据

    def on_message(self, ws, message):
        """收到消息时的回调函数"""
        data = json.loads(message)
        # 这里先不做处理，直接存起来
        self.data_queue.append(data)
        # 我个人习惯，在这里加一个简单的打印，方便调试
        # print(f"收到数据: {data['lastUpdateId']}")

    def on_error(self, ws, error):
        print(f"连接出错: {error}")

    def on_close(self, ws, close_status_code, close_msg):
        print("连接已关闭")

    def on_open(self, ws):
        print("连接已建立")
        # 可以在这里发送订阅消息，但本例中URL已经指定了订阅内容

    def start(self):
        """启动WebSocket连接"""
        websocket.enableTrace(False)  # 关闭调试日志
        self.ws = websocket.WebSocketApp(
            self.url,
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )
        # 运行在一个单独的线程中，避免阻塞主程序
        wst = threading.Thread(target=self.ws.run_forever)
        wst.daemon = True
        wst.start()

    def stop(self):
        if self.ws:
            self.ws.close()

# 使用示例
feed = RealTimeDataFeed("btcusdt")
feed.start()

# 让程序跑一会儿，收集数据
import time
time.sleep(5)

# 看看收集了多少条数据
print(f"共收到 {len(feed.data_queue)} 条原始数据")

我的小技巧：WebSocket连接在公网上偶尔会断开。我曾经在实盘交易中，因为网络抖动导致连接断了，但程序没感知到，结果策略一直用旧数据在跑。后来我加了一个心跳检测（Ping/Pong），每30秒检查一次连接状态，断了就自动重连。这个一定要加上。

3.3 数据清洗：把“脏数据”变成“干净数据”

从WebSocket拿到的原始数据，不能直接用。为什么？因为交易所推送的数据里，可能包含重复数据、过期数据，甚至偶尔会有格式错误。你想想看，如果拿一个错误的价格去算VWAP，那整个策略就偏了。

清洗步骤，我一般分三步走：

去重：检查数据的唯一ID（比如updateId），如果ID重复，直接丢弃。
排序：虽然WebSocket是顺序推送的，但偶尔会因为网络原因导致后发的数据先到。所以，按时间戳或ID排序很重要。
过滤异常值：价格突然跳空太大，或者买卖价差（Spread）异常，这些数据要标记出来，不能参与计算。

来看代码实现：

class DataCleaner:
    def __init__(self):
        self.last_update_id = 0
        self.clean_data = []

    def clean(self, raw_data):
        """
        清洗单条原始数据
        返回清洗后的数据，如果数据无效则返回None
        """
        # 1. 检查数据结构是否完整
        if 'lastUpdateId' not in raw_data or 'bids' not in raw_data:
            return None

        current_id = raw_data['lastUpdateId']

        # 2. 去重：如果ID小于等于上一次的ID，说明是重复或过期数据
        if current_id <= self.last_update_id:
            return None

        # 3. 更新最新的ID
        self.last_update_id = current_id

        # 4. 提取并格式化买卖盘数据
        # 原始数据中，bids和asks是列表，每个元素是 [价格, 数量]
        bids = [[float(price), float(qty)] for price, qty in raw_data['bids']]
        asks = [[float(price), float(qty)] for price, qty in raw_data['asks']]

        # 5. 过滤异常：检查买卖价差是否合理
        best_bid = bids[0][0] if bids else 0
        best_ask = asks[0][0] if asks else 0
        spread = best_ask - best_bid

        # 如果价差为负或者价差过大（比如超过价格的1%），视为异常
        if spread <= 0 or spread / best_bid > 0.01:
            print(f"警告：检测到异常价差 {spread:.2f}")
            return None

        # 6. 返回清洗后的数据
        cleaned = {
            'timestamp': raw_data.get('E', int(time.time() * 1000)),
            'update_id': current_id,
            'bids': bids,
            'asks': asks,
            'spread': spread
        }
        self.clean_data.append(cleaned)
        return cleaned

# 使用示例
cleaner = DataCleaner()
for raw in feed.data_queue:
    result = cleaner.clean(raw)
    if result:
        print(f"清洗成功: 最新价差 {result['spread']:.2f}")

避坑指南：我曾经在清洗数据时，只做了去重，没做排序。结果有一次网络波动，导致几个数据包乱序到达。我按顺序处理，结果用了一个过时的价格去更新了VWAP，导致策略在那一分钟里出现了明显的偏差。从那以后，我每次都会在清洗前，先按update_id排个序。

3.4 预处理：为VWAP计算做准备

数据清洗干净了，但还不能直接用。VWAP计算需要的是“价格”和“成交量”的乘积累加。而交易所推送的深度数据，是各个档位的挂单量和挂单价。我们需要把这些数据，转换成我们需要的格式。

预处理的核心任务有两个：

计算中间价：用买一价和卖一价的平均值，作为当前的市场价格。
估算成交量：深度数据里没有实时成交信息，但我们可以用最近几笔的成交数据（Trade数据）来估算。或者，简单一点，用买一和卖一的挂单量之和，作为当前的市场深度。

来看一个简单的预处理函数：

def preprocess_for_vwap(cleaned_data):
    """
    将清洗后的深度数据，预处理成VWAP计算需要的格式
    """
    # 计算中间价
    best_bid = cleaned_data['bids'][0][0]
    best_ask = cleaned_data['asks'][0][0]
    mid_price = (best_bid + best_ask) / 2

    # 估算当前时刻的成交量（这里用买一和卖一的挂单量之和作为近似）
    bid_volume = cleaned_data['bids'][0][1]
    ask_volume = cleaned_data['asks'][0][1]
    estimated_volume = bid_volume + ask_volume

    # 返回一个字典，包含VWAP计算所需的最小信息
    vwap_input = {
        'timestamp': cleaned_data['timestamp'],
        'price': mid_price,
        'volume': estimated_volume
    }
    return vwap_input

# 应用预处理
vwap_data_list = []
for clean in cleaner.clean_data:
    vwap_input = preprocess_for_vwap(clean)
    vwap_data_list.append(vwap_input)

print(f"预处理完成，共 {len(vwap_data_list)} 条可用数据")

3.5 本章知识体系总览

下面这张图，帮你把整个流程串起来。从WebSocket连接到最终拿到可用的VWAP输入数据，每一步都不能少。

整个流程走下来，你会发现，实时数据获取其实不难，难的是把每一个细节都处理好。尤其是清洗和预处理，很多人会忽略，觉得“数据来了直接用就行”。但我在实盘里吃过亏，所以现在每一步都做得特别仔细。

📚 课程目录

3、实时数据获取：使用WebSocket获取实时行情数据、数据清洗与预处理

3.1 为什么非得用WebSocket？

3.2 实战：用Python连接WebSocket行情

3.3 数据清洗：把“脏数据”变成“干净数据”

3.4 预处理：为VWAP计算做准备

3.5 本章知识体系总览