4、Python实现OBI计算：编写OBI计算函数、处理多档位数据、性能优化

好，咱们进入实战环节。

前面几章我们把OBI的理论讲透了，也分析了它在不同市场环境下的表现。但说一千道一万，最终还是要落到代码上。这一章，我带你手写一个生产级别的OBI计算模块。

4.1 从最简单的OBI开始

先别急着上多档位。我个人的习惯是，先写一个最朴素的版本，跑通再说。

单档位OBI，就是只看买一和卖一。公式很简单：

OBI = (买一量 - 卖一量) / (买一量 + 卖一量)

嗯，这里要注意：分母不能为零。虽然极端情况很少，但我在实盘回测时遇到过几次，某个冷门币种突然没有卖单了，分母直接归零，程序当场崩溃。所以，防御性编程是必须的。

def calculate_obi_single(bid_volume: float, ask_volume: float) -> float:
    """
    计算单档位订单簿不平衡度
    :param bid_volume: 买一量
    :param ask_volume: 卖一量
    :return: OBI值，范围[-1, 1]
    """
    total = bid_volume + ask_volume
    if total == 0:
        return 0.0  # 没有挂单，返回中性值
    return (bid_volume - ask_volume) / total

这个函数，说白了就是小学数学。但别小看它，很多量化策略的底层逻辑就靠这个值来驱动。

4.2 多档位OBI：加权才是灵魂

单档位有个问题——它只看表面。大户如果想操纵盘口，完全可以挂一个巨大的买一，然后悄悄在买二、买三出货。你如果只看买一，就会被骗进去。

所以，我们需要多档位数据。我个人习惯用前5档，因为大多数交易所的深度快照都提供这个数据，而且计算量可控。

加权方式有很多种，我推荐两种：

  • 线性加权：第1档权重最高，依次递减。比如权重序列 [5, 4, 3, 2, 1]。
  • 指数加权：衰减更快，更贴近真实盘口的影响力。比如权重序列 [1, 0.5, 0.25, 0.125, 0.0625]。

我在项目中遇到过一个问题：用线性加权时，如果某档位突然出现巨量挂单，OBI会剧烈抖动。后来我改用指数加权，平滑了很多。你想想看，离当前价格越远的挂单，对价格的直接影响本来就小，指数衰减更符合逻辑。

def calculate_obi_multi(bids: list, asks: list, weights: list = None) -> float:
    """
    计算多档位加权OBI
    :param bids: 买单列表，每个元素为 (价格, 数量)
    :param asks: 卖单列表，每个元素为 (价格, 数量)
    :param weights: 权重列表，默认指数衰减 [1, 0.5, 0.25, 0.125, 0.0625]
    :return: 加权OBI值
    """
    if weights is None:
        weights = [1, 0.5, 0.25, 0.125, 0.0625]
    
    # 确保档位数一致
    depth = min(len(bids), len(asks), len(weights))
    if depth == 0:
        return 0.0
    
    weighted_bid = 0.0
    weighted_ask = 0.0
    
    for i in range(depth):
        weighted_bid += bids[i][1] * weights[i]
        weighted_ask += asks[i][1] * weights[i]
    
    total = weighted_bid + weighted_ask
    if total == 0:
        return 0.0
    
    return (weighted_bid - weighted_ask) / total

这段代码，我建议你直接复制到自己的工具库里。它处理了档位数不一致的情况——有些交易所只返回3档，有些返回10档，我们取最小值，避免索引越界。

4.3 性能优化：别让OBI成为瓶颈

如果你只是做日频交易，上面的代码完全够用。但如果你做高频，或者同时监控几百个交易对，那就要考虑性能了。

我曾经在回测时，用上面的函数处理一年的tick数据，跑了整整两个小时。后来优化后，同样的数据只用了15分钟。差距在哪？

4.3.1 向量化计算

Python的for循环很慢。能用NumPy就用NumPy。

import numpy as np

def calculate_obi_vectorized(bid_volumes: np.ndarray, ask_volumes: np.ndarray, weights: np.ndarray) -> float:
    """
    向量化计算多档位OBI
    """
    weighted_bid = np.dot(bid_volumes, weights)
    weighted_ask = np.dot(ask_volumes, weights)
    total = weighted_bid + weighted_ask
    
    if total == 0:
        return 0.0
    return (weighted_bid - weighted_ask) / total

