8. 订单流失衡（Order Flow Imbalance）：失衡率的计算，失衡与价格反转信号，Python实现与回测

各位同学，今天我们来聊聊订单流失衡。这个概念，说白了就是看买卖双方谁在「用力」。我刚开始做订单流分析时，总觉得成交量就是一切。后来发现，成交量的方向比大小更重要。嗯，这就是失衡率的由来。

8.1 什么是订单流失衡？

订单流失衡，英文叫 Order Flow Imbalance，简称 OFI。它衡量的是主动买单和主动卖单之间的差值。

你想想看，如果主动买单远大于主动卖单，说明买方在抢筹。反之，卖方在砸盘。但这里有个关键点——不是看绝对值，而是看相对值。

我在项目中遇到过一个问题：直接用成交量算失衡，结果噪音很大。后来我改用「逐笔成交」数据，效果好了很多。为什么呢？因为逐笔数据能区分主动和被动。

8.2 失衡率的计算方法

计算失衡率，其实就三步：

1. 获取逐笔成交数据：包括时间、价格、成交量、买卖方向
2. 按时间窗口聚合：比如1分钟、5分钟、或者每100笔
3. 计算失衡率：用上面的公式

这里有个细节要注意。我曾经踩过一个坑：把「被动单」也算进去了。结果失衡率永远接近0，因为买卖总量总是相等的。记住，我们只算主动单！

8.3 失衡与价格反转信号

失衡率怎么用？说白了就是找「背离」。

正常情况下，失衡率和价格是同向的：

• 失衡率上升 → 价格应该上涨
• 失衡率下降 → 价格应该下跌

但真正有价值的信号，是「背离」：

为什么会这样？因为价格是由「边际交易者」推动的。当价格创新高时，如果买方力量已经跟不上，说明上涨是「虚的」。嗯，这就是反转信号。

我记得有一次做回测，发现失衡背离信号的胜率能达到65%以上。但要注意，这个信号在趋势行情中容易失效。说白了，趋势行情中「背离可以一直背离」。

8.4 Python实现

好了，上代码。我们用一个简单的类来实现失衡率计算和信号生成。

import pandas as pd
import numpy as np

class OrderFlowImbalance:
    """订单流失衡分析器"""
    
    def __init__(self, window=20):
        self.window = window  # 计算窗口
        
    def calculate_ofi(self, df):
        """
        计算失衡率
        df需要包含: time, price, volume, side (1=主动买, -1=主动卖)
        """
        # 按时间聚合
        df['buy_volume'] = df['volume'] * (df['side'] == 1)
        df['sell_volume'] = df['volume'] * (df['side'] == -1)
        
        # 计算失衡率
        df['ofi'] = (df['buy_volume'] - df['sell_volume']) / \
                    (df['buy_volume'] + df['sell_volume'] + 1e-10)
        
        return df
    
    def detect_divergence(self, df):
        """
        检测背离信号
        返回: 1=看涨背离, -1=看跌背离, 0=无信号
        """
        # 计算价格和失衡率的滚动极值
        df['price_high'] = df['price'].rolling(self.window).max()
        df['price_low'] = df['price'].rolling(self.window).min()
        df['ofi_high'] = df['ofi'].rolling(self.window).max()
        df['ofi_low'] = df['ofi'].rolling(self.window).min()
        
        # 看跌背离：价格新高，OFI未新高
        bearish = (df['price'] == df['price_high']) & \
                  (df['ofi'] < df['ofi_high'] * 0.8)
        
        # 看涨背离：价格新低，OFI未新低
        bullish = (df['price'] == df['price_low']) & \
                  (df['ofi'] > df['ofi_low'] * 1.2)
        
        df['signal'] = 0
        df.loc[bullish, 'signal'] = 1
        df.loc[bearish, 'signal'] = -1
        
        return df

8.5 回测框架

回测是检验策略的唯一标准。我们写一个简单的回测函数：

def backtest_ofi_strategy(df, initial_capital=100000):
    """
    基于失衡背离的回测
    """
    capital = initial_capital
    position = 0
    trades = []
    
    for i in range(len(df)):
        if df.iloc[i]['signal'] == 1:  # 看涨信号
            if position == 0:
                position = capital / df.iloc[i]['price']
                capital = 0
                trades.append({
                    'time': df.iloc[i]['time'],
                    'type': 'buy',
                    'price': df.iloc[i]['price'],
                    'shares': position
                })
                
        elif df.iloc[i]['signal'] == -1:  # 看跌信号
            if position > 0:
                capital = position * df.iloc[i]['price']
                position = 0
                trades.append({
                    'time': df.iloc[i]['time'],
                    'type': 'sell',
                    'price': df.iloc[i]['price'],
                    'shares': 0
                })
    
    # 计算最终收益
    final_value = capital + position * df.iloc[-1]['price']
    return final_value, trades

8.6 实战经验总结

最后，分享几点我的实战经验：

场景	失衡率特征	交易建议
开盘前30分钟	波动大，噪音多	建议用更大的窗口（50+）
收盘前1小时	趋势性强	背离信号更可靠
重大消息前后	失衡率极端化	建议暂停交易
横盘震荡	失衡率在0附近	背离信号容易假突破

我个人习惯把失衡率和其他指标结合使用。比如加上成交量确认，或者配合盘口深度。单一指标总会有失效的时候，组合使用才能提高胜率。

好了，关于订单流失衡，我们就讲到这里。记住，失衡率不是预测工具，而是「当下」的测量工具。它告诉你现在谁在主导，但未来怎么走，还需要结合其他信息。