. Вдохновившись недавней статьей на Veles Finance "«Bomberman»: стратегия для крипторынка с индикаторами BOP, Mean Reversion и Дончиана", я решил воплотить эту идею в жизнь. Не просто в теории, а в коде: создал полноценного алгобота на Python, который автоматизирует торговлю, тестирует параметры и визуализирует результаты.

Этот бот прозрачен: использует открытые данные с binance, классические индикаторы технического анализа и строгий walk-forward бэктест, чтобы избежать look-ahead bias (смещения в будущее).

В статье разберем логику стратегии, архитектуру бота, ключевые функции и реальные результаты на исторических данных BTC/USDT. Если вы программист с интересом к финансам или трейдер, жаждущий автоматизации, — добро пожаловать. Мы пройдемся по коду, формулам и рискам.

Почему Bomberman

Стратегия "Bomberman", описанная в оригинальной статье, черпает вдохновение из аркадного хита 1980-х. Вкратце, основная идея самой стратегии в комбинации трех индикаторов на разных таймфреймах:

• BOP (Balance of Power) на 30-минутном (M30) — измеряет баланс сил между быками и медведями.

• Mean Reversion Channel на 15-минутном (M15) — канал возврата к среднему для зон перепроданности/перекупленности.

• Donchian Channel на 5-минутном (M5) — прорывы для подтверждения импульса.

Бот реализует это с учетом шортов (коротких позиций), левериджа и строгого риск-менеджмента (1% на сделку). В отличие от ручной торговли, бот вычисляет индикаторы только на исторических данных до текущего бара, имитируя реальную торговлю без подглядывания в будущее. Это не просто бэктест — это симуляция, готовая к деплою на реальном API. Достаточно будет просто добавить подгрузку актуальных данных в бота и открытие сделок с помощью клиента для биржи (ранее писали собственный клиент для bingX).

Логика стратегии: Вход, выход и "бомбы"

Основа бота — функция backtest_one, которая симулирует торговлю. Давайте разберем правила стратегии. Все расчеты на основе OHLCV-данных (Open, High, Low, Close, Volume).

1. Индикаторы

Бот вычисляет индикаторы динамически, чтобы избежать предвзятости. Вот формулы:

• BOP (Balance of Power):

  

• Mean Reversion Channel (на M15):

  

• Donchian Channel (на M5):

  

2. Условия входа: "Подрыв барьера"

Толерантность tol (0.5–1.5%) добавляет гибкости для приближения к уровням.

• Лонг (LONG):

  • BOP > 0 (быки доминируют на M30).

  • Цена ≤ Lower × (1 + tol) (перепроданность на M15).

  • Цена > предыдущий Donchian_high (прорыв на M5).

• Шорт (SHORT):

  • BOP < 0 (медведи доминируют).

  • Цена ≥ Upper × (1 - tol) (перекупленность).

  • Цена < предыдущий Donchian_low (прорыв вниз).

Размер позиции расчитываем исходя из указанного риска и капитала. Это в первую очередь нужно при реализации реальной торговле, для бектеста пока что достаточно получить прибыль в %.

3. Условия выхода: "Взрыв и отступление"

Выход минимизирует убытки и фиксирует прибыль:

• Для лонга:

  • Тейк-профит: цена ≥ Upper (возврат к сопротивлению).

  • Стоп-лосс: цена ≤ entry × 0.99 (1% убыток).

• Для шорта:

  • Тейк-профит: цена ≤ Lower.

  • Стоп-лосс: цена ≥ entry × 1.01 (1% убыток).

Эта логика обеспечивает баланс: mean reversion ловит отскоки, Donchian — трендовые прорывы, BOP — фильтр тренда. В "both" режиме бот торгует и лонг, и шорт.

Архитектура бота: Функции и модули

Бот — модульный скрипт на ~300 строк Python. Зависимости: ccxt (для данных), pandas/numpy (анализ), matplotlib (визуализация). Нет ML или сложных фреймворков — чистый TA-Lib стиль, но самописный для контроля.

Для начала сделаем импорты:

import ccxt
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import warnings
from datetime import datetime
import time

1. Загрузка данных: fetch_binance_ohlcv

Публичный API binance(без ключей). Фетчит OHLCV за N баров назад:

from binance.client import Client
client = Client()

def fetch_binance_ohlcv(symbol='ETHUSDT', interval='15m', total_bars=50000, client=client):
    limit = 1000
    data = []
    current_end = None
    while len(data) < total_bars:
        bars_to_fetch = min(limit, total_bars - len(data))
        try:
            klines = client.futures_klines(
                symbol=symbol.upper(),
                interval=interval,
                limit=bars_to_fetch,
                endTime=current_end
            )
        except Exception as e:
            print("Ошибка Binance API:", e)
            break
        if not klines:
            break
        data = klines + data  # prepend
        current_end = klines[0][0] - 1
        time.sleep(0.1)

    if not data:
        return pd.DataFrame()

    df = pd.DataFrame(data, columns=[
        'timestamp','open','high','low','close','volume',
        'close_time','quote_asset_volume','number_of_trades',
        'taker_buy_base','taker_buy_quote','ignore'
    ])
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    df[['open','high','low','close','volume']] = df[['open','high','low','close','volume']].astype(float)
    df = df.drop_duplicates('timestamp').sort_values('timestamp').reset_index(drop=True)
    df = df.rename(columns={'timestamp': 'time'})
    return df[["time", "open", "high", "low", "close", "volume"]]

Синхронизирует таймфреймы: ~2000 баров M30 ≈ 4000 M15 ≈ 12000 M5 (~1 год данных). Обработка дубликатов и NaN.

2. Вычисление индикаторов: add_bop, add_mean_reversion, add_donchian

Простые rolling-операции на pandas. Для каждого бара (данные свечи) режем DF до current_time и пересчитываем только прошлое. Также необходимо синхронизовать таймфреймы, так как нам нужно чтобы все индексы свечей совпадали и не было ненужных нам смещений.

def add_bop(df: pd.DataFrame, smooth: int = 1) -> pd.DataFrame:
    df = df.copy()
    df['bop'] = (df['close'] - df['open']) / (df['high'] - df['low']).replace(0, np.nan)
    if smooth > 1:
        df['bop'] = df['bop'].rolling(smooth).mean()
    return df


def add_mean_reversion(df: pd.DataFrame, period: int = 20, mult: float = 2.0) -> pd.DataFrame:
    df = df.copy()
    df['sma'] = df['close'].rolling(period).mean()
    df['std'] = df['close'].rolling(period).std()
    df['upper'] = df['sma'] + mult * df['std']
    df['lower'] = df['sma'] - mult * df['std']
    return df


def add_donchian(df: pd.DataFrame, period: int = 20) -> pd.DataFrame:
    df = df.copy()
    df['donchian_high'] = df['high'].rolling(period).max()
    df['donchian_low']  = df['low'].rolling(period).min()
    return df


# ------------------------------------------------------------------
# 3. Синхронизация (walk-forward)
# ------------------------------------------------------------------
def sync_timeframes(df30, df15, df5):
    df = df30.copy()
    df = df.join(df15[['upper', 'lower']], how='left')
    df = df.join(df5[['donchian_high', 'donchian_low']], how='left')
    cols = ['upper', 'lower', 'donchian_high', 'donchian_low']
    df[cols] = df[cols].ffill()  # Только прошлые данные
    return df

Три таймфрейма по этим индикатором (M30 → M15 → M5) создают иерархическую фильтрацию:

1. M30: "Кто сильнее?"

2. M15: "Где перекупленность?"

3. M5: "Есть ли прорыв?"

Такой мультитаймфреймовый подход - основа стратегии.

3. Бэктест: backtest_one

Walk-forward симуляция: цикл по барам M30, слайс данных + индикаторы + сигналы. Возвращает словарь с результатом и основыми данными.

def backtest_one(df: pd.DataFrame,
                 tol: float,
                 direction: str = "both",
                 initial_capital: float = 10_000,
                 leverage: int = 5,
                 risk_per_trade: float = 0.01) -> dict:

    capital = initial_capital
    position = 0  # 1 = long, -1 = short, 0 = flat
    entry_price = 0.0
    size = 0.0
    equity = [capital]
    trades = []

    for i in range(50, len(df)):
        row = df.iloc[i]
        prev = df.iloc[i-1]
        price = row['close']

        # --- ВХОД ---
        if position == 0:
            # LONG
            if direction in ["long", "both"]:
                long_cond = (
                    row['bop'] > 0 and
                    price <= row['lower'] * (1 + tol) and
                    price > prev['donchian_high']
                )
                if long_cond:
                    size = (capital * risk_per_trade * leverage) / price
                    entry_price = price
                    position = 1
                    trades.append(f"LONG  at {price:.5f}")

            # SHORT
            if direction in ["short", "both"]:
                short_cond = (
                    row['bop'] < 0 and
                    price >= row['upper'] * (1 - tol) and
                    price < prev['donchian_low']
                )
                if short_cond:
                    size = (capital * risk_per_trade * leverage) / price
                    entry_price = price
                    position = -1
                    trades.append(f"SHORT at {price:.5f}")

        # --- ВЫХОД ---
        if position == 1:  # LONG
            if price >= row['upper']:
                pnl = size * (price - entry_price) * leverage - size * 0.0005
                capital += pnl
                position = 0
                trades.append(f"EXIT LONG  at {price:.5f} | PnL: {pnl:+.2f}")
            elif price <= entry_price * 0.99:
                pnl = size * (price - entry_price) * leverage - size * 0.0005
                capital += pnl
                position = 0
                trades.append(f"STOP LONG  at {price:.5f} | PnL: {pnl:+.2f}")

        if position == -1:  # SHORT
            if price <= row['lower']:
                pnl = (size * (entry_price - price) * leverage) - size * 0.0005
                capital += pnl
                position = 0
                trades.append(f"EXIT SHORT at {price:.5f} | PnL: {pnl:+.2f}")
            elif price >= entry_price * 1.01:
                pnl = size * (entry_price - price) * leverage - size * 0.0005
                capital += pnl
                position = 0
                trades.append(f"STOP SHORT at {price:.5f} | PnL: {pnl:+.2f}")

        equity.append(capital)

    total_ret = (capital - initial_capital) / initial_capital

    return {
        'final_capital': capital,
        'total_return': total_ret,
        'equity': equity,
        'trades': trades
    }

Обратите внимание на расчёт прибыли:

pnl = size (price - entry_price) leverage - size * 0.005

Комиссия учтена и она равняется size * 0.005 (для bingx со скидкой на комиссию taker 0.0025=0.25%). Для binance, bybit - 0.01, что будет уже не так приятно, но в рамках не самого большого количества сделок всё еще терпимо.

4. Оптимизация: grid_search

Грид по 5+ параметрам (576 комбинаций в полном режиме, но для теста — подмножество). Скрипт выполняется довольно быстро ввиду не самого большого количества свечей. Если хотите доработать его и добавить новые фильтры и перебор большего числа параметров - несложно можно добавить multiprocessing (параллелизм).

def grid_search(df30, df15, df5):
    param_grid = {
        'mr_period': [15, 20],
        'mr_mult'  : [2.0, 2.5],
        'dc_period': [15, 20],
        'bop_smooth': [1],
        'tol'      : [0.015],
        'direction': ['both']
    }

    results = []
    total = np.prod([len(v) for v in param_grid.values()])
    print(f"Грид-поиск: {total} комбинаций...")

    cnt = 0
    for mr_p in param_grid['mr_period']:
        for mr_m in param_grid['mr_mult']:
            for dc_p in param_grid['dc_period']:
                for bop_s in param_grid['bop_smooth']:
                    for tol in param_grid['tol']:
                        for dir_ in param_grid['direction']:
                            cnt += 1
                            print(f"[{cnt}/{total}] MR={mr_p}*{mr_m}, DC={dc_p}, BOP_s={bop_s}, tol={tol*100:.1f}%, {dir_}")

                            df_mr = add_mean_reversion(df15.copy(), mr_p, mr_m)
                            df_dc = add_donchian(df5.copy(), dc_p)
                            df_bop = add_bop(df30.copy(), bop_s)

                            df_test = sync_timeframes(df_bop, df_mr, df_dc)
                            res = backtest_one(df_test, tol, direction=dir_)

                            results.append({
                                'mr_period': mr_p,
                                'mr_mult': mr_m,
                                'dc_period': dc_p,
                                'bop_smooth': bop_s,
                                'tolerance': tol,
                                'direction': dir_,
                                'final_capital': res['final_capital'],
                                'total_return': res['total_return'],
                            })

    return pd.DataFrame(results)

По сути, мы здесь просто для каждой комбинации прогоняем бектест и собираем результаты - далее будем их сравнивать и выбирать лучшее.

5. Визуализация: plot_best

Построение графика для лучшей комбинации:

• Верхний: Цена + каналы + сигналы (стрелки лонг/шорт).

• Нижний: Equity-кривой. Matplotlib генерирует интерактивный plot.

def plot_best(df30, df15, df5, best_row):
    df_mr = add_mean_reversion(df15.copy(),
                               int(best_row['mr_period']),
                               best_row['mr_mult'])
    df_dc = add_donchian(df5.copy(),
                         int(best_row['dc_period']))
    df_bop = add_bop(df30.copy(),
                     int(best_row['bop_smooth']))

    df = sync_timeframes(df_bop, df_mr, df_dc)

    # Пересчёт с лучшими параметрами
    res = backtest_one(df,
                       best_row['tolerance'],
                       direction=best_row['direction'])

    # Сигналы
    df['signal'] = 0
    for i, trade in enumerate(res['trades']):
        if 'LONG' in trade or 'SHORT' in trade:
            timestamp = df.index[50 + i // 2]
            df.loc[timestamp, 'signal'] = 1 if 'LONG' in trade else -1

    plt.figure(figsize=(16, 10))
    ax1 = plt.subplot(3, 1, 1)
    ax1.plot(df.index, df['close'], label='BTC/USDT', color='steelblue')
    ax1.plot(df.index, df['lower'], '--', color='green', label='Lower')
    ax1.plot(df.index, df['upper'], '--', color='red', label='Upper')
    ax1.scatter(df.index[df['signal'] == 1], df['close'][df['signal'] == 1],
                marker='^', color='lime', s=100, label='LONG')
    ax1.scatter(df.index[df['signal'] == -1], df['close'][df['signal'] == -1],
                marker='v', color='red', s=100, label='SHORT')
    ax1.set_title(f'Bomberman | {best_row["direction"].upper()} | '
                  f'Годовых: {best_row["annualized"]*100:.1f}%')
    ax1.legend()


    ax3 = plt.subplot(3, 1, 3)
    ax3.plot(res['equity'], color='gold')
    ax3.set_title(f'Капитал: ${res["final_capital"]:,.0f}')
    plt.tight_layout()
    plt.show()

Запустим нашу стратегию

Запустим подгрузку свечей по всем тф и используем наши функции:

if __name__ == "__main__":
    print("Загрузка данных...")
    df_30m = fetch_binance_ohlcv('BTCUSDT', '30m', 5000)
    df_15m = fetch_binance_ohlcv('BTCUSDT', '15m', 10000)
    df_5m  = fetch_binance_ohlcv('BTCUSDT', '5m',  30000)
    print(f"  30m: {len(df_30m)} | 15m: {len(df_15m)} | 5m: {len(df_5m)}")

    results_df = grid_search(df_30m, df_15m, df_5m)

    top5 = results_df.sort_values('annualized', ascending=False).head(5)
    print("\n" + "="*80)
    print("ТОП-5 ПАРАМЕТРОВ (с LONG/SHORT/BOTH)")
    print("="*80)
    print(top5[['direction', 'mr_period', 'mr_mult', 'dc_period',
                'bop_smooth', 'tolerance', 'annualized', 'final_capital']])

    best = top5.iloc[0]
    print(f"\nВизуализация лучшего: {best['direction']} | "
          f"Годовых: {best['annualized']*100:.1f}%")
    plot_best(df_30m, df_15m, df_5m, best)

Результаты

На исторических данных BTC/USDT бот показал солидные цифры. В полном грид-серче (576 runs) лучшая комбинация дала 38% прибыли за период в 30000 5m свечей, что равно приблизительно 104 дням. Это отличный результат за такой небольшой период.

Конечно, можно добавлять дополнительные фильтры, делать больше расчётов. Но по графику доходности мы видим, что она стабильна - так что работоспособность стратегии подтвердилась!