你看，三行代码搞定。而且NumPy底层是C语言实现的，速度比纯Python快几十倍。

4.3.2 缓存权重数组

权重数组在每次计算时都是固定的。别在函数里重复生成，把它定义成全局常量或者类属性。

# 全局常量，只初始化一次
OBI_WEIGHTS = np.array([1, 0.5, 0.25, 0.125, 0.0625], dtype=np.float64)

这个优化虽然小，但如果你每秒计算几百次，积少成多也很可观。

4.3.3 避免频繁的类型转换

从交易所API拿到的数据，通常是Python列表或者字典。每次计算前都转成NumPy数组，其实也有开销。我的做法是：在数据接收层就统一转成NumPy数组，后续计算全部用数组操作。

小技巧： 如果你用Pandas做回测，可以把OBI计算写成apply函数，但记得用swifter或者numba加速。我个人的经验是，numba在循环计算上表现更好，尤其是处理逐笔tick数据时。

4.4 完整的多档位OBI计算类

最后，我给出一个生产级别的封装。它包含了缓存、向量化、异常处理，以及一个简单的历史记录功能。

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

class OBICalculator:
    """订单簿不平衡度计算器"""
    
    def __init__(self, depth: int = 5, decay: str = 'exponential'):
        self.depth = depth
        self.history = []  # 保存最近N次OBI值
        
        if decay == 'exponential':
            self.weights = np.array([0.5 ** i for i in range(depth)], dtype=np.float64)
        elif decay == 'linear':
            self.weights = np.array(list(range(depth, 0, -1)), dtype=np.float64)
        else:
            raise ValueError("decay参数必须是 'exponential' 或 'linear'")
        
        # 归一化权重，使得总和为1
        self.weights /= self.weights.sum()
    
    def calculate(self, bids: List[Tuple[float, float]], asks: List[Tuple[float, float]]) -> float:
        """
        计算当前快照的OBI
        """
        # 提取数量并转为NumPy数组
        bid_volumes = np.array([b[1] for b in bids[:self.depth]], dtype=np.float64)
        ask_volumes = np.array([a[1] for a in asks[:self.depth]], dtype=np.float64)
        
        # 补齐到固定长度（如果档位不足）
        if len(bid_volumes) < self.depth:
            bid_volumes = np.pad(bid_volumes, (0, self.depth - len(bid_volumes)), 'constant')
        if len(ask_volumes) < self.depth:
            ask_volumes = np.pad(ask_volumes, (0, self.depth - len(ask_volumes)), 'constant')
        
        weighted_bid = np.dot(bid_volumes, self.weights)
        weighted_ask = np.dot(ask_volumes, self.weights)
        
        total = weighted_bid + weighted_ask
        obi = 0.0 if total == 0 else (weighted_bid - weighted_ask) / total
        
        # 记录历史
        self.history.append(obi)
        if len(self.history) > 1000:
            self.history.pop(0)
        
        return obi
    
    def get_smoothed_obi(self, window: int = 10) -> float:
        """获取平滑后的OBI（移动平均）"""
        if len(self.history) < window:
            return self.history[-1] if self.history else 0.0
        return np.mean(self.history[-window:])
核心要点：
  • 权重归一化：让OBI始终保持在[-1, 1]区间，方便与其他指标对比
  • 档位补齐：某些交易所返回的深度不足5档时，用0填充，避免计算偏差
  • 历史记录：用于后续的平滑处理，减少噪声
避坑指南： 我曾经在实盘中发现，某些交易所的深度快照中，同一价格可能出现多笔挂单。比如买一有两个不同的订单，价格相同但数量分开。这时候一定要先按价格聚合，否则OBI会算错。建议在数据预处理阶段就做好聚合。

4.5 本章知识体系

下面这张图，帮你梳理了OBI计算的完整流程：

OBI计算核心流程 原始深度数据 数据预处理 价格聚合 · 档位补齐 加权计算 线性/指数权重 OBI值输出 范围[-1, 1] 性能优化 向量化 · 缓存 · Numba 平滑处理（可选） 图：OBI计算从原始数据到最终输出的完整链路

嗯，这一章的内容就到这里。代码不多，但每一行都是实战中打磨出来的。你把这个OBI计算器写好，后面讲策略构建的时候，直接拿来用就行。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →