Поиск паттернов Order Book Imbalance (OBI) на Python

Исходный код

В высокочастотной торговле (HFT) и количественном анализе классические свечные паттерны уступают место анализу микроструктуры рынка. Одним из наиболее сильных краткосрочных предикторов движения цены является дисбаланс лимитных заявок в стакане — Order Book Imbalance (OBI). Он отражает перекос между спросом и предложением на глубоких уровнях стакана (Level 2).

В отличие от анализа распределения доходностей, который мы рассматривали в статье Распределение доходностей и тест Шапиро-Уилка на Python, анализ OBI работает непосредственно с потоком ордеров (Order Flow) и позволяет находить неэффективности на субсекундных интервалах. Для поиска более сложных геометрических паттернов в стакане также можно применять алгоритмы сопоставления временных рядов, о которых рассказано в материале Поиск ценовых паттернов с помощью Dynamic Time Warping (DTW) в Pandas.

Ниже представлен готовый к продакшену скрипт на Python, который генерирует синтетический поток данных Level 2 (для демонстрации), рассчитывает взвешенный OBI по нескольким уровням стакана с экспоненциальным затуханием весов и генерирует торговые сигналы при превышении заданного порога дисбаланса.

import numpy as np
import pandas as pd

def generate_l2_order_book(ticks=500, levels=5):
    np.random.seed(42)
    mid_price = 100.0
    data = []
    
    for t in range(ticks):
        mid_price += np.random.normal(0, 0.02)
        row = {
            'timestamp': pd.Timestamp('2023-10-01') + pd.Timedelta(milliseconds=t*100),
            'mid_price': mid_price
        }
        
        for lvl in range(levels):
            spread_offset = 0.01 * (lvl + 1)
            row[f'bid_price_{lvl}'] = round(mid_price - spread_offset, 2)
            row[f'bid_vol_{lvl}'] = np.random.randint(10, 500)
            row[f'ask_price_{lvl}'] = round(mid_price + spread_offset, 2)
            row[f'ask_vol_{lvl}'] = np.random.randint(10, 500)
            
        data.append(row)
        
    return pd.DataFrame(data)

def calculate_obi(df, levels=5, decay=0.8):
    bid_vols = np.array([df[f'bid_vol_{lvl}'].values for lvl in range(levels)])
    ask_vols = np.array([df[f'ask_vol_{lvl}'].values for lvl in range(levels)])
    
    weights = np.array([decay**i for i in range(levels)])
    weights /= weights.sum()
    
    weighted_bids = np.dot(weights, bid_vols)
    weighted_asks = np.dot(weights, ask_vols)
    
    obi = (weighted_bids - weighted_asks) / (weighted_bids + weighted_asks)
    return obi

def find_imbalance_signals(df, obi_series, threshold=0.4):
    signals = pd.Series(0, index=df.index)
    signals[obi_series > threshold] = 1
    signals[obi_series < -threshold] = -1
    return signals

if __name__ == '__main__':
    df = generate_l2_order_book(ticks=1000, levels=5)
    df['obi'] = calculate_obi(df, levels=5, decay=0.75)
    df['signal'] = find_imbalance_signals(df, df['obi'], threshold=0.3)
    
    print(df[['timestamp', 'mid_price', 'obi', 'signal']].head(10))

Разбор параметров

• levels: Количество уровней стакана (Level 2), используемых для расчета. Чем глубже стакан мы анализируем, тем более устойчив показатель к единичным крупным заявкам на спреде.
• decay: Коэффициент экспоненциального затухания весов для уровней стакана. Значение 0.75 означает, что первый уровень (лучший Bid/Ask) имеет наибольший вес, а последующие уровни вносят затухающий вклад.
• threshold: Пороговое значение дисбаланса (от -1 до 1), при превышении которого генерируется сигнал. Например, значение 0.3 указывает на то, что объем покупателей превышает объем продавцов более чем на 30%.

Как запустить

Для запуска скрипта вам понадобятся библиотеки pandas и numpy. Установите их с помощью команды:

pip install pandas numpy

Сохраните код в файл obi_trader.py и запустите его. Скрипт сгенерирует синтетический поток котировок Level 2, рассчитает динамический OBI и выведет первые строки датафрейма с сигналами в консоль.

Помните, что высокочастотные стратегии на основе микроструктуры требуют жесткого контроля рисков и учета проскальзываний (slippage). Перед запуском подобных алгоритмов в реальную торговлю обязательно ознакомьтесь с методологией оценки рисков в статье Бэктестинг Мартингейл-стратегий: Расчет Risk of Ruin на Python.