4、信号计算引擎：编写Python类实时计算买卖压力差、累积失衡度、加权失衡比

好，咱们进入正题。

前面几章我们聊了订单簿的基础结构，也讲了怎么抓取和清洗数据。但说实话，那些都只是准备工作。真正让策略跑起来、让信号变成交易决策的，是这一章要讲的——信号计算引擎。

我个人习惯把信号计算引擎比作「订单簿的翻译官」。原始数据是密密麻麻的挂单，普通人看着眼晕。但引擎能把这些数字翻译成三个核心指标：买卖压力差、累积失衡度、加权失衡比。这三个指标，基本能覆盖90%的订单簿失衡场景。

4.1 三个指标，各司其职

先别急着写代码。咱们得先搞清楚，这三个指标到底在算什么。

指标名称	英文缩写	核心含义	典型用法
买卖压力差	PSD (Pressure Spread Difference)	当前档位买盘总量 vs 卖盘总量的差值	判断当前瞬间的多空力量对比
累积失衡度	CID (Cumulative Imbalance Degree)	从第1档到第N档的累积买卖量差值	判断深度方向上的持续失衡
加权失衡比	WIR (Weighted Imbalance Ratio)	按距离最优价的距离加权后的失衡比例	消除「挂单陷阱」，更真实反映意图

嗯，这里要注意。很多新手只盯着买卖压力差看，觉得正数就是买盘强，负数就是卖盘强。其实不然。我在项目中遇到过好几次，买卖压力差明明是正的，价格却往下砸。为什么？因为有人在第5档、第6档偷偷挂了大单，压力差没算进去。这就是为什么我们需要累积失衡度和加权失衡比。

4.2 代码实现：OrderBookImbalance 类

好，直接上代码。我习惯把信号计算引擎封装成一个类，这样在回测和实盘里都能复用。

import numpy as np
from typing import List, Tuple, Optional

class OrderBookImbalance:
    """
    订单簿失衡信号计算引擎
    支持实时计算买卖压力差、累积失衡度、加权失衡比
    """
    
    def __init__(self, depth: int = 10):
        """
        :param depth: 计算深度，默认取前10档
        """
        self.depth = depth
        self._last_psd = 0.0
        self._last_cid = 0.0
        self._last_wir = 0.0
        
    def compute_psd(self, bids: List[Tuple[float, float]], 
                    asks: List[Tuple[float, float]]) -> float:
        """
        计算买卖压力差 (Pressure Spread Difference)
        
        :param bids: 买盘数据，格式 [(price, volume), ...]
        :param asks: 卖盘数据，格式 [(price, volume), ...]
        :return: 压力差值，正数表示买盘强，负数表示卖盘强
        """
        # 取前depth档
        bids_slice = bids[:self.depth]
        asks_slice = asks[:self.depth]
        
        # 计算总买量和总卖量
        total_bid_vol = sum(v for _, v in bids_slice)
        total_ask_vol = sum(v for _, v in asks_slice)
        
        # 压力差 = 买盘总量 - 卖盘总量
        psd = total_bid_vol - total_ask_vol
        
        self._last_psd = psd
        return psd
    
    def compute_cid(self, bids: List[Tuple[float, float]], 
                    asks: List[Tuple[float, float]], 
                    levels: int = 5) -> float:
        """
        计算累积失衡度 (Cumulative Imbalance Degree)
        
        :param bids: 买盘数据
        :param asks: 卖盘数据
        :param levels: 累积计算的档位数
        :return: 累积失衡度
        """
        # 取前levels档
        bids_slice = bids[:levels]
        asks_slice = asks[:levels]
        
        # 逐档累积差值
        cumulative_diff = 0.0
        for i in range(min(len(bids_slice), len(asks_slice))):
            bid_vol = bids_slice[i][1]
            ask_vol = asks_slice[i][1]
            cumulative_diff += (bid_vol - ask_vol)
        
        self._last_cid = cumulative_diff
        return cumulative_diff
    
    def compute_wir(self, bids: List[Tuple[float, float]], 
                    asks: List[Tuple[float, float]], 
                    decay_factor: float = 0.9) -> float:
        """
        计算加权失衡比 (Weighted Imbalance Ratio)
        
        使用指数衰减权重，越靠近最优价的档位权重越大
        
        :param bids: 买盘数据
        :param asks: 卖盘数据
        :param decay_factor: 衰减因子，默认0.9
        :return: 加权失衡比，范围[-1, 1]
        """
        # 计算加权买量
        weighted_bid = 0.0
        for i, (_, vol) in enumerate(bids[:self.depth]):
            weight = decay_factor ** i
            weighted_bid += vol * weight
        
        # 计算加权卖量
        weighted_ask = 0.0
        for i, (_, vol) in enumerate(asks[:self.depth]):
            weight = decay_factor ** i
            weighted_ask += vol * weight
        
        # 防止除零
        total = weighted_bid + weighted_ask
        if total == 0:
            return 0.0
        
        # 加权失衡比 = (加权买量 - 加权卖量) / (加权买量 + 加权卖量)
        wir = (weighted_bid - weighted_ask) / total
        
        self._last_wir = wir
        return wir
    
    def compute_all(self, bids: List[Tuple[float, float]], 
                    asks: List[Tuple[float, float]]) -> dict:
        """
        一次性计算所有指标
        """
        return {
            'psd': self.compute_psd(bids, asks),
            'cid': self.compute_cid(bids, asks),
            'wir': self.compute_wir(bids, asks)
        }

