. Вдохновившись недавней статьей на 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) создают иерархическую фильтрацию:
M30: "Кто сильнее?"
M15: "Где перекупленность?"
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 дням. Это отличный результат за такой небольшой период.
Конечно, можно добавлять дополнительные фильтры, делать больше расчётов. Но по графику доходности мы видим, что она стабильна - так что работоспособность стратегии подтвердилась!