4.3 三个指标的计算逻辑详解

代码贴完了，咱们拆开揉碎了讲一讲。

4.3.1 买卖压力差（PSD）

这个最简单。就是把买一到买十的总量加起来，减去卖一到卖十的总量。说白了就是「当前所有明牌的多头兵力减去空头兵力」。正值说明买盘人多，负值说明卖盘人多。

但我要提醒你——这个指标有个致命缺陷。它把所有档位的挂单量一视同仁。你想想看，第1档挂1个BTC和第10档挂1个BTC，对价格的冲击能一样吗？显然不一样。所以光看PSD是不够的。

4.3.2 累积失衡度（CID）

CID解决了「只看表面」的问题。它从第1档开始，逐档累加买卖差值。比如你算前5档的CID，就是第1档差值 + 第2档差值 + ... + 第5档差值。

这样做的好处是，你能看到失衡是在浅层还是深层。我曾经在ETH的订单簿上发现，PSD是正的（买盘总量大），但CID从第3档开始就变成负的了。这说明什么？说明有人在浅层挂小单诱多，深层偷偷挂大单准备砸盘。这种信号，光看PSD根本发现不了。

4.3.3 加权失衡比（WIR）

WIR是我个人最看重的指标。它引入了权重概念——离最优价越近的档位，权重越大。用指数衰减的方式，让第1档的权重是第2档的0.9倍，第2档是第3档的0.9倍，以此类推。

为什么这么做？因为离最优价越近的挂单，越能反映交易者的真实意图。有人愿意在买一价挂单，说明他真的想在这个价位买入。而在买十价挂单的人，可能只是挂个钓鱼单，根本不指望成交。

4.4 信号计算引擎的完整流程图

下面这张图，是我做这个引擎时的设计思路。你看一眼就能明白整个计算流程。

4.5 实战中的使用技巧

代码和流程图都有了，最后聊几个实战技巧。

• 多时间框架验证：别只看当前tick的失衡值。我习惯把过去10个tick的PSD、CID、WIR存下来，算一个滚动均值。如果当前值偏离均值超过2个标准差，才认为是有效信号。
• 结合成交量验证：失衡信号出来之后，如果成交量没有同步放大，很可能是假信号。我在项目中遇到过好几次，失衡值很大但成交量萎缩，结果价格根本没动。
• 参数自适应：不同币种、不同行情阶段，depth和decay_factor应该不一样。波动大的时候depth设小一点（比如5档），波动小的时候设大一点（比如20档）。

